Update officer permissions daily

config/cron.sh

@@ -0,0 +1,6 @@
+# This script adds commands to the current crontab
+
+# run the daily script at 1am every morning
+# TODO: make sure timezone is PST
+crontab -l | { cat; echo "0 1 * * * /home/csss-site/csss-site-backend/src/cron/daily.py"; } | crontab -
+

config/export_secrets.sh

@@ -0,0 +1,10 @@
+# TODO: only fill out this file in production
+export GMAIL_USERNAME = "todo"
+export GMAIL_PASSWORD = "todo"
+export GOOGLE_DRIVE_TOKEN = "todo"
+export GITHUB_TOKEN = "todo"
+export DISCORD_TOKEN = "todo"
+
+export CSSS_GUILD_ID = "todo"
+export SFU_API_TOKEN = "todo"
+

src/admin/email.py

@@ -0,0 +1,24 @@
+import os
+import smtplib
+
+# TODO: set this up
+GMAIL_PASSWORD = os.environ['GMAIL_PASSWORD']
+GMAIL_ADDRESS = "[email protected]"
+
+# TODO: look into sending emails from an sfu maillist (this might be painful)
+def send_email(
+    recipient_address: str,
+    subject: str,
+    contents: str,
+):
+    mail = smtplib.SMTP('smtp.gmail.com', 587)
+    mail.ehlo()
+    mail.starttls()
+    mail.login(GMAIL_ADDRESS, GMAIL_PASSWORD)
+
+    header = f"To: {recipient_address}\nFrom: {GMAIL_USERNAME}\nSubject: {subject}"
+    content = header + content
+
+    mail.sendmail(GMAIL_ADDRESS, recipient_address, content)
+    mail.quit()
+

src/cron/daily.py

@@ -0,0 +1,71 @@
+"""This module gets called by cron every day"""
+
+import asyncio
+import logging
+
+from database import _db_session
+from officers.crud import officer_terms
+
+import github
+import google
+
+_logger = logging.getLogger(__name__)
+
+async def update_google_permissions(db_session):
+    google_permissions = google.current_permissions()
+    #one_year_ago = datetime.today() - timedelta(days=365)
+
+    # TODO: for performance, only include officers with recent end-date (1 yr)
+    # but measure performance first
+    all_officer_terms = await all_officer_terms(db_session)
+    for term in all_officer_terms:
+        if utils.is_active(term):
+            # TODO: if google drive permission is not active, update them
+            pass
+        else:
+            # TODO: if google drive permissions are active, remove them
+            pass
+
+    _logger.info("updated google permissions")
+
+async def update_github_permissions(db_session):
+    github_permissions, team_id_map = github.all_permissions()
+
+    all_officer_terms = await all_officer_terms(db_session)
+    for term in all_officer_terms:
+        new_teams = (
+            # move all active officers to their respective teams
+            github.officer_teams(term.position)
+            if utils.is_active(term)
+            # move all inactive officers to the past_officers github organization
+            else ["past_officers"]
+        )
+        if term.username not in github_permissions:
+            user = get_user_by_username(term.username)
+            github.invite_user(
+                user.id,
+                [team_id_map[team] for team in new_teams], 
+            )
+        else:
+            github.set_user_teams(
+                term.username,
+                github_permissions[term.username].teams,
+                new_teams
+            )
+
+    _logger.info("updated github permissions")
+
+async def update_permissions():
+    db_session = _db_session()
+
+    update google_permissions(db_session)
+    db_session.commit()
+
+    update_github_permissions(db_session)
+    db_session.commit()
+
+    _logger.info("all permissions updated")
+
+if __name__ == "__main__":
+    asyncio.run(update_permissions())
+

src/github/__init__.py

@@ -0,0 +1,101 @@
+# TODO: does this allow importing anything from the module?
+
+from github.internals import list_members, list_teams 
+from admin.email import send_email
+
+from officers.constants import OfficerPosition
+
+# Rules:
+# - all past officers will be members of the github org
+# - all past officers will be put in past_officers team
+# - all current officers will be put in the officers team
+# - 
+
+
+# TODO: move this to github.constants.py
+GITHUB_TEAMS = {
+    "doa": "auto",
+    "election_officer": "auto",
+
+    "officers": "auto",
+    # TODO: create the past_officers team
+    "past_officers": "auto",
+
+    "w3_committee": "manual",
+    "wall_e": "manual",
+}
+AUTO_GITHUB_TEAMS = [
+    team
+    for (name, kind) in GITHUB_TEAMS.items()
+    if kind == "auto"
+]
+
+def officer_teams(position: str) -> list[str]:
+    if position == OfficerPosition.DIRECTOR_OF_ARCHIVES:
+        return ["doa", "officers"]
+    elif position == OfficerPosition.ELECTIONS_OFFICER:
+        return ["election_officer", "officers"]
+    else:
+        return ["officers"]
+
+# TODO: move these functions to github.public.py
+
+def all_permissions() -> dict[str, GithubUserPermissions]:
+    """
+    return a list of members in the organization (org) & their permissions
+    """
+
+    member_list = list_members()
+    member_name_list = { member.name for member in member_list }
+
+    team_list = []
+    for team in list_teams():
+        if team.name not in GITHUB_TEAMS.keys():
+            _logger.warning(f"Found unexpected github team {team.name}")
+            continue
+        elif GITHUB_TEAMS[team.name] == "manual":
+            continue
+
+        team_list += [team]
+
+    team_name_list = [team.name for team in team_list]
+    for team_name in AUTO_GITHUB_TEAMS:
+        if team_name not in team_name_list:
+            # TODO: send email for all errors & warnings 
+            # send_email("[email protected]", "ERROR: Missing Team", "...") 
+            _logger.error(f"Could not find 'auto' team {team_name} in organization")
+
+    user_permissions = {
+        user.username: GithubUserPermissions(user.username, [])
+        for user in member_list
+    }
+    for team in team_list: 
+        team_members = list_team_members(team.slug)
+        for member in team_members:
+            if member.name not in member_name_list:
+                _logger.warning(f"Found unexpected team_member={member.name} in team_slug={team.slug} not in the organization")
+                continue
+            user_permissions[member.username].teams += [team.slug]
+
+    # create a mapping between team name & team id, for use in creating invitations
+    team_id_map = {}
+    for team in team_list:
+        team_id_map[team.slug] = team.id
+
+    return user_permissions, team_id_map
+
+def set_user_teams(username: str, old_teams: list[str], new_teams: list[str]):
+    for team_slug in old_teams:
+        if team_slug not in new_teams:
+            remove_user_from_team(term.username, team_slug)
+
+    for team_slug in new_teams:
+        if team_slug not in old_teams:
+            # TODO: what happens when adding a user to a team who is not part of the github org yet?
+            add_user_to_team(term.username, team_slug)
+
+def invite_user(github_username: str, teams: str):
+    # invite this user to the github organization
+    # TODO: is an invited user considered a member of the organization?
+    pass
+