diff --git a/README.md b/README.md index 124e81b..77b6134 100644 --- a/README.md +++ b/README.md @@ -20,6 +20,7 @@ The basic principle: all settings reside in YAML configuration files which will * Manage GitHub organization owners * Manage GitHub teams, their members, maintainers and settings * Support of parent/child teams +* Assign GitHub organization roles to teams, e.g. `app_manager` * Manage teams' permissions on organizations' repositories * Invite members to the organization if they aren't part of it yet * Warn about unmanaged teams @@ -64,6 +65,7 @@ Access tokens and apps need the following permissions: * Metadata: read * Organization permissions: * Administration: read and write + * Custom organization roles: read * Members: read and write You can set the required secrets in `config/app.yaml` or via environment variables (`GITHUB_TOKEN` or `GITHUB_APP_ID` and `GITHUB_APP_PRIVATE_KEY`). diff --git a/config/example/teams/myteams.yaml b/config/example/teams/myteams.yaml index 5ea61b1..67c3c46 100644 --- a/config/example/teams/myteams.yaml +++ b/config/example/teams/myteams.yaml @@ -27,6 +27,13 @@ Project Maintainers: # Make this a child team of the provided parent. Note: You should first # describe parents in this file, then children parent: Maintainers + # Organisation roles that shall be assigned to this team. Any other + # organisation roles of the team are removed. Predefined roles are: + # all_repo_read, all_repo_triage, all_repo_write, all_repo_maintain, + # all_repo_admin, app_manager, ci_cd_admin, security_manager, + # open_source_license_manager + roles: + - app_manager member: - octocat repos: diff --git a/gh_org_mgr/_config.py b/gh_org_mgr/_config.py index a8d7cb7..abcebbf 100644 --- a/gh_org_mgr/_config.py +++ b/gh_org_mgr/_config.py @@ -82,6 +82,9 @@ "oneOf": [{"type": "null"}, {"type": "array", "items": {"type": "string"}}] }, "parent": {"type": "string"}, + "roles": { + "oneOf": [{"type": "null"}, {"type": "array", "items": {"type": "string"}}] + }, "repos": { "type": "object", "propertyNames": {"type": "string"}, diff --git a/gh_org_mgr/_gh_api.py b/gh_org_mgr/_gh_api.py index aa3d8c1..a20b895 100644 --- a/gh_org_mgr/_gh_api.py +++ b/gh_org_mgr/_gh_api.py @@ -25,6 +25,32 @@ def get_github_secrets_from_env(env_variable: str, secret: str | int) -> str: return str(secret) +def run_rest_api_request( + method: str, url: str, token: str +) -> tuple[HTTPStatus, dict | list | None]: + """Run a request against the GitHub REST API. Returns the HTTP status code + and the JSON body of the response, or None if there was no body (e.g. for + successful PUT/DELETE requests). + """ + headers = { + "Authorization": f"Bearer {token}", + "Accept": "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28", + } + try: + request = requests.request(method=method, url=url, headers=headers, timeout=10) + except requests.exceptions.RequestException: + logging.exception("Request to the GitHub API failed") + sys.exit(1) + + # Get JSON result. Note that some requests (e.g. successful PUT/DELETE) have no body + json_return: dict | list | None = None + with contextlib.suppress(requests.exceptions.JSONDecodeError): + json_return = request.json() + + return HTTPStatus(request.status_code), json_return + + # Function to execute GraphQL query def run_graphql_query(query: str, variables: dict[str, Any], token: str) -> dict | str: """Run a query against the GitHub GraphQL API.""" diff --git a/gh_org_mgr/_gh_org.py b/gh_org_mgr/_gh_org.py index cc9a25c..4c0ab47 100644 --- a/gh_org_mgr/_gh_org.py +++ b/gh_org_mgr/_gh_org.py @@ -7,6 +7,7 @@ import logging import sys from dataclasses import asdict, dataclass, field +from http import HTTPStatus from github import Auth, Github, GithubIntegration from github.GithubException import ( @@ -21,7 +22,7 @@ from github.Team import Team from jwt.exceptions import InvalidKeyError -from ._gh_api import get_github_secrets_from_env, run_graphql_query +from ._gh_api import get_github_secrets_from_env, run_graphql_query, run_rest_api_request from ._helpers import ( compare_two_dicts, compare_two_lists, @@ -47,6 +48,8 @@ class GHorg: current_teams: dict[Team, dict] = field(default_factory=dict) current_teams_str: list[str] = field(default_factory=list) configured_teams: dict[str, dict | None] = field(default_factory=dict) + org_roles: dict[str, int] = field(default_factory=dict) + team_slugs: dict[str, str] = field(default_factory=dict) newly_added_users: list[NamedUser] = field(default_factory=list) current_repos_teams: dict[Repository, dict[Team, str]] = field(default_factory=dict) graphql_repos_collaborators: dict[str, list[dict]] = field(default_factory=dict) @@ -732,6 +735,155 @@ def get_members_without_team( for user in members_without_team: self.stats.remove_member_without_team(user=user.login, removed=False) + # -------------------------------------------------------------------------- + # Organisation roles + # -------------------------------------------------------------------------- + def _org_roles_api_request(self, method: str, path: str = "") -> dict | list: + """Run a request against the organisation roles API and return the JSON + response. Exits the program if the request fails. + """ + url = f"{self.org.url}/organization-roles{path}" + status, response = run_rest_api_request(method, url, self.gh_token) + if status in (HTTPStatus.OK, HTTPStatus.NO_CONTENT): + return response if response is not None else {} + + logging.critical( + "Request to the organisation roles API failed: %s %s (HTTP status %s). " + "Does your token have permission to read/manage organisation roles? Response: %s", + method, + url, + status, + response, + ) + sys.exit(1) + + def _get_current_org_roles(self) -> None: + """Get all organisation roles of the organisation as a name-to-ID dict.""" + response = self._org_roles_api_request("GET", "?per_page=100") + if not isinstance(response, dict): + logging.critical("Unexpected organisation roles response: %s", response) + sys.exit(1) + self.org_roles = {role["name"]: role["id"] for role in response["roles"]} + logging.debug("Found %s organisation roles: %s", len(self.org_roles), self.org_roles) + + def _get_current_teams_org_roles(self) -> dict[str, set[str]]: + """Get the current organisation roles of all teams. As the API only + allows querying the teams per role, iterate over all roles and invert + the result into a dict of team slug to set of role names. + """ + teams_roles: dict[str, set[str]] = {} + for role_name, role_id in self.org_roles.items(): + response = self._org_roles_api_request("GET", f"/{role_id}/teams?per_page=100") + if not isinstance(response, list): + logging.critical("Unexpected organisation role teams response: %s", response) + sys.exit(1) + for team in response: + teams_roles.setdefault(team["slug"], set()).add(role_name) + + return teams_roles + + def validate_configured_team_roles(self) -> None: + """Check that all configured team roles exist in the organisation. + Fail fast before any write operations happen. + """ + # Collect all configured roles. If none are configured, skip validation + configured_roles: dict[str, list[str]] = {} + for team_name, team_config in self.configured_teams.items(): + if team_config and (roles := team_config.get("roles")): + configured_roles[team_name] = list(roles) + if not configured_roles: + logging.debug("No organisation roles configured for any team. Skipping validation") + return + + self._get_current_org_roles() + + # Find all configured roles that do not exist in the organisation + invalid_roles: list[str] = [ + f"Team '{team_name}': role '{role}'" + for team_name, roles in configured_roles.items() + for role in roles + if role not in self.org_roles + ] + + if invalid_roles: + logging.critical( + "The following configured organisation roles do not exist in the " + "organisation '%s'. Cannot continue:\n%s", + self.org.login, + "\n".join(f"- {item}" for item in invalid_roles), + ) + sys.exit(1) + + def _get_team_slug(self, team_name: str) -> str | None: + """Get the slug of a team from the GitHub API. The name-to-slug mapping + is fetched once and cached. Never derive the slug from the team name. + Returns None if the team does not exist. + """ + if not self.team_slugs: + # Reuse the team objects if they have been fetched already, + # otherwise get them from the API + if self.current_teams: + self.team_slugs = {team.name: team.slug for team in self.current_teams} + else: + for team in self.org.get_teams(): + self.team_slugs[team.name] = team.slug + + return self.team_slugs.get(team_name) + + def sync_org_roles(self, dry: bool = False) -> None: + """Synchronise the organisation roles of all teams. + """ + # Ensure we know the role IDs of the organisation + if not self.org_roles: + self._get_current_org_roles() + + # Get the current organisation roles of all teams. Needs one API + # request per organisation role + current_teams_roles = self._get_current_teams_org_roles() + + for team_name, team_config in self.configured_teams.items(): + configured_roles = set((team_config or {}).get("roles") or []) + + # Get the team's slug from the API. If the team does not exist, + # fail fast in a real run; in a dry-run it may still be created + team_slug = self._get_team_slug(team_name) + if team_slug is None: + if not dry: + logging.critical( + "The team '%s' does not exist in the organisation '%s'. Cannot continue.", + team_name, + self.org.login, + ) + sys.exit(1) + logging.info( + "Team '%s' does not exist yet, probably because it should be created " + "but this is a dry-run. Treating its current organisation roles as empty", + team_name, + ) + current_roles: set[str] = set() + else: + current_roles = current_teams_roles.get(team_slug, set()) + + if configured_roles == current_roles: + logging.info("Organisation roles of team '%s' are in sync, no changes", team_name) + continue + + # Assign missing roles to the team + for role in configured_roles - current_roles: + logging.info("Assigning organisation role '%s' to team '%s'", role, team_name) + self.stats.add_org_role_to_team(team=team_name, role=role) + if not dry: + self._org_roles_api_request("PUT", f"/teams/{team_slug}/{self.org_roles[role]}") + + # Remove unconfigured roles from the team + for role in current_roles - configured_roles: + logging.info("Removing organisation role '%s' from team '%s'", role, team_name) + self.stats.remove_org_role_from_team(team=team_name, role=role) + if not dry: + self._org_roles_api_request( + "DELETE", f"/teams/{team_slug}/{self.org_roles[role]}" + ) + # -------------------------------------------------------------------------- # Repos # -------------------------------------------------------------------------- diff --git a/gh_org_mgr/_setup_team.py b/gh_org_mgr/_setup_team.py index 2463d64..93ddae1 100644 --- a/gh_org_mgr/_setup_team.py +++ b/gh_org_mgr/_setup_team.py @@ -14,6 +14,7 @@ ${team_name}: # parent: # repos: + # roles: # maintainer: member: """ diff --git a/gh_org_mgr/_stats.py b/gh_org_mgr/_stats.py index b10aa8a..0f83ee9 100644 --- a/gh_org_mgr/_stats.py +++ b/gh_org_mgr/_stats.py @@ -22,6 +22,8 @@ class TeamChanges: changed_members_role: list[str] = field(default_factory=list) removed_members: list[str] = field(default_factory=list) pending_members: list[str] = field(default_factory=list) + added_org_roles: list[str] = field(default_factory=list) + removed_org_roles: list[str] = field(default_factory=list) @dataclass @@ -105,6 +107,17 @@ def remove_member_without_team(self, user: str, removed: bool) -> None: if removed: self.removed_members.append(user) + # -------------------------------------------------------------------------- + # Organisation roles + # -------------------------------------------------------------------------- + def add_org_role_to_team(self, team: str, role: str) -> None: + """Organisation role has been assigned to a team.""" + self.update_team(team_name=team, added_org_roles=role) + + def remove_org_role_from_team(self, team: str, role: str) -> None: + """Organisation role has been removed from a team.""" + self.update_team(team_name=team, removed_org_roles=role) + # -------------------------------------------------------------------------- # Repos # -------------------------------------------------------------------------- @@ -233,6 +246,14 @@ def print_changes( # noqa: C901, PLR0912, PLR0915 output += " ā³ Pending members:\n" for item in tchanges.pending_members: output += f" - {item}\n" + if tchanges.added_org_roles: + output += " šŸŽ­ Added organisation roles:\n" + for item in tchanges.added_org_roles: + output += f" - {item}\n" + if tchanges.removed_org_roles: + output += " šŸŽ­ Removed organisation roles:\n" + for item in tchanges.removed_org_roles: + output += f" - {item}\n" if self.repos: output += "\nšŸ“‚ Repository Changes:\n" for repo, rchanges in self.repos.items(): diff --git a/gh_org_mgr/manage.py b/gh_org_mgr/manage.py index 6c0de9c..14b2bd5 100644 --- a/gh_org_mgr/manage.py +++ b/gh_org_mgr/manage.py @@ -5,6 +5,7 @@ """Manage a GitHub Organization, its teams, repository permissions, and more.""" import argparse +import contextlib import logging import sys @@ -93,6 +94,9 @@ def main() -> None: """Main function.""" + with contextlib.suppress(AttributeError): + sys.stdout.reconfigure(encoding="utf-8") # ty: ignore[unresolved-attribute] + # Process arguments args = parser.parse_args() @@ -130,6 +134,10 @@ def main() -> None: # Get current rate limit org.ratelimit() + # Validate configured team roles before any changes are made + log_progress("Validating configured team roles...") + org.validate_configured_team_roles() + # Synchronise organisation owners log_progress("Synchronising organisation owners...") org.sync_org_owners(dry=args.dry, force=args.force) @@ -145,6 +153,9 @@ def main() -> None: # Synchronise the team memberships log_progress("Synchronising team memberships...") org.sync_teams_members(dry=args.dry) + # Synchronise the organisation roles of the teams + log_progress("Synchronising team organisation roles...") + org.sync_org_roles(dry=args.dry) # Report and act on teams that are not configured locally log_progress("Checking for unconfigured teams...") org.get_and_delete_unconfigured_teams(