quay/endpoints/api/__init__.py

import logging
import datetime
import json

from calendar import timegm
from email.utils import formatdate
from functools import partial, wraps

from enum import Enum
from flask import Blueprint, Response, request, make_response, jsonify, session, url_for
from flask_restful import Resource, abort, Api, reqparse
from flask_restful.utils.cors import crossdomain
from jsonschema import validate, ValidationError

import features

from app import app, metric_queue
from data import model
from auth.permissions import (ReadRepositoryPermission, ModifyRepositoryPermission,
                              AdministerRepositoryPermission, UserReadPermission,
                              UserAdminPermission)
from auth import scopes
from auth.auth_context import get_authenticated_user, get_validated_oauth_token
from auth.decorators import process_oauth
from endpoints.csrf import csrf_protect
from endpoints.exception import (ApiException, Unauthorized, InvalidRequest, InvalidResponse,
                                 FreshLoginRequired, NotFound)
from endpoints.decorators import check_anon_protection
from endpoints.decorated import (handle_dme, handle_emailexception, handle_configexception,
                                 handle_too_many_login_attempts)
from util.config.provider.baseprovider import CannotWriteConfigException
from util.metrics.metricqueue import time_decorator
from util.names import parse_namespace_repository
from util.pagination import encrypt_page_token, decrypt_page_token
from util.useremails import CannotSendEmailException

logger = logging.getLogger(__name__)
api_bp = Blueprint('api', __name__)


class ApiExceptionHandlingApi(Api):
  @crossdomain(origin='*', headers=['Authorization', 'Content-Type'])
  def handle_error(self, error):
    return super(ApiExceptionHandlingApi, self).handle_error(error)


api = ApiExceptionHandlingApi()
api.init_app(api_bp)
api.decorators = [csrf_protect(),
                  crossdomain(origin='*', headers=['Authorization', 'Content-Type']),
                  process_oauth, time_decorator(api_bp.name, metric_queue)]


def resource(*urls, **kwargs):
  def wrapper(api_resource):
    if not api_resource:
      return None

    api.add_resource(api_resource, *urls, **kwargs)
    return api_resource
  return wrapper


def show_if(value):
  def f(inner):
    if not value:
      return None

    return inner
  return f


def hide_if(value):
  def f(inner):
    if value:
      return None

    return inner
  return f


def truthy_bool(param):
  return param not in {False, 'false', 'False', '0', 'FALSE', '', 'null'}


def format_date(date):
  """ Output an RFC822 date format. """
  if date is None:
    return None
  return formatdate(timegm(date.utctimetuple()))


def add_method_metadata(name, value):
  def modifier(func):
    if func is None:
      return None

    if '__api_metadata' not in dir(func):
      func.__api_metadata = {}
    func.__api_metadata[name] = value
    return func
  return modifier


def method_metadata(func, name):
  if func is None:
    return None

  if '__api_metadata' in dir(func):
    return func.__api_metadata.get(name, None)
  return None



nickname = partial(add_method_metadata, 'nickname')
related_user_resource = partial(add_method_metadata, 'related_user_resource')
internal_only = add_method_metadata('internal', True)


def path_param(name, description):
  def add_param(func):
    if not func:
      return func

    if '__api_path_params' not in dir(func):
      func.__api_path_params = {}
    func.__api_path_params[name] = {
      'name': name,
      'description': description
    }
    return func
  return add_param


def query_param(name, help_str, type=reqparse.text_type, default=None,
                choices=(), required=False):
  def add_param(func):
    if '__api_query_params' not in dir(func):
      func.__api_query_params = []
    func.__api_query_params.append({
      'name': name,
      'type': type,
      'help': help_str,
      'default': default,
      'choices': choices,
      'required': required,
      'location': ('args')
    })
    return func
  return add_param

def page_support(page_token_kwarg='page_token', parsed_args_kwarg='parsed_args'):
  def inner(func):
    """ Adds pagination support to an API endpoint. The decorated API will have an
        added query parameter named 'next_page'. Works in tandem with the
        modelutil paginate method.
    """
    @wraps(func)
    @query_param('next_page', 'The page token for the next page', type=str)
    def wrapper(self, *args, **kwargs):
      # Note: if page_token is None, we'll receive the first page of results back.
      page_token = decrypt_page_token(kwargs[parsed_args_kwarg]['next_page'])
      kwargs[page_token_kwarg] = page_token

      (result, next_page_token) = func(self, *args, **kwargs)
      if next_page_token is not None:
        result['next_page'] = encrypt_page_token(next_page_token)

      return result
    return wrapper
  return inner

def parse_args(kwarg_name='parsed_args'):
  def inner(func):
    @wraps(func)
    def wrapper(self, *args, **kwargs):
      if '__api_query_params' not in dir(func):
        abort(500)

      parser = reqparse.RequestParser()
      for arg_spec in func.__api_query_params:
        parser.add_argument(**arg_spec)
      kwargs[kwarg_name] = parser.parse_args()

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

def parse_repository_name(func):
  @wraps(func)
  def wrapper(repository, *args, **kwargs):
    (namespace, repository) = parse_namespace_repository(repository, app.config['LIBRARY_NAMESPACE'])
    return func(namespace, repository, *args, **kwargs)
  return wrapper


class ApiResource(Resource):
  method_decorators = [check_anon_protection]

  def options(self):
    return None, 200


class RepositoryParamResource(ApiResource):
  method_decorators = [check_anon_protection, parse_repository_name]


def disallow_for_app_repositories(func):
  @wraps(func)
  def wrapped(self, namespace, repository, *args, **kwargs):
    # Lookup the repository with the given namespace and name and ensure it is not an application
    # repository.
    repo = model.repository.get_repository(namespace, repository, kind_filter='application')
    if repo:
      abort(501)

    return func(self, namespace, repository, *args, **kwargs)

  return wrapped


def require_repo_permission(permission_class, scope, allow_public=False):
  def wrapper(func):
    @add_method_metadata('oauth2_scope', scope)
    @wraps(func)
    def wrapped(self, namespace, repository, *args, **kwargs):
      logger.debug('Checking permission %s for repo: %s/%s', permission_class, namespace,
                   repository)
      permission = permission_class(namespace, repository)
      if (permission.can() or
          (allow_public and
           model.repository.repository_is_public(namespace, repository))):
        return func(self, namespace, repository, *args, **kwargs)
      raise Unauthorized()
    return wrapped
  return wrapper


require_repo_read = require_repo_permission(ReadRepositoryPermission, scopes.READ_REPO, True)
require_repo_write = require_repo_permission(ModifyRepositoryPermission, scopes.WRITE_REPO)
require_repo_admin = require_repo_permission(AdministerRepositoryPermission, scopes.ADMIN_REPO)


def require_user_permission(permission_class, scope=None):
  def wrapper(func):
    @add_method_metadata('oauth2_scope', scope)
    @wraps(func)
    def wrapped(self, *args, **kwargs):
      user = get_authenticated_user()
      if not user:
        raise Unauthorized()

      logger.debug('Checking permission %s for user %s', permission_class, user.username)
      permission = permission_class(user.username)
      if permission.can():
        return func(self, *args, **kwargs)
      raise Unauthorized()
    return wrapped
  return wrapper


require_user_read = require_user_permission(UserReadPermission, scopes.READ_USER)
require_user_admin = require_user_permission(UserAdminPermission, scopes.ADMIN_USER)


def verify_not_prod(func):
  @add_method_metadata('enterprise_only', True)
  @wraps(func)
  def wrapped(*args, **kwargs):
    # Verify that we are not running on a production (i.e. hosted) stack. If so, we fail.
    # This should never happen (because of the feature-flag on SUPER_USERS), but we want to be
    # absolutely sure.
    if app.config['SERVER_HOSTNAME'].find('quay.io') >= 0:
      logger.error('!!! Super user method called IN PRODUCTION !!!')
      raise NotFound()

    return func(*args, **kwargs)

  return wrapped


def require_fresh_login(func):
  @add_method_metadata('requires_fresh_login', True)
  @wraps(func)
  def wrapped(*args, **kwargs):
    user = get_authenticated_user()
    if not user:
      raise Unauthorized()

    oauth_token = get_validated_oauth_token()
    if oauth_token:
      return func(*args, **kwargs)

    logger.debug('Checking fresh login for user %s', user.username)

    last_login = session.get('login_time', datetime.datetime.min)
    valid_span = datetime.datetime.now() - datetime.timedelta(minutes=10)

    if not user.password_hash or last_login >= valid_span:
      return func(*args, **kwargs)

    raise FreshLoginRequired()
  return wrapped


def require_scope(scope_object):
  def wrapper(func):
    @add_method_metadata('oauth2_scope', scope_object)
    @wraps(func)
    def wrapped(*args, **kwargs):
      return func(*args, **kwargs)
    return wrapped
  return wrapper


def validate_json_request(schema_name):
  def wrapper(func):
    @add_method_metadata('request_schema', schema_name)
    @wraps(func)
    def wrapped(self, *args, **kwargs):
      schema = self.schemas[schema_name]
      try:
        json_data = request.get_json()
        if json_data is None:
          raise InvalidRequest('Missing JSON body')

        validate(json_data, schema)
        return func(self, *args, **kwargs)
      except ValidationError as ex:
        raise InvalidRequest(ex.message)
    return wrapped
  return wrapper


def request_error(exception=None, **kwargs):
  data = kwargs.copy()
  message = 'Request error.'
  if exception:
    message = exception.message
  message = data.pop('message', message)
  raise InvalidRequest(message, data)


def log_action(kind, user_or_orgname, metadata=None, repo=None):
  if not metadata:
    metadata = {}

  oauth_token = get_validated_oauth_token()
  if oauth_token:
    metadata['oauth_token_id'] = oauth_token.id
    metadata['oauth_token_application_id'] = oauth_token.application.client_id
    metadata['oauth_token_application'] = oauth_token.application.name

  performer = get_authenticated_user()
  model.log.log_action(kind, user_or_orgname, performer=performer, ip=request.remote_addr,
                       metadata=metadata, repository=repo)


def define_json_response(schema_name):
  def wrapper(func):
    @add_method_metadata('response_schema', schema_name)
    @wraps(func)
    def wrapped(self, *args, **kwargs):
      schema = self.schemas[schema_name]
      resp = func(self, *args, **kwargs)

      if app.config['TESTING']:
        try:
          validate(resp, schema)
        except ValidationError as ex:
          raise InvalidResponse(ex.message)

      return resp
    return wrapped
  return wrapper


def disallow_under_trust(func):
  """ Disallows the decorated operation for repository when it has trust enabled.
  """
  @wraps(func)
  def wrapper(self, *args, **kwargs):
    if features.SIGNING:
      namespace = args[0]
      repository = args[1]

      repo = model.repository.get_repository(namespace, repository)
      if repo is not None and repo.trust_enabled:
        raise InvalidRequest('Cannot call this method on a repostory with trust enabled')

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


import endpoints.api.billing
import endpoints.api.build
import endpoints.api.discovery
import endpoints.api.error
import endpoints.api.globalmessages
import endpoints.api.image
import endpoints.api.logs
import endpoints.api.manifest
import endpoints.api.organization
import endpoints.api.permission
import endpoints.api.prototype
import endpoints.api.repository
import endpoints.api.repositorynotification
import endpoints.api.repoemail
import endpoints.api.repotoken
import endpoints.api.robot
import endpoints.api.search
import endpoints.api.suconfig
import endpoints.api.superuser
import endpoints.api.tag
import endpoints.api.team
import endpoints.api.trigger
import endpoints.api.user
import endpoints.api.secscan
import endpoints.api.signing