quay/auth/decorators.py

import logging

from functools import wraps
from uuid import UUID
from datetime import datetime
from base64 import b64decode

from flask import request, session
from flask.sessions import SecureCookieSessionInterface, BadSignature
from flask_login import current_user
from flask_principal import identity_changed, Identity

import scopes

from app import app, authentication, metric_queue
from auth_context import (set_authenticated_user, set_validated_token, set_grant_context,
                          set_validated_oauth_token)
from data import model
from endpoints.exception import InvalidToken, ExpiredToken
from permissions import QuayDeferredPermissionUser
from util.http import abort


logger = logging.getLogger(__name__)

SIGNATURE_PREFIX = 'sigv2='

def _load_user_from_cookie():
  if not current_user.is_anonymous:
    try:
      # Attempt to parse the user uuid to make sure the cookie has the right value type
      UUID(current_user.get_id())
    except ValueError:
      return None

    logger.debug('Loading user from cookie: %s', current_user.get_id())
    db_user = current_user.db_user()
    if db_user is not None:
      # Don't allow disabled users to login.
      if not db_user.enabled:
        return None

      set_authenticated_user(db_user)
      loaded = QuayDeferredPermissionUser.for_user(db_user)
      identity_changed.send(app, identity=loaded)
      return db_user

  return None


def _validate_and_apply_oauth_token(token):
  validated = model.oauth.validate_access_token(token)
  if not validated:
    logger.warning('OAuth access token could not be validated: %s', token)
    metric_queue.authentication_count.Inc(labelvalues=['oauth', False])
    raise InvalidToken('OAuth access token could not be validated: {token}'.format(token=token))
  elif validated.expires_at <= datetime.utcnow():
    logger.info('OAuth access with an expired token: %s', token)
    metric_queue.authentication_count.Inc(labelvalues=['oauth', False])
    raise ExpiredToken('OAuth access token has expired: {token}'.format(token=token))

  # Don't allow disabled users to login.
  if not validated.authorized_user.enabled:
    metric_queue.authentication_count.Inc(labelvalues=['oauth', False])
    return False

  # We have a valid token
  scope_set = scopes.scopes_from_scope_string(validated.scope)
  logger.debug('Successfully validated oauth access token: %s with scope: %s', token,
               scope_set)

  set_authenticated_user(validated.authorized_user)
  set_validated_oauth_token(validated)

  new_identity = QuayDeferredPermissionUser.for_user(validated.authorized_user, scope_set)
  identity_changed.send(app, identity=new_identity)
  metric_queue.authentication_count.Inc(labelvalues=['oauth', True])
  return True


def _parse_basic_auth_header(auth):
  normalized = [part.strip() for part in auth.split(' ') if part]
  if normalized[0].lower() != 'basic' or len(normalized) != 2:
    logger.debug('Invalid basic auth format.')
    return None

  logger.debug('Found basic auth header: %s', auth)
  try:
    credentials = [part.decode('utf-8') for part in b64decode(normalized[1]).split(':', 1)]
  except TypeError:
    logger.exception('Exception when parsing basic auth header')
    return None

  if len(credentials) != 2:
    logger.debug('Invalid basic auth credential format.')
    return None

  return credentials


def _process_basic_auth(auth):
  credentials = _parse_basic_auth_header(auth)
  if credentials is None:
    return

  if credentials[0] == '$token':
    # Use as token auth
    try:
      token = model.token.load_token_data(credentials[1])
      logger.debug('Successfully validated token: %s', credentials[1])
      set_validated_token(token)
      identity_changed.send(app, identity=Identity(token.code, 'token'))
      metric_queue.authentication_count.Inc(labelvalues=['token', True])
      return True

    except model.DataModelException:
      logger.debug('Invalid token: %s', credentials[1])
      metric_queue.authentication_count.Inc(labelvalues=['token', False])

  elif credentials[0] == '$oauthtoken':
    oauth_token = credentials[1]
    return _validate_and_apply_oauth_token(oauth_token)

  elif '+' in credentials[0]:
    logger.debug('Trying robot auth with credentials %s', str(credentials))
    # Use as robot auth
    try:
      robot = model.user.verify_robot(credentials[0], credentials[1])
      logger.debug('Successfully validated robot: %s', credentials[0])
      set_authenticated_user(robot)

      deferred_robot = QuayDeferredPermissionUser.for_user(robot)
      identity_changed.send(app, identity=deferred_robot)
      metric_queue.authentication_count.Inc(labelvalues=['robot', True])
      return True
    except model.InvalidRobotException:
      logger.debug('Invalid robot or password for robot: %s', credentials[0])
      metric_queue.authentication_count.Inc(labelvalues=['robot', False])

  else:
    (authenticated, _) = authentication.verify_and_link_user(credentials[0], credentials[1],
                                                             basic_auth=True)
    if authenticated:
      logger.debug('Successfully validated user: %s', authenticated.username)
      set_authenticated_user(authenticated)

      new_identity = QuayDeferredPermissionUser.for_user(authenticated)
      identity_changed.send(app, identity=new_identity)
      metric_queue.authentication_count.Inc(labelvalues=['user', True])
      return True
    else:
      metric_queue.authentication_count.Inc(labelvalues=['user', False])

  # We weren't able to authenticate via basic auth.
  logger.debug('Basic auth present but could not be validated.')
  return False


def has_basic_auth(username):
  auth = request.headers.get('authorization', '')
  if not auth:
    return False

  credentials = _parse_basic_auth_header(auth)
  if not credentials:
    return False

  (authenticated, _) = authentication.verify_and_link_user(credentials[0], credentials[1],
                                                           basic_auth=True)
  if not authenticated:
    return False

  return authenticated.username == username


def generate_signed_token(grants, user_context):
  ser = SecureCookieSessionInterface().get_signing_serializer(app)
  data_to_sign = {
    'grants': grants,
    'user_context': user_context,
  }

  encrypted = ser.dumps(data_to_sign)
  return '{0}{1}'.format(SIGNATURE_PREFIX, encrypted)


def _process_signed_grant(auth):
  normalized = [part.strip() for part in auth.split(' ') if part]
  if normalized[0].lower() != 'token' or len(normalized) != 2:
    logger.debug('Not a token: %s', auth)
    return False

  if not normalized[1].startswith(SIGNATURE_PREFIX):
    logger.debug('Not a signed grant token: %s', auth)
    return False

  encrypted = normalized[1][len(SIGNATURE_PREFIX):]
  ser = SecureCookieSessionInterface().get_signing_serializer(app)

  try:
    token_data = ser.loads(encrypted, max_age=app.config['SIGNED_GRANT_EXPIRATION_SEC'])
  except BadSignature:
    logger.warning('Signed grant could not be validated: %s', encrypted)
    metric_queue.authentication_count.Inc(labelvalues=['signed', False])
    abort(401, message='Signed grant could not be validated: %(auth)s', issue='invalid-auth-token',
          auth=auth)

  logger.debug('Successfully validated signed grant with data: %s', token_data)

  loaded_identity = Identity(None, 'signed_grant')

  if token_data['user_context']:
    set_grant_context({
      'user': token_data['user_context'],
      'kind': 'user',
    })

  loaded_identity.provides.update(token_data['grants'])
  identity_changed.send(app, identity=loaded_identity)
  metric_queue.authentication_count.Inc(labelvalues=['signed', True])
  return True


def process_oauth(func):
  @wraps(func)
  def wrapper(*args, **kwargs):
    auth = request.headers.get('authorization', '')
    if auth:
      normalized = [part.strip() for part in auth.split(' ') if part]
      if normalized[0].lower() != 'bearer' or len(normalized) != 2:
        logger.debug('Invalid oauth bearer token format.')
        return func(*args, **kwargs)

      token = normalized[1]
      _validate_and_apply_oauth_token(token)
    elif _load_user_from_cookie() is None:
      logger.debug('No auth header or login cookie.')
    return func(*args, **kwargs)
  return wrapper


def process_auth(func):
  @wraps(func)
  def wrapper(*args, **kwargs):
    auth = request.headers.get('authorization', '')

    if auth:
      logger.debug('Validating auth header: %s', auth)
      _process_signed_grant(auth)
      _process_basic_auth(auth)
    else:
      logger.debug('No auth header.')

    return func(*args, **kwargs)
  return wrapper


def process_auth_or_cookie(func):
  @wraps(func)
  def wrapper(*args, **kwargs):
    auth = request.headers.get('authorization', '')

    if auth:
      logger.debug('Validating auth header: %s', auth)
      _process_basic_auth(auth)
    else:
      logger.debug('No auth header.')
      _load_user_from_cookie()

    return func(*args, **kwargs)
  return wrapper


def require_session_login(func):
  @wraps(func)
  def wrapper(*args, **kwargs):
    loaded = _load_user_from_cookie()
    if loaded is None or loaded.organization:
      abort(401, message='Method requires login and no valid login could be loaded.')
    return func(*args, **kwargs)
  return wrapper


def extract_namespace_repo_from_session(func):
  @wraps(func)
  def wrapper(*args, **kwargs):
    if 'namespace' not in session or 'repository' not in session:
      logger.error('Unable to load namespace or repository from session: %s', session)
      abort(400, message='Missing namespace in request')

    return func(session['namespace'], session['repository'], *args, **kwargs)
  return wrapper