import atexit from operator import truediv import os import time import json import github3 from distutils.util import strtobool import threading import sys import traceback from concurrent.futures import ThreadPoolExecutor from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger from flask import Flask from githubapp import ( GitHubApp, DirectoryClient, CRON_INTERVAL, TEST_MODE, ADD_MEMBER, REMOVE_ORG_MEMBERS_WITHOUT_TEAM, USER_SYNC_ATTRIBUTE, SYNCMAP_ONLY, ) app = Flask(__name__) github_app = GitHubApp(app) # Schedule a full sync scheduler = BackgroundScheduler(daemon=True) scheduler.start() atexit.register(lambda: scheduler.shutdown(wait=False)) @github_app.on("team.created") def sync_new_team(): """ Sync a new team when it is created :return: """ owner = github_app.payload["organization"]["login"] team_id = github_app.payload["team"]["id"] if os.environ["USER_DIRECTORY"].upper() == "AAD": # Azure APIs don't currently support case insensitive searching slug = github_app.payload["team"]["name"].replace(" ", "-") else: slug = github_app.payload["team"]["slug"] client = github_app.installation_client sync_team(client=client, owner=owner, team_id=team_id, slug=slug) def sync_team(client=None, owner=None, team_id=None, slug=None): """ Prepare the team sync :param client: :param owner: :param team_id: :param slug: :return: """ print("-------------------------------") print(f"Processing Team: {slug}") try: org = client.organization(owner) team = org.team(team_id) custom_map, group_prefix, ignore_users = load_custom_map() try: directory_group = get_directory_from_slug(slug, custom_map, org) # If we're filtering on group prefix, skip if the group doesn't match if group_prefix.length() > 0 and not directory_group.startswith( tuple(group_prefix) ): print(f"skipping team {team.slug} - not in group prefix") return directory_members = directory_group_members(group=directory_group) except Exception as e: directory_members = [] traceback.print_exc(file=sys.stderr) team_members = github_team_members( client=client, owner=owner, team_id=team_id, attribute=USER_SYNC_ATTRIBUTE, ignore_users=ignore_users, ) compare = compare_members( group=directory_members, team=team_members, attribute=USER_SYNC_ATTRIBUTE ) if TEST_MODE: print(f"TEST_MODE: Pending changes for team {team.slug}:") print(json.dumps(compare, indent=2)) else: try: execute_sync(org=org, team=team, slug=slug, state=compare) except (AssertionError, ValueError) as e: if strtobool(os.environ["OPEN_ISSUE_ON_FAILURE"]): open_issue(client=client, slug=slug, message=e) raise Exception(f"Team {team.slug} sync failed: {e}") print(f"Processing Team Successful: {team.slug}") except Exception: traceback.print_exc(file=sys.stderr) raise def directory_group_members(group=None): """ Look up members of a group in your user directory :param group: The name of the group to query in your directory server :type group: str :return: group_members :rtype: list """ try: directory = DirectoryClient() members = directory.get_group_members(group_name=group) group_members = [member for member in members] except Exception as e: group_members = [] traceback.print_exc(file=sys.stderr) return group_members def github_team_info(client=None, owner=None, team_id=None): """ Look up team info in GitHub :param client: :param owner: :param team_id: :return: """ org = client.organization(owner) return org.team(team_id) def github_team_members( client=None, owner=None, team_id=None, attribute="username", ignore_users=[] ): """ Look up members of a given team in GitHub :param client: :param owner: :param team_id: :param attribute: :type owner: str :type team_id: int :type attribute: str :return: team_members :rtype: list """ team_members = [] team = github_team_info(client=client, owner=owner, team_id=team_id) if attribute == "email": for m in team.members(): user = client.user(m.login) team_members.append( { "username": str(user.login), "email": str(user.email), } ) else: for member in team.members(): team_members.append({"username": str(member), "email": ""}) return [m for m in team_members if m["username"] not in ignore_users] def compare_members(group, team, attribute="username"): """ Compare users in GitHub and the User Directory to see which users need to be added or removed :param group: :param team: :param attribute: :return: sync_state :rtype: dict """ directory_list = [x[attribute].casefold() for x in group] github_list = [x[attribute].casefold() for x in team] add_users = list(set(directory_list) - set(github_list)) remove_users = list(set(github_list) - set(directory_list)) sync_state = { "directory": group, "github": team, "action": {"add": add_users, "remove": remove_users}, } return sync_state def execute_sync(org, team, slug, state): """ Perform the synchronization :param org: :param team: :param slug: :param state: :return: """ total_changes = len(state["action"]["remove"]) + len(state["action"]["add"]) if len(state["directory"]) == 0: message = f"{os.environ.get('USER_DIRECTORY', 'LDAP').upper()} group returned empty: {slug}" raise ValueError(message) elif int(total_changes) > int(os.environ.get("CHANGE_THRESHOLD", 25)): message = "Skipping sync for {}.
".format(slug) message += "Total number of changes ({}) would exceed the change threshold ({}).".format( str(total_changes), str(os.environ.get("CHANGE_THRESHOLD", 25)) ) message += "
Please investigate this change and increase your threshold if this is accurate." raise AssertionError(message) else: for user in state["action"]["add"]: # Validate that user is in org if org.is_member(user) or ADD_MEMBER: try: print(f"Adding {user} to {slug}") team.add_or_update_membership(user) except github3.exceptions.NotFoundError: print(f"User: {user} not found") pass else: print(f"Skipping {user} as they are not part of the org") for user in state["action"]["remove"]: print(f"Removing {user} from {slug}") team.revoke_membership(user) def open_issue(client, slug, message): """ Open an issue with the failed sync details :param client: Our installation client :param slug: Team slug :param message: Error message to detail :return: """ repo_for_issues = os.environ["REPO_FOR_ISSUES"] owner = repo_for_issues.split("/")[0] repository = repo_for_issues.split("/")[1] assignee = os.environ["ISSUE_ASSIGNEE"] client.create_issue( owner=owner, repository=repository, assignee=assignee, title="Team sync failed for @{}/{}".format(owner, slug), body=str(message), ) def load_custom_map(file="syncmap.yml"): """ Custom team synchronization :param file: :return: """ syncmap = {} ignore_users = [] if os.path.isfile(file): from yaml import load, Loader with open(file, "r") as f: data = load(f, Loader=Loader) for d in data["mapping"]: if "org" in d: syncmap[(d["org"], d["github"])] = d["directory"] else: syncmap[d["github"]] = d["directory"] group_prefix = data.get("group_prefix", []) ignore_users = data.get("ignore_users", []) return (syncmap, group_prefix, ignore_users) def get_app_installations(): """ Get a list of installations for this app :return: """ with app.app_context() as ctx: try: c = ctx.push() gh = GitHubApp(c) installations = gh.app_client.app_installations finally: ctx.pop() return installations @scheduler.scheduled_job( trigger=CronTrigger.from_crontab(CRON_INTERVAL), id="sync_all_teams" ) def sync_all_teams(): """ Lookup teams in a GitHub org and synchronize all teams with your user directory :return: """ print(f'Syncing all teams: {time.strftime("%A, %d. %B %Y %I:%M:%S %p")}') installations = get_app_installations() custom_map, _ = load_custom_map() futures = [] install_count = 0 with ThreadPoolExecutor(max_workers=10) as exe: for i in installations(): install_count += 1 print("========================================================") print(f"## Processing Organization: {i.account['login']}") print("========================================================") with app.app_context() as ctx: try: gh = GitHubApp(ctx.push()) client = gh.app_installation(installation_id=i.id) org = client.organization(i.account["login"]) for team in org.teams(): futures.append( exe.submit(sync_team_helper, team, custom_map, client, org) ) except Exception as e: print(f"DEBUG: {e}") finally: ctx.pop() if not install_count: raise Exception(f"No installation defined for APP_ID {os.getenv('APP_ID')}") for future in futures: future.result() if REMOVE_ORG_MEMBERS_WITHOUT_TEAM: remove_org_members_without_team(installations) print(f'Syncing all teams successful: {time.strftime("%A, %d. %B %Y %I:%M:%S %p")}') def remove_org_members_without_team(installations): for i in installations(): with app.app_context() as ctx: try: gh = GitHubApp(ctx.push()) client = gh.app_installation(installation_id=i.id) org = client.organization(i.account["login"]) org_members = [member for member in org.members()] team_members = [ member for team in org.teams() for member in team.members() ] remove_members = list(set(org_members) - set(team_members)) for member in remove_members: print(f"Removing {member}") if not TEST_MODE: org.remove_membership(str(member)) except Exception as e: print(f"DEBUG: {e}") finally: ctx.pop() def sync_team_helper(team, custom_map, client, org): print(f"Organization: {org.login}") try: if SYNCMAP_ONLY and not is_team_in_map(team.slug, custom_map, org): print(f"skipping team {team.slug} - not in sync map") return sync_team( client=client, owner=org.login, team_id=team.id, slug=team.slug, ) except Exception as e: print(f"Organization: {org.login}") print(f"Unable to sync team: {team.slug}") print(f"DEBUG: {e}") def is_team_in_map(slug, custom_map, org): key_with_org = (org.login, slug) key_without_org = slug if key_with_org in custom_map or key_without_org in custom_map: return True else: return False def get_directory_from_slug(slug, custom_map, org): if not is_team_in_map(slug, custom_map, org): return slug elif (org.login, slug) in custom_map: return custom_map[(org.login, slug)] elif slug in custom_map: return custom_map[slug] if "FLASK_APP" in os.environ: thread = threading.Thread(target=sync_all_teams) thread.start() if __name__ == "__main__": if "FLASK_APP" in os.environ: app.run( host=os.environ.get("FLASK_RUN_HOST", "0.0.0.0"), port=os.environ.get("FLASK_RUN_PORT", "5000"), ) else: sync_all_teams()