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

Enable support in OIDC for endpoints without user info support

Browse source 
The user info endpoint is apparently optional.
This commit is contained in:
Joseph Schorr 2017-07-21 15:56:46 -04:00
parent 9676d7d8c7
commit 751598056e
3 changed files with 100 additions and 100 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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)}