quay/endpoints/v2/v2auth.py

import logging
import re

from cachetools import lru_cache
from flask import request, jsonify, abort

from app import app, userevents, instance_keys
from auth.auth_context import get_authenticated_user, get_validated_token, get_validated_oauth_token
from auth.decorators import process_basic_auth
from auth.permissions import (ModifyRepositoryPermission, ReadRepositoryPermission,
                              CreateRepositoryPermission, AdministerRepositoryPermission)
from endpoints.decorators import anon_protect
from endpoints.v2 import v2_bp
from endpoints.v2.errors import InvalidLogin, NameInvalid, InvalidRequest, Unsupported, Unauthorized
from data.interfaces.v2 import pre_oci_model as model
from util.cache import no_cache
from util.names import parse_namespace_repository, REPOSITORY_NAME_REGEX
from util.security.registry_jwt import generate_bearer_token, build_context_and_subject, QUAY_TUF_ROOT, SIGNER_TUF_ROOT

logger = logging.getLogger(__name__)


TOKEN_VALIDITY_LIFETIME_S = 60 * 60  # 1 hour
SCOPE_REGEX_TEMPLATE = r'^repository:((?:{}\/)?((?:[\.a-zA-Z0-9_\-]+\/)*[\.a-zA-Z0-9_\-]+)):((?:push|pull|\*)(?:,(?:push|pull|\*))*)$'

@lru_cache(maxsize=1)
def get_scope_regex():
  hostname = re.escape(app.config['SERVER_HOSTNAME'])
  scope_regex_string = SCOPE_REGEX_TEMPLATE.format(hostname)
  return re.compile(scope_regex_string)


@v2_bp.route('/auth')
@process_basic_auth
@no_cache
@anon_protect
def generate_registry_jwt(auth_result):
  """
  This endpoint will generate a JWT conforming to the Docker Registry v2 Auth Spec:
  https://docs.docker.com/registry/spec/auth/token/
  """
  audience_param = request.args.get('service')
  logger.debug('Request audience: %s', audience_param)

  scope_param = request.args.get('scope') or ''
  logger.debug('Scope request: %s', scope_param)

  user = get_authenticated_user()
  logger.debug('Authenticated user: %s', user)

  token = get_validated_token()
  logger.debug('Authenticated token: %s', token)

  oauthtoken = get_validated_oauth_token()
  logger.debug('Authenticated OAuth token: %s', oauthtoken)

  auth_header = request.headers.get('authorization', '')
  auth_credentials_sent = bool(auth_header)
  if auth_credentials_sent and not user and not token:
    # The auth credentials sent for the user are invalid.
    raise InvalidLogin(auth_result.error_message)

  access = []
  user_event_data = {
    'action': 'login',
  }
  tuf_root = QUAY_TUF_ROOT

  if len(scope_param) > 0:
    match = get_scope_regex().match(scope_param)
    if match is None:
      logger.debug('Match: %s', match)
      logger.debug('len: %s', len(scope_param))
      logger.warning('Unable to decode repository and actions: %s', scope_param)
      raise InvalidRequest('Unable to decode repository and actions: %s' % scope_param)

    logger.debug('Match: %s', match.groups())

    registry_and_repo = match.group(1)
    namespace_and_repo = match.group(2)
    actions = match.group(3).split(',')

    lib_namespace = app.config['LIBRARY_NAMESPACE']
    namespace, reponame = parse_namespace_repository(namespace_and_repo, lib_namespace)

    # Ensure that we are never creating an invalid repository.
    if not REPOSITORY_NAME_REGEX.match(reponame):
      logger.debug('Found invalid repository name in auth flow: %s', reponame)
      if len(namespace_and_repo.split('/')) > 1:
        msg = 'Nested repositories are not supported. Found: %s' % namespace_and_repo
        raise NameInvalid(message=msg)

      raise NameInvalid(message='Invalid repository name: %s' % namespace_and_repo)

    final_actions = []

    repo = model.get_repository(namespace, reponame)

    repo_is_public = repo is not None and repo.is_public
    invalid_repo_message = ''
    if repo is not None and repo.kind != 'image':
      invalid_repo_message = (('This repository is for managing %s resources ' +
                               'and not container images.') % repo.kind)

    if 'push' in actions:
      # If there is no valid user or token, then the repository cannot be
      # accessed.
      if user is not None or token is not None:
        # Lookup the repository. If it exists, make sure the entity has modify
        # permission. Otherwise, make sure the entity has create permission.
        if repo:
          if ModifyRepositoryPermission(namespace, reponame).can():
            if repo.kind != 'image':
              raise Unsupported(message=invalid_repo_message)

            final_actions.append('push')
          else:
            logger.debug('No permission to modify repository %s/%s', namespace, reponame)
        else:
          if CreateRepositoryPermission(namespace).can() and user is not None:
            logger.debug('Creating repository: %s/%s', namespace, reponame)
            model.create_repository(namespace, reponame, user)
            final_actions.append('push')
          else:
            logger.debug('No permission to create repository %s/%s', namespace, reponame)

    if 'pull' in actions:
      # Grant pull if the user can read the repo or it is public.
      if ReadRepositoryPermission(namespace, reponame).can() or repo_is_public:
        if repo is not None and repo.kind != 'image':
          raise Unsupported(message=invalid_repo_message)

        final_actions.append('pull')
      else:
        logger.debug('No permission to pull repository %s/%s', namespace, reponame)

    if '*' in actions:
      # Grant * user is admin
      if AdministerRepositoryPermission(namespace, reponame).can():
        if repo is not None and repo.kind != 'image':
          raise Unsupported(message=invalid_repo_message)

        final_actions.append('*')
      else:
        logger.debug("No permission to administer repository %s/%s", namespace, reponame)

    # Add the access for the JWT.
    access.append({
      'type': 'repository',
      'name': registry_and_repo,
      'actions': final_actions,
    })

    # Set the user event data for the auth.
    if 'push' in final_actions:
      user_action = 'push_start'
    elif 'pull' in final_actions:
      user_action = 'pull_start'
    else:
      user_action = 'login'

    user_event_data = {
      'action': user_action,
      'repository': reponame,
      'namespace': namespace,
    }

    tuf_root = get_tuf_root(namespace, reponame)

  elif user is None and token is None:
    # In this case, we are doing an auth flow, and it's not an anonymous pull
    logger.debug('No user and no token sent for empty scope list')
    raise Unauthorized()

  # Send the user event.
  if user is not None:
    event = userevents.get_event(user.username)
    event.publish_event_data('docker-cli', user_event_data)

  # Build the signed JWT.
  context, subject = build_context_and_subject(user=user, token=token, oauthtoken=oauthtoken, tuf_root=tuf_root)
  token = generate_bearer_token(audience_param, subject, context, access,
                                TOKEN_VALIDITY_LIFETIME_S, instance_keys)
  return jsonify({'token': token})


def get_tuf_root(namespace, reponame):
  # Users with write access to a repo will see signer-rooted TUF metadata
  if ModifyRepositoryPermission(namespace, reponame).can():
    return SIGNER_TUF_ROOT
  return QUAY_TUF_ROOT