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,31 @@ 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']
+            msg = "Added User to " + command_team
+
+            # If this team is a team with special permissions, promote the
+            # user to the appropriate permission
+            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
+
+            # Only perform promotion if it is actually a promotion.
+            if promoted_level > user.permissions_level:
+                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 +438,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 +453,39 @@ 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 the user is being removed from a team with special
+            # permisisons, figure out a demotion strategy.
+            demoted_level = None
+            if command_team == self.config.github_team_leads:
+                # If the user is currently an admin, we only demote this user
+                # if it is currently NOT and admin team member
+                if user.permissions_level == Permissions.admin \
+                        and len(self.config.github_team_admin) > 0:
+                    admins = get_team_by_name(
+                        self.facade, self.config.github_team_admin)
+                    if not admins.has_member(user.github_id):
+                        demoted_level = Permissions.member
+                else:
+                    demoted_level = Permissions.member
+            if command_team == self.config.github_team_admin:
+                # If the user is being removed from the admin team, we demote
+                # this user to team_lead if this user is a member of the leads
+                # team, otherwise we demote to member.
+                demoted_level = Permissions.member
+                if len(self.config.github_team_leads) > 0:
+                    leads = get_team_by_name(
+                        self.facade, self.config.github_team_leads)
+                    if leads.has_member(user.github_id):
+                        demoted_level = Permissions.team_lead
+
+            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 +510,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 +544,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 +557,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 +566,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 +587,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 +656,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 +707,43 @@ 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:
+                    if user.permissions_level < t['permission']:
+                        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 +755,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)

app/model/permissions.py

@@ -2,7 +2,34 @@
 from enum import Enum
 
 
-class Permissions(Enum):
+class OrderedEnum(Enum):
+    """
+    Comparable enum -
+    copied from https://docs.python.org/3/library/enum.html#orderedenum
+    """
+
+    def __ge__(self, other):
+        if self.__class__ is other.__class__:
+            return self.value >= other.value
+        return NotImplemented
+
+    def __gt__(self, other):
+        if self.__class__ is other.__class__:
+            return self.value > other.value
+        return NotImplemented
+
+    def __le__(self, other):
+        if self.__class__ is other.__class__:
+            return self.value <= other.value
+        return NotImplemented
+
+    def __lt__(self, other):
+        if self.__class__ is other.__class__:
+            return self.value < other.value
+        return NotImplemented
+
+
+class Permissions(OrderedEnum):
     """Enum to represent possible permissions levels."""
 
     member = 1

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}