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

Merge pull request #2817 from coreos-inc/joseph.schorr/QUAY-688/dex-fixes

Browse source 
Dex Fixes
This commit is contained in:
josephschorr 2017-08-01 13:26:50 -04:00 committed by GitHub
commit 83e9dfac7a
6 changed files with 105 additions and 104 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
endpoints/api/repository.py
View file

@ -330,7 +330,8 @@ class RepositoryTrust(RepositoryParamResource):
tags, _ = tuf_metadata_api.get_default_tags_with_expiration(namespace, repository)
if tags and not tuf_metadata_api.delete_metadata(namespace, repository):
raise DownstreamIssue({'message': 'Unable to delete downstream trust metadata'})
raise DownstreamIssue('Unable to delete downstream trust metadata')
values = request.get_json()
model.set_trust(namespace, repository, values['trust_enabled'])

2
endpoints/api/secscan.py
View file

@ -47,7 +47,7 @@ def _security_status_for_image(namespace, repository, repo_image, include_vulner
else:
data = secscan_api.get_layer_data(repo_image, include_features=True)
except APIRequestFailure as arf:
raise DownstreamIssue({'message': arf.message})
raise DownstreamIssue(arf.message)
if data is None:
raise NotFound()

4
endpoints/exception.py
View file

@ -129,5 +129,5 @@ class NotFound(ApiException):
class DownstreamIssue(ApiException):
def __init__(self, payload=None):
ApiException.__init__(self, ApiErrorType.downstream_issue, 520, 'Downstream Issue', payload)
def __init__(self, error_description, payload=None):
ApiException.__init__(self, ApiErrorType.downstream_issue, 520, error_description, payload)

7
oauth/oidc.py
View file

@ -72,7 +72,7 @@ class OIDCLoginService(OAuthService):
return self._oidc_config().get('userinfo_endpoint')
def validate(self):
return bool(self.user_endpoint())
return bool(self.token_endpoint())
def validate_client_id_and_secret(self, http_client, app_config):
# TODO: find a way to verify client secret too.
@ -119,11 +119,16 @@ class OIDCLoginService(OAuthService):
logger.exception('Could not load public key during OIDC decode: %s', pke.message)
raise OAuthLoginException('Could find public OIDC key')
# If there is a user endpoint, use it to retrieve the user's information. Otherwise, we use
# the decoded ID token.
if self.user_endpoint():
# Retrieve the user information.
try:
user_info = self.get_user_info(http_client, access_token)
except OAuthGetUserInfoException as oge:
raise OAuthLoginException(oge.message)
else:
user_info = decoded_id_token
# Verify subs.
if user_info['sub'] != decoded_id_token['sub']:

181
oauth/test/test_oidc.py
View file

@ -14,7 +14,17 @@ from jwkest.jwk import RSAKey
from oauth.oidc import OIDCLoginService, OAuthLoginException
@pytest.fixture()
@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(scope="module")
def http_client():
sess = requests.Session()
adapter = requests.adapters.HTTPAdapter(pool_connections=100,
@ -23,12 +33,32 @@ def http_client():
sess.mount('https://', adapter)
return sess
@pytest.fixture(scope="module")
def valid_code():
return 'validcode'
@pytest.fixture(params=[True, False])
def app_config(http_client, request):
def mailing_feature(request):
return request.param
@pytest.fixture(params=[True, False])
def email_verified(request):
return request.param
@pytest.fixture(params=[True, False])
def userinfo_supported(request):
return request.param
@pytest.fixture(params=["someusername", None])
def preferred_username(request):
return request.param
@pytest.fixture()
def app_config(http_client, mailing_feature):
return {
'PREFERRED_URL_SCHEME': 'http',
'SERVER_HOSTNAME': 'localhost',
'FEATURE_MAILING': request.param,
'FEATURE_MAILING': mailing_feature,
'SOMEOIDC_TEST_SERVICE': {
'CLIENT_ID': 'foo',
@ -47,35 +77,26 @@ def oidc_service(app_config):
return OIDCLoginService(app_config, 'SOMEOIDC_TEST_SERVICE')
@pytest.fixture()
def discovery_content():
def discovery_content(userinfo_supported):
return {
'scopes_supported': ['profile'],
'authorization_endpoint': 'http://fakeoidc/authorize',
'token_endpoint': 'http://fakeoidc/token',
'userinfo_endpoint': 'http://fakeoidc/userinfo',
'userinfo_endpoint': 'http://fakeoidc/userinfo' if userinfo_supported else None,
'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()
def userinfo_content(preferred_username, email_verified):
return {
'id': 'somekey',
'private_key': private_key.exportKey('PEM'),
'jwk': jwk,
'sub': 'cooluser',
'preferred_username': preferred_username,
'email': 'foo@example.com',
'email_verified': email_verified,
}
@pytest.fixture()
def id_token(oidc_service, signing_key, app_config):
def id_token(oidc_service, signing_key, userinfo_content, app_config):
token_data = {
'iss': oidc_service.config['OIDC_SERVER'],
'aud': oidc_service.client_id(),
@ -85,6 +106,8 @@ def id_token(oidc_service, signing_key, app_config):
'sub': 'cooluser',
}
token_data.update(userinfo_content)
token_headers = {
'kid': signing_key['id'],
}
@ -92,8 +115,12 @@ def id_token(oidc_service, signing_key, app_config):
return jwt.encode(token_data, signing_key['private_key'], 'RS256', headers=token_headers)
@pytest.fixture()
def valid_code():
return 'validcode'
def discovery_handler(discovery_content):
@urlmatch(netloc=r'fakeoidc', path=r'.+openid.+')
def handler(_, __):
return json.dumps(discovery_content)
return handler
@pytest.fixture()
def token_handler(oidc_service, id_token, valid_code):
@ -146,25 +173,14 @@ def emptykeys_jwks_handler():
return handler
@pytest.fixture(params=["someusername", None])
def preferred_username(request):
return request.param
@pytest.fixture
def userinfo_handler(oidc_service, preferred_username):
def userinfo_handler(oidc_service, userinfo_content):
@urlmatch(netloc=r'fakeoidc', path=r'/userinfo')
def handler(_, req):
if req.headers.get('Authorization') != 'Bearer sometoken':
return {'status_code': 401, 'content': 'Missing expected header'}
content = {
'sub': 'cooluser',
'preferred_username':preferred_username,
'email': 'foo@example.com',
'email_verified': True,
}
return {'status_code': 200, 'content': json.dumps(content)}
return {'status_code': 200, 'content': json.dumps(userinfo_content)}
return handler
@ -174,39 +190,26 @@ def invalidsub_userinfo_handler(oidc_service):
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):
def test_discovery(oidc_service, http_client, discovery_content, 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']
auth = discovery_content['authorization_endpoint'] + '?response_type=code&'
assert oidc_service.authorize_endpoint() == auth
assert oidc_service.token_endpoint() == discovery_content['token_endpoint']
assert oidc_service.user_endpoint() == discovery_content['userinfo_endpoint']
assert oidc_service.get_login_scopes() == discovery_content['scopes_supported']
def test_public_config(oidc_service, discovery_handler):
with HTTMock(discovery_handler):
@ -222,46 +225,13 @@ def test_exchange_code_invalidcode(oidc_service, discovery_handler, app_config,
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):
valid_code, userinfo_supported):
# Skip when userinfo is not supported.
if not userinfo_supported:
return
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):
@ -274,3 +244,28 @@ def test_exchange_code_missingkey(oidc_service, discovery_handler, app_config, h
# Should fail because the key is missing.
with pytest.raises(OAuthLoginException):
oidc_service.exchange_code_for_login(app_config, http_client, valid_code, '')
def test_exchange_code_validcode(oidc_service, discovery_handler, app_config, http_client,
token_handler, userinfo_handler, jwks_handler, valid_code,
preferred_username, mailing_feature, email_verified):
with HTTMock(jwks_handler, token_handler, userinfo_handler, discovery_handler):
if mailing_feature and not email_verified:
# Should fail because there isn't a verified email address.
with pytest.raises(OAuthLoginException):
oidc_service.exchange_code_for_login(app_config, http_client, valid_code, '')
else:
# Should succeed.
lid, lusername, lemail = oidc_service.exchange_code_for_login(app_config, http_client,
valid_code, '')
assert lid == 'cooluser'
if email_verified:
assert lemail == 'foo@example.com'
else:
assert lemail is None
if preferred_username is not None:
assert lusername == preferred_username
else:
assert lusername == lid

2
util/config/validators/test/test_validate_oidc.py
View file

@ -27,7 +27,7 @@ def test_validate_oidc_login(app):
def handler(_, __):
url_hit[0] = True
data = {
'userinfo_endpoint': 'foobar',
'token_endpoint': 'foobar',
}
return {'status_code': 200, 'content': json.dumps(data)}