import logging from flask import request from endpoints.api import (resource, nickname, require_repo_admin, RepositoryParamResource, log_action, validate_json_request, NotFound, path_param) 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//tokens/') @path_param('repository', 'The full path of the repository. e.g. namespace/name') class RepositoryTokenList(RepositoryParamResource): """ Resource for creating and listing repository tokens. """ schemas = { 'NewToken': { 'id': 'NewToken', 'type': 'object', 'description': 'Description of a new token.', 'required':[ 'friendlyName', ], 'properties': { 'friendlyName': { 'type': 'string', 'description': 'Friendly name to help identify the token', }, }, }, } @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//tokens/') @path_param('repository', 'The full path of the repository. e.g. namespace/name') @path_param('code', 'The token code') class RepositoryToken(RepositoryParamResource): """ Resource for managing individual tokens. """ schemas = { 'TokenPermission': { 'id': 'TokenPermission', 'type': 'object', 'description': 'Description of a token permission', 'required': [ 'role', ], 'properties': { 'role': { 'type': 'string', 'description': 'Role to use for the token', 'enum': [ 'read', 'write', 'admin', ], }, }, }, } @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: raise NotFound() 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