quay/oauth/test/test_oidc.py

# 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(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,
                                          pool_maxsize=100)
  sess.mount('http://', adapter)
  sess.mount('https://', adapter)
  return sess

@pytest.fixture(scope="module")
def valid_code():
  return 'validcode'

@pytest.fixture(params=[True, False])
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': mailing_feature,

    'SOMEOIDC_LOGIN_CONFIG': {
      '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_LOGIN_CONFIG')

@pytest.fixture()
def discovery_content(userinfo_supported):
  return {
    'scopes_supported': ['profile'],
    'authorization_endpoint': 'http://fakeoidc/authorize',
    'token_endpoint': 'http://fakeoidc/token',
    'userinfo_endpoint': 'http://fakeoidc/userinfo' if userinfo_supported else None,
    'jwks_uri': 'http://fakeoidc/jwks',
  }

@pytest.fixture()
def userinfo_content(preferred_username, email_verified):
  return {
    'sub': 'cooluser',
    'preferred_username': preferred_username,
    'email': 'foo@example.com',
    'email_verified': email_verified,
  }

@pytest.fixture()
def id_token(oidc_service, signing_key, userinfo_content, 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_data.update(userinfo_content)

  token_headers = {
    'kid': signing_key['id'],
  }

  return jwt.encode(token_data, signing_key['private_key'], 'RS256', headers=token_headers)

@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()
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
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'}

    return {'status_code': 200, 'content': json.dumps(userinfo_content)}

  return handler

@pytest.fixture()
def invalidsub_userinfo_handler(oidc_service):
  @urlmatch(netloc=r'fakeoidc', path=r'/userinfo')
  def handler(_, __):
    content = {
      'sub': 'invalidsub',
    }

    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_content, discovery_handler):
  with HTTMock(discovery_handler):
    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):
    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_invalidsub(oidc_service, discovery_handler, app_config, http_client,
                                  token_handler, invalidsub_userinfo_handler, jwks_handler,
                                  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):
      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, '')

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