Takes org list for promotion, filters orgs updated/output by this subset. Other refactoring

Author: aegilops

.gitignore

@@ -1,6 +1,5 @@
-# Token file used in local testing
-token.txt
-unmanaged_orgs.txt
+# Token file, orgs list and security managers list
+*.txt
 
 # CSV files
 *.csv

README.md

@@ -7,7 +7,8 @@ The scripts will give you a list of all organizations in the enterprise as a CSV
 > [!NOTE]
 > This is an _unofficial_ tool created by Field Security Specialists, and is not officially supported by GitHub.
 
-:information_source: This uses the [security manager role](https://docs.github.com/en/organizations/managing-peoples-access-to-your-organization-with-roles/managing-security-managers-in-your-organization) and parts of the GraphQL API that is available in GitHub.com (free/pro/teams and enterprise), as well as GitHub Enterprise Server versions 3.5 and higher.
+> [!NOTE]
+> This uses the [security manager role](https://docs.github.com/en/organizations/managing-peoples-access-to-your-organization-with-roles/managing-security-managers-in-your-organization) and parts of the GraphQL API that is available in GitHub.com (free/pro/teams and enterprise), as well as GitHub Enterprise Server versions 3.5 and higher.
 
 ## Scripts
 
@@ -30,19 +31,26 @@ The scripts will give you a list of all organizations in the enterprise as a CSV
     pip install -r requirements.txt
     ```
 
-1. Choose inputs as arguments to the script as follows:
+1. Choose inputs as arguments to the scripts as follows:
 
-    - the server URL (for GHES, EMU, or data residency) in `--github-url`
+    - The server URL (for GHES, EMU, or data residency) in `--github-url`.
       - For GHEC this is not required.
-    - call the script with the correct GitHub PAT
-      - place it in `GITHUB_TOKEN` in your environment, or
-      - create a file and save your token there to read it, and call the script with the `--token-file` argument
-    - use the enterprise slug as the first argument in the promote/demote scripts
-      - this is string URL version of the enterprise identity.  It's available in the enterprise admin url (for cloud and server), e.g. `https://github.com/enterprises/ENTERPRISE-SLUG-HERE`.
-    - for the security manager team script:
-      - use the list of orgs output by `org-admin-promote.py` in `--unmanaged-orgs`
-      - put the name of the security manager team and the team members to add in `--team-name` and `--team-members`.
-      - If you are using GHES 3.15 or below, please use the `--legacy` flag to use the legacy security managers API.
+    - Call the scripts with the correct GitHub PAT:
+      - Place it in `GITHUB_TOKEN` in your environment, or
+      - create a file and save your token there to read it, and call the script with the `--token-file` argument.
+    - See progress with the `--progress` flag.
+    - Promote/demote scripts:
+      - Limit the promotion to a subset of organization slugs/names using the `--orgs` or `--orgs-file` arguments.
+        - For `--orgs/-o`, list them space separated after the argument.
+        - For `--orgs-file/-f`, put a new-line separated list of organizations in a file and provide the path.
+      - Use the enterprise slug as the first argument:
+        - This is string URL version of the enterprise identity. It's available in the enterprise admin url (for cloud and server), e.g. `https://github.com/enterprises/ENTERPRISE-SLUG-HERE`.
+      - By default, a list of all of the organizations in scope, and the unmanaged set, will be output to `all_orgs.csv` and `unmanaged_orgs.txt` respectively.
+        - You can use the `--orgs-csv` and `--unmanaged-orgs` arguments to place these elsewhere.
+    - Security manager team script:
+      - Put the name of the security manager team and the team members to add in `--team-name` and `--team-members`.
+      - If you are using GHES 3.15 or below, use the `--legacy` flag to use the legacy security managers API.
+      - Use the list of orgs output by `org-admin-promote.py` in `--unmanaged-orgs`, if you changed the output path.
 
 1. Run them in the following order:
 

manage-sec-team.py

@@ -57,6 +57,11 @@ def add_args(parser) -> None:
         action="store_true",
         help="Use legacy API endpoints to manage the security managers",
     )
+    parser.add_argument(
+        "--progress",
+        action="store_true",
+        help="Show progress information",
+    )
     parser.add_argument(
         "--debug",
         "-d",
@@ -71,6 +76,7 @@ def make_security_managers_team(
     api_url: str,
     headers: dict[str, str],
     legacy=False,
+    progress=False,
 ) -> None:
     """Create or update the security managers team in the specified organization."""
     security_manager_role_id: str | None = None
@@ -105,7 +111,8 @@ def make_security_managers_team(
 
     # Create the team if it doesn't exist
     if sec_team_name not in teams_list:
-        LOG.info("Creating team {}".format(sec_team_name))
+        if progress:
+            LOG.info("Creating team {}".format(sec_team_name))
         try:
             teams.create_team(api_url, headers, org_name, sec_team_name)
         except Exception as e:
@@ -130,11 +137,12 @@ def make_security_managers_team(
                 security_manager_role_id,
                 legacy=legacy,
             )
-            LOG.info(
-                "✓ Team {} updated as a security manager for {}".format(
-                    sec_team_name, org_name
+            if progress:
+                LOG.info(
+                    "✓ Team {} updated as a security manager for {}".format(
+                        sec_team_name, org_name
+                    )
                 )
-            )
         else:
             LOG.debug(
                 "✓ Team {} already has the security manager role for {}".format(
@@ -153,14 +161,16 @@ def add_security_managers_to_team(
     sec_team_members: list[str],
     api_url: str,
     headers: dict[str, str],
+    progress: bool = False
 ) -> None:
     """Add security managers to the specified team in the organization."""
     # Get the list of org members, adding the missing ones to the org
     org_members = organizations.list_org_users(api_url, headers, org_name)
     org_members_list = [member["login"] for member in org_members]
     for username in sec_team_members:
         if username not in org_members_list:
-            LOG.info("Adding {} to {}".format(username, org_name))
+            if progress:
+                LOG.info("Adding {} to {}".format(username, org_name))
             try:
                 organizations.add_org_user(api_url, headers, org_name, username)
             except Exception as e:
@@ -232,8 +242,12 @@ def main() -> None:
 
     sec_team_members = []
     if args.sec_team_members_file:
-        with open(args.sec_team_members_file, "r") as f:
-            sec_team_members = [line.strip() for line in f if line.strip()]
+        sec_team_members = util.read_lines(args.sec_team_members_file)
+
+        if not sec_team_members:
+            LOG.error("⨯ No security team members found in file")
+            return
+
     elif args.sec_team_members:
         sec_team_members = args.sec_team_members
     else:
@@ -254,10 +268,10 @@ def main() -> None:
         org_name = org["login"]
 
         make_security_managers_team(
-            org_name, args.sec_team_name, api_url, headers, legacy=args.legacy
+            org_name, args.sec_team_name, api_url, headers, legacy=args.legacy, progress=args.progress
         )
         add_security_managers_to_team(
-            org_name, args.sec_team_name, sec_team_members, api_url, headers
+            org_name, args.sec_team_name, sec_team_members, api_url, headers, progress=args.progress
         )
 
 

org-admin-demote.py

@@ -40,12 +40,17 @@ def add_args(parser: ArgumentParser) -> None:
         required=False,
         help="File containing a GitHub Personal Access Token with admin:enterprise and read:org scope (or use GITHUB_TOKEN)",
     )
-
     parser.add_argument(
         "--unmanaged-orgs",
         default="unmanaged_orgs.txt",
         help="Path to newline-delimited list of organization IDs to demote from (default: unmanaged_orgs.txt)",
     )
+    parser.add_argument(
+        "--progress",
+        "-p",
+        action="store_true",
+        help="Show progress during demotion",
+    )
     parser.add_argument(
         "--debug",
         "-d",
@@ -54,24 +59,19 @@ def add_args(parser: ArgumentParser) -> None:
     )
 
 
-def read_unmanaged_org_ids(path: str) -> List[str]:
-    """Return a list of non-empty lines from the unmanaged orgs file."""
-    with open(path, "r", encoding="utf-8") as f:
-        return [line.strip() for line in f if line.strip()]
-
-
 def demote_admin(
-    api_url: str, headers: dict[str, str], enterprise_id: str, org_ids: Iterable[str]
+    api_url: str, headers: dict[str, str], enterprise_id: str, org_ids: Iterable[str], progress: bool = False
 ) -> None:
     """Demote the enterprise admin from each organization ID provided."""
     org_ids_list = list(org_ids)
     LOG.info("Total count of orgs to demote admin from: {}".format(len(org_ids_list)))
     for i, org_id in enumerate(org_ids_list):
-        LOG.info(
-            "Removing from organization: {} [{}/{}]".format(
-                org_id, i + 1, len(org_ids_list)
+        if progress:
+            LOG.info(
+                "Removing from organization: {} [{}/{}]".format(
+                    org_id, i + 1, len(org_ids_list)
+                )
             )
-        )
         enterprises.promote_admin(
             api_url, headers, enterprise_id, org_id, "UNAFFILIATED"
         )
@@ -101,8 +101,13 @@ def main() -> None:
         api_url, args.enterprise_slug, headers
     )
 
-    unmanaged_orgs = read_unmanaged_org_ids(args.unmanaged_orgs)
-    demote_admin(api_url, headers, enterprise_id, unmanaged_orgs)
+    unmanaged_orgs = util.read_lines(args.unmanaged_orgs)
+
+    if not unmanaged_orgs:
+        LOG.error("⨯ No unmanaged organizations found to demote admin from")
+        return
+
+    demote_admin(api_url, headers, enterprise_id, unmanaged_orgs, args.progress)
 
 
 if __name__ == "__main__":  # pragma: no cover

org-admin-promote.py

@@ -32,6 +32,19 @@ def add_args(parser: ArgumentParser) -> None:
         "enterprise_slug",
         help="Enterprise slug (after /enterprises/ in URL)",
     )
+    parser.add_argument(
+        "--orgs",
+        "-o",
+        nargs="*",
+        required=False,
+        help="List of organization slugs to promote on (default: all organizations)",
+    )
+    parser.add_argument(
+        "--orgs-file",
+        "-f",
+        required=False,
+        help="Path to file containing organization slugs to promote - line separated (default: all organizations)",
+    )
     parser.add_argument(
         "--github-url",
         required=False,
@@ -52,6 +65,12 @@ def add_args(parser: ArgumentParser) -> None:
         default="all_orgs.csv",
         help="Output CSV file listing all organizations in the enterprise (default: all_orgs.csv)",
     )
+    parser.add_argument(
+        "--progress",
+        "-p",
+        action="store_true",
+        help="Show progress during promotion",
+    )
     parser.add_argument(
         "--debug",
         "-d",
@@ -70,10 +89,17 @@ def write_unmanaged_orgs(path: str, unmanaged_org_ids: List[str]) -> None:
 
 
 def promote_all(
-    api_url: str, headers: dict[str, str], enterprise_slug: str, unmanaged_out: str
+    api_url: str,
+    headers: dict[str, str],
+    enterprise_slug: str,
+    orgs_subset: set[str] | None,
+    unmanaged_out: str,
+    progress: bool = False,
 ) -> List[str] | None:
     """
     Promote the enterprise admin to owner on all unmanaged organizations.
+
+    If a subset of organizations is provided, only attempt to promote on those; otherwise, try on all organizations.
     """
     total_org_count = organizations.get_total_count(api_url, enterprise_slug, headers)
     if total_org_count == 0:
@@ -85,31 +111,32 @@ def promote_all(
             "⨯ Total count of organizations returned by the query is different from the expected count"
         )
         return None
-    LOG.info(
-        "Total count of organizations returned by the query is: {}".format(
-            total_org_count
-        )
-    )
+    LOG.info("Total organizations: {}".format(total_org_count))
+
+    if orgs_subset is not None:
+        orgs = [org for org in orgs if org["node"]["login"] in orgs_subset]
+
+        LOG.info("Organizations in scope: {}".format(len(orgs)))
+
     enterprise_id = enterprises.get_enterprise_id(api_url, enterprise_slug, headers)
     unmanaged_orgs = [
         org["node"]["id"] for org in orgs if not org["node"]["viewerCanAdminister"]
     ]
-    LOG.info(
-        "Total count of unmanaged organizations to be promoted on: {}".format(
-            len(unmanaged_orgs)
-        )
-    )
+    if not unmanaged_orgs:
+        LOG.info("No organizations to promote on")
+        return []
+
+    LOG.info("Unmanaged organizations to promote on: {}".format(len(unmanaged_orgs)))
     for i, org_id in enumerate(unmanaged_orgs):
-        LOG.info(
-            "Promoting to owner on organization: {} [{}/{}]".format(
-                org_id, i + 1, len(unmanaged_orgs)
+        if progress:
+            LOG.info(
+                "Promoting to owner on organization: {} [{}/{}]".format(
+                    org_id, i + 1, len(unmanaged_orgs)
+                )
             )
-        )
         enterprises.promote_admin(api_url, headers, enterprise_id, org_id, "OWNER")
     write_unmanaged_orgs(unmanaged_out, unmanaged_orgs)
-    LOG.info(
-        "Total count of newly managed organizations is: {}".format(len(unmanaged_orgs))
-    )
+    LOG.info("Promoted on organizations: {}".format(len(unmanaged_orgs)))
     return unmanaged_orgs
 
 
@@ -132,12 +159,21 @@ def main() -> None:
         "Authorization": f"token {github_pat}",
     }
 
+    orgs_subset_list: list[str] | None = (
+        args.orgs or util.read_lines(args.orgs_file) or None
+    )
+    orgs_subset: set[str] | None = (
+        set(orgs_subset_list) if orgs_subset_list is not None else None
+    )
+
     if (
         promote_all(
             api_url,
             headers,
             args.enterprise_slug,
+            orgs_subset,
             args.unmanaged_orgs,
+            args.progress
         )
         is None
     ):
@@ -146,6 +182,11 @@ def main() -> None:
 
     # Refresh and write all orgs CSV after promotions
     orgs = organizations.list_orgs(api_url, args.enterprise_slug, headers)
+
+    # Filter by the list of orgs, if provided
+    if orgs_subset is not None:
+        orgs = [org for org in orgs if org["node"]["login"] in orgs_subset]
+
     organizations.write_orgs_to_csv(orgs, args.orgs_csv)
 
 

src/util.py

@@ -4,6 +4,10 @@
 
 import os
 from urllib.parse import urlparse
+import logging
+
+
+LOG = logging.getLogger(__name__)
 
 
 def read_token(token_file: str | None) -> str | None:
@@ -27,6 +31,21 @@ def read_token(token_file: str | None) -> str | None:
     return token
 
 
+def read_lines(input_path: str | None) -> list[str] | None:
+    """
+    Read a file and return a list of lines.
+    """
+    if not input_path:
+        return None
+
+    try:
+        with open(input_path, "r", encoding="utf-8") as f:
+            return [line.strip() for line in f.readlines() if line.strip()]
+    except (FileNotFoundError, IsADirectoryError) as err:
+        LOG.error(f"⨯ File error: {input_path}: {err}")
+        raise err
+
+
 def add_request_headers(headers: dict[str, str]) -> dict[str, str]:
     """
     Add required headers to the request headers.