quay/endpoints/api/team.py

from flask import request

from endpoints.api import (resource, nickname, ApiResource, validate_json_request, request_error,
                           log_action, Unauthorized, NotFound, internal_only, require_scope,
                           path_param, query_param, truthy_bool, parse_args, require_user_admin,
                           show_if)
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 app import avatar

import features

def try_accept_invite(code, user):
  (team, inviter) = model.confirm_team_invite(code, user)

  model.delete_matching_notifications(user, 'org_team_invite', code=code)

  orgname = team.organization.username
  log_action('org_team_member_invite_accepted', orgname, {
    'member': user.username,
    'team': team.name,
    'inviter': inviter.username
  })

  return team


def handle_addinvite_team(inviter, team, user=None, email=None):
  invite = model.add_or_invite_to_team(inviter, team, user, email,
    requires_invite = features.MAILING)
  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,
    'avatar': avatar.compute_hash(member.email, name=member.username) 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',
      'avatar': avatar.compute_hash(invite.email),
      'invited': True
    }


@resource('/v1/organization/<orgname>/team/<teamname>')
@path_param('orgname', 'The name of the organization')
@path_param('teamname', 'The name of the team')
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',
        },
      },
    },
  }

  @require_scope(scopes.ORG_ADMIN)
  @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()

  @require_scope(scopes.ORG_ADMIN)
  @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')
@path_param('orgname', 'The name of the organization')
@path_param('teamname', 'The name of the team')
class TeamMemberList(ApiResource):
  """ Resource for managing the list of members for a team. """
  @require_scope(scopes.ORG_ADMIN)
  @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>')
@path_param('orgname', 'The name of the organization')
@path_param('teamname', 'The name of the team')
@path_param('membername', 'The username of the team member')
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, {
        'user': membername,
        '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 a member of a team. If the user is merely invited to join
        the team, then the invite is removed instead.
    """
    permission = AdministerOrganizationPermission(orgname)
    if permission.can():
      # Remote the user from the team.
      invoking_user = get_authenticated_user().username

      # Find the team.
      try:
        team = model.get_organization_team(orgname, teamname)
      except model.InvalidTeamException:
        raise NotFound()

      # Find the member.
      member = model.get_user(membername)
      if not member:
        raise NotFound()

      # First attempt to delete an invite for the user to this team. If none found,
      # then we try to remove the user directly.
      if model.delete_team_user_invite(team, member):
        log_action('org_delete_team_member_invite', orgname, {
          'user': membername,
          'team': teamname,
          'member': membername
        })
        return 'Deleted', 204

      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/organization/<orgname>/team/<teamname>/invite/<email>')
@show_if(features.MAILING)
class InviteTeamMember(ApiResource):
  """ Resource for inviting a team member via email address. """
  @require_scope(scopes.ORG_ADMIN)
  @nickname('inviteTeamMemberEmail')
  def put(self, orgname, teamname, email):
    """ Invites an email address to an existing team. """
    permission = AdministerOrganizationPermission(orgname)
    if permission.can():
      team = None

      # Find the team.
      try:
        team = model.get_organization_team(orgname, teamname)
      except model.InvalidTeamException:
        raise NotFound()

      # Invite the email to the team.
      inviter = get_authenticated_user()
      invite = handle_addinvite_team(inviter, team, email=email)
      log_action('org_invite_team_member', orgname, {
        'email': email,
        'team': teamname,
        'member': email
      })
      return invite_view(invite)

    raise Unauthorized()

  @require_scope(scopes.ORG_ADMIN)
  @nickname('deleteTeamMemberEmailInvite')
  def delete(self, orgname, teamname, email):
    """ Delete an invite of an email address to join a team. """
    permission = AdministerOrganizationPermission(orgname)
    if permission.can():
      team = None

      # Find the team.
      try:
        team = model.get_organization_team(orgname, teamname)
      except model.InvalidTeamException:
        raise NotFound()

      # Delete the invite.
      model.delete_team_email_invite(team, email)
      log_action('org_delete_team_member_invite', orgname, {
        'email': email,
        'team': teamname,
        'member': email
      })
      return 'Deleted', 204

    raise Unauthorized()


@resource('/v1/teaminvite/<code>')
@internal_only
@show_if(features.MAILING)
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.
    team = try_accept_invite(code, get_authenticated_user())
    if not team:
      raise NotFound()

    orgname = team.organization.username
    return {
      'org': orgname,
      'team': team.name
    }

  @nickname('declineOrganizationTeamInvite')
  @require_user_admin
  def delete(self, code):
    """ Delete an existing member of a team. """
    (team, inviter) = model.delete_team_invite(code, get_authenticated_user())

    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