quay/util/config/validators/validate_jwt.py

import os
from data.users.externaljwt import ExternalJWTAuthN
from util.config.validators import BaseValidator, ConfigValidationException

class JWTAuthValidator(BaseValidator):
  name = "jwt"

  @classmethod
  def validate(cls, validator_context, public_key_path=None):
    """ Validates the JWT authentication system. """
    config = validator_context.config
    user = validator_context.user
    user_password = validator_context.user_password
    http_client = validator_context.http_client
    jwt_auth_max = validator_context.jwt_auth_max
    config_provider = validator_context.config_provider

    if config.get('AUTHENTICATION_TYPE', 'Database') != 'JWT':
      return

    verify_endpoint = config.get('JWT_VERIFY_ENDPOINT')
    query_endpoint = config.get('JWT_QUERY_ENDPOINT', None)
    getuser_endpoint = config.get('JWT_GETUSER_ENDPOINT', None)

    issuer = config.get('JWT_AUTH_ISSUER')

    if not verify_endpoint:
      raise ConfigValidationException('Missing JWT Verification endpoint')

    if not issuer:
      raise ConfigValidationException('Missing JWT Issuer ID')


    # TODO(jschorr): fix this
    return

    override_config_directory = os.path.join(config_provider.get_config_root(), '../stack/')

    # Try to instatiate the JWT authentication mechanism. This will raise an exception if
    # the key cannot be found.
    users = ExternalJWTAuthN(verify_endpoint, query_endpoint, getuser_endpoint, issuer,
                             override_config_directory,
                             http_client,
                             jwt_auth_max,
                             public_key_path=public_key_path,
                             requires_email=config.get('FEATURE_MAILING', True))

    # Verify that the superuser exists. If not, raise an exception.
    username = user.username
    (result, err_msg) = users.verify_credentials(username, user_password)
    if not result:
      msg = ('Verification of superuser %s failed: %s. \n\nThe user either does not ' +
             'exist in the remote authentication system ' +
             'OR JWT auth is misconfigured') % (username, err_msg)
      raise ConfigValidationException(msg)

    # If the query endpoint exists, ensure we can query to find the current user and that we can
    # look up users directly.
    if query_endpoint:
      (results, _, err_msg) = users.query_users(username)
      if not results:
        err_msg = err_msg or ('Could not find users matching query: %s' % username)
        raise ConfigValidationException('Query endpoint is misconfigured or not returning ' +
                                        'proper users: %s' % err_msg)

      # Make sure the get user endpoint is also configured.
      if not getuser_endpoint:
        raise ConfigValidationException('The lookup user endpoint must be configured if the ' +
                                        'query endpoint is set')

      (result, err_msg) = users.get_user(username)
      if not result:
        err_msg = err_msg or ('Could not find user %s' % username)
        raise ConfigValidationException('Lookup endpoint is misconfigured or not returning ' +
                                        'properly: %s' % err_msg)