This repository has been archived on 2020-03-24. You can view files and clone it, but cannot push or open issues or pull requests.
quay/endpoints/api/robot.py
EvB 43aed7c6f4 fix(endpoints/api): return empty 204 resp
Return an empty body on API requests with status code 204, which
means "No content". Incorrect 'Deleted' responses were being
returned after successful DELETE operations despite the "No Content"
definition of 204.
2016-12-14 16:22:39 -05:00

261 lines
9.8 KiB
Python

""" Manage user and organization robot accounts. """
from endpoints.api import (resource, nickname, ApiResource, log_action, related_user_resource,
require_user_admin, require_scope, path_param, parse_args,
truthy_bool, query_param)
from endpoints.exception import Unauthorized
from auth.permissions import AdministerOrganizationPermission, OrganizationMemberPermission
from auth.auth_context import get_authenticated_user
from auth import scopes
from data import model
from data.database import User, Team, Repository, FederatedLogin
from util.names import format_robot_username
from flask import abort
from app import avatar
def robot_view(name, token):
return {
'name': name,
'token': token
}
def permission_view(permission):
return {
'repository': {
'name': permission.repository.name,
'is_public': permission.repository.visibility.name == 'public'
},
'role': permission.role.name
}
def robots_list(prefix, include_permissions=False):
tuples = model.user.list_entity_robot_permission_teams(prefix,
include_permissions=include_permissions)
robots = {}
robot_teams = set()
for robot_tuple in tuples:
robot_name = robot_tuple.get(User.username)
if not robot_name in robots:
robots[robot_name] = {
'name': robot_name,
'token': robot_tuple.get(FederatedLogin.service_ident)
}
if include_permissions:
robots[robot_name].update({
'teams': [],
'repositories': []
})
if include_permissions:
team_name = robot_tuple.get(Team.name)
repository_name = robot_tuple.get(Repository.name)
if team_name is not None:
check_key = robot_name + ':' + team_name
if not check_key in robot_teams:
robot_teams.add(check_key)
robots[robot_name]['teams'].append({
'name': team_name,
'avatar': avatar.get_data(team_name, team_name, 'team')
})
if repository_name is not None:
if not repository_name in robots[robot_name]['repositories']:
robots[robot_name]['repositories'].append(repository_name)
return {'robots': robots.values()}
@resource('/v1/user/robots')
class UserRobotList(ApiResource):
""" Resource for listing user robots. """
@require_user_admin
@nickname('getUserRobots')
@parse_args()
@query_param('permissions',
'Whether to include repostories and teams in which the robots have permission.',
type=truthy_bool, default=False)
def get(self, parsed_args):
""" List the available robots for the user. """
user = get_authenticated_user()
return robots_list(user.username, include_permissions=parsed_args.get('permissions', False))
@resource('/v1/user/robots/<robot_shortname>')
@path_param('robot_shortname',
'The short name for the robot, without any user or organization prefix')
class UserRobot(ApiResource):
""" Resource for managing a user's robots. """
@require_user_admin
@nickname('getUserRobot')
def get(self, robot_shortname):
""" Returns the user's robot with the specified name. """
parent = get_authenticated_user()
robot, password = model.user.get_robot(robot_shortname, parent)
return robot_view(robot.username, password)
@require_user_admin
@nickname('createUserRobot')
def put(self, robot_shortname):
""" Create a new user robot with the specified name. """
parent = get_authenticated_user()
robot, password = model.user.create_robot(robot_shortname, parent)
log_action('create_robot', parent.username, {'robot': robot_shortname})
return robot_view(robot.username, password), 201
@require_user_admin
@nickname('deleteUserRobot')
def delete(self, robot_shortname):
""" Delete an existing robot. """
parent = get_authenticated_user()
model.user.delete_robot(format_robot_username(parent.username, robot_shortname))
log_action('delete_robot', parent.username, {'robot': robot_shortname})
return '', 204
@resource('/v1/organization/<orgname>/robots')
@path_param('orgname', 'The name of the organization')
@related_user_resource(UserRobotList)
class OrgRobotList(ApiResource):
""" Resource for listing an organization's robots. """
@require_scope(scopes.ORG_ADMIN)
@nickname('getOrgRobots')
@parse_args()
@query_param('permissions',
'Whether to include repostories and teams in which the robots have permission.',
type=truthy_bool, default=False)
def get(self, orgname, parsed_args):
""" List the organization's robots. """
permission = OrganizationMemberPermission(orgname)
if permission.can():
return robots_list(orgname, include_permissions=parsed_args.get('permissions', False))
raise Unauthorized()
@resource('/v1/organization/<orgname>/robots/<robot_shortname>')
@path_param('orgname', 'The name of the organization')
@path_param('robot_shortname',
'The short name for the robot, without any user or organization prefix')
@related_user_resource(UserRobot)
class OrgRobot(ApiResource):
""" Resource for managing an organization's robots. """
@require_scope(scopes.ORG_ADMIN)
@nickname('getOrgRobot')
def get(self, orgname, robot_shortname):
""" Returns the organization's robot with the specified name. """
permission = AdministerOrganizationPermission(orgname)
if permission.can():
parent = model.organization.get_organization(orgname)
robot, password = model.user.get_robot(robot_shortname, parent)
return robot_view(robot.username, password)
raise Unauthorized()
@require_scope(scopes.ORG_ADMIN)
@nickname('createOrgRobot')
def put(self, orgname, robot_shortname):
""" Create a new robot in the organization. """
permission = AdministerOrganizationPermission(orgname)
if permission.can():
parent = model.organization.get_organization(orgname)
robot, password = model.user.create_robot(robot_shortname, parent)
log_action('create_robot', orgname, {'robot': robot_shortname})
return robot_view(robot.username, password), 201
raise Unauthorized()
@require_scope(scopes.ORG_ADMIN)
@nickname('deleteOrgRobot')
def delete(self, orgname, robot_shortname):
""" Delete an existing organization robot. """
permission = AdministerOrganizationPermission(orgname)
if permission.can():
model.user.delete_robot(format_robot_username(orgname, robot_shortname))
log_action('delete_robot', orgname, {'robot': robot_shortname})
return '', 204
raise Unauthorized()
@resource('/v1/user/robots/<robot_shortname>/permissions')
@path_param('robot_shortname',
'The short name for the robot, without any user or organization prefix')
class UserRobotPermissions(ApiResource):
""" Resource for listing the permissions a user's robot has in the system. """
@require_user_admin
@nickname('getUserRobotPermissions')
def get(self, robot_shortname):
""" Returns the list of repository permissions for the user's robot. """
parent = get_authenticated_user()
robot, _ = model.user.get_robot(robot_shortname, parent)
permissions = model.permission.list_robot_permissions(robot.username)
return {
'permissions': [permission_view(permission) for permission in permissions]
}
@resource('/v1/organization/<orgname>/robots/<robot_shortname>/permissions')
@path_param('orgname', 'The name of the organization')
@path_param('robot_shortname',
'The short name for the robot, without any user or organization prefix')
@related_user_resource(UserRobotPermissions)
class OrgRobotPermissions(ApiResource):
""" Resource for listing the permissions an org's robot has in the system. """
@require_user_admin
@nickname('getOrgRobotPermissions')
def get(self, orgname, robot_shortname):
""" Returns the list of repository permissions for the org's robot. """
permission = AdministerOrganizationPermission(orgname)
if permission.can():
parent = model.organization.get_organization(orgname)
robot, _ = model.user.get_robot(robot_shortname, parent)
permissions = model.permission.list_robot_permissions(robot.username)
return {
'permissions': [permission_view(permission) for permission in permissions]
}
abort(403)
@resource('/v1/user/robots/<robot_shortname>/regenerate')
@path_param('robot_shortname',
'The short name for the robot, without any user or organization prefix')
class RegenerateUserRobot(ApiResource):
""" Resource for regenerate an organization's robot's token. """
@require_user_admin
@nickname('regenerateUserRobotToken')
def post(self, robot_shortname):
""" Regenerates the token for a user's robot. """
parent = get_authenticated_user()
robot, password = model.user.regenerate_robot_token(robot_shortname, parent)
log_action('regenerate_robot_token', parent.username, {'robot': robot_shortname})
return robot_view(robot.username, password)
@resource('/v1/organization/<orgname>/robots/<robot_shortname>/regenerate')
@path_param('orgname', 'The name of the organization')
@path_param('robot_shortname',
'The short name for the robot, without any user or organization prefix')
@related_user_resource(RegenerateUserRobot)
class RegenerateOrgRobot(ApiResource):
""" Resource for regenerate an organization's robot's token. """
@require_scope(scopes.ORG_ADMIN)
@nickname('regenerateOrgRobotToken')
def post(self, orgname, robot_shortname):
""" Regenerates the token for an organization robot. """
permission = AdministerOrganizationPermission(orgname)
if permission.can():
parent = model.organization.get_organization(orgname)
robot, password = model.user.regenerate_robot_token(robot_shortname, parent)
log_action('regenerate_robot_token', orgname, {'robot': robot_shortname})
return robot_view(robot.username, password)
raise Unauthorized()