43aed7c6f4
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.
281 lines
9.8 KiB
Python
281 lines
9.8 KiB
Python
""" Manage repository permissions. """
|
|
|
|
import logging
|
|
|
|
from flask import request
|
|
|
|
from app import avatar
|
|
from endpoints.api import (resource, nickname, require_repo_admin, RepositoryParamResource,
|
|
log_action, request_error, validate_json_request, path_param)
|
|
from endpoints.exception import NotFound
|
|
from data import model
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def role_view(repo_perm_obj):
|
|
return {
|
|
'role': repo_perm_obj.role.name,
|
|
}
|
|
|
|
def wrap_role_view_user(role_json, user):
|
|
role_json['name'] = user.username
|
|
role_json['is_robot'] = user.robot
|
|
if not user.robot:
|
|
role_json['avatar'] = avatar.get_data_for_user(user)
|
|
return role_json
|
|
|
|
|
|
def wrap_role_view_org(role_json, user, org_members):
|
|
role_json['is_org_member'] = user.robot or user.username in org_members
|
|
return role_json
|
|
|
|
|
|
def wrap_role_view_team(role_json, team):
|
|
role_json['name'] = team.name
|
|
role_json['avatar'] = avatar.get_data_for_team(team)
|
|
return role_json
|
|
|
|
|
|
@resource('/v1/repository/<apirepopath:repository>/permissions/team/')
|
|
@path_param('repository', 'The full path of the repository. e.g. namespace/name')
|
|
class RepositoryTeamPermissionList(RepositoryParamResource):
|
|
""" Resource for repository team permissions. """
|
|
@require_repo_admin
|
|
@nickname('listRepoTeamPermissions')
|
|
def get(self, namespace, repository):
|
|
""" List all team permission. """
|
|
repo_perms = model.permission.get_all_repo_teams(namespace, repository)
|
|
|
|
def wrapped_role_view(repo_perm):
|
|
return wrap_role_view_team(role_view(repo_perm), repo_perm.team)
|
|
|
|
return {
|
|
'permissions': {repo_perm.team.name: wrapped_role_view(repo_perm)
|
|
for repo_perm in repo_perms}
|
|
}
|
|
|
|
|
|
@resource('/v1/repository/<apirepopath:repository>/permissions/user/')
|
|
@path_param('repository', 'The full path of the repository. e.g. namespace/name')
|
|
class RepositoryUserPermissionList(RepositoryParamResource):
|
|
""" Resource for repository user permissions. """
|
|
@require_repo_admin
|
|
@nickname('listRepoUserPermissions')
|
|
def get(self, namespace, repository):
|
|
""" List all user permissions. """
|
|
# Lookup the organization (if any).
|
|
org = None
|
|
try:
|
|
org = model.organization.get_organization(namespace) # Will raise an error if not org
|
|
except model.InvalidOrganizationException:
|
|
# This repository isn't under an org
|
|
pass
|
|
|
|
# Determine how to wrap the role(s).
|
|
def wrapped_role_view(repo_perm):
|
|
return wrap_role_view_user(role_view(repo_perm), repo_perm.user)
|
|
|
|
role_view_func = wrapped_role_view
|
|
|
|
if org:
|
|
org_members = model.organization.get_organization_member_set(namespace)
|
|
current_func = role_view_func
|
|
|
|
def wrapped_role_org_view(repo_perm):
|
|
return wrap_role_view_org(current_func(repo_perm), repo_perm.user,
|
|
org_members)
|
|
|
|
role_view_func = wrapped_role_org_view
|
|
|
|
# Load and return the permissions.
|
|
repo_perms = model.user.get_all_repo_users(namespace, repository)
|
|
return {
|
|
'permissions': {perm.user.username: role_view_func(perm)
|
|
for perm in repo_perms}
|
|
}
|
|
|
|
|
|
@resource('/v1/repository/<apirepopath:repository>/permissions/user/<username>/transitive')
|
|
@path_param('repository', 'The full path of the repository. e.g. namespace/name')
|
|
@path_param('username', 'The username of the user to which the permissions apply')
|
|
class RepositoryUserTransitivePermission(RepositoryParamResource):
|
|
""" Resource for retrieving whether a user has access to a repository, either directly
|
|
or via a team. """
|
|
@require_repo_admin
|
|
@nickname('getUserTransitivePermission')
|
|
def get(self, namespace, repository, username):
|
|
""" Get the fetch the permission for the specified user. """
|
|
user = model.user.get_user(username)
|
|
if not user:
|
|
raise NotFound
|
|
|
|
repo = model.repository.get_repository(namespace, repository)
|
|
if not repo:
|
|
raise NotFound
|
|
|
|
permissions = list(model.permission.get_user_repo_permissions(user, repo))
|
|
return {
|
|
'permissions': [role_view(permission) for permission in permissions]
|
|
}
|
|
|
|
|
|
@resource('/v1/repository/<apirepopath:repository>/permissions/user/<username>')
|
|
@path_param('repository', 'The full path of the repository. e.g. namespace/name')
|
|
@path_param('username', 'The username of the user to which the permission applies')
|
|
class RepositoryUserPermission(RepositoryParamResource):
|
|
""" Resource for managing individual user permissions. """
|
|
schemas = {
|
|
'UserPermission': {
|
|
'type': 'object',
|
|
'description': 'Description of a user permission.',
|
|
'required': [
|
|
'role',
|
|
],
|
|
'properties': {
|
|
'role': {
|
|
'type': 'string',
|
|
'description': 'Role to use for the user',
|
|
'enum': [
|
|
'read',
|
|
'write',
|
|
'admin',
|
|
],
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
@require_repo_admin
|
|
@nickname('getUserPermissions')
|
|
def get(self, namespace, repository, username):
|
|
""" Get the Fetch the permission for the specified user. """
|
|
logger.debug('Get repo: %s/%s permissions for user %s', namespace, repository, username)
|
|
perm = model.permission.get_user_reponame_permission(username, namespace, repository)
|
|
perm_view = wrap_role_view_user(role_view(perm), perm.user)
|
|
|
|
try:
|
|
model.organization.get_organization(namespace)
|
|
org_members = model.organization.get_organization_member_set(namespace)
|
|
perm_view = wrap_role_view_org(perm_view, perm.user, org_members)
|
|
except model.InvalidOrganizationException:
|
|
# This repository is not part of an organization
|
|
pass
|
|
|
|
return perm_view
|
|
|
|
@require_repo_admin
|
|
@nickname('changeUserPermissions')
|
|
@validate_json_request('UserPermission')
|
|
def put(self, namespace, repository, username): # Also needs to respond to post
|
|
""" Update the perimssions for an existing repository. """
|
|
new_permission = request.get_json()
|
|
|
|
logger.debug('Setting permission to: %s for user %s', new_permission['role'], username)
|
|
|
|
try:
|
|
perm = model.permission.set_user_repo_permission(username, namespace, repository,
|
|
new_permission['role'])
|
|
except model.DataModelException as ex:
|
|
raise request_error(exception=ex)
|
|
|
|
perm_view = wrap_role_view_user(role_view(perm), perm.user)
|
|
|
|
try:
|
|
model.organization.get_organization(namespace)
|
|
org_members = model.organization.get_organization_member_set(namespace)
|
|
perm_view = wrap_role_view_org(perm_view, perm.user, org_members)
|
|
except model.InvalidOrganizationException:
|
|
# This repository is not part of an organization
|
|
pass
|
|
except model.DataModelException as ex:
|
|
raise request_error(exception=ex)
|
|
|
|
log_action('change_repo_permission', namespace,
|
|
{'username': username, 'repo': repository,
|
|
'role': new_permission['role']},
|
|
repo=model.repository.get_repository(namespace, repository))
|
|
|
|
return perm_view, 200
|
|
|
|
@require_repo_admin
|
|
@nickname('deleteUserPermissions')
|
|
def delete(self, namespace, repository, username):
|
|
""" Delete the permission for the user. """
|
|
try:
|
|
model.permission.delete_user_permission(username, namespace, repository)
|
|
except model.DataModelException as ex:
|
|
raise request_error(exception=ex)
|
|
|
|
log_action('delete_repo_permission', namespace,
|
|
{'username': username, 'repo': repository},
|
|
repo=model.repository.get_repository(namespace, repository))
|
|
|
|
return '', 204
|
|
|
|
|
|
@resource('/v1/repository/<apirepopath:repository>/permissions/team/<teamname>')
|
|
@path_param('repository', 'The full path of the repository. e.g. namespace/name')
|
|
@path_param('teamname', 'The name of the team to which the permission applies')
|
|
class RepositoryTeamPermission(RepositoryParamResource):
|
|
""" Resource for managing individual team permissions. """
|
|
schemas = {
|
|
'TeamPermission': {
|
|
'type': 'object',
|
|
'description': 'Description of a team permission.',
|
|
'required': [
|
|
'role',
|
|
],
|
|
'properties': {
|
|
'role': {
|
|
'type': 'string',
|
|
'description': 'Role to use for the team',
|
|
'enum': [
|
|
'read',
|
|
'write',
|
|
'admin',
|
|
],
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
@require_repo_admin
|
|
@nickname('getTeamPermissions')
|
|
def get(self, namespace, repository, teamname):
|
|
""" Fetch the permission for the specified team. """
|
|
logger.debug('Get repo: %s/%s permissions for team %s', namespace, repository, teamname)
|
|
perm = model.permission.get_team_reponame_permission(teamname, namespace, repository)
|
|
return role_view(perm)
|
|
|
|
@require_repo_admin
|
|
@nickname('changeTeamPermissions')
|
|
@validate_json_request('TeamPermission')
|
|
def put(self, namespace, repository, teamname):
|
|
""" Update the existing team permission. """
|
|
new_permission = request.get_json()
|
|
|
|
logger.debug('Setting permission to: %s for team %s', new_permission['role'], teamname)
|
|
|
|
perm = model.permission.set_team_repo_permission(teamname, namespace, repository,
|
|
new_permission['role'])
|
|
|
|
log_action('change_repo_permission', namespace,
|
|
{'team': teamname, 'repo': repository,
|
|
'role': new_permission['role']},
|
|
repo=model.repository.get_repository(namespace, repository))
|
|
|
|
return wrap_role_view_team(role_view(perm), perm.team), 200
|
|
|
|
@require_repo_admin
|
|
@nickname('deleteTeamPermissions')
|
|
def delete(self, namespace, repository, teamname):
|
|
""" Delete the permission for the specified team. """
|
|
model.permission.delete_team_permission(teamname, namespace, repository)
|
|
|
|
log_action('delete_repo_permission', namespace,
|
|
{'team': teamname, 'repo': repository},
|
|
repo=model.repository.get_repository(namespace, repository))
|
|
|
|
return '', 204
|