feat: set teams that automatically grant permissions

app/controller/command/commands/team.py

@@ -4,8 +4,9 @@
 from argparse import ArgumentParser, _SubParsersAction
 from app.controller import ResponseTuple
 from app.controller.command.commands.base import Command
+from app.model.permissions import Permissions
 from db.facade import DBFacade
-from db.utils import get_team_by_name
+from db.utils import get_team_by_name, get_team_members
 from interface.github import GithubAPIException, GithubInterface
 from interface.slack import SlackAPIError
 from interface.gcp import GCPInterface
@@ -275,27 +276,28 @@ def view_helper(self, team_name) -> ResponseTuple:
         :return: error message if team not found,
                  otherwise return team information
         """
-        teams = self.facade.query(Team, [('github_team_name', team_name)])
-        if len(teams) != 1:
+        try:
+            team = get_team_by_name(self.facade, team_name)
+            team_leads_set = team.team_leads
+            team_leads_list = list(map(lambda i: ('github_user_id',
+                                                  str(i)), team_leads_set))
+            team_leads: List[User] = []
+            if team_leads_list:
+                team_leads = self.facade.query_or(User, team_leads_list)
+            names = set(map(lambda m: m.github_username, team_leads))
+            team.team_leads = names
+
+            members_set = team.members
+            members_list = list(map(lambda i: ('github_user_id',
+                                               str(i)), members_set))
+            members: List[User] = []
+            if members_list:
+                members = self.facade.query_or(User, members_list)
+            names = set(map(lambda m: m.github_username, members))
+            team.members = names
+            return {'attachments': [team.get_attachment()]}, 200
+        except LookupError:
             return self.lookup_error, 200
-        team_leads_set = teams[0].team_leads
-        team_leads_list = list(map(lambda i: ('github_user_id',
-                                              str(i)), team_leads_set))
-        team_leads: List[User] = []
-        if team_leads_list:
-            team_leads = self.facade.query_or(User, team_leads_list)
-        names = set(map(lambda m: m.github_username, team_leads))
-        teams[0].team_leads = names
-
-        members_set = teams[0].members
-        members_list = list(map(lambda i: ('github_user_id',
-                                           str(i)), members_set))
-        members: List[User] = []
-        if members_list:
-            members = self.facade.query_or(User, members_list)
-        names = set(map(lambda m: m.github_username, members))
-        teams[0].members = names
-        return {'attachments': [teams[0].get_attachment()]}, 200
 
     def create_helper(self, param_list, user_id) -> ResponseTuple:
         """
@@ -385,19 +387,29 @@ def add_helper(self, param_list, user_id) -> ResponseTuple:
         """
         try:
             command_user = self.facade.retrieve(User, user_id)
-            teams = self.facade.query(Team, [('github_team_name',
-                                              param_list['team_name'])])
-            if len(teams) != 1:
-                return self.lookup_error, 200
-            team = teams[0]
+            command_team = param_list['team_name']
+            team = get_team_by_name(self.facade, command_team)
             if not check_permissions(command_user, team):
                 return self.permission_error, 200
 
             user = self.facade.retrieve(User, param_list['username'])
             team.add_member(user.github_id)
             self.gh.add_team_member(user.github_username, team.github_team_id)
             self.facade.store(team)
-            msg = "Added User to " + param_list['team_name']
+
+            # If this team is a team with special permissions, promote the user
+            promoted_level = Permissions.member
+            if command_team == self.config.github_team_admin:
+                promoted_level = Permissions.admin
+            elif command_team == self.config.github_team_leads:
+                promoted_level = Permissions.team_lead
+
+            msg = "Added User to " + command_team
+            if promoted_level != Permissions.member:
+                logging.info(f"Promoting {command_user} to {promoted_level}")
+                user.permissions_level = promoted_level
+                self.facade.store(user)
+                msg += f" and promoted user to {promoted_level}"
             ret = {'attachments': [team.get_attachment()], 'text': msg}
             return ret, 200
 
@@ -424,11 +436,8 @@ def remove_helper(self, param_list, user_id) -> ResponseTuple:
         """
         try:
             command_user = self.facade.retrieve(User, user_id)
-            teams = self.facade.query(Team, [('github_team_name',
-                                              param_list['team_name'])])
-            if len(teams) != 1:
-                return self.lookup_error, 200
-            team = teams[0]
+            command_team = param_list['team_name']
+            team = get_team_by_name(self.facade, command_team)
             if not check_permissions(command_user, team):
                 return self.permission_error, 200
 
@@ -442,7 +451,34 @@ def remove_helper(self, param_list, user_id) -> ResponseTuple:
             self.gh.remove_team_member(user.github_username,
                                        team.github_team_id)
             self.facade.store(team)
-            msg = "Removed User from " + param_list['team_name']
+
+            msg = "Removed User from " + command_team
+
+            # If this team is a team with special permissions, demote the user
+            demoted_level = None
+            if command_team == self.config.github_team_leads:
+                # Leads can only be demoted to member
+                demoted_level = Permissions.member
+            if command_team == self.config.github_team_admin:
+                # If this team is admin, we need to check what they should be
+                # demoted to
+                if len(self.config.github_team_leads) > 0:
+                    try:
+                        leads = get_team_by_name(self.facade, 'leads')
+                        if leads.has_member(user.github_id):
+                            demoted_level = Permissions.team_lead
+                        else:
+                            demoted_level = Permissions.member
+                    except LookupError:
+                        demoted_level = Permissions.member
+                else:
+                    demoted_level = Permissions.member
+
+            if demoted_level is not None:
+                logging.info(f"Demoting {command_user} to member")
+                user.permissions_level = demoted_level
+                self.facade.store(user)
+                msg += " and demoted user"
             ret = {'attachments': [team.get_attachment()], 'text': msg}
             return ret, 200
 
@@ -467,14 +503,11 @@ def edit_helper(self, param_list, user_id) -> ResponseTuple:
         """
         try:
             command_user = self.facade.retrieve(User, user_id)
-            teams = self.facade.query(Team, [('github_team_name',
-                                              param_list['team_name'])])
-            if len(teams) != 1:
-                return self.lookup_error, 200
-            team = teams[0]
+            command_team = param_list['team_name']
+            team = get_team_by_name(self.facade, command_team)
             if not check_permissions(command_user, team):
                 return self.permission_error, 200
-            msg = f"Team edited: {param_list['team_name']}, "
+            msg = f"Team edited: {command_team}, "
             if param_list['name'] is not None:
                 msg += f"name: {param_list['name']}, "
                 team.display_name = param_list['name']
@@ -504,11 +537,8 @@ def lead_helper(self, param_list, user_id) -> ResponseTuple:
         """
         try:
             command_user = self.facade.retrieve(User, user_id)
-            teams = self.facade.query(Team, [('github_team_name',
-                                              param_list['team_name'])])
-            if len(teams) != 1:
-                return self.lookup_error, 200
-            team = teams[0]
+            command_team = param_list['team_name']
+            team = get_team_by_name(self.facade, command_team)
             if not check_permissions(command_user, team):
                 return self.permission_error, 200
             user = self.facade.retrieve(User, param_list["username"])
@@ -520,7 +550,7 @@ def lead_helper(self, param_list, user_id) -> ResponseTuple:
                     team.discard_team_lead(user.github_id)
                 self.facade.store(team)
                 msg = f"User removed as team lead from" \
-                      f" {param_list['team_name']}"
+                      f" {command_team}"
             else:
                 if not team.has_member(user.github_id):
                     team.add_member(user.github_id)
@@ -529,7 +559,7 @@ def lead_helper(self, param_list, user_id) -> ResponseTuple:
                 team.add_team_lead(user.github_id)
                 self.facade.store(team)
                 msg = f"User added as team lead to" \
-                      f" {param_list['team_name']}"
+                      f" {command_team}"
             ret = {'attachments': [team.get_attachment()], 'text': msg}
             return ret, 200
         except LookupError:
@@ -550,11 +580,7 @@ def delete_helper(self, team_name, user_id) -> ResponseTuple:
         """
         try:
             command_user = self.facade.retrieve(User, user_id)
-            teams = self.facade.query(Team, [('github_team_name',
-                                              team_name)])
-            if len(teams) != 1:
-                return self.lookup_error, 200
-            team = teams[0]
+            team = get_team_by_name(self.facade, team_name)
             if not check_permissions(command_user, team):
                 return self.permission_error, 200
             self.facade.delete(Team, team.github_team_id)
@@ -623,6 +649,9 @@ def refresh_helper(self, user_id) -> ResponseTuple:
             # add all members (if not already added) to the 'all' team
             self.refresh_all_team()
 
+            # promote members inside special teams
+            self.refresh_all_rocket_permissions()
+
             # enforce Drive permissions
             self.refresh_all_drive_permissions()
         except GithubAPIException as e:
@@ -671,6 +700,42 @@ def refresh_all_team(self):
         else:
             logging.error(f'Could not create {all_name}. Aborting.')
 
+    def refresh_all_rocket_permissions(self):
+        """
+        Refresh Rocket permissions for members in teams like
+        GITHUB_ADMIN_TEAM_NAME and GITHUB_LEADS_TEAM_NAME.
+
+        It only ever promotes users, and does not demote users.
+        """
+        # provide teams from low permissions level to high
+        teams = [
+            {
+                'name': self.config.github_team_leads,
+                'permission': Permissions.team_lead,
+            },
+            {
+                'name': self.config.github_team_admin,
+                'permission': Permissions.admin,
+            },
+        ]
+        for t in teams:
+            if len(t['name']) == 0:
+                continue
+
+            team = None
+            try:
+                team = get_team_by_name(self.facade, t['name'])
+            except LookupError:
+                t_id = str(self.gh.org_create_team(t['name']))
+                logging.info(f'team {t["name"]} created')
+                self.facade.store(Team(t_id, t['name'], t['name']))
+
+            if team is not None:
+                team_members = get_team_members(team)
+                for user in team_members:
+                    user.permissions_level = t['permission']
+                    self.facade.store(user)
+
     def refresh_all_drive_permissions(self):
         """
         Refresh Google Drive permissions for all teams. If no GCP client
@@ -682,5 +747,23 @@ def refresh_all_drive_permissions(self):
             return
 
         all_teams: List[Team] = self.facade.query(Team)
+        leads_team: Team = None
+        admin_team: Team = None
         for t in all_teams:
+            if t.github_team_name == self.config.github_team_leads:
+                leads_team = t
+                continue
+            if t.github_team_name == self.config.github_team_admin:
+                admin_team = t
+                continue
             sync_team_email_perms(self.gcp, self.facade, t)
+
+        # Workaround for https://github.com/ubclaunchpad/rocket2/issues/497:
+        # We sort the teams such that special-permissions teams are sync'd
+        # last, so that inherited permissions are not overwritten in child
+        # folders.
+        #
+        # TODO: If a proper fix is implemented, remove this and related code
+        for t in [leads_team, admin_team]:
+            if t is not None:
+                sync_team_email_perms(self.gcp, self.facade, t)

config/__init__.py

@@ -21,6 +21,8 @@ class Config:
         'GITHUB_APP_ID': 'github_app_id',
         'GITHUB_ORG_NAME': 'github_org_name',
         'GITHUB_DEFAULT_TEAM_NAME': 'github_team_all',
+        'GITHUB_ADMIN_TEAM_NAME': 'github_team_admin',
+        'GITHUB_LEADS_TEAM_NAME': 'github_team_leads',
         'GITHUB_WEBHOOK_ENDPT': 'github_webhook_endpt',
         'GITHUB_WEBHOOK_SECRET': 'github_webhook_secret',
         'GITHUB_KEY': 'github_key',
@@ -39,6 +41,8 @@ class Config:
     OPTIONALS = {
         'AWS_LOCAL': 'False',
         'GITHUB_DEFAULT_TEAM_NAME': 'all',
+        'GITHUB_ADMIN_TEAM_NAME': '',
+        'GITHUB_LEADS_TEAM_NAME': '',
         'GCP_SERVICE_ACCOUNT_CREDENTIALS': '',
         'GCP_SERVICE_ACCOUNT_SUBJECT': '',
     }
@@ -84,6 +88,8 @@ def _set_attrs(self):
         self.github_app_id = ''
         self.github_org_name = ''
         self.github_team_all = ''
+        self.github_team_admin = ''
+        self.github_team_leads = ''
         self.github_webhook_endpt = ''
         self.github_webhook_secret = ''
         self.github_key = ''

db/utils.py

@@ -35,6 +35,15 @@ def get_team_by_name(dbf: DBFacade, gh_team_name: str) -> Team:
         return teams[0]
 
 
+def get_team_members(dbf: DBFacade, team: Team) -> List[User]:
+    """
+    Query users that are members of the given team.
+
+    :return: Users that belong to the team
+    """
+    return get_users_by_ghid(dbf, list(team.members))
+
+
 def get_users_by_ghid(dbf: DBFacade, gh_ids: List[str]) -> List[User]:
     """
     Query users by github user id.

docker-compose.yml

@@ -28,6 +28,9 @@ services:
             - GITHUB_ORG_NAME=${GITHUB_ORG_NAME}
             - GITHUB_WEBHOOK_ENDPT=${GITHUB_WEBHOOK_ENDPT}
             - GITHUB_WEBHOOK_SECRET=${GITHUB_WEBHOOK_SECRET}
+            - GITHUB_DEFAULT_TEAM_NAME=${GITHUB_DEFAULT_TEAM_NAME}
+            - GITHUB_ADMIN_TEAM_NAME=${GITHUB_ADMIN_TEAM_NAME}
+            - GITHUB_LEADS_TEAM_NAME=${GITHUB_LEADS_TEAM_NAME}
             - GITHUB_KEY=${GITHUB_KEY}
             - AWS_ACCESS_KEYID=${AWS_ACCESS_KEYID}
             - AWS_SECRET_KEY=${AWS_SECRET_KEY}

docs/Config.rst

@@ -72,6 +72,29 @@ GITHUB_ORG_NAME
 The name of your Github organization (the string in the URL whenever you
 go to the organization.
 
+GITHUB_DEFAULT_TEAM_NAME
+------------------------
+
+The name of the GitHub team in your organization that all users should
+be added to. Optional, defaults to ``all``.
+
+GITHUB_ADMIN_TEAM_NAME
+----------------------
+
+The name of the GitHub team in your organization that should be automatically
+promoted to Rocket administrators. Optional.
+
+Note that this does not mean all Rocket administrators will be added to this
+team.
+
+GITHUB_LEADS_TEAM_NAME
+----------------------
+
+The name of the GitHub team in your organization that should be automatically
+promoted to Rocket team leads. Optional.
+
+Note that this does not mean all Rocket team leads will be added to this team.
+
 GITHUB_WEBHOOK_ENDPT
 --------------------
 

interface/gcp_utils.py

@@ -3,6 +3,7 @@
 from typing import List, Optional
 from interface.gcp import GCPInterface
 from db import DBFacade
+from db.utils import get_team_members
 from app.model import User, Team
 
 
@@ -43,17 +44,11 @@ def sync_team_email_perms(gcp: Optional[GCPInterface],
         return
 
     # Generate who to share with
+    team_members = get_team_members(db, team)
     emails: List[str] = []
-    for github_id in team.members:
-        users = db.query(User, [('github_user_id', github_id)])
-        if len(users) != 1:
-            logging.warn(f"None/multiple users for GitHub ID {github_id}")
-
-        # For now, naiively iterate over all users, due to
-        # https://github.com/ubclaunchpad/rocket2/issues/493
-        for user in users:
-            if len(user.email) > 0:
-                emails.append(user.email)
+    for user in team_members:
+        if len(user.email) > 0:
+            emails.append(user.email)
 
     # Sync permissions
     if len(emails) > 0: