Taking a cert bundle for verification from the args (#30)

* Taking a cert bundle for verification from the args

* Formatted with Black

Author: aegilops

README.md

@@ -98,3 +98,15 @@ Removing ghe-admin from security-managers
   - [requests](https://pypi.org/project/requests/) is a simple and popular HTTP library.
   - [defusedcsv](https://github.com/raphaelm/defusedcsv) is used over `csv` to mitigate spreadsheet application exploitation in older versions.
 - The `.csv` files and `.txt` files are in the `.gitignore` file to avoid accidental commits into the repo.
+
+## TLS / custom certificates
+
+If you are running against a GitHub Enterprise Server instance that uses a self‑signed certificate or an internal certificate authority, you can provide a custom root CA (or certificate bundle) with the `--ca-bundle` argument on all three scripts:
+
+```shell
+./org-admin-promote.py ENTERPRISE_SLUG --github-url https://ghe.example.com --ca-bundle /path/to/internal-ca.pem
+./manage-sec-team.py --github-url https://ghe.example.com --ca-bundle /path/to/internal-ca.pem --sec-team-members alice bob
+./org-admin-demote.py ENTERPRISE_SLUG --github-url https://ghe.example.com --ca-bundle /path/to/internal-ca.pem
+```
+
+The value passed must be a readable PEM file containing the certificate(s). If omitted, the Python default trust store is used.

manage-sec-team.py

@@ -68,6 +68,11 @@ def add_args(parser) -> None:
         action="store_true",
         help="Enable debug logging",
     )
+    parser.add_argument(
+        "--ca-bundle",
+        required=False,
+        help="Path to a custom CA certificate or bundle (PEM) to trust for TLS (self-signed/internal CAs)",
+    )
 
 
 def make_security_managers_team(
@@ -77,13 +82,14 @@ def make_security_managers_team(
     headers: dict[str, str],
     legacy=False,
     progress=False,
+    verify: str | bool | None = True,
 ) -> None:
     """Create or update the security managers team in the specified organization."""
     security_manager_role_id: str | None = None
 
     if not legacy:
         org_roles: dict[str, Any] = organizations.list_org_roles(
-            api_url, headers, org_name
+            api_url, headers, org_name, verify=verify
         )
 
         # Check if the "security manager" role exists
@@ -106,15 +112,15 @@ def make_security_managers_team(
         security_manager_role_id = security_manager_role_id_list[0]
 
     # Get the list of teams
-    teams_info = teams.list_teams(api_url, headers, org_name)
+    teams_info = teams.list_teams(api_url, headers, org_name, verify=verify)
     teams_list = [team["name"] for team in teams_info]
 
     # Create the team if it doesn't exist
     if sec_team_name not in teams_list:
         if progress:
             LOG.info("Creating team {}".format(sec_team_name))
         try:
-            teams.create_team(api_url, headers, org_name, sec_team_name)
+            teams.create_team(api_url, headers, org_name, sec_team_name, verify=verify)
         except Exception as e:
             LOG.error("⨯ Failed to create team {}: {}".format(sec_team_name, e))
 
@@ -128,6 +134,7 @@ def make_security_managers_team(
             sec_team_name,
             security_manager_role_id,
             legacy=legacy,
+            verify=verify,
         ):
             teams.change_team_role(
                 api_url,
@@ -136,6 +143,7 @@ def make_security_managers_team(
                 sec_team_name,
                 security_manager_role_id,
                 legacy=legacy,
+                verify=verify,
             )
             if progress:
                 LOG.info(
@@ -162,17 +170,22 @@ def add_security_managers_to_team(
     api_url: str,
     headers: dict[str, str],
     progress: bool = False,
+    verify: str | bool | None = True,
 ) -> 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 = organizations.list_org_users(
+        api_url, headers, org_name, verify=verify
+    )
     org_members_list = [member["login"] for member in org_members]
     for username in sec_team_members:
         if username not in org_members_list:
             if progress:
                 LOG.info("Adding {} to {}".format(username, org_name))
             try:
-                organizations.add_org_user(api_url, headers, org_name, username)
+                organizations.add_org_user(
+                    api_url, headers, org_name, username, verify=verify
+                )
             except Exception as e:
                 LOG.error(
                     "⨯ Failed to add user {} to org {}: {}".format(
@@ -182,15 +195,17 @@ def add_security_managers_to_team(
                 return
 
     # Get the list of team members, adding the missing ones to the team and removing the extra ones
-    team_members = teams.list_team_members(api_url, headers, org_name, sec_team_name)
+    team_members = teams.list_team_members(
+        api_url, headers, org_name, sec_team_name, verify=verify
+    )
     team_members_list = [member["login"] for member in team_members]
     for username in team_members_list:
         if username not in sec_team_members:
             if progress:
                 LOG.info("Removing {} from {}".format(username, sec_team_name))
             try:
                 teams.remove_team_member(
-                    api_url, headers, org_name, sec_team_name, username
+                    api_url, headers, org_name, sec_team_name, username, verify=verify
                 )
             except Exception as e:
                 LOG.error(
@@ -205,7 +220,7 @@ def add_security_managers_to_team(
                 LOG.info("Adding {} to {}".format(username, sec_team_name))
             try:
                 teams.add_team_member(
-                    api_url, headers, org_name, sec_team_name, username
+                    api_url, headers, org_name, sec_team_name, username, verify=verify
                 )
             except Exception as e:
                 LOG.error(
@@ -260,6 +275,13 @@ def main() -> None:
 
     api_url = util.rest_api_url_from_server_url(args.github_url)
 
+    # Optional custom CA bundle / cert file
+    verify: str | bool | None = True
+    try:
+        verify = util.validate_ca_bundle(args.ca_bundle)
+    except FileNotFoundError:
+        return
+
     # Set up the headers
     headers = {
         "Authorization": "token {}".format(github_pat),
@@ -276,6 +298,7 @@ def main() -> None:
             headers,
             legacy=args.legacy,
             progress=args.progress,
+            verify=verify,
         )
         add_security_managers_to_team(
             org_name,
@@ -284,6 +307,7 @@ def main() -> None:
             api_url,
             headers,
             progress=args.progress,
+            verify=verify,
         )
 
 

org-admin-demote.py

@@ -57,6 +57,11 @@ def add_args(parser: ArgumentParser) -> None:
         action="store_true",
         help="Enable debug logging",
     )
+    parser.add_argument(
+        "--ca-bundle",
+        required=False,
+        help="Path to a custom CA certificate or bundle (PEM) for TLS verification (self-signed/internal roots)",
+    )
 
 
 def demote_admin(
@@ -65,6 +70,7 @@ def demote_admin(
     enterprise_id: str,
     org_ids: Iterable[str],
     progress: bool = False,
+    verify: str | bool | None = True,
 ) -> None:
     """Demote the enterprise admin from each organization ID provided."""
     org_ids_list = list(org_ids)
@@ -77,7 +83,7 @@ def demote_admin(
                 )
             )
         enterprises.promote_admin(
-            api_url, headers, enterprise_id, org_id, "UNAFFILIATED"
+            api_url, headers, enterprise_id, org_id, "UNAFFILIATED", verify=verify
         )
 
 
@@ -101,8 +107,15 @@ def main() -> None:
         "Authorization": f"token {github_pat}",
     }
 
+    # Optional custom CA bundle / cert file
+    verify: str | bool | None = True
+    try:
+        verify = util.validate_ca_bundle(args.ca_bundle)
+    except FileNotFoundError:
+        return
+
     enterprise_id = enterprises.get_enterprise_id(
-        api_url, args.enterprise_slug, headers
+        api_url, args.enterprise_slug, headers, verify=verify
     )
 
     unmanaged_orgs = util.read_lines(args.unmanaged_orgs)
@@ -111,7 +124,9 @@ def main() -> None:
         LOG.error("⨯ No unmanaged organizations found to demote admin from")
         return
 
-    demote_admin(api_url, headers, enterprise_id, unmanaged_orgs, args.progress)
+    demote_admin(
+        api_url, headers, enterprise_id, unmanaged_orgs, args.progress, verify=verify
+    )
 
 
 if __name__ == "__main__":  # pragma: no cover

org-admin-promote.py

@@ -77,6 +77,11 @@ def add_args(parser: ArgumentParser) -> None:
         action="store_true",
         help="Enable debug logging",
     )
+    parser.add_argument(
+        "--ca-bundle",
+        required=False,
+        help="Path to a custom CA certificate or bundle (PEM) for TLS verification (self-signed/internal roots)",
+    )
 
 
 def write_unmanaged_orgs(path: str, unmanaged_org_ids: List[str]) -> None:
@@ -95,17 +100,20 @@ def promote_all(
     orgs_subset: set[str] | None,
     unmanaged_out: str,
     progress: bool = False,
+    verify: str | bool | None = True,
 ) -> 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)
+    total_org_count = organizations.get_total_count(
+        api_url, enterprise_slug, headers, verify=verify
+    )
     if total_org_count == 0:
         LOG.warning("⚠️ No organizations found.")
         return []
-    orgs = organizations.list_orgs(api_url, enterprise_slug, headers)
+    orgs = organizations.list_orgs(api_url, enterprise_slug, headers, verify=verify)
     if len(orgs) != total_org_count:
         LOG.error(
             "⨯ Total count of organizations returned by the query is different from the expected count"
@@ -118,7 +126,9 @@ def promote_all(
 
         LOG.info("Organizations in scope: {}".format(len(orgs)))
 
-    enterprise_id = enterprises.get_enterprise_id(api_url, enterprise_slug, headers)
+    enterprise_id = enterprises.get_enterprise_id(
+        api_url, enterprise_slug, headers, verify=verify
+    )
     unmanaged_orgs = [
         org["node"]["id"] for org in orgs if not org["node"]["viewerCanAdminister"]
     ]
@@ -134,7 +144,9 @@ def promote_all(
                     org_id, i + 1, len(unmanaged_orgs)
                 )
             )
-        enterprises.promote_admin(api_url, headers, enterprise_id, org_id, "OWNER")
+        enterprises.promote_admin(
+            api_url, headers, enterprise_id, org_id, "OWNER", verify=verify
+        )
     write_unmanaged_orgs(unmanaged_out, unmanaged_orgs)
     LOG.info("Promoted on organizations: {}".format(len(unmanaged_orgs)))
     return unmanaged_orgs
@@ -159,6 +171,13 @@ def main() -> None:
         "Authorization": f"token {github_pat}",
     }
 
+    # Optional custom CA bundle / cert file
+    verify: str | bool | None = True
+    try:
+        verify = util.validate_ca_bundle(args.ca_bundle)
+    except FileNotFoundError:
+        return
+
     orgs_subset_list: list[str] | None = (
         args.orgs or util.read_lines(args.orgs_file) or None
     )
@@ -174,14 +193,17 @@ def main() -> None:
             orgs_subset,
             args.unmanaged_orgs,
             args.progress,
+            verify=verify,
         )
         is None
     ):
         LOG.error("⨯ Promotion failed")
         return
 
     # Refresh and write all orgs CSV after promotions
-    orgs = organizations.list_orgs(api_url, args.enterprise_slug, headers)
+    orgs = organizations.list_orgs(
+        api_url, args.enterprise_slug, headers, verify=verify
+    )
 
     # Filter by the list of orgs, if provided
     if orgs_subset is not None:

src/enterprises.py

@@ -10,7 +10,10 @@
 
 
 def get_enterprise_id(
-    api_endpoint: str, enterprise_slug: str, headers: dict[str, str]
+    api_endpoint: str,
+    enterprise_slug: str,
+    headers: dict[str, str],
+    verify: str | bool | None = True,
 ) -> str:
     """
     Get the ID of an enterprise by its slug.
@@ -28,6 +31,7 @@ def get_enterprise_id(
         api_endpoint,
         json={"query": enterprise_query},
         headers=add_request_headers(headers),
+        verify=verify,
     )
     response.raise_for_status()
     return response.json()["data"]["enterprise"]["id"]
@@ -58,6 +62,7 @@ def promote_admin(
     enterprise_id: str,
     org_id: str,
     role: str,
+    verify: str | bool | None = True,
 ) -> dict[str, Any]:
     """
     Promote an enterprise admin to an organization owner.
@@ -67,6 +72,7 @@ def promote_admin(
         api_endpoint,
         json={"query": promote_query},
         headers=add_request_headers(headers),
+        verify=verify,
     )
     response.raise_for_status()
     return response.json()