quay/endpoints/api/repotoken.py

import logging

from flask import request
from flask.ext.restful import abort

from endpoints.api import (resource, nickname, require_repo_admin, RepositoryParamResource,
                           log_action, validate_json_request)
from data import model


logger = logging.getLogger(__name__)


def token_view(token_obj):
  return {
    'friendlyName': token_obj.friendly_name,
    'code': token_obj.code,
    'role': token_obj.role.name,
  }


@resource('/v1/repository/<path:repository>/tokens/')
class RepositoryTokenList(RepositoryParamResource):
  """ Resource for creating and listing repository tokens. """
  schemas = {
    'NewToken': {
      'id': 'NewToken',
      'type': 'object',
      'description': 'Description of a new token.',
      'required': True,
      'properties': {
        'friendlyName': {
          'type': 'string',
          'description': 'Friendly name to help identify the token.',
          'required': True,
        },
      },
    },
  }

  @require_repo_admin
  @nickname('listRepoTokens')
  def get(self, namespace, repository):
    """ List the tokens for the specified repository. """
    tokens = model.get_repository_delegate_tokens(namespace, repository)

    return {
      'tokens': {token.code: token_view(token) for token in tokens}
    }

  @require_repo_admin
  @nickname('createToken')
  @validate_json_request('NewToken')
  def post(self, namespace, repository):
    """ Create a new repository token. """
    token_params = request.get_json()

    token = model.create_delegate_token(namespace, repository,
                                        token_params['friendlyName'])

    log_action('add_repo_accesstoken', namespace,
               {'repo': repository, 'token': token_params['friendlyName']},
               repo = model.get_repository(namespace, repository))

    return token_view(token), 201


@resource('/v1/repository/<path:repository>/tokens/<code>')
class RepositoryToken(RepositoryParamResource):
  """ Resource for managing individual tokens. """
  schemas = {
    'TokenPermission': {
      'id': 'TokenPermission',
      'type': 'object',
      'description': 'Description of a token permission.',
      'required': True,
      'properties': {
        'role': {
          'type': 'string',
          'description': 'Role to use for the token',
          'enum': [
            'read',
            'write',
            'admin',
          ],
          'required': True,
        },
      },
    },
  }
  @require_repo_admin
  @nickname('getTokens')
  def get(self, namespace, repository, code):
    """ Fetch the specified repository token information. """
    try:
      perm = model.get_repo_delegate_token(namespace, repository, code)
    except model.InvalidTokenException:
      abort(404)

    return token_view(perm)

  @require_repo_admin
  @nickname('changeToken')
  @validate_json_request('TokenPermission')
  def put(self, namespace, repository, code):
    """ Update the permissions for the specified repository token. """
    new_permission = request.get_json()

    logger.debug('Setting permission to: %s for code %s' %
                 (new_permission['role'], code))

    token = model.set_repo_delegate_token_role(namespace, repository, code,
                                               new_permission['role'])

    log_action('change_repo_permission', namespace,
               {'repo': repository, 'token': token.friendly_name, 'code': code,
                'role': new_permission['role']},
               repo = model.get_repository(namespace, repository))

    return token_view(token)

  @require_repo_admin
  @nickname('deleteToken')
  def delete(self, namespace, repository, code):
    """ Delete the repository token. """
    token = model.delete_delegate_token(namespace, repository, code)

    log_action('delete_repo_accesstoken', namespace,
               {'repo': repository, 'token': token.friendly_name,
                'code': code},
               repo = model.get_repository(namespace, repository))

    return 'Deleted', 204