""" 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, validate_json_request, max_json_size, ) from endpoints.api.robot_models_pre_oci import pre_oci_model as model from endpoints.exception import Unauthorized from auth.permissions import AdministerOrganizationPermission, OrganizationMemberPermission from auth.auth_context import get_authenticated_user from auth import scopes from util.names import format_robot_username from flask import abort, request CREATE_ROBOT_SCHEMA = { "type": "object", "description": "Optional data for creating a robot", "properties": { "description": { "type": "string", "description": "Optional text description for the robot", "maxLength": 255, }, "unstructured_metadata": { "type": "object", "description": "Optional unstructured metadata for the robot", }, }, } ROBOT_MAX_SIZE = 1024 * 1024 # 1 KB. def robots_list(prefix, include_permissions=False, include_token=False, limit=None): robots = model.list_entity_robot_permission_teams( prefix, limit=limit, include_token=include_token, include_permissions=include_permissions ) return {"robots": [robot.to_dict(include_token=include_token) for robot in robots]} @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 repositories and teams in which the robots have permission.", type=truthy_bool, default=False, ) @query_param( "token", "If false, the robot's token is not returned.", type=truthy_bool, default=True ) @query_param("limit", "If specified, the number of robots to return.", type=int, default=None) def get(self, parsed_args): """ List the available robots for the user. """ user = get_authenticated_user() return robots_list( user.username, include_token=parsed_args.get("token", True), include_permissions=parsed_args.get("permissions", False), limit=parsed_args.get("limit"), ) @resource("/v1/user/robots/") @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. """ schemas = { "CreateRobot": CREATE_ROBOT_SCHEMA, } @require_user_admin @nickname("getUserRobot") def get(self, robot_shortname): """ Returns the user's robot with the specified name. """ parent = get_authenticated_user() robot = model.get_user_robot(robot_shortname, parent) return robot.to_dict(include_metadata=True, include_token=True) @require_user_admin @nickname("createUserRobot") @max_json_size(ROBOT_MAX_SIZE) @validate_json_request("CreateRobot", optional=True) def put(self, robot_shortname): """ Create a new user robot with the specified name. """ parent = get_authenticated_user() create_data = request.get_json() or {} robot = model.create_user_robot( robot_shortname, parent, create_data.get("description"), create_data.get("unstructured_metadata"), ) log_action( "create_robot", parent.username, { "robot": robot_shortname, "description": create_data.get("description"), "unstructured_metadata": create_data.get("unstructured_metadata"), }, ) return robot.to_dict(include_metadata=True, include_token=True), 201 @require_user_admin @nickname("deleteUserRobot") def delete(self, robot_shortname): """ Delete an existing robot. """ parent = get_authenticated_user() model.delete_robot(format_robot_username(parent.username, robot_shortname)) log_action("delete_robot", parent.username, {"robot": robot_shortname}) return "", 204 @resource("/v1/organization//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, ) @query_param( "token", "If false, the robot's token is not returned.", type=truthy_bool, default=True ) @query_param("limit", "If specified, the number of robots to return.", type=int, default=None) def get(self, orgname, parsed_args): """ List the organization's robots. """ permission = OrganizationMemberPermission(orgname) if permission.can(): include_token = AdministerOrganizationPermission(orgname).can() and parsed_args.get( "token", True ) include_permissions = AdministerOrganizationPermission( orgname ).can() and parsed_args.get("permissions", False) return robots_list( orgname, include_permissions=include_permissions, include_token=include_token, limit=parsed_args.get("limit"), ) raise Unauthorized() @resource("/v1/organization//robots/") @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. """ schemas = { "CreateRobot": CREATE_ROBOT_SCHEMA, } @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(): robot = model.get_org_robot(robot_shortname, orgname) return robot.to_dict(include_metadata=True, include_token=True) raise Unauthorized() @require_scope(scopes.ORG_ADMIN) @nickname("createOrgRobot") @max_json_size(ROBOT_MAX_SIZE) @validate_json_request("CreateRobot", optional=True) def put(self, orgname, robot_shortname): """ Create a new robot in the organization. """ permission = AdministerOrganizationPermission(orgname) if permission.can(): create_data = request.get_json() or {} robot = model.create_org_robot( robot_shortname, orgname, create_data.get("description"), create_data.get("unstructured_metadata"), ) log_action( "create_robot", orgname, { "robot": robot_shortname, "description": create_data.get("description"), "unstructured_metadata": create_data.get("unstructured_metadata"), }, ) return robot.to_dict(include_metadata=True, include_token=True), 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.delete_robot(format_robot_username(orgname, robot_shortname)) log_action("delete_robot", orgname, {"robot": robot_shortname}) return "", 204 raise Unauthorized() @resource("/v1/user/robots//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.get_user_robot(robot_shortname, parent) permissions = model.list_robot_permissions(robot.name) return {"permissions": [permission.to_dict() for permission in permissions]} @resource("/v1/organization//robots//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(): robot = model.get_org_robot(robot_shortname, orgname) permissions = model.list_robot_permissions(robot.name) return {"permissions": [permission.to_dict() for permission in permissions]} abort(403) @resource("/v1/user/robots//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 = model.regenerate_user_robot_token(robot_shortname, parent) log_action("regenerate_robot_token", parent.username, {"robot": robot_shortname}) return robot.to_dict(include_token=True) @resource("/v1/organization//robots//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(): robot = model.regenerate_org_robot_token(robot_shortname, orgname) log_action("regenerate_robot_token", orgname, {"robot": robot_shortname}) return robot.to_dict(include_token=True) raise Unauthorized()