235 lines
7.7 KiB
Python
235 lines
7.7 KiB
Python
import logging
|
|
|
|
from flask import request
|
|
|
|
from endpoints.api import (resource, nickname, require_repo_admin, RepositoryParamResource,
|
|
log_action, request_error, validate_json_request)
|
|
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['is_robot'] = user.robot
|
|
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
|
|
|
|
|
|
@resource('/v1/repository/<path:repository>/permissions/team/')
|
|
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.get_all_repo_teams(namespace, repository)
|
|
|
|
return {
|
|
'permissions': {repo_perm.team.name: role_view(repo_perm)
|
|
for repo_perm in repo_perms}
|
|
}
|
|
|
|
|
|
@resource('/v1/repository/<path:repository>/permissions/user/')
|
|
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.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.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.get_all_repo_users(namespace, repository)
|
|
return {
|
|
'permissions': {perm.user.username: role_view_func(perm)
|
|
for perm in repo_perms}
|
|
}
|
|
|
|
|
|
@resource('/v1/repository/<path:repository>/permissions/user/<username>')
|
|
class RepositoryUserPermission(RepositoryParamResource):
|
|
""" Resource for managing individual user permissions. """
|
|
schemas = {
|
|
'UserPermission': {
|
|
'id': 'UserPermission',
|
|
'type': 'object',
|
|
'description': 'Description of a user permission.',
|
|
'required': True,
|
|
'properties': {
|
|
'role': {
|
|
'type': 'string',
|
|
'description': 'Role to use for the user',
|
|
'enum': [
|
|
'read',
|
|
'write',
|
|
'admin',
|
|
],
|
|
'required': True,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
@require_repo_admin
|
|
@nickname('getUserPermission')
|
|
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.get_user_reponame_permission(username, namespace, repository)
|
|
perm_view = wrap_role_view_user(role_view(perm), perm.user)
|
|
|
|
try:
|
|
model.get_organization(namespace)
|
|
org_members = model.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))
|
|
|
|
perm = model.set_user_repo_permission(username, namespace, repository,
|
|
new_permission['role'])
|
|
perm_view = wrap_role_view_user(role_view(perm), perm.user)
|
|
|
|
try:
|
|
model.get_organization(namespace)
|
|
org_members = model.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:
|
|
return request_error(exception=ex)
|
|
|
|
log_action('change_repo_permission', namespace,
|
|
{'username': username, 'repo': repository,
|
|
'role': new_permission['role']},
|
|
repo=model.get_repository(namespace, repository))
|
|
|
|
return perm_view, 200 # 201 for post
|
|
|
|
@require_repo_admin
|
|
@nickname('deleteUserPermissions')
|
|
def delete(namespace, repository, username):
|
|
""" Delete the permission for the user. """
|
|
try:
|
|
model.delete_user_permission(username, namespace, repository)
|
|
except model.DataModelException as ex:
|
|
return request_error(exception=ex)
|
|
|
|
log_action('delete_repo_permission', namespace,
|
|
{'username': username, 'repo': repository},
|
|
repo=model.get_repository(namespace, repository))
|
|
|
|
return 'Deleted', 204
|
|
|
|
|
|
@resource('/v1/repository/<path:repository>/permissions/team/<teamname>')
|
|
class RepositoryTeamPermission(RepositoryParamResource):
|
|
""" Resource for managing individual team permissions. """
|
|
schemas = {
|
|
'TeamPermission': {
|
|
'id': 'TeamPermission',
|
|
'type': 'object',
|
|
'description': 'Description of a team permission.',
|
|
'required': True,
|
|
'properties': {
|
|
'role': {
|
|
'type': 'string',
|
|
'description': 'Role to use for the team',
|
|
'enum': [
|
|
'read',
|
|
'write',
|
|
'admin',
|
|
],
|
|
'required': True,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
@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.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.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.get_repository(namespace, repository))
|
|
|
|
return role_view(perm), 200 # Should be 201 for post
|
|
|
|
@require_repo_admin
|
|
@nickname('deleteTeamPermissions')
|
|
def delete(self, namespace, repository, teamname):
|
|
""" Delete the permission for the specified team. """
|
|
model.delete_team_permission(teamname, namespace, repository)
|
|
|
|
log_action('delete_repo_permission', namespace,
|
|
{'team': teamname, 'repo': repository},
|
|
repo=model.get_repository(namespace, repository))
|
|
|
|
return 'Deleted', 204
|