vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Merge pull request #2300 from coreos-inc/openid-connect

Browse source 
OpenID Connect support and OAuth login refactoring
This commit is contained in:
josephschorr 2017-01-31 18:14:44 -05:00 committed by GitHub
commit 01ec22b362
36 changed files with 1623 additions and 983 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.dockerignore
View file

@ -20,3 +20,4 @@ coverage
.cache
.npm-debug.log
test/__pycache__
__pycache__

22
app.py
View file

@ -25,6 +25,9 @@ from data.queue import WorkQueue, BuildMetricQueueReporter
from data.userevent import UserEventsBuilderModule
from data.userfiles import Userfiles
from data.users import UserAuthentication
from oauth.services.github import GithubOAuthService
from oauth.services.gitlab import GitLabOAuthService
from oauth.loginmanager import OAuthLoginManager
from storage import Storage
from util import get_app_url
from util.saas.analytics import Analytics
@ -32,18 +35,13 @@ from util.saas.useranalytics import UserAnalytics
from util.saas.exceptionlog import Sentry
from util.names import urn_generator
from util.config.configutil import generate_secret_key
from util.config.oauth import (GoogleOAuthConfig, GithubOAuthConfig, GitLabOAuthConfig,
DexOAuthConfig)
from util.config.provider import get_config_provider
from util.config.superusermanager import SuperUserManager
from util.label_validator import LabelValidator
from util.license import LicenseValidator, LICENSE_FILENAME
from util.license import LicenseValidator
from util.metrics.metricqueue import MetricQueue
from util.metrics.prometheus import PrometheusPlugin
from util.names import urn_generator
from util.saas.analytics import Analytics
from util.saas.cloudwatch import start_cloudwatch_sender
from util.saas.exceptionlog import Sentry
from util.secscan.api import SecurityScannerAPI
from util.security.instancekeys import InstanceKeys
from util.security.signing import Signer
@ -204,13 +202,11 @@ license_validator.start()
start_cloudwatch_sender(metric_queue, app)
github_login = GithubOAuthConfig(app.config, 'GITHUB_LOGIN_CONFIG')
github_trigger = GithubOAuthConfig(app.config, 'GITHUB_TRIGGER_CONFIG')
gitlab_trigger = GitLabOAuthConfig(app.config, 'GITLAB_TRIGGER_CONFIG')
google_login = GoogleOAuthConfig(app.config, 'GOOGLE_LOGIN_CONFIG')
dex_login = DexOAuthConfig(app.config, 'DEX_LOGIN_CONFIG')
github_trigger = GithubOAuthService(app.config, 'GITHUB_TRIGGER_CONFIG')
gitlab_trigger = GitLabOAuthService(app.config, 'GITLAB_TRIGGER_CONFIG')
oauth_apps = [github_login, github_trigger, gitlab_trigger, google_login, dex_login]
oauth_login = OAuthLoginManager(app.config)
oauth_apps = [github_trigger, gitlab_trigger]
image_replication_queue = WorkQueue(app.config['REPLICATION_QUEUE_NAME'], tf,
has_namespace=False, metric_queue=metric_queue)
@ -243,7 +239,7 @@ model.config.register_image_cleanup_callback(secscan_api.cleanup_layers)
@login_manager.user_loader
def load_user(user_uuid):
logger.debug('User loader loading deferred user with uuid: %s' % user_uuid)
logger.debug('User loader loading deferred user with uuid: %s', user_uuid)
return LoginWrappedDBUser(user_uuid)
class LoginWrappedDBUser(UserMixin):

16
data/model/user.py
View file

@ -369,7 +369,7 @@ def update_user_metadata(user, given_name=None, family_name=None, company=None):
remove_user_prompt(user, UserPromptTypes.ENTER_COMPANY)
def create_federated_user(username, email, service_name, service_ident,
def create_federated_user(username, email, service_id, service_ident,
set_password_notification, metadata={},
email_required=True, prompts=tuple()):
prompts = set(prompts)
@ -379,7 +379,11 @@ def create_federated_user(username, email, service_name, service_ident,
new_user.verified = True
new_user.save()
service = LoginService.get(LoginService.name == service_name)
try:
service = LoginService.get(LoginService.name == service_id)
except LoginService.DoesNotExist:
service = LoginService.create(name=service_id)
FederatedLogin.create(user=new_user, service=service,
service_ident=service_ident,
metadata_json=json.dumps(metadata))
@ -390,20 +394,20 @@ def create_federated_user(username, email, service_name, service_ident,
return new_user
def attach_federated_login(user, service_name, service_ident, metadata={}):
service = LoginService.get(LoginService.name == service_name)
def attach_federated_login(user, service_id, service_ident, metadata={}):
service = LoginService.get(LoginService.name == service_id)
FederatedLogin.create(user=user, service=service, service_ident=service_ident,
metadata_json=json.dumps(metadata))
return user
def verify_federated_login(service_name, service_ident):
def verify_federated_login(service_id, service_ident):
try:
found = (FederatedLogin
.select(FederatedLogin, User)
.join(LoginService)
.switch(FederatedLogin).join(User)
.where(FederatedLogin.service_ident == service_ident, LoginService.name == service_name)
.where(FederatedLogin.service_ident == service_ident, LoginService.name == service_id)
.get())
return found.user
except FederatedLogin.DoesNotExist:

63
endpoints/api/user.py
View file

@ -11,7 +11,9 @@ from peewee import IntegrityError
import features
from app import app, billing as stripe, authentication, avatar, user_analytics, all_queues
from app import (app, billing as stripe, authentication, avatar, user_analytics, all_queues,
oauth_login)
from auth import scopes
from auth.auth_context import get_authenticated_user
from auth.permissions import (AdministerOrganizationPermission, CreateRepositoryPermission,
@ -24,11 +26,12 @@ from endpoints.api import (ApiResource, nickname, resource, validate_json_reques
query_param, require_scope, format_date, show_if,
require_fresh_login, path_param, define_json_response,
RepositoryParamResource, page_support)
from endpoints.exception import NotFound, InvalidToken
from endpoints.exception import NotFound, InvalidToken, InvalidRequest, DownstreamIssue
from endpoints.api.subscribe import subscribe
from endpoints.common import common_login
from endpoints.csrf import generate_csrf_token, OAUTH_CSRF_TOKEN_NAME
from endpoints.decorators import anon_allowed
from oauth.oidc import DiscoveryFailureException
from util.useremails import (send_confirmation_email, send_recovery_email, send_change_email,
send_password_changed, send_org_recovery_email)
from util.names import parse_single_urn
@ -692,26 +695,62 @@ class Signout(ApiResource):
return {'success': True}
@resource('/v1/externaltoken')
@resource('/v1/externallogin/<service_id>')
@internal_only
class GenerateExternalToken(ApiResource):
""" Resource for generating a token for external login. """
@nickname('generateExternalLoginToken')
def post(self):
""" Generates a CSRF token explicitly for OIDC/OAuth-associated login. """
return {'token': generate_csrf_token(OAUTH_CSRF_TOKEN_NAME)}
class ExternalLoginInformation(ApiResource):
""" Resource for both setting a token for external login and returning its authorization
url.
"""
schemas = {
'GetLogin': {
'type': 'object',
'description': 'Information required to an retrieve external login URL.',
'required': [
'kind',
],
'properties': {
'kind': {
'type': 'string',
'description': 'The kind of URL',
'enum': ['login', 'attach'],
},
},
},
}
@resource('/v1/detachexternal/<servicename>')
@nickname('retrieveExternalLoginAuthorizationUrl')
@anon_allowed
@validate_json_request('GetLogin')
def post(self, service_id):
""" Generates the auth URL and CSRF token explicitly for OIDC/OAuth-associated login. """
login_service = oauth_login.get_service(service_id)
if login_service is None:
raise InvalidRequest()
csrf_token = generate_csrf_token(OAUTH_CSRF_TOKEN_NAME)
kind = request.get_json()['kind']
redirect_suffix = '/attach' if kind == 'attach' else ''
try:
login_scopes = login_service.get_login_scopes()
auth_url = login_service.get_auth_url(app.config, redirect_suffix, csrf_token, login_scopes)
return {'auth_url': auth_url}
except DiscoveryFailureException as dfe:
logger.exception('Could not discovery OAuth endpoint information')
raise DownstreamIssue(dfe.message)
@resource('/v1/detachexternal/<service_id>')
@show_if(features.DIRECT_LOGIN)
@internal_only
class DetachExternal(ApiResource):
""" Resource for detaching an external login. """
@require_user_admin
@nickname('detachExternalLogin')
def post(self, servicename):
def post(self, service_id):
""" Request that the current user be detached from the external login service. """
model.user.detach_external_login(get_authenticated_user(), servicename)
model.user.detach_external_login(get_authenticated_user(), service_id)
return {'success': True}

15
endpoints/common.py
View file

@ -16,7 +16,7 @@ from flask_principal import identity_changed
import endpoints.decorated # Register the various exceptions via decorators.
import features
from app import app, oauth_apps, LoginWrappedDBUser, user_analytics, license_validator
from app import app, oauth_apps, oauth_login, LoginWrappedDBUser, user_analytics, license_validator
from auth import scopes
from auth.permissions import QuayDeferredPermissionUser
from config import frontend_visible_config
@ -189,6 +189,18 @@ def render_page_template(name, route_data=None, **kwargs):
cache_buster = cachebusters.get(filename, random_string()) if not debugging else 'debugging'
yield (filename, cache_buster)
def get_external_login_config():
login_config = []
for login_service in oauth_login.services:
login_config.append({
'id': login_service.service_id(),
'title': login_service.service_name(),
'config': login_service.get_public_config(),
'icon': login_service.get_icon(),
})
return login_config
def get_oauth_config():
oauth_config = {}
for oauth_app in oauth_apps:
@ -215,6 +227,7 @@ def render_page_template(name, route_data=None, **kwargs):
feature_set=features.get_features(),
config_set=frontend_visible_config(app.config),
oauth_set=get_oauth_config(),
external_login_set=get_external_login_config(),
scope_set=scopes.app_scopes(app.config),
vuln_priority_set=PRIORITY_LEVELS,
enterprise_logo=app.config.get('ENTERPRISE_LOGO_URL', ''),

342
endpoints/oauthlogin.py
View file

@ -1,19 +1,19 @@
import logging
import requests
import uuid
from flask import request, redirect, url_for, Blueprint
from flask_login import current_user
from peewee import IntegrityError
import features
from app import app, analytics, get_app_url, github_login, google_login, dex_login
from app import app, analytics, get_app_url, oauth_login
from auth.auth_context import get_authenticated_user
from auth.process import require_session_login
from data import model
from endpoints.common import common_login, route_show_if
from endpoints.common import common_login
from endpoints.web import index
from endpoints.csrf import csrf_protect, OAUTH_CSRF_TOKEN_NAME
from util.security.jwtutil import decode, InvalidTokenError
from oauth.login import OAuthLoginException
from util.validation import generate_valid_usernames
logger = logging.getLogger(__name__)
@ -22,7 +22,9 @@ oauthlogin = Blueprint('oauthlogin', __name__)
oauthlogin_csrf_protect = csrf_protect(OAUTH_CSRF_TOKEN_NAME, 'state', all_methods=True)
def render_ologin_error(service_name, error_message=None, register_redirect=False):
def _render_ologin_error(service_name, error_message=None, register_redirect=False):
""" Returns a Flask response indicating an OAuth error. """
user_creation = bool(features.USER_CREATION and features.DIRECT_LOGIN)
error_info = {
'reason': 'ologinerror',
@ -37,30 +39,19 @@ def render_ologin_error(service_name, error_message=None, register_redirect=Fals
resp.status_code = 400
return resp
def _conduct_oauth_login(service_id, service_name, user_id, username, email, metadata=None):
""" Conducts login from the result of an OAuth service's login flow. """
def get_user(service, token):
token_param = {
'access_token': token,
'alt': 'json',
}
got_user = client.get(service.user_endpoint(), params=token_param)
if got_user.status_code != requests.codes.ok:
return {}
return got_user.json()
def conduct_oauth_login(service, user_id, username, email, metadata=None):
service_name = service.service_name()
to_login = model.user.verify_federated_login(service_name.lower(), user_id)
to_login = model.user.verify_federated_login(service_id, user_id)
if not to_login:
# See if we can create a new user.
if not features.USER_CREATION:
error_message = 'User creation is disabled. Please contact your administrator'
return render_ologin_error(service_name, error_message)
return _render_ologin_error(service_name, error_message)
# Try to create the user
try:
# Generate a valid username.
new_username = None
for valid in generate_valid_usernames(username):
if model.user.get_user_or_org(valid):
@ -69,8 +60,13 @@ def conduct_oauth_login(service, user_id, username, email, metadata=None):
new_username = valid
break
# Generate a valid email. If the email is None and the MAILING feature is turned
# off, simply place in a fake email address.
if email is None and not features.MAILING:
email = '%s@fake.example.com' % (str(uuid.uuid4()))
prompts = model.user.get_default_user_prompts(features)
to_login = model.user.create_federated_user(new_username, email, service_name.lower(),
to_login = model.user.create_federated_user(new_username, email, service_id,
user_id, set_password_notification=True,
metadata=metadata or {},
prompts=prompts)
@ -84,10 +80,10 @@ def conduct_oauth_login(service, user_id, username, email, metadata=None):
message = message + "\nPlease log in with your username and password and "
message = message + "associate your %s account to use it in the future." % (service_name, )
return render_ologin_error(service_name, message, register_redirect=True)
return _render_ologin_error(service_name, message, register_redirect=True)
except model.DataModelException as ex:
return render_ologin_error(service_name, ex.message)
return _render_ologin_error(service_name, ex.message)
if common_login(to_login):
if model.user.has_user_prompts(to_login):
@ -95,277 +91,79 @@ def conduct_oauth_login(service, user_id, username, email, metadata=None):
else:
return redirect(url_for('web.index'))
return render_ologin_error(service_name)
return _render_ologin_error(service_name)
def get_email_username(user_data):
username = user_data['email']
at = username.find('@')
if at > 0:
username = username[0:at]
def _register_service(login_service):
""" Registers the given login service, adding its callback and attach routes to the blueprint. """
return username
@oauthlogin.route('/google/callback', methods=['GET'])
@route_show_if(features.GOOGLE_LOGIN)
@oauthlogin_csrf_protect
def google_oauth_callback():
@oauthlogin_csrf_protect
def callback_func():
# Check for a callback error.
error = request.args.get('error', None)
if error:
return render_ologin_error('Google', error)
return _render_ologin_error(login_service.service_name(), error)
# Exchange the OAuth code for login information.
code = request.args.get('code')
token = google_login.exchange_code_for_token(app.config, client, code, form_encode=True)
if token is None:
return render_ologin_error('Google')
try:
lid, lusername, lemail = login_service.exchange_code_for_login(app.config, client, code, '')
except OAuthLoginException as ole:
logger.exception('Got login exception')
return _render_ologin_error(login_service.service_name(), ole.message)
user_data = get_user(google_login, token)
if not user_data or not user_data.get('id', None) or not user_data.get('email', None):
return render_ologin_error('Google')
if not user_data.get('verified_email', False):
return render_ologin_error(
'Google',
'A verified e-mail address is required for login. Please verify your ' +
'e-mail address in Google and try again.',
)
username = get_email_username(user_data)
# Conduct login.
metadata = {
'service_username': user_data['email']
'service_username': lusername
}
return conduct_oauth_login(google_login, user_data['id'], username, user_data['email'],
metadata=metadata)
return _conduct_oauth_login(login_service.service_id(), login_service.service_name(), lid,
lusername, lemail, metadata=metadata)
@oauthlogin.route('/github/callback', methods=['GET'])
@route_show_if(features.GITHUB_LOGIN)
@oauthlogin_csrf_protect
def github_oauth_callback():
@require_session_login
@oauthlogin_csrf_protect
def attach_func():
# Check for a callback error.
error = request.args.get('error', None)
if error:
return render_ologin_error('GitHub', error)
return _render_ologin_error(login_service.service_name(), error)
# Exchange the OAuth code.
# Exchange the OAuth code for login information.
code = request.args.get('code')
token = github_login.exchange_code_for_token(app.config, client, code)
if token is None:
return render_ologin_error('GitHub')
# Retrieve the user's information.
user_data = get_user(github_login, token)
if not user_data or 'login' not in user_data:
return render_ologin_error('GitHub')
username = user_data['login']
github_id = user_data['id']
v3_media_type = {
'Accept': 'application/vnd.github.v3'
}
token_param = {
'access_token': token,
}
# Retrieve the user's orgnizations (if organization filtering is turned on)
if github_login.allowed_organizations() is not None:
get_orgs = client.get(github_login.orgs_endpoint(), params=token_param,
headers={'Accept': 'application/vnd.github.moondragon+json'})
organizations = set([org.get('login').lower() for org in get_orgs.json()])
matching_organizations = organizations & set(github_login.allowed_organizations())
if not matching_organizations:
err = """You are not a member of an allowed GitHub organization.
Please contact your system administrator if you believe this is in error."""
return render_ologin_error('GitHub', err)
# Find the e-mail address for the user: we will accept any email, but we prefer the primary
get_email = client.get(github_login.email_endpoint(), params=token_param,
headers=v3_media_type)
if get_email.status_code / 100 != 2:
return render_ologin_error('GitHub')
found_email = None
for user_email in get_email.json():
if not github_login.is_enterprise() and not user_email['verified']:
continue
found_email = user_email['email']
if user_email['primary']:
break
if found_email is None:
err = 'There is no verified e-mail address attached to the GitHub account.'
return render_ologin_error('GitHub', err)
try:
lid, lusername, _ = login_service.exchange_code_for_login(app.config, client, code, '/attach')
except OAuthLoginException as ole:
return _render_ologin_error(login_service.service_name(), ole.message)
# Conduct attach.
metadata = {
'service_username': username
'service_username': lusername
}
return conduct_oauth_login(github_login, github_id, username, found_email, metadata=metadata)
@oauthlogin.route('/google/callback/attach', methods=['GET'])
@route_show_if(features.GOOGLE_LOGIN)
@require_session_login
@oauthlogin_csrf_protect
def google_oauth_attach():
code = request.args.get('code')
token = google_login.exchange_code_for_token(app.config, client, code,
redirect_suffix='/attach', form_encode=True)
if token is None:
return render_ologin_error('Google')
user_data = get_user(google_login, token)
if not user_data or not user_data.get('id', None):
return render_ologin_error('Google')
if not user_data.get('verified_email', False):
return render_ologin_error(
'Google',
'A verified e-mail address is required for login. Please verify your ' +
'e-mail address in Google and try again.',
)
google_id = user_data['id']
user_obj = current_user.db_user()
username = get_email_username(user_data)
metadata = {
'service_username': user_data['email']
}
user_obj = get_authenticated_user()
try:
model.user.attach_federated_login(user_obj, 'google', google_id, metadata=metadata)
except IntegrityError:
err = 'Google account %s is already attached to a %s account' % (
username, app.config['REGISTRY_TITLE_SHORT'])
return render_ologin_error('Google', err)
return redirect(url_for('web.user_view', path=user_obj.username, tab='external'))
@oauthlogin.route('/github/callback/attach', methods=['GET'])
@route_show_if(features.GITHUB_LOGIN)
@require_session_login
@oauthlogin_csrf_protect
def github_oauth_attach():
code = request.args.get('code')
token = github_login.exchange_code_for_token(app.config, client, code)
if token is None:
return render_ologin_error('GitHub')
user_data = get_user(github_login, token)
if not user_data:
return render_ologin_error('GitHub')
github_id = user_data['id']
user_obj = current_user.db_user()
username = user_data['login']
metadata = {
'service_username': username
}
try:
model.user.attach_federated_login(user_obj, 'github', github_id, metadata=metadata)
except IntegrityError:
err = 'Github account %s is already attached to a %s account' % (
username, app.config['REGISTRY_TITLE_SHORT'])
return render_ologin_error('GitHub', err)
return redirect(url_for('web.user_view', path=user_obj.username, tab='external'))
def decode_user_jwt(token, oidc_provider):
try:
return decode(token, oidc_provider.get_public_key(), algorithms=['RS256'],
audience=oidc_provider.client_id(),
issuer=oidc_provider.issuer)
except InvalidTokenError:
# Public key may have expired. Try to retrieve an updated public key and use it to decode.
return decode(token, oidc_provider.get_public_key(force_refresh=True), algorithms=['RS256'],
audience=oidc_provider.client_id(),
issuer=oidc_provider.issuer)
@oauthlogin.route('/dex/callback', methods=['GET', 'POST'])
@route_show_if(features.DEX_LOGIN)
@oauthlogin_csrf_protect
def dex_oauth_callback():
error = request.values.get('error', None)
if error:
return render_ologin_error(dex_login.public_title, error)
code = request.values.get('code')
if not code:
return render_ologin_error(dex_login.public_title, 'Missing OAuth code')
token = dex_login.exchange_code_for_token(app.config, client, code, client_auth=True,
form_encode=True)
if token is None:
return render_ologin_error(dex_login.public_title)
try:
payload = decode_user_jwt(token, dex_login)
except InvalidTokenError:
logger.exception('Exception when decoding returned JWT')
return render_ologin_error(
dex_login.public_title,
'Could not decode response. Please contact your system administrator about this error.',
)
username = get_email_username(payload)
metadata = {}
dex_id = payload['sub']
email_address = payload['email']
if not payload.get('email_verified', False):
return render_ologin_error(
dex_login.public_title,
'A verified e-mail address is required for login. Please verify your ' +
'e-mail address in %s and try again.' % dex_login.public_title,
)
return conduct_oauth_login(dex_login, dex_id, username, email_address,
model.user.attach_federated_login(user_obj, login_service.service_id(), lid,
metadata=metadata)
@oauthlogin.route('/dex/callback/attach', methods=['GET', 'POST'])
@route_show_if(features.DEX_LOGIN)
@require_session_login
@oauthlogin_csrf_protect
def dex_oauth_attach():
code = request.args.get('code')
token = dex_login.exchange_code_for_token(app.config, client, code, redirect_suffix='/attach',
client_auth=True, form_encode=True)
if token is None:
return render_ologin_error(dex_login.public_title)
try:
payload = decode_user_jwt(token, dex_login)
except InvalidTokenError:
logger.exception('Exception when decoding returned JWT')
return render_ologin_error(
dex_login.public_title,
'Could not decode response. Please contact your system administrator about this error.',
)
user_obj = current_user.db_user()
dex_id = payload['sub']
metadata = {}
try:
model.user.attach_federated_login(user_obj, 'dex', dex_id, metadata=metadata)
except IntegrityError:
err = '%s account is already attached to a %s account' % (dex_login.public_title,
app.config['REGISTRY_TITLE_SHORT'])
return render_ologin_error(dex_login.public_title, err)
err = '%s account %s is already attached to a %s account' % (
login_service.service_name(), lusername, app.config['REGISTRY_TITLE_SHORT'])
return _render_ologin_error(login_service.service_name(), err)
return redirect(url_for('web.user_view', path=user_obj.username, tab='external'))
oauthlogin.add_url_rule('/%s/callback' % login_service.service_id(),
'%s_oauth_callback' % login_service.service_id(),
callback_func,
methods=['GET'])
oauthlogin.add_url_rule('/%s/callback/attach' % login_service.service_id(),
'%s_oauth_attach' % login_service.service_id(),
attach_func,
methods=['GET'])
# Register the routes for each of the login services.
for current_service in oauth_login.services:
_register_service(current_service)

2
external_libraries.py
View file

@ -22,7 +22,7 @@ EXTERNAL_JS = [
]
EXTERNAL_CSS = [
'netdna.bootstrapcdn.com/font-awesome/4.6.0/css/font-awesome.css',
'netdna.bootstrapcdn.com/font-awesome/4.7.0/css/font-awesome.css',
'netdna.bootstrapcdn.com/bootstrap/3.3.2/css/bootstrap.min.css',
'fonts.googleapis.com/css?family=Source+Sans+Pro:300,400,700',
's3.amazonaws.com/cdn.core-os.net/icons/core-icons.css',

0
oauth/__init__.py Normal file
View file

155
oauth/base.py Normal file
View file

@ -0,0 +1,155 @@
import logging
import urllib
from abc import ABCMeta, abstractmethod
from six import add_metaclass
from util import get_app_url
logger = logging.getLogger(__name__)
class OAuthExchangeCodeException(Exception):
""" Exception raised if a code exchange fails. """
pass
class OAuthGetUserInfoException(Exception):
""" Exception raised if a call to get user information fails. """
pass
@add_metaclass(ABCMeta)
class OAuthService(object):
""" A base class for defining an external service, exposed via OAuth. """
def __init__(self, config, key_name):
self.key_name = key_name
self.config = config.get(key_name) or {}
@abstractmethod
def service_id(self):
""" The internal ID for this service. Must match the URL portion for the service, e.g. `github`
"""
pass
@abstractmethod
def service_name(self):
""" The user-readable name for the service, e.g. `GitHub`"""
pass
@abstractmethod
def token_endpoint(self):
""" The endpoint at which the OAuth code can be exchanged for a token. """
pass
@abstractmethod
def user_endpoint(self):
""" The endpoint at which user information can be looked up. """
pass
@abstractmethod
def validate_client_id_and_secret(self, http_client, app_config):
""" Performs validation of the client ID and secret, raising an exception on failure. """
pass
@abstractmethod
def authorize_endpoint(self):
""" Endpoint for authorization. """
pass
def requires_form_encoding(self):
""" Returns True if form encoding is necessary for the exchange_code_for_token call. """
return False
def client_id(self):
return self.config.get('CLIENT_ID')
def client_secret(self):
return self.config.get('CLIENT_SECRET')
def get_auth_url(self, app_config, redirect_suffix, csrf_token, scopes):
""" Retrieves the authorization URL for this login service. """
redirect_uri = '%s/oauth2/%s/callback%s' % (get_app_url(app_config), self.service_id(),
redirect_suffix)
params = {
'client_id': self.client_id(),
'redirect_uri': redirect_uri,
'scope': ' '.join(scopes),
'state': csrf_token,
}
authorize_url = '%s%s' % (self.authorize_endpoint(), urllib.urlencode(params))
return authorize_url
def get_redirect_uri(self, app_config, redirect_suffix=''):
return '%s://%s/oauth2/%s/callback%s' % (app_config['PREFERRED_URL_SCHEME'],
app_config['SERVER_HOSTNAME'],
self.service_id(),
redirect_suffix)
def get_user_info(self, http_client, token):
token_param = {
'access_token': token,
'alt': 'json',
}
got_user = http_client.get(self.user_endpoint(), params=token_param)
if got_user.status_code // 100 != 2:
raise OAuthGetUserInfoException('Non-2XX response code for user_info call: %s' %
got_user.status_code)
user_info = got_user.json()
if user_info is None:
raise OAuthGetUserInfoException()
return user_info
def exchange_code_for_token(self, app_config, http_client, code, form_encode=False,
redirect_suffix='', client_auth=False):
""" Exchanges an OAuth access code for the associated OAuth token. """
json_data = self.exchange_code(app_config, http_client, code, form_encode, redirect_suffix,
client_auth)
access_token = json_data.get('access_token', None)
if access_token is None:
logger.debug('Got successful get_access_token response with missing token: %s', json_data)
raise OAuthExchangeCodeException('Missing `access_token` in OAuth response')
return access_token
def exchange_code(self, app_config, http_client, code, form_encode=False, redirect_suffix='',
client_auth=False):
""" Exchanges an OAuth access code for associated OAuth token and other data. """
payload = {
'code': code,
'grant_type': 'authorization_code',
'redirect_uri': self.get_redirect_uri(app_config, redirect_suffix)
}
headers = {
'Accept': 'application/json'
}
auth = None
if client_auth:
auth = (self.client_id(), self.client_secret())
else:
payload['client_id'] = self.client_id()
payload['client_secret'] = self.client_secret()
token_url = self.token_endpoint()
if form_encode:
get_access_token = http_client.post(token_url, data=payload, headers=headers, auth=auth)
else:
get_access_token = http_client.post(token_url, params=payload, headers=headers, auth=auth)
if get_access_token.status_code // 100 != 2:
logger.debug('Got get_access_token response %s', get_access_token.text)
raise OAuthExchangeCodeException('Got non-2XX response for code exchange: %s' %
get_access_token.status_code)
json_data = get_access_token.json()
if not json_data:
raise OAuthExchangeCodeException('Got non-JSON response for code exchange')
if 'error' in json_data:
raise OAuthExchangeCodeException(json_data.get('error_description', json_data['error']))
return json_data

96
oauth/login.py Normal file
View file

@ -0,0 +1,96 @@
import logging
from abc import ABCMeta, abstractmethod
from six import add_metaclass
import features
from oauth.base import OAuthService, OAuthExchangeCodeException, OAuthGetUserInfoException
logger = logging.getLogger(__name__)
class OAuthLoginException(Exception):
""" Exception raised if a login operation fails. """
pass
@add_metaclass(ABCMeta)
class OAuthLoginService(OAuthService):
""" A base class for defining an OAuth-compliant service that can be used for, amongst other
things, login and authentication. """
@abstractmethod
def login_enabled(self):
""" Returns true if the login service is enabled. """
pass
@abstractmethod
def get_login_service_id(self, user_info):
""" Returns the internal ID for the given user under this login service. """
pass
@abstractmethod
def get_login_service_username(self, user_info):
""" Returns the username for the given user under this login service. """
pass
@abstractmethod
def get_verified_user_email(self, app_config, http_client, token, user_info):
""" Returns the verified email address for the given user, if any or None if none. """
pass
@abstractmethod
def get_icon(self):
""" Returns the icon to display for this login service. """
pass
@abstractmethod
def get_login_scopes(self):
""" Returns the list of scopes for login for this service. """
pass
def service_verify_user_info_for_login(self, app_config, http_client, token, user_info):
""" Performs service-specific verification of user information for login. On failure, a service
should raise a OAuthLoginService.
"""
# By default, does nothing.
pass
def exchange_code_for_login(self, app_config, http_client, code, redirect_suffix):
""" Exchanges the given OAuth access code for user information on behalf of a user trying to
login or attach their account. Raises a OAuthLoginService exception on failure. Returns
a tuple consisting of (service_id, service_username, email)
"""
# Retrieve the token for the OAuth code.
try:
token = self.exchange_code_for_token(app_config, http_client, code,
redirect_suffix=redirect_suffix,
form_encode=self.requires_form_encoding())
except OAuthExchangeCodeException as oce:
raise OAuthLoginException(oce.message)
# Retrieve the user's information with the token.
try:
user_info = self.get_user_info(http_client, token)
except OAuthGetUserInfoException as oge:
raise OAuthLoginException(oge.message)
if user_info.get('id', None) is None:
logger.debug('Got user info response %s', user_info)
raise OAuthLoginException('Missing `id` column in returned user information')
# Perform any custom verification for this login service.
self.service_verify_user_info_for_login(app_config, http_client, token, user_info)
# Retrieve the user's email address (if necessary).
email_address = self.get_verified_user_email(app_config, http_client, token, user_info)
if features.MAILING and email_address is None:
raise OAuthLoginException('A verified email address is required to login with this service')
service_user_id = self.get_login_service_id(user_info)
service_username = self.get_login_service_username(user_info)
logger.debug('Completed successful exchange for service %s: %s, %s, %s',
self.service_id(), service_user_id, service_username, email_address)
return (service_user_id, service_username, email_address)

31
oauth/loginmanager.py Normal file
View file

@ -0,0 +1,31 @@
from oauth.services.github import GithubOAuthService
from oauth.services.google import GoogleOAuthService
from oauth.oidc import OIDCLoginService
CUSTOM_LOGIN_SERVICES = {
'GITHUB_LOGIN_CONFIG': GithubOAuthService,
'GOOGLE_LOGIN_CONFIG': GoogleOAuthService,
}
class OAuthLoginManager(object):
""" Helper class which manages all registered OAuth login services. """
def __init__(self, config):
self.services = []
# Register the endpoints for each of the OAuth login services.
for key in config.keys():
# All keys which end in _LOGIN_CONFIG setup a login service.
if key.endswith('_LOGIN_CONFIG'):
if key in CUSTOM_LOGIN_SERVICES:
custom_service = CUSTOM_LOGIN_SERVICES[key](config, key)
if custom_service.login_enabled(config):
self.services.append(custom_service)
else:
self.services.append(OIDCLoginService(config, key))
def get_service(self, service_id):
for service in self.services:
if service.service_id() == service_id:
return service
return None

237
oauth/oidc.py Normal file
View file

@ -0,0 +1,237 @@
import time
import json
import logging
import urlparse
import jwt
from cachetools import lru_cache
from cachetools.ttl import TTLCache
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.serialization import load_der_public_key
from jwkest.jwk import KEYS
from oauth.base import OAuthService, OAuthExchangeCodeException, OAuthGetUserInfoException
from oauth.login import OAuthLoginException
from util.security.jwtutil import decode, InvalidTokenError
logger = logging.getLogger(__name__)
OIDC_WELLKNOWN = ".well-known/openid-configuration"
PUBLIC_KEY_CACHE_TTL = 3600 # 1 hour
ALLOWED_ALGORITHMS = ['RS256']
JWT_CLOCK_SKEW_SECONDS = 30
class DiscoveryFailureException(Exception):
""" Exception raised when OIDC discovery fails. """
pass
class PublicKeyLoadException(Exception):
""" Exception raised if loading the OIDC public key fails. """
pass
class OIDCLoginService(OAuthService):
""" Defines a generic service for all OpenID-connect compatible login services. """
def __init__(self, config, key_name):
super(OIDCLoginService, self).__init__(config, key_name)
self._public_key_cache = TTLCache(1, PUBLIC_KEY_CACHE_TTL, missing=self._load_public_key)
self._id = key_name[0:key_name.find('_')].lower()
self._http_client = config['HTTPCLIENT']
self._mailing = config.get('FEATURE_MAILING', False)
def service_id(self):
return self._id
def service_name(self):
return self.config.get('SERVICE_NAME', self.service_id())
def get_icon(self):
return self.config.get('SERVICE_ICON', 'fa-user-circle')
def get_login_scopes(self):
default_scopes = ['openid']
if self.user_endpoint() is not None:
default_scopes.append('profile')
if self._mailing:
default_scopes.append('email')
return self._oidc_config().get('scopes_supported', default_scopes)
def authorize_endpoint(self):
return self._oidc_config().get('authorization_endpoint', '') + '?response_type=code&'
def token_endpoint(self):
return self._oidc_config().get('token_endpoint')
def user_endpoint(self):
return self._oidc_config().get('userinfo_endpoint')
def validate_client_id_and_secret(self, http_client, app_config):
# TODO: find a way to verify client secret too.
check_auth_url = http_client.get(self.get_auth_url())
if check_auth_url.status_code // 100 != 2:
raise Exception('Got non-200 status code for authorization endpoint')
def requires_form_encoding(self):
return True
def get_public_config(self):
return {
'CLIENT_ID': self.client_id(),
'OIDC': True,
}
def exchange_code_for_login(self, app_config, http_client, code, redirect_suffix):
# Exchange the code for the access token and id_token
try:
json_data = self.exchange_code(app_config, http_client, code,
redirect_suffix=redirect_suffix,
form_encode=self.requires_form_encoding())
except OAuthExchangeCodeException as oce:
raise OAuthLoginException(oce.message)
# Make sure we received both.
access_token = json_data.get('access_token', None)
if access_token is None:
logger.debug('Missing access_token in response: %s', json_data)
raise OAuthLoginException('Missing `access_token` in OIDC response')
id_token = json_data.get('id_token', None)
if id_token is None:
logger.debug('Missing id_token in response: %s', json_data)
raise OAuthLoginException('Missing `id_token` in OIDC response')
# Decode the id_token.
try:
decoded_id_token = self._decode_user_jwt(id_token)
except InvalidTokenError as ite:
logger.exception('Got invalid token error on OIDC decode: %s', ite.message)
raise OAuthLoginException('Could not decode OIDC token')
except PublicKeyLoadException as pke:
logger.exception('Could not load public key during OIDC decode: %s', pke.message)
raise OAuthLoginException('Could find public OIDC key')
# Retrieve the user information.
try:
user_info = self.get_user_info(http_client, access_token)
except OAuthGetUserInfoException as oge:
raise OAuthLoginException(oge.message)
# Verify subs.
if user_info['sub'] != decoded_id_token['sub']:
raise OAuthLoginException('Mismatch in `sub` returned by OIDC user info endpoint')
# Check if we have a verified email address.
email_address = user_info.get('email') if user_info.get('email_verified') else None
if self._mailing:
if email_address is None:
raise OAuthLoginException('A verified email address is required to login with this service')
# Check for a preferred username.
lusername = user_info.get('preferred_username') or user_info.get('sub')
return decoded_id_token['sub'], lusername, email_address
@property
def _issuer(self):
return self.config.get('OIDC_ISSUER', self.config['OIDC_SERVER'])
@lru_cache(maxsize=1)
def _oidc_config(self):
if self.config.get('OIDC_SERVER'):
return self._load_oidc_config_via_discovery(self.config.get('DEBUGGING', False))
else:
return {}
def _load_oidc_config_via_discovery(self, is_debugging):
""" Attempts to load the OIDC config via the OIDC discovery mechanism. If is_debugging is True,
non-secure connections are alllowed. Raises an DiscoveryFailureException on failure.
"""
oidc_server = self.config['OIDC_SERVER']
if not oidc_server.startswith('https://') and not is_debugging:
raise DiscoveryFailureException('OIDC server must be accessed over SSL')
discovery_url = urlparse.urljoin(oidc_server, OIDC_WELLKNOWN)
discovery = self._http_client.get(discovery_url, timeout=5, verify=is_debugging is False)
if discovery.status_code // 100 != 2:
logger.debug('Got %s response for OIDC discovery: %s', discovery.status_code, discovery.text)
raise DiscoveryFailureException("Could not load OIDC discovery information")
try:
return json.loads(discovery.text)
except ValueError:
logger.exception('Could not parse OIDC discovery for url: %s', discovery_url)
raise DiscoveryFailureException("Could not parse OIDC discovery information")
def _decode_user_jwt(self, token):
""" Decodes the given JWT under the given provider and returns it. Raises an InvalidTokenError
exception on an invalid token or a PublicKeyLoadException if the public key could not be
loaded for decoding.
"""
# Find the key to use.
headers = jwt.get_unverified_header(token)
kid = headers.get('kid', None)
if kid is None:
raise InvalidTokenError('Missing `kid` header')
try:
return decode(token, self._get_public_key(kid), algorithms=ALLOWED_ALGORITHMS,
audience=self.client_id(),
issuer=self._issuer,
leeway=JWT_CLOCK_SKEW_SECONDS,
options=dict(require_nbf=False))
except InvalidTokenError:
# Public key may have expired. Try to retrieve an updated public key and use it to decode.
return decode(token, self._get_public_key(kid, force_refresh=True),
algorithms=ALLOWED_ALGORITHMS,
audience=self.client_id(),
issuer=self._issuer,
leeway=JWT_CLOCK_SKEW_SECONDS,
options=dict(require_nbf=False))
def _get_public_key(self, kid, force_refresh=False):
""" Retrieves the public key for this handler with the given kid. Raises a
PublicKeyLoadException on failure. """
# If force_refresh is true, we expire all the items in the cache by setting the time to
# the current time + the expiration TTL.
if force_refresh:
self._public_key_cache.expire(time=time.time() + PUBLIC_KEY_CACHE_TTL)
# Retrieve the public key from the cache. If the cache does not contain the public key,
# it will internally call _load_public_key to retrieve it and then save it.
return self._public_key_cache[kid]
def _load_public_key(self, kid):
""" Loads the public key for this handler from the OIDC service. Raises PublicKeyLoadException
on failure.
"""
keys_url = self._oidc_config()['jwks_uri']
# Load the keys.
try:
keys = KEYS()
keys.load_from_url(keys_url, verify=not self.config.get('DEBUGGING', False))
except Exception as ex:
logger.exception('Exception loading public key')
raise PublicKeyLoadException(ex.message)
# Find the matching key.
keys_found = keys.by_kid(kid)
if len(keys_found) == 0:
raise PublicKeyLoadException('Public key %s not found' % kid)
rsa_keys = [key for key in keys_found if key.kty == 'RSA']
if len(rsa_keys) == 0:
raise PublicKeyLoadException('No RSA form of public key %s not found' % kid)
matching_key = rsa_keys[0]
matching_key.deserialize()
# Reload the key so that we can give a key *instance* to PyJWT to work around its weird parsing
# issues.
return load_der_public_key(matching_key.key.exportKey('DER'), backend=default_backend())

0
oauth/services/__init__.py Normal file
View file

179
oauth/services/github.py Normal file
View file

@ -0,0 +1,179 @@
import logging
from oauth.login import OAuthLoginService, OAuthLoginException
from util import slash_join
logger = logging.getLogger(__name__)
class GithubOAuthService(OAuthLoginService):
def __init__(self, config, key_name):
super(GithubOAuthService, self).__init__(config, key_name)
def login_enabled(self, config):
return config.get('FEATURE_GITHUB_LOGIN', False)
def service_id(self):
return 'github'
def service_name(self):
if self.is_enterprise():
return 'GitHub Enterprise'
return 'GitHub'
def get_icon(self):
return 'fa-github'
def get_login_scopes(self):
if self.config.get('ORG_RESTRICT'):
return ['user:email', 'read:org']
return ['user:email']
def allowed_organizations(self):
if not self.config.get('ORG_RESTRICT', False):
return None
allowed = self.config.get('ALLOWED_ORGANIZATIONS', None)
if allowed is None:
return None
return [org.lower() for org in allowed]
def get_public_url(self, suffix):
return slash_join(self._endpoint(), suffix)
def _endpoint(self):
return self.config.get('GITHUB_ENDPOINT', 'https://github.com')
def is_enterprise(self):
return self._api_endpoint().find('.github.com') < 0
def authorize_endpoint(self):
return slash_join(self._endpoint(), '/login/oauth/authorize') + '?'
def token_endpoint(self):
return slash_join(self._endpoint(), '/login/oauth/access_token')
def _api_endpoint(self):
return self.config.get('API_ENDPOINT', slash_join(self._endpoint(), '/api/v3/'))
def api_endpoint(self):
endpoint = self._api_endpoint()
if endpoint.endswith('/'):
return endpoint[0:-1]
return endpoint
def user_endpoint(self):
return slash_join(self._api_endpoint(), 'user')
def email_endpoint(self):
return slash_join(self._api_endpoint(), 'user/emails')
def orgs_endpoint(self):
return slash_join(self._api_endpoint(), 'user/orgs')
def validate_client_id_and_secret(self, http_client, app_config):
# First: Verify that the github endpoint is actually Github by checking for the
# X-GitHub-Request-Id here.
api_endpoint = self._api_endpoint()
result = http_client.get(api_endpoint, auth=(self.client_id(), self.client_secret()), timeout=5)
if not 'X-GitHub-Request-Id' in result.headers:
raise Exception('Endpoint is not a Github (Enterprise) installation')
# Next: Verify the client ID and secret.
# Note: The following code is a hack until such time as Github officially adds an API endpoint
# for verifying a {client_id, client_secret} pair. That being said, this hack was given to us
# *by a Github Engineer*, so I think it is okay for the time being :)
#
# TODO(jschorr): Replace with the real API call once added.
#
# Hitting the endpoint applications/{client_id}/tokens/foo will result in the following
# behavior IF the client_id is given as the HTTP username and the client_secret as the HTTP
# password:
# - If the {client_id, client_secret} pair is invalid in some way, we get a 401 error.
# - If the pair is valid, then we get a 404 because the 'foo' token does not exists.
validate_endpoint = slash_join(api_endpoint, 'applications/%s/tokens/foo' % self.client_id())
result = http_client.get(validate_endpoint, auth=(self.client_id(), self.client_secret()),
timeout=5)
return result.status_code == 404
def validate_organization(self, organization_id, http_client):
org_endpoint = slash_join(self._api_endpoint(), 'orgs/%s' % organization_id.lower())
result = http_client.get(org_endpoint,
headers={'Accept': 'application/vnd.github.moondragon+json'},
timeout=5)
return result.status_code == 200
def get_public_config(self):
return {
'CLIENT_ID': self.client_id(),
'AUTHORIZE_ENDPOINT': self.authorize_endpoint(),
'GITHUB_ENDPOINT': self._endpoint(),
'ORG_RESTRICT': self.config.get('ORG_RESTRICT', False)
}
def get_login_service_id(self, user_info):
return user_info['id']
def get_login_service_username(self, user_info):
return user_info['login']
def get_verified_user_email(self, app_config, http_client, token, user_info):
v3_media_type = {
'Accept': 'application/vnd.github.v3'
}
token_param = {
'access_token': token,
}
# Find the e-mail address for the user: we will accept any email, but we prefer the primary
get_email = http_client.get(self.email_endpoint(), params=token_param, headers=v3_media_type)
if get_email.status_code // 100 != 2:
raise OAuthLoginException('Got non-2XX status code for emails endpoint: %s' %
get_email.status_code)
verified_emails = [email for email in get_email.json() if email['verified']]
primary_emails = [email for email in get_email.json() if email['primary']]
# Special case: We don't care about whether an e-mail address is "verified" under GHE.
if self.is_enterprise() and not verified_emails:
verified_emails = primary_emails
allowed_emails = (primary_emails or verified_emails or [])
return allowed_emails[0]['email'] if len(allowed_emails) > 0 else None
def service_verify_user_info_for_login(self, app_config, http_client, token, user_info):
# Retrieve the user's orgnizations (if organization filtering is turned on)
if self.allowed_organizations() is None:
return
moondragon_media_type = {
'Accept': 'application/vnd.github.moondragon+json'
}
token_param = {
'access_token': token,
}
get_orgs = http_client.get(self.orgs_endpoint(), params=token_param,
headers=moondragon_media_type)
if get_orgs.status_code // 100 != 2:
logger.debug('get_orgs response: %s', get_orgs.json())
raise OAuthLoginException('Got non-2XX response for org lookup: %s' %
get_orgs.status_code)
organizations = set([org.get('login').lower() for org in get_orgs.json()])
matching_organizations = organizations & set(self.allowed_organizations())
if not matching_organizations:
logger.debug('Found organizations %s, but expected one of %s', organizations,
self.allowed_organizations())
err = """You are not a member of an allowed GitHub organization.
Please contact your system administrator if you believe this is in error."""
raise OAuthLoginException(err)

60
oauth/services/gitlab.py Normal file
View file

@ -0,0 +1,60 @@
from oauth.base import OAuthService
from util import slash_join
class GitLabOAuthService(OAuthService):
def __init__(self, config, key_name):
super(GitLabOAuthService, self).__init__(config, key_name)
def service_id(self):
return 'gitlab'
def service_name(self):
return 'GitLab'
def _endpoint(self):
return self.config.get('GITLAB_ENDPOINT', 'https://gitlab.com')
def user_endpoint(self):
raise NotImplementedError
def api_endpoint(self):
return self._endpoint()
def get_public_url(self, suffix):
return slash_join(self._endpoint(), suffix)
def authorize_endpoint(self):
return slash_join(self._endpoint(), '/oauth/authorize')
def token_endpoint(self):
return slash_join(self._endpoint(), '/oauth/token')
def validate_client_id_and_secret(self, http_client, app_config):
# We validate the client ID and secret by hitting the OAuth token exchange endpoint with
# the real client ID and secret, but a fake auth code to exchange. Gitlab's implementation will
# return `invalid_client` as the `error` if the client ID or secret is invalid; otherwise, it
# will return another error.
url = self.token_endpoint()
redirect_uri = self.get_redirect_uri(app_config, redirect_suffix='trigger')
data = {
'code': 'fakecode',
'client_id': self.client_id(),
'client_secret': self.client_secret(),
'grant_type': 'authorization_code',
'redirect_uri': redirect_uri
}
# We validate by checking the error code we receive from this call.
result = http_client.post(url, data=data, timeout=5)
value = result.json()
if not value:
return False
return value.get('error', '') != 'invalid_client'
def get_public_config(self):
return {
'CLIENT_ID': self.client_id(),
'AUTHORIZE_ENDPOINT': self.authorize_endpoint(),
'GITLAB_ENDPOINT': self._endpoint(),
}

79
oauth/services/google.py Normal file
View file

@ -0,0 +1,79 @@
from oauth.login import OAuthLoginService
def _get_email_username(email_address):
username = email_address
at = username.find('@')
if at > 0:
username = username[0:at]
return username
class GoogleOAuthService(OAuthLoginService):
def __init__(self, config, key_name):
super(GoogleOAuthService, self).__init__(config, key_name)
def login_enabled(self, config):
return config.get('FEATURE_GOOGLE_LOGIN', False)
def service_id(self):
return 'google'
def service_name(self):
return 'Google'
def get_icon(self):
return 'fa-google'
def get_login_scopes(self):
return ['openid', 'email']
def authorize_endpoint(self):
return 'https://accounts.google.com/o/oauth2/auth?response_type=code&'
def token_endpoint(self):
return 'https://accounts.google.com/o/oauth2/token'
def user_endpoint(self):
return 'https://www.googleapis.com/oauth2/v1/userinfo'
def requires_form_encoding(self):
return True
def validate_client_id_and_secret(self, http_client, app_config):
# To verify the Google client ID and secret, we hit the
# https://www.googleapis.com/oauth2/v3/token endpoint with an invalid request. If the client
# ID or secret are invalid, we get returned a 403 Unauthorized. Otherwise, we get returned
# another response code.
url = 'https://www.googleapis.com/oauth2/v3/token'
data = {
'code': 'fakecode',
'client_id': self.client_id(),
'client_secret': self.client_secret(),
'grant_type': 'authorization_code',
'redirect_uri': 'http://example.com'
}
result = http_client.post(url, data=data, timeout=5)
return result.status_code != 401
def get_public_config(self):
return {
'CLIENT_ID': self.client_id(),
'AUTHORIZE_ENDPOINT': self.authorize_endpoint()
}
def get_login_service_id(self, user_info):
return user_info['id']
def get_login_service_username(self, user_info):
return _get_email_username(user_info['email'])
def get_verified_user_email(self, app_config, http_client, token, user_info):
if not user_info.get('verified_email', False):
return None
return user_info['email']
def service_verify_user_info_for_login(self, app_config, http_client, token, user_info):
# Nothing to do.
pass

62
oauth/test/test_loginmanager.py Normal file
View file

@ -0,0 +1,62 @@
from oauth.loginmanager import OAuthLoginManager
from oauth.services.github import GithubOAuthService
from oauth.services.google import GoogleOAuthService
from oauth.oidc import OIDCLoginService
def test_login_manager_github():
config = {
'FEATURE_GITHUB_LOGIN': True,
'GITHUB_LOGIN_CONFIG': {},
}
loginmanager = OAuthLoginManager(config)
assert len(loginmanager.services) == 1
assert isinstance(loginmanager.services[0], GithubOAuthService)
def test_github_disabled():
config = {
'GITHUB_LOGIN_CONFIG': {},
}
loginmanager = OAuthLoginManager(config)
assert len(loginmanager.services) == 0
def test_login_manager_google():
config = {
'FEATURE_GOOGLE_LOGIN': True,
'GOOGLE_LOGIN_CONFIG': {},
}
loginmanager = OAuthLoginManager(config)
assert len(loginmanager.services) == 1
assert isinstance(loginmanager.services[0], GoogleOAuthService)
def test_google_disabled():
config = {
'GOOGLE_LOGIN_CONFIG': {},
}
loginmanager = OAuthLoginManager(config)
assert len(loginmanager.services) == 0
def test_oidc():
config = {
'SOMECOOL_LOGIN_CONFIG': {},
'HTTPCLIENT': None,
}
loginmanager = OAuthLoginManager(config)
assert len(loginmanager.services) == 1
assert isinstance(loginmanager.services[0], OIDCLoginService)
def test_multiple_oidc():
config = {
'SOMECOOL_LOGIN_CONFIG': {},
'ANOTHER_LOGIN_CONFIG': {},
'HTTPCLIENT': None,
}
loginmanager = OAuthLoginManager(config)
assert len(loginmanager.services) == 2
assert isinstance(loginmanager.services[0], OIDCLoginService)
assert isinstance(loginmanager.services[1], OIDCLoginService)

273
oauth/test/test_oidc.py Normal file
View file

@ -0,0 +1,273 @@
# pylint: disable=redefined-outer-name, unused-argument, invalid-name, missing-docstring, too-many-arguments
import json
import time
import urlparse
import jwt
import pytest
import requests
from httmock import urlmatch, HTTMock
from Crypto.PublicKey import RSA
from jwkest.jwk import RSAKey
from oauth.oidc import OIDCLoginService, OAuthLoginException
@pytest.fixture()
def http_client():
sess = requests.Session()
adapter = requests.adapters.HTTPAdapter(pool_connections=100,
pool_maxsize=100)
sess.mount('http://', adapter)
sess.mount('https://', adapter)
return sess
@pytest.fixture(params=[True, False])
def app_config(http_client, request):
return {
'PREFERRED_URL_SCHEME': 'http',
'SERVER_HOSTNAME': 'localhost',
'FEATURE_MAILING': request.param,
'SOMEOIDC_TEST_SERVICE': {
'CLIENT_ID': 'foo',
'CLIENT_SECRET': 'bar',
'SERVICE_NAME': 'Some Cool Service',
'SERVICE_ICON': 'http://some/icon',
'OIDC_SERVER': 'http://fakeoidc',
'DEBUGGING': True,
},
'HTTPCLIENT': http_client,
}
@pytest.fixture()
def oidc_service(app_config):
return OIDCLoginService(app_config, 'SOMEOIDC_TEST_SERVICE')
@pytest.fixture()
def discovery_content():
return {
'scopes_supported': ['profile'],
'authorization_endpoint': 'http://fakeoidc/authorize',
'token_endpoint': 'http://fakeoidc/token',
'userinfo_endpoint': 'http://fakeoidc/userinfo',
'jwks_uri': 'http://fakeoidc/jwks',
}
@pytest.fixture()
def discovery_handler(discovery_content):
@urlmatch(netloc=r'fakeoidc', path=r'.+openid.+')
def handler(_, __):
return json.dumps(discovery_content)
return handler
@pytest.fixture(scope="module") # Slow to generate, only do it once.
def signing_key():
private_key = RSA.generate(2048)
jwk = RSAKey(key=private_key.publickey()).serialize()
return {
'id': 'somekey',
'private_key': private_key.exportKey('PEM'),
'jwk': jwk,
}
@pytest.fixture()
def id_token(oidc_service, signing_key, app_config):
token_data = {
'iss': oidc_service.config['OIDC_SERVER'],
'aud': oidc_service.client_id(),
'nbf': int(time.time()),
'iat': int(time.time()),
'exp': int(time.time() + 600),
'sub': 'cooluser',
}
token_headers = {
'kid': signing_key['id'],
}
return jwt.encode(token_data, signing_key['private_key'], 'RS256', headers=token_headers)
@pytest.fixture()
def valid_code():
return 'validcode'
@pytest.fixture()
def token_handler(oidc_service, id_token, valid_code):
@urlmatch(netloc=r'fakeoidc', path=r'/token')
def handler(_, request):
params = urlparse.parse_qs(request.body)
if params.get('redirect_uri')[0] != 'http://localhost/oauth2/someoidc/callback':
return {'status_code': 400, 'content': 'Invalid redirect URI'}
if params.get('client_id')[0] != oidc_service.client_id():
return {'status_code': 401, 'content': 'Invalid client id'}
if params.get('client_secret')[0] != oidc_service.client_secret():
return {'status_code': 401, 'content': 'Invalid client secret'}
if params.get('code')[0] != valid_code:
return {'status_code': 401, 'content': 'Invalid code'}
if params.get('grant_type')[0] != 'authorization_code':
return {'status_code': 400, 'content': 'Invalid authorization type'}
content = {
'access_token': 'sometoken',
'id_token': id_token,
}
return {'status_code': 200, 'content': json.dumps(content)}
return handler
@pytest.fixture()
def jwks_handler(signing_key):
def jwk_with_kid(kid, jwk):
jwk = jwk.copy()
jwk.update({'kid': kid})
return jwk
@urlmatch(netloc=r'fakeoidc', path=r'/jwks')
def handler(_, __):
content = {'keys': [jwk_with_kid(signing_key['id'], signing_key['jwk'])]}
return {'status_code': 200, 'content': json.dumps(content)}
return handler
@pytest.fixture()
def emptykeys_jwks_handler():
@urlmatch(netloc=r'fakeoidc', path=r'/jwks')
def handler(_, __):
content = {'keys': []}
return {'status_code': 200, 'content': json.dumps(content)}
return handler
@pytest.fixture(params=["someusername", None])
def preferred_username(request):
return request.param
@pytest.fixture
def userinfo_handler(oidc_service, preferred_username):
@urlmatch(netloc=r'fakeoidc', path=r'/userinfo')
def handler(_, __):
content = {
'sub': 'cooluser',
'preferred_username':preferred_username,
'email': 'foo@example.com',
'email_verified': True,
}
return {'status_code': 200, 'content': json.dumps(content)}
return handler
@pytest.fixture()
def invalidsub_userinfo_handler(oidc_service):
@urlmatch(netloc=r'fakeoidc', path=r'/userinfo')
def handler(_, __):
content = {
'sub': 'invalidsub',
'preferred_username': 'someusername',
'email': 'foo@example.com',
'email_verified': True,
}
return {'status_code': 200, 'content': json.dumps(content)}
return handler
@pytest.fixture()
def missingemail_userinfo_handler(oidc_service, preferred_username):
@urlmatch(netloc=r'fakeoidc', path=r'/userinfo')
def handler(_, __):
content = {
'sub': 'cooluser',
'preferred_username': preferred_username,
}
return {'status_code': 200, 'content': json.dumps(content)}
return handler
def test_basic_config(oidc_service):
assert oidc_service.service_id() == 'someoidc'
assert oidc_service.service_name() == 'Some Cool Service'
assert oidc_service.get_icon() == 'http://some/icon'
def test_discovery(oidc_service, http_client, discovery_handler):
with HTTMock(discovery_handler):
assert oidc_service.authorize_endpoint() == 'http://fakeoidc/authorize?response_type=code&'
assert oidc_service.token_endpoint() == 'http://fakeoidc/token'
assert oidc_service.user_endpoint() == 'http://fakeoidc/userinfo'
assert oidc_service.get_login_scopes() == ['profile']
def test_public_config(oidc_service, discovery_handler):
with HTTMock(discovery_handler):
assert oidc_service.get_public_config()['OIDC']
assert oidc_service.get_public_config()['CLIENT_ID'] == 'foo'
assert 'CLIENT_SECRET' not in oidc_service.get_public_config()
assert 'bar' not in oidc_service.get_public_config().values()
def test_exchange_code_invalidcode(oidc_service, discovery_handler, app_config, http_client,
token_handler):
with HTTMock(token_handler, discovery_handler):
with pytest.raises(OAuthLoginException):
oidc_service.exchange_code_for_login(app_config, http_client, 'testcode', '')
def test_exchange_code_validcode(oidc_service, discovery_handler, app_config, http_client,
token_handler, userinfo_handler, jwks_handler, valid_code,
preferred_username):
with HTTMock(jwks_handler, token_handler, userinfo_handler, discovery_handler):
lid, lusername, lemail = oidc_service.exchange_code_for_login(app_config, http_client,
valid_code, '')
assert lid == 'cooluser'
assert lemail == 'foo@example.com'
if preferred_username is not None:
assert lusername == preferred_username
else:
assert lusername == lid
def test_exchange_code_missingemail(oidc_service, discovery_handler, app_config, http_client,
token_handler, missingemail_userinfo_handler, jwks_handler,
valid_code, preferred_username):
with HTTMock(jwks_handler, token_handler, missingemail_userinfo_handler, discovery_handler):
if app_config['FEATURE_MAILING']:
# Should fail because there is no valid email address.
with pytest.raises(OAuthLoginException):
oidc_service.exchange_code_for_login(app_config, http_client, valid_code, '')
else:
# Should succeed because, while there is no valid email address, it isn't necessary with
# mailing disabled.
lid, lusername, lemail = oidc_service.exchange_code_for_login(app_config, http_client,
valid_code, '')
assert lid == 'cooluser'
assert lemail is None
if preferred_username is not None:
assert lusername == preferred_username
else:
assert lusername == lid
def test_exchange_code_invalidsub(oidc_service, discovery_handler, app_config, http_client,
token_handler, invalidsub_userinfo_handler, jwks_handler,
valid_code):
with HTTMock(jwks_handler, token_handler, invalidsub_userinfo_handler, discovery_handler):
# Should fail because the sub of the user info doesn't match that returned by the id_token.
with pytest.raises(OAuthLoginException):
oidc_service.exchange_code_for_login(app_config, http_client, valid_code, '')
def test_exchange_code_missingkey(oidc_service, discovery_handler, app_config, http_client,
token_handler, userinfo_handler, emptykeys_jwks_handler,
valid_code):
with HTTMock(emptykeys_jwks_handler, token_handler, userinfo_handler, discovery_handler):
# Should fail because the key is missing.
with pytest.raises(OAuthLoginException):
oidc_service.exchange_code_for_login(app_config, http_client, valid_code, '')

9
static/directives/external-login-button.html
View file

@ -1,14 +1,13 @@
<span class="external-login-button-element">
<a ng-class="isLink ? '' : 'btn btn-primary btn-block'"
ng-if="providerInfo.enabled" ng-click="startSignin()" style="margin-bottom: 10px"
ng-click="startSignin()" style="margin-bottom: 10px"
ng-disabled="signingIn">
<img ng-src="{{ providerInfo.icon().url }}" ng-if="providerInfo.icon().url">
<i class="fa" ng-class="providerInfo.icon().icon" ng-if="providerInfo.icon().icon"></i>
<span class="icon-image-view" value="{{ provider.icon }}"></span>
<span class="login-text" ng-if="action != 'attach'" style="vertical-align: middle">
<span class="prefix">Sign in with&nbsp;</span><span class="suffix">{{ providerInfo.title() }}</span>
<span class="prefix">Sign in with&nbsp;</span><span class="suffix">{{ provider.title }}</span>
</span>
<span class="login-text" ng-if="action == 'attach'" style="vertical-align: middle">
Attach to {{ providerInfo.title() }}
Attach to {{ provider.title }}
</span>
</a>
</span>

17
static/directives/external-logins-manager.html
View file

@ -14,27 +14,24 @@
<tr class="external-auth-provider" ng-repeat="provider in EXTERNAL_LOGINS">
<td class="external-auth-provider-title">
<img ng-src="{{ provider.icon().url }}" ng-if="provider.icon().url">
<i class="fa" ng-class="provider.icon().icon" ng-if="provider.icon().icon"></i>
{{ provider.title() }}
<span class="icon-image-view" value="{{ provider.icon }}"></span>
{{ provider.title }}
</td>
<td>
<span ng-if="externalLoginInfo[provider.id]">
Attached to {{ provider.title() }} account
<b ng-if="provider.hasUserInfo">
<a ng-href="{{ provider.getUserInfo(externalLoginInfo[provider.id]).endpoint }}" ng-safenewtab>
{{ provider.getUserInfo(externalLoginInfo[provider.id]).username }}
</a>
Attached to {{ provider.title }} account
<b ng-if="externalLoginInfo[provider.id].metadata.service_username">
{{ externalLoginInfo[provider.id].metadata.service_username }}
</b>
</span>
<span class="empty" ng-if="!externalLoginInfo[provider.id]">
Not attached to {{ provider.title() }}
Not attached to {{ provider.title }}
</span>
</td>
<td>
<span class="external-login-button" provider="{{ provider.id }}" action="attach" is-link="true"
<span class="external-login-button" provider="provider" action="attach" is-link="true"
ng-if="!externalLoginInfo[provider.id]"></span>
<a ng-if="externalLoginInfo[provider.id] && Features.DIRECT_LOGIN"
ng-click="detachExternalLogin(provider.id)">Detach Account</a>

4
static/directives/icon-image-view.html Normal file
View file

@ -0,0 +1,4 @@
<span class="icon-image-view-element">
<img ng-src="{{ value }}" ng-if="value.indexOf('http') == 0">
<i class="fa" ng-class="value" ng-if="value.indexOf('http') < 0"></i>
</span>

2
static/directives/signin-form.html
View file

@ -6,7 +6,7 @@
</h4>
<div class="external-logins" quay-show="EXTERNAL_LOGINS.length" ng-class="EXTERNAL_LOGINS.length > 2 ? 'smaller': 'larger'">
<div class="external-login-button" provider="{{ provider.id }}" redirect-url="redirectUrl"
<div class="external-login-button" provider="provider" redirect-url="redirectUrl"
sign-in-started="markStarted()" ng-repeat="provider in EXTERNAL_LOGINS" is-link="true"></div>
</div>

9
static/js/directives/ui/external-login-button.js
View file

@ -12,18 +12,15 @@ angular.module('quay').directive('externalLoginButton', function () {
'signInStarted': '&signInStarted',
'redirectUrl': '=redirectUrl',
'isLink': '=isLink',
'provider': '@provider',
'provider': '=provider',
'action': '@action'
},
controller: function($scope, $timeout, $interval, ApiService, KeyService, CookieService, ExternalLoginService) {
$scope.signingIn = false;
$scope.providerInfo = ExternalLoginService.getProvider($scope.provider);
$scope.startSignin = function() {
$scope.signInStarted({'service': $scope.provider});
ApiService.generateExternalLoginToken().then(function(data) {
var url = ExternalLoginService.getLoginUrl($scope.provider, $scope.action || 'login');
url = url + '&state=' + encodeURIComponent(data['token']);
ExternalLoginService.getLoginUrl($scope.provider, $scope.action || 'login', function(url) {
// Save the redirect URL in a cookie so that we can redirect back after the service returns to us.
var redirectURL = $scope.redirectUrl || window.location.toString();
@ -35,7 +32,7 @@ angular.module('quay').directive('externalLoginButton', function () {
$timeout(function() {
document.location = url;
}, 250);
}, ApiService.errorDisplay('Could not perform sign in'));
});
};
}
};

4
static/js/directives/ui/external-logins-manager.js
View file

@ -34,11 +34,11 @@ angular.module('quay').directive('externalLoginsManager', function () {
}
});
$scope.detachExternalLogin = function(kind) {
$scope.detachExternalLogin = function(service_id) {
if (!Features.DIRECT_LOGIN) { return; }
var params = {
'servicename': kind
'service_id': service_id
};
ApiService.detachExternalLogin(null, params).then(function() {

4
static/js/directives/ui/header-bar.js
View file

@ -16,7 +16,9 @@ angular.module('quay').directive('headerBar', function () {
PlanService, ApiService, NotificationService, Config, Features,
DocumentationService, ExternalLoginService) {
$scope.externalSigninUrl = ExternalLoginService.getSingleSigninUrl();
ExternalLoginService.getSingleSigninUrl(function(url) {
$scope.externalSigninUrl = url;
});
var hotkeysAdded = false;
var userUpdated = function(cUser) {

18
static/js/directives/ui/icon-image-view.js Normal file
View file

@ -0,0 +1,18 @@
/**
* An element which displays either an icon or an image, depending on the value.
*/
angular.module('quay').directive('iconImageView', function () {
var directiveDefinitionObject = {
priority: 0,
templateUrl: '/static/directives/icon-image-view.html',
replace: false,
transclude: true,
restrict: 'C',
scope: {
'value': '@value'
},
controller: function($scope, $element) {
}
};
return directiveDefinitionObject;
});

3
static/js/pages/signin.js
View file

@ -11,9 +11,10 @@
function SignInCtrl($scope, $location, ExternalLoginService, Features) {
$scope.redirectUrl = '/';
var singleUrl = ExternalLoginService.getSingleSigninUrl();
ExternalLoginService.getSingleSigninUrl(function(singleUrl) {
if (singleUrl) {
document.location = singleUrl;
}
});
}
})();

129
static/js/services/external-login-service.js
View file

@ -1,131 +1,42 @@
/**
* Service which exposes the supported external logins.
*/
angular.module('quay').factory('ExternalLoginService', ['KeyService', 'Features', 'Config',
function(KeyService, Features, Config) {
angular.module('quay').factory('ExternalLoginService', ['Features', 'Config', 'ApiService',
function(Features, Config, ApiService) {
var externalLoginService = {};
externalLoginService.getLoginUrl = function(service, action) {
var serviceInfo = externalLoginService.getProvider(service);
if (!serviceInfo) { return ''; }
externalLoginService.EXTERNAL_LOGINS = window.__external_login || [];
var loginUrl = KeyService.getConfiguration(serviceInfo.key, 'AUTHORIZE_ENDPOINT');
var clientId = KeyService.getConfiguration(serviceInfo.key, 'CLIENT_ID');
externalLoginService.getLoginUrl = function(loginService, action, callback) {
var errorDisplay = ApiService.errorDisplay('Could not load external login service ' +
'information. Please contact your service ' +
'administrator.')
var scope = serviceInfo.scopes();
var redirectUri = Config.getUrl('/oauth2/' + service + '/callback');
if (action == 'attach') {
redirectUri += '/attach';
}
var url = loginUrl + 'client_id=' + clientId + '&scope=' + scope + '&redirect_uri=' +
redirectUri;
return url;
var params = {
'service_id': loginService['id']
};
var DEX = {
id: 'dex',
key: 'DEX_LOGIN_CONFIG',
title: function() {
return KeyService.getConfiguration('DEX_LOGIN_CONFIG', 'OIDC_TITLE');
},
icon: function() {
return {'url': KeyService.getConfiguration('DEX_LOGIN_CONFIG', 'OIDC_LOGO') };
},
scopes: function() {
return 'openid email profile'
},
enabled: Features.DEX_LOGIN
var data = {
'kind': action
};
var GITHUB = {
id: 'github',
key: 'GITHUB_LOGIN_CONFIG',
title: function() {
return KeyService.isEnterprise('github') ? 'GitHub Enterprise' : 'GitHub';
},
icon: function() {
return {'icon': 'fa-github'};
},
hasUserInfo: true,
getUserInfo: function(service_info) {
username = service_info['metadata']['service_username'];
return {
'username': username,
'endpoint': KeyService['githubEndpoint'] + username
}
},
scopes: function() {
var scopes = 'user:email';
if (KeyService.getConfiguration('GITHUB_LOGIN_CONFIG', 'ORG_RESTRICT')) {
scopes += ' read:org';
}
return scopes;
},
enabled: Features.GITHUB_LOGIN
};
var GOOGLE = {
id: 'google',
key: 'GOOGLE_LOGIN_CONFIG',
title: function() {
return 'Google';
},
icon: function() {
return {'icon': 'fa-google'};
},
scopes: function() {
return 'openid email';
},
enabled: Features.GOOGLE_LOGIN
};
externalLoginService.ALL_EXTERNAL_LOGINS = [
DEX, GITHUB, GOOGLE
];
externalLoginService.EXTERNAL_LOGINS = externalLoginService.ALL_EXTERNAL_LOGINS.filter(function(el) {
return el.enabled;
});
externalLoginService.getProvider = function(providerId) {
for (var i = 0; i < externalLoginService.EXTERNAL_LOGINS.length; ++i) {
var current = externalLoginService.EXTERNAL_LOGINS[i];
if (current.id == providerId) {
return current;
}
}
return null;
ApiService.retrieveExternalLoginAuthorizationUrl(data, params).then(function(resp) {
callback(resp['auth_url']);
}, errorDisplay);
};
externalLoginService.hasSingleSignin = function() {
return externalLoginService.EXTERNAL_LOGINS.length == 1 && !Features.DIRECT_LOGIN;
};
externalLoginService.getSingleSigninUrl = function() {
// If there is a single external login service and direct login is disabled,
// then redirect to the external login directly.
if (externalLoginService.hasSingleSignin()) {
return externalLoginService.getLoginUrl(externalLoginService.EXTERNAL_LOGINS[0].id);
externalLoginService.getSingleSigninUrl = function(callback) {
if (!externalLoginService.hasSingleSignin()) {
return callback(null);
}
return null;
// If there is a single external login service and direct login is disabled,
// then redirect to the external login directly.
externalLoginService.getLoginUrl(externalLoginService.EXTERNAL_LOGINS[0], 'login', callback);
};
return externalLoginService;

1
templates/base.html
View file

@ -39,6 +39,7 @@
window.__features = {{ feature_set|tojson|safe }};
window.__config = {{ config_set|tojson|safe }};
window.__oauth = {{ oauth_set|tojson|safe }};
window.__external_login = {{ external_login_set|tojson|safe }};
window.__auth_scopes = {{ scope_set|tojson|safe }};
window.__vuln_priority = {{ vuln_priority_set|tojson|safe }}
window.__token = '{{ csrf_token() }}';

22
test/test_api_security.py
View file

@ -28,7 +28,7 @@ from endpoints.api.repositorynotification import RepositoryNotification, Reposit
from endpoints.api.user import (PrivateRepositories, ConvertToOrganization, Recovery, Signout,
Signin, User, UserAuthorizationList, UserAuthorization, UserNotification,
VerifyUser, DetachExternal, StarredRepositoryList, StarredRepository,
ClientKey)
ClientKey, ExternalLoginInformation)
from endpoints.api.repotoken import RepositoryToken, RepositoryTokenList
from endpoints.api.prototype import PermissionPrototype, PermissionPrototypeList
from endpoints.api.logs import UserLogs, OrgLogs, RepositoryLogs
@ -505,10 +505,28 @@ class TestSignin(ApiTestCase):
self._run_test('POST', 403, 'devtable', {u'username': 'E9RY', u'password': 'LQ0N'})
class TestExternalLoginInformation(ApiTestCase):
def setUp(self):
ApiTestCase.setUp(self)
self._set_url(ExternalLoginInformation, service_id='someservice')
def test_post_anonymous(self):
self._run_test('POST', 400, None, {})
def test_post_freshuser(self):
self._run_test('POST', 400, 'freshuser', {})
def test_post_reader(self):
self._run_test('POST', 400, 'reader', {})
def test_post_devtable(self):
self._run_test('POST', 400, 'devtable', {})
class TestDetachExternal(ApiTestCase):
def setUp(self):
ApiTestCase.setUp(self)
self._set_url(DetachExternal, servicename='someservice')
self._set_url(DetachExternal, service_id='someservice')
def test_post_anonymous(self):
self._run_test('POST', 401, None, {})

142
test/test_endpoints.py
View file

@ -8,7 +8,6 @@ import base64
from urllib import urlencode
from urlparse import urlparse, urlunparse, parse_qs
from datetime import datetime, timedelta
from httmock import urlmatch, HTTMock
import jwt
@ -25,16 +24,9 @@ from endpoints.api.user import Signin
from endpoints.keyserver import jwk_with_kid
from endpoints.csrf import OAUTH_CSRF_TOKEN_NAME
from endpoints.web import web as web_bp
from endpoints.oauthlogin import oauthlogin as oauthlogin_bp
from initdb import setup_database_for_testing, finished_database_for_testing
from test.helpers import assert_action_logged
try:
app.register_blueprint(oauthlogin_bp, url_prefix='/oauth')
except ValueError:
# This blueprint was already registered
pass
try:
app.register_blueprint(web_bp, url_prefix='')
except ValueError:
@ -129,140 +121,6 @@ class EndpointTestCase(unittest.TestCase):
self.assertEquals(rv.status_code, 200)
class OAuthLoginTestCase(EndpointTestCase):
def invoke_oauth_tests(self, callback_endpoint, attach_endpoint, service_name, service_ident,
new_username):
# Test callback.
created = self.invoke_oauth_test(callback_endpoint, service_name, service_ident, new_username)
# Delete the created user.
model.user.delete_user(created, [])
# Test attach.
self.login('devtable', 'password')
self.invoke_oauth_test(attach_endpoint, service_name, service_ident, 'devtable')
def invoke_oauth_test(self, endpoint_name, service_name, service_ident, username):
# No CSRF.
self.getResponse('oauthlogin.' + endpoint_name, expected_code=403)
# Invalid CSRF.
self.getResponse('oauthlogin.' + endpoint_name, state='somestate', expected_code=403)
# Valid CSRF, invalid code.
self.getResponse('oauthlogin.' + endpoint_name, state='someoauthtoken',
code='invalidcode', expected_code=400)
# Valid CSRF, valid code.
self.getResponse('oauthlogin.' + endpoint_name, state='someoauthtoken',
code='somecode', expected_code=302)
# Ensure the user was added/modified.
found_user = model.user.get_user(username)
self.assertIsNotNone(found_user)
federated_login = model.user.lookup_federated_login(found_user, service_name)
self.assertIsNotNone(federated_login)
self.assertEquals(federated_login.service_ident, service_ident)
return found_user
def test_google_oauth(self):
@urlmatch(netloc=r'accounts.google.com', path='/o/oauth2/token')
def account_handler(_, request):
if request.body.find("code=somecode") > 0:
content = {'access_token': 'someaccesstoken'}
return py_json.dumps(content)
else:
return {'status_code': 400, 'content': '{"message": "Invalid code"}'}
@urlmatch(netloc=r'www.googleapis.com', path='/oauth2/v1/userinfo')
def user_handler(_, __):
content = {
'id': 'someid',
'email': 'someemail@example.com',
'verified_email': True,
}
return py_json.dumps(content)
with HTTMock(account_handler, user_handler):
self.invoke_oauth_tests('google_oauth_callback', 'google_oauth_attach', 'google',
'someid', 'someemail')
def test_github_oauth(self):
@urlmatch(netloc=r'github.com', path='/login/oauth/access_token')
def account_handler(url, _):
if url.query.find("code=somecode") > 0:
content = {'access_token': 'someaccesstoken'}
return py_json.dumps(content)
else:
return {'status_code': 400, 'content': '{"message": "Invalid code"}'}
@urlmatch(netloc=r'github.com', path='/api/v3/user')
def user_handler(_, __):
content = {
'id': 'someid',
'login': 'someusername'
}
return py_json.dumps(content)
@urlmatch(netloc=r'github.com', path='/api/v3/user/emails')
def email_handler(_, __):
content = [{
'email': 'someemail@example.com',
'verified': True,
'primary': True,
}]
return py_json.dumps(content)
with HTTMock(account_handler, email_handler, user_handler):
self.invoke_oauth_tests('github_oauth_callback', 'github_oauth_attach', 'github',
'someid', 'someusername')
def test_dex_oauth(self):
# TODO(jschorr): Add tests for invalid and expired keys.
# Generate a public/private key pair for the OIDC transaction.
private_key = RSA.generate(2048)
jwk = RSAKey(key=private_key.publickey()).serialize()
token = jwt.encode({
'iss': 'https://oidcserver/',
'aud': 'someclientid',
'sub': 'someid',
'exp': int(time.time()) + 60,
'iat': int(time.time()),
'nbf': int(time.time()),
'email': 'someemail@example.com',
'email_verified': True,
}, private_key.exportKey('PEM'), 'RS256')
@urlmatch(netloc=r'oidcserver', path='/.well-known/openid-configuration')
def wellknown_handler(url, _):
return py_json.dumps({
'authorization_endpoint': 'http://oidcserver/auth',
'token_endpoint': 'http://oidcserver/token',
'jwks_uri': 'http://oidcserver/keys',
})
@urlmatch(netloc=r'oidcserver', path='/token')
def account_handler(url, request):
if request.body.find("code=somecode") > 0:
return py_json.dumps({
'access_token': token,
})
else:
return {'status_code': 400, 'content': '{"message": "Invalid code"}'}
@urlmatch(netloc=r'oidcserver', path='/keys')
def keys_handler(_, __):
return py_json.dumps({
"keys": [jwk],
})
with HTTMock(wellknown_handler, account_handler, keys_handler):
self.invoke_oauth_tests('dex_oauth_callback', 'dex_oauth_attach', 'dex',
'someid', 'someemail')
class WebEndpointTestCase(EndpointTestCase):
def test_index(self):
self.getResponse('web.index')

6
test/test_github.py
View file

@ -1,6 +1,6 @@
import unittest
from util.config.oauth import GithubOAuthConfig
from oauth.services.github import GithubOAuthService
class TestGithub(unittest.TestCase):
def test_basic_enterprise_config(self):
@ -12,7 +12,7 @@ class TestGithub(unittest.TestCase):
}
}
github_trigger = GithubOAuthConfig(config, 'GITHUB_TRIGGER_CONFIG')
github_trigger = GithubOAuthService(config, 'GITHUB_TRIGGER_CONFIG')
self.assertTrue(github_trigger.is_enterprise())
self.assertEquals('https://github.somedomain.com/login/oauth/authorize?', github_trigger.authorize_endpoint())
self.assertEquals('https://github.somedomain.com/login/oauth/access_token', github_trigger.token_endpoint())
@ -33,7 +33,7 @@ class TestGithub(unittest.TestCase):
}
}
github_trigger = GithubOAuthConfig(config, 'GITHUB_TRIGGER_CONFIG')
github_trigger = GithubOAuthService(config, 'GITHUB_TRIGGER_CONFIG')
self.assertTrue(github_trigger.is_enterprise())
self.assertEquals('https://github.somedomain.com/login/oauth/authorize?', github_trigger.authorize_endpoint())
self.assertEquals('https://github.somedomain.com/login/oauth/access_token', github_trigger.token_endpoint())

175
test/test_oauth_login.py Normal file
View file

@ -0,0 +1,175 @@
import json as py_json
import time
import unittest
import jwt
from Crypto.PublicKey import RSA
from httmock import urlmatch, HTTMock
from jwkest.jwk import RSAKey
from app import app
from data import model
from endpoints.oauthlogin import oauthlogin as oauthlogin_bp
from test.test_endpoints import EndpointTestCase
try:
app.register_blueprint(oauthlogin_bp, url_prefix='/oauth2')
except ValueError:
# This blueprint was already registered
pass
class OAuthLoginTestCase(EndpointTestCase):
def invoke_oauth_tests(self, callback_endpoint, attach_endpoint, service_name, service_ident,
new_username):
# Test callback.
created = self.invoke_oauth_test(callback_endpoint, service_name, service_ident, new_username)
# Delete the created user.
model.user.delete_user(created, [])
# Test attach.
self.login('devtable', 'password')
self.invoke_oauth_test(attach_endpoint, service_name, service_ident, 'devtable')
def invoke_oauth_test(self, endpoint_name, service_name, service_ident, username):
# No CSRF.
self.getResponse('oauthlogin.' + endpoint_name, expected_code=403)
# Invalid CSRF.
self.getResponse('oauthlogin.' + endpoint_name, state='somestate', expected_code=403)
# Valid CSRF, invalid code.
self.getResponse('oauthlogin.' + endpoint_name, state='someoauthtoken',
code='invalidcode', expected_code=400)
# Valid CSRF, valid code.
self.getResponse('oauthlogin.' + endpoint_name, state='someoauthtoken',
code='somecode', expected_code=302)
# Ensure the user was added/modified.
found_user = model.user.get_user(username)
self.assertIsNotNone(found_user)
federated_login = model.user.lookup_federated_login(found_user, service_name)
self.assertIsNotNone(federated_login)
self.assertEquals(federated_login.service_ident, service_ident)
return found_user
def test_google_oauth(self):
@urlmatch(netloc=r'accounts.google.com', path='/o/oauth2/token')
def account_handler(_, request):
if request.body.find("code=somecode") > 0:
content = {'access_token': 'someaccesstoken'}
return py_json.dumps(content)
else:
return {'status_code': 400, 'content': '{"message": "Invalid code"}'}
@urlmatch(netloc=r'www.googleapis.com', path='/oauth2/v1/userinfo')
def user_handler(_, __):
content = {
'id': 'someid',
'email': 'someemail@example.com',
'verified_email': True,
}
return py_json.dumps(content)
with HTTMock(account_handler, user_handler):
self.invoke_oauth_tests('google_oauth_callback', 'google_oauth_attach', 'google',
'someid', 'someemail')
def test_github_oauth(self):
@urlmatch(netloc=r'github.com', path='/login/oauth/access_token')
def account_handler(url, _):
if url.query.find("code=somecode") > 0:
content = {'access_token': 'someaccesstoken'}
return py_json.dumps(content)
else:
return {'status_code': 400, 'content': '{"message": "Invalid code"}'}
@urlmatch(netloc=r'github.com', path='/api/v3/user')
def user_handler(_, __):
content = {
'id': 'someid',
'login': 'someusername'
}
return py_json.dumps(content)
@urlmatch(netloc=r'github.com', path='/api/v3/user/emails')
def email_handler(_, __):
content = [{
'email': 'someemail@example.com',
'verified': True,
'primary': True,
}]
return py_json.dumps(content)
with HTTMock(account_handler, email_handler, user_handler):
self.invoke_oauth_tests('github_oauth_callback', 'github_oauth_attach', 'github',
'someid', 'someusername')
def test_oidc_auth(self):
private_key = RSA.generate(2048)
generatedjwk = RSAKey(key=private_key.publickey()).serialize()
kid = 'somekey'
private_pem = private_key.exportKey('PEM')
token_data = {
'iss': app.config['TESTOIDC_LOGIN_CONFIG']['OIDC_SERVER'],
'aud': app.config['TESTOIDC_LOGIN_CONFIG']['CLIENT_ID'],
'nbf': int(time.time()),
'iat': int(time.time()),
'exp': int(time.time() + 600),
'sub': 'cooluser',
}
token_headers = {
'kid': kid,
}
id_token = jwt.encode(token_data, private_pem, 'RS256', headers=token_headers)
@urlmatch(netloc=r'fakeoidc', path='/token')
def token_handler(_, request):
if request.body.find("code=somecode") >= 0:
content = {'access_token': 'someaccesstoken', 'id_token': id_token}
return py_json.dumps(content)
else:
return {'status_code': 400, 'content': '{"message": "Invalid code"}'}
@urlmatch(netloc=r'fakeoidc', path='/user')
def user_handler(_, __):
content = {
'sub': 'cooluser',
'preferred_username': 'someusername',
'email': 'someemail@example.com',
'email_verified': True,
}
return py_json.dumps(content)
@urlmatch(netloc=r'fakeoidc', path='/jwks')
def jwks_handler(_, __):
jwk = generatedjwk.copy()
jwk.update({'kid': kid})
content = {'keys': [jwk]}
return py_json.dumps(content)
@urlmatch(netloc=r'fakeoidc', path='.+openid.+')
def discovery_handler(_, __):
content = {
'scopes_supported': ['profile'],
'authorization_endpoint': 'http://fakeoidc/authorize',
'token_endpoint': 'http://fakeoidc/token',
'userinfo_endpoint': 'http://fakeoidc/userinfo',
'jwks_uri': 'http://fakeoidc/jwks',
}
return py_json.dumps(content)
with HTTMock(discovery_handler, jwks_handler, token_handler, user_handler):
self.invoke_oauth_tests('testoidc_oauth_callback', 'testoidc_oauth_attach', 'testoidc',
'cooluser', 'someusername')
if __name__ == '__main__':
unittest.main()

12
test/testconfig.py
View file

@ -76,13 +76,17 @@ class TestConfig(DefaultConfig):
PROMETHEUS_AGGREGATOR_URL = None
GITHUB_LOGIN_CONFIG = {}
GOOGLE_LOGIN_CONFIG = {}
FEATURE_GITHUB_LOGIN = True
FEATURE_GOOGLE_LOGIN = True
FEATURE_DEX_LOGIN = True
DEX_LOGIN_CONFIG = {
'CLIENT_ID': 'someclientid',
'OIDC_SERVER': 'https://oidcserver/',
TESTOIDC_LOGIN_CONFIG = {
'CLIENT_ID': 'foo',
'CLIENT_SECRET': 'bar',
'OIDC_SERVER': 'http://fakeoidc',
'DEBUGGING': True,
}
RECAPTCHA_SITE_KEY = 'somekey'

370
util/config/oauth.py
View file

@ -1,370 +0,0 @@
import urlparse
import json
import logging
import time
from cachetools import TTLCache
from cachetools.func import lru_cache
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.serialization import load_der_public_key
from jwkest.jwk import KEYS
from util import slash_join
logger = logging.getLogger(__name__)
class OAuthConfig(object):
def __init__(self, config, key_name):
self.key_name = key_name
self.config = config.get(key_name) or {}
def service_name(self):
raise NotImplementedError
def token_endpoint(self):
raise NotImplementedError
def user_endpoint(self):
raise NotImplementedError
def validate_client_id_and_secret(self, http_client, app_config):
raise NotImplementedError
def client_id(self):
return self.config.get('CLIENT_ID')
def client_secret(self):
return self.config.get('CLIENT_SECRET')
def get_redirect_uri(self, app_config, redirect_suffix=''):
return '%s://%s/oauth2/%s/callback%s' % (app_config['PREFERRED_URL_SCHEME'],
app_config['SERVER_HOSTNAME'],
self.service_name().lower(),
redirect_suffix)
def exchange_code_for_token(self, app_config, http_client, code, form_encode=False,
redirect_suffix='', client_auth=False):
payload = {
'code': code,
'grant_type': 'authorization_code',
'redirect_uri': self.get_redirect_uri(app_config, redirect_suffix)
}
headers = {
'Accept': 'application/json'
}
auth = None
if client_auth:
auth = (self.client_id(), self.client_secret())
else:
payload['client_id'] = self.client_id()
payload['client_secret'] = self.client_secret()
token_url = self.token_endpoint()
if form_encode:
get_access_token = http_client.post(token_url, data=payload, headers=headers, auth=auth)
else:
get_access_token = http_client.post(token_url, params=payload, headers=headers, auth=auth)
if get_access_token.status_code / 100 != 2:
return None
json_data = get_access_token.json()
if not json_data:
return None
return json_data.get('access_token', None)
class GithubOAuthConfig(OAuthConfig):
def __init__(self, config, key_name):
super(GithubOAuthConfig, self).__init__(config, key_name)
def service_name(self):
return 'GitHub'
def allowed_organizations(self):
if not self.config.get('ORG_RESTRICT', False):
return None
allowed = self.config.get('ALLOWED_ORGANIZATIONS', None)
if allowed is None:
return None
return [org.lower() for org in allowed]
def get_public_url(self, suffix):
return slash_join(self._endpoint(), suffix)
def _endpoint(self):
return self.config.get('GITHUB_ENDPOINT', 'https://github.com')
def is_enterprise(self):
return self._endpoint().find('.github.com') < 0
def authorize_endpoint(self):
return slash_join(self._endpoint(), '/login/oauth/authorize') + '?'
def token_endpoint(self):
return slash_join(self._endpoint(), '/login/oauth/access_token')
def _api_endpoint(self):
return self.config.get('API_ENDPOINT', slash_join(self._endpoint(), '/api/v3/'))
def api_endpoint(self):
endpoint = self._api_endpoint()
if endpoint.endswith('/'):
return endpoint[0:-1]
return endpoint
def user_endpoint(self):
return slash_join(self._api_endpoint(), 'user')
def email_endpoint(self):
return slash_join(self._api_endpoint(), 'user/emails')
def orgs_endpoint(self):
return slash_join(self._api_endpoint(), 'user/orgs')
def validate_client_id_and_secret(self, http_client, app_config):
# First: Verify that the github endpoint is actually Github by checking for the
# X-GitHub-Request-Id here.
api_endpoint = self._api_endpoint()
result = http_client.get(api_endpoint, auth=(self.client_id(), self.client_secret()), timeout=5)
if not 'X-GitHub-Request-Id' in result.headers:
raise Exception('Endpoint is not a Github (Enterprise) installation')
# Next: Verify the client ID and secret.
# Note: The following code is a hack until such time as Github officially adds an API endpoint
# for verifying a {client_id, client_secret} pair. That being said, this hack was given to us
# *by a Github Engineer*, so I think it is okay for the time being :)
#
# TODO(jschorr): Replace with the real API call once added.
#
# Hitting the endpoint applications/{client_id}/tokens/foo will result in the following
# behavior IF the client_id is given as the HTTP username and the client_secret as the HTTP
# password:
# - If the {client_id, client_secret} pair is invalid in some way, we get a 401 error.
# - If the pair is valid, then we get a 404 because the 'foo' token does not exists.
validate_endpoint = slash_join(api_endpoint, 'applications/%s/tokens/foo' % self.client_id())
result = http_client.get(validate_endpoint, auth=(self.client_id(), self.client_secret()),
timeout=5)
return result.status_code == 404
def validate_organization(self, organization_id, http_client):
org_endpoint = slash_join(self._api_endpoint(), 'orgs/%s' % organization_id.lower())
result = http_client.get(org_endpoint,
headers={'Accept': 'application/vnd.github.moondragon+json'},
timeout=5)
return result.status_code == 200
def get_public_config(self):
return {
'CLIENT_ID': self.client_id(),
'AUTHORIZE_ENDPOINT': self.authorize_endpoint(),
'GITHUB_ENDPOINT': self._endpoint(),
'ORG_RESTRICT': self.config.get('ORG_RESTRICT', False)
}
class GoogleOAuthConfig(OAuthConfig):
def __init__(self, config, key_name):
super(GoogleOAuthConfig, self).__init__(config, key_name)
def service_name(self):
return 'Google'
def authorize_endpoint(self):
return 'https://accounts.google.com/o/oauth2/auth?response_type=code&'
def token_endpoint(self):
return 'https://accounts.google.com/o/oauth2/token'
def user_endpoint(self):
return 'https://www.googleapis.com/oauth2/v1/userinfo'
def validate_client_id_and_secret(self, http_client, app_config):
# To verify the Google client ID and secret, we hit the
# https://www.googleapis.com/oauth2/v3/token endpoint with an invalid request. If the client
# ID or secret are invalid, we get returned a 403 Unauthorized. Otherwise, we get returned
# another response code.
url = 'https://www.googleapis.com/oauth2/v3/token'
data = {
'code': 'fakecode',
'client_id': self.client_id(),
'client_secret': self.client_secret(),
'grant_type': 'authorization_code',
'redirect_uri': 'http://example.com'
}
result = http_client.post(url, data=data, timeout=5)
return result.status_code != 401
def get_public_config(self):
return {
'CLIENT_ID': self.client_id(),
'AUTHORIZE_ENDPOINT': self.authorize_endpoint()
}
class GitLabOAuthConfig(OAuthConfig):
def __init__(self, config, key_name):
super(GitLabOAuthConfig, self).__init__(config, key_name)
def _endpoint(self):
return self.config.get('GITLAB_ENDPOINT', 'https://gitlab.com')
def api_endpoint(self):
return self._endpoint()
def get_public_url(self, suffix):
return slash_join(self._endpoint(), suffix)
def service_name(self):
return 'GitLab'
def authorize_endpoint(self):
return slash_join(self._endpoint(), '/oauth/authorize')
def token_endpoint(self):
return slash_join(self._endpoint(), '/oauth/token')
def validate_client_id_and_secret(self, http_client, app_config):
url = self.token_endpoint()
redirect_uri = self.get_redirect_uri(app_config, redirect_suffix='trigger')
data = {
'code': 'fakecode',
'client_id': self.client_id(),
'client_secret': self.client_secret(),
'grant_type': 'authorization_code',
'redirect_uri': redirect_uri
}
# We validate by checking the error code we receive from this call.
result = http_client.post(url, data=data, timeout=5)
value = result.json()
if not value:
return False
return value.get('error', '') != 'invalid_client'
def get_public_config(self):
return {
'CLIENT_ID': self.client_id(),
'AUTHORIZE_ENDPOINT': self.authorize_endpoint(),
'GITLAB_ENDPOINT': self._endpoint(),
}
OIDC_WELLKNOWN = ".well-known/openid-configuration"
PUBLIC_KEY_CACHE_TTL = 3600 # 1 hour
class OIDCConfig(OAuthConfig):
def __init__(self, config, key_name):
super(OIDCConfig, self).__init__(config, key_name)
self._public_key_cache = TTLCache(1, PUBLIC_KEY_CACHE_TTL, missing=self._get_public_key)
self._config = config
self._http_client = config['HTTPCLIENT']
@lru_cache(maxsize=1)
def _oidc_config(self):
if self.config.get('OIDC_SERVER'):
return self._load_via_discovery(self._config.get('DEBUGGING', False))
else:
return {}
def _load_via_discovery(self, is_debugging):
oidc_server = self.config['OIDC_SERVER']
if not oidc_server.startswith('https://') and not is_debugging:
raise Exception('OIDC server must be accessed over SSL')
discovery_url = urlparse.urljoin(oidc_server, OIDC_WELLKNOWN)
discovery = self._http_client.get(discovery_url, timeout=5)
if discovery.status_code / 100 != 2:
raise Exception("Could not load OIDC discovery information")
try:
return json.loads(discovery.text)
except ValueError:
logger.exception('Could not parse OIDC discovery for url: %s', discovery_url)
raise Exception("Could not parse OIDC discovery information")
def authorize_endpoint(self):
return self._oidc_config().get('authorization_endpoint', '') + '?'
def token_endpoint(self):
return self._oidc_config().get('token_endpoint')
def user_endpoint(self):
return None
def validate_client_id_and_secret(self, http_client, app_config):
pass
def get_public_config(self):
return {
'CLIENT_ID': self.client_id(),
'AUTHORIZE_ENDPOINT': self.authorize_endpoint()
}
@property
def issuer(self):
return self.config.get('OIDC_ISSUER', self.config['OIDC_SERVER'])
def get_public_key(self, force_refresh=False):
""" Retrieves the public key for this handler. """
# If force_refresh is true, we expire all the items in the cache by setting the time to
# the current time + the expiration TTL.
if force_refresh:
self._public_key_cache.expire(time=time.time() + PUBLIC_KEY_CACHE_TTL)
# Retrieve the public key from the cache. If the cache does not contain the public key,
# it will internally call _get_public_key to retrieve it and then save it. The None is
# a random key chose to be stored in the cache, and could be anything.
return self._public_key_cache[None]
def _get_public_key(self, _):
""" Retrieves the public key for this handler. """
keys_url = self._oidc_config()['jwks_uri']
keys = KEYS()
keys.load_from_url(keys_url)
if not list(keys):
raise Exception('No keys provided by OIDC provider')
rsa_key = list(keys)[0]
rsa_key.deserialize()
# Reload the key so that we can give a key *instance* to PyJWT to work around its weird parsing
# issues.
return load_der_public_key(rsa_key.key.exportKey('DER'), backend=default_backend())
class DexOAuthConfig(OIDCConfig):
def service_name(self):
return 'Dex'
@property
def public_title(self):
return self.get_public_config()['OIDC_TITLE']
def get_public_config(self):
return {
'CLIENT_ID': self.client_id(),
'AUTHORIZE_ENDPOINT': self.authorize_endpoint(),
# TODO(jschorr): This should ideally come from the Dex side.
'OIDC_TITLE': 'Dex',
'OIDC_LOGO': 'https://tectonic.com/assets/ico/favicon-96x96.png'
}

10
util/config/validator.py
View file

@ -22,7 +22,9 @@ from data.users.externaljwt import ExternalJWTAuthN
from data.users.externalldap import LDAPConnection, LDAPUsers
from data.users.keystone import get_keystone_users
from storage import get_storage_driver
from util.config.oauth import GoogleOAuthConfig, GithubOAuthConfig, GitLabOAuthConfig
from oauth.services.github import GithubOAuthService
from oauth.services.google import GoogleOAuthService
from oauth.services.gitlab import GitLabOAuthService
from util.secscan.api import SecurityScannerAPI
from util.registry.torrent import torrent_jwt
from util.security.signing import SIGNING_ENGINES
@ -159,7 +161,7 @@ def _validate_gitlab(config, user_obj, _):
raise ConfigValidationException('Missing Client Secret')
client = app.config['HTTPCLIENT']
oauth = GitLabOAuthConfig(config, 'GITLAB_TRIGGER_CONFIG')
oauth = GitLabOAuthService(config, 'GITLAB_TRIGGER_CONFIG')
result = oauth.validate_client_id_and_secret(client, app.config)
if not result:
raise ConfigValidationException('Invalid client id or client secret')
@ -193,7 +195,7 @@ def _validate_github_with_key(config_key, config):
'organization')
client = app.config['HTTPCLIENT']
oauth = GithubOAuthConfig(config, config_key)
oauth = GithubOAuthService(config, config_key)
result = oauth.validate_client_id_and_secret(client, app.config)
if not result:
raise ConfigValidationException('Invalid client id or client secret')
@ -239,7 +241,7 @@ def _validate_google_login(config, user_obj, _):
raise ConfigValidationException('Missing Client Secret')
client = app.config['HTTPCLIENT']
oauth = GoogleOAuthConfig(config, 'GOOGLE_LOGIN_CONFIG')
oauth = GoogleOAuthService(config, 'GOOGLE_LOGIN_CONFIG')
result = oauth.validate_client_id_and_secret(client, app.config)
if not result:
raise ConfigValidationException('Invalid client id or client secret')