280 lines
9.1 KiB
Python
280 lines
9.1 KiB
Python
from flask import request
|
|
|
|
from endpoints.api import (resource, nickname, ApiResource, validate_json_request, request_error,
|
|
log_action, Unauthorized, NotFound, internal_only, require_scope,
|
|
query_param, truthy_bool, parse_args, require_user_admin)
|
|
from auth.permissions import AdministerOrganizationPermission, ViewTeamPermission
|
|
from auth.auth_context import get_authenticated_user
|
|
from auth import scopes
|
|
from data import model
|
|
from util.useremails import send_org_invite_email
|
|
from util.gravatar import compute_hash
|
|
|
|
def handle_addinvite_team(inviter, team, user=None, email=None):
|
|
invite = model.add_or_invite_to_team(inviter, team, user, email)
|
|
if not invite:
|
|
# User was added to the team directly.
|
|
return
|
|
|
|
orgname = team.organization.username
|
|
if user:
|
|
model.create_notification('org_team_invite', user, metadata = {
|
|
'code': invite.invite_token,
|
|
'inviter': inviter.username,
|
|
'org': orgname,
|
|
'team': team.name
|
|
})
|
|
|
|
send_org_invite_email(user.username if user else email, user.email if user else email,
|
|
orgname, team.name, inviter.username, invite.invite_token)
|
|
return invite
|
|
|
|
def team_view(orgname, team):
|
|
view_permission = ViewTeamPermission(orgname, team.name)
|
|
role = model.get_team_org_role(team).name
|
|
return {
|
|
'id': team.id,
|
|
'name': team.name,
|
|
'description': team.description,
|
|
'can_view': view_permission.can(),
|
|
'role': role
|
|
}
|
|
|
|
def member_view(member, invited=False):
|
|
return {
|
|
'name': member.username,
|
|
'kind': 'user',
|
|
'is_robot': member.robot,
|
|
'gravatar': compute_hash(member.email) if not member.robot else None,
|
|
'invited': invited,
|
|
}
|
|
|
|
|
|
def invite_view(invite):
|
|
if invite.user:
|
|
return member_view(invite.user, invited=True)
|
|
else:
|
|
return {
|
|
'email': invite.email,
|
|
'kind': 'invite',
|
|
'gravatar': compute_hash(invite.email),
|
|
'invited': True
|
|
}
|
|
|
|
|
|
@resource('/v1/organization/<orgname>/team/<teamname>')
|
|
@internal_only
|
|
class OrganizationTeam(ApiResource):
|
|
""" Resource for manging an organization's teams. """
|
|
schemas = {
|
|
'TeamDescription': {
|
|
'id': 'TeamDescription',
|
|
'type': 'object',
|
|
'description': 'Description of a team',
|
|
'required': [
|
|
'role',
|
|
],
|
|
'properties': {
|
|
'role': {
|
|
'type': 'string',
|
|
'description': 'Org wide permissions that should apply to the team',
|
|
'enum': [
|
|
'member',
|
|
'creator',
|
|
'admin',
|
|
],
|
|
},
|
|
'description': {
|
|
'type': 'string',
|
|
'description': 'Markdown description for the team',
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
@nickname('updateOrganizationTeam')
|
|
@validate_json_request('TeamDescription')
|
|
def put(self, orgname, teamname):
|
|
""" Update the org-wide permission for the specified team. """
|
|
edit_permission = AdministerOrganizationPermission(orgname)
|
|
if edit_permission.can():
|
|
team = None
|
|
|
|
details = request.get_json()
|
|
is_existing = False
|
|
try:
|
|
team = model.get_organization_team(orgname, teamname)
|
|
is_existing = True
|
|
except model.InvalidTeamException:
|
|
# Create the new team.
|
|
description = details['description'] if 'description' in details else ''
|
|
role = details['role'] if 'role' in details else 'member'
|
|
|
|
org = model.get_organization(orgname)
|
|
team = model.create_team(teamname, org, role, description)
|
|
log_action('org_create_team', orgname, {'team': teamname})
|
|
|
|
if is_existing:
|
|
if ('description' in details and
|
|
team.description != details['description']):
|
|
team.description = details['description']
|
|
team.save()
|
|
log_action('org_set_team_description', orgname,
|
|
{'team': teamname, 'description': team.description})
|
|
|
|
if 'role' in details:
|
|
role = model.get_team_org_role(team).name
|
|
if role != details['role']:
|
|
team = model.set_team_org_permission(team, details['role'],
|
|
get_authenticated_user().username)
|
|
log_action('org_set_team_role', orgname, {'team': teamname, 'role': details['role']})
|
|
|
|
return team_view(orgname, team), 200
|
|
|
|
raise Unauthorized()
|
|
|
|
@nickname('deleteOrganizationTeam')
|
|
def delete(self, orgname, teamname):
|
|
""" Delete the specified team. """
|
|
permission = AdministerOrganizationPermission(orgname)
|
|
if permission.can():
|
|
model.remove_team(orgname, teamname, get_authenticated_user().username)
|
|
log_action('org_delete_team', orgname, {'team': teamname})
|
|
return 'Deleted', 204
|
|
|
|
raise Unauthorized()
|
|
|
|
|
|
@resource('/v1/organization/<orgname>/team/<teamname>/members')
|
|
@internal_only
|
|
class TeamMemberList(ApiResource):
|
|
""" Resource for managing the list of members for a team. """
|
|
@parse_args
|
|
@query_param('includePending', 'Whether to include pending members', type=truthy_bool, default=False)
|
|
@nickname('getOrganizationTeamMembers')
|
|
def get(self, args, orgname, teamname):
|
|
""" Retrieve the list of members for the specified team. """
|
|
view_permission = ViewTeamPermission(orgname, teamname)
|
|
edit_permission = AdministerOrganizationPermission(orgname)
|
|
|
|
if view_permission.can():
|
|
team = None
|
|
try:
|
|
team = model.get_organization_team(orgname, teamname)
|
|
except model.InvalidTeamException:
|
|
raise NotFound()
|
|
|
|
members = model.get_organization_team_members(team.id)
|
|
invites = []
|
|
|
|
if args['includePending'] and edit_permission.can():
|
|
invites = model.get_organization_team_member_invites(team.id)
|
|
|
|
data = {
|
|
'members': [member_view(m) for m in members] + [invite_view(i) for i in invites],
|
|
'can_edit': edit_permission.can()
|
|
}
|
|
|
|
return data
|
|
|
|
raise Unauthorized()
|
|
|
|
|
|
@resource('/v1/organization/<orgname>/team/<teamname>/members/<membername>')
|
|
class TeamMember(ApiResource):
|
|
""" Resource for managing individual members of a team. """
|
|
@require_scope(scopes.ORG_ADMIN)
|
|
@nickname('updateOrganizationTeamMember')
|
|
def put(self, orgname, teamname, membername):
|
|
""" Adds or invites a member to an existing team. """
|
|
permission = AdministerOrganizationPermission(orgname)
|
|
if permission.can():
|
|
team = None
|
|
user = None
|
|
|
|
# Find the team.
|
|
try:
|
|
team = model.get_organization_team(orgname, teamname)
|
|
except model.InvalidTeamException:
|
|
raise NotFound()
|
|
|
|
# Find the user.
|
|
user = model.get_user(membername)
|
|
if not user:
|
|
raise request_error(message='Unknown user')
|
|
|
|
# Add or invite the user to the team.
|
|
inviter = get_authenticated_user()
|
|
invite = handle_addinvite_team(inviter, team, user=user)
|
|
if not invite:
|
|
log_action('org_add_team_member', orgname, {'member': membername, 'team': teamname})
|
|
return member_view(user, invited=False)
|
|
|
|
# User was invited.
|
|
log_action('org_invite_team_member', orgname, {'member': membername, 'team': teamname})
|
|
return member_view(user, invited=True)
|
|
|
|
raise Unauthorized()
|
|
|
|
@require_scope(scopes.ORG_ADMIN)
|
|
@nickname('deleteOrganizationTeamMember')
|
|
def delete(self, orgname, teamname, membername):
|
|
""" Delete an existing member of a team. """
|
|
permission = AdministerOrganizationPermission(orgname)
|
|
if permission.can():
|
|
# Remote the user from the team.
|
|
invoking_user = get_authenticated_user().username
|
|
model.remove_user_from_team(orgname, teamname, membername, invoking_user)
|
|
log_action('org_remove_team_member', orgname, {'member': membername, 'team': teamname})
|
|
return 'Deleted', 204
|
|
|
|
raise Unauthorized()
|
|
|
|
|
|
@resource('/v1/teaminvite/<code>')
|
|
@internal_only
|
|
class TeamMemberInvite(ApiResource):
|
|
""" Resource for managing invites to jon a team. """
|
|
@require_user_admin
|
|
@nickname('acceptOrganizationTeamInvite')
|
|
def put(self, code):
|
|
""" Accepts an invite to join a team in an organization. """
|
|
# Accept the invite for the current user.
|
|
try:
|
|
(team, inviter) = model.confirm_team_invite(code, get_authenticated_user())
|
|
except model.DataModelException:
|
|
raise NotFound()
|
|
|
|
model.delete_matching_notifications(get_authenticated_user(), 'org_team_invite', code=code)
|
|
|
|
orgname = team.organization.username
|
|
log_action('org_team_member_invite_accepted', orgname, {
|
|
'member': get_authenticated_user().username,
|
|
'team': team.name,
|
|
'inviter': inviter.username
|
|
})
|
|
|
|
return {
|
|
'org': orgname,
|
|
'team': team.name
|
|
}
|
|
|
|
@nickname('declineOrganizationTeamInvite')
|
|
@require_user_admin
|
|
def delete(self, code):
|
|
""" Delete an existing member of a team. """
|
|
try:
|
|
(team, inviter) = model.delete_team_invite(code, get_authenticated_user())
|
|
except model.DataModelException:
|
|
raise NotFound()
|
|
|
|
model.delete_matching_notifications(get_authenticated_user(), 'org_team_invite', code=code)
|
|
|
|
orgname = team.organization.username
|
|
log_action('org_team_member_invite_declined', orgname, {
|
|
'member': get_authenticated_user().username,
|
|
'team': team.name,
|
|
'inviter': inviter.username
|
|
})
|
|
|
|
return 'Deleted', 204
|