audit-labs/audit-tools

  1"""
  2Collect org membership, access, and permission data.
  3
  4Covers:
  5- Org member roster with roles (owner vs member)
  6- Outside collaborators
  7- Privileged access (admin permission on any repo)
  8- Pending org invitations
  9- Team memberships and repo permissions
 10- Full per-repo permission matrix
 11- Members with 2FA disabled
 12"""
 13
 14import sys
 15from datetime import datetime, timezone
 16
 17import requests
 18
 19from .api import paginate
 20
 21
 22def _permission_level(perms):
 23    for level in ("admin", "maintain", "push", "triage", "pull"):
 24        if perms.get(level):
 25            return "write" if level == "push" else level
 26    return "unknown"
 27
 28
 29def _repos(org, cfg):
 30    return paginate(f"https://api.github.com/orgs/{org}/repos", cfg)
 31
 32
 33def fetch_repo_collaborators(org, cfg):
 34    """
 35    Fetch collaborators for every repo once (affiliation=all).
 36    Returns a list of dicts: {repo, visibility, collaborators}.
 37    Repos that 403 are skipped with a warning.
 38    This cache is passed into outside_collaborators, privileged_access,
 39    and permission_matrix to avoid redundant API calls.
 40    """
 41    repos = _repos(org, cfg)
 42    results = []
 43    for repo in repos:
 44        repo_name = repo["name"]
 45        try:
 46            collabs = paginate(
 47                f"https://api.github.com/repos/{org}/{repo_name}/collaborators",
 48                cfg,
 49                {"affiliation": "all"},
 50            )
 51        except requests.HTTPError as e:
 52            if e.response is not None and e.response.status_code == 403:
 53                print(
 54                    f"  Skipping {repo_name}: collaborators endpoint returned 403",
 55                    file=sys.stderr,
 56                )
 57                continue
 58            raise
 59        results.append(
 60            {
 61                "repo": repo_name,
 62                "visibility": repo["visibility"],
 63                "collaborators": collabs,
 64            }
 65        )
 66    return results
 67
 68
 69def member_roster(org, cfg):
 70    members = paginate(
 71        f"https://api.github.com/orgs/{org}/members", cfg, {"role": "all"}
 72    )
 73    owners = {
 74        m["login"]
 75        for m in paginate(
 76            f"https://api.github.com/orgs/{org}/members", cfg, {"role": "owner"}
 77        )
 78    }
 79    return [
 80        {
 81            "login": m["login"],
 82            "org_role": "owner" if m["login"] in owners else "member",
 83            "profile_url": m["html_url"],
 84        }
 85        for m in members
 86    ]
 87
 88
 89def two_factor_disabled(org, cfg):
 90    """
 91    List org members who do not have 2FA enabled.
 92    Requires the token to have read:org scope.
 93    Note: GitHub only exposes this to org owners.
 94    """
 95    members = paginate(
 96        f"https://api.github.com/orgs/{org}/members",
 97        cfg,
 98        {"filter": "2fa_disabled"},
 99    )
100    return [
101        {
102            "login": m["login"],
103            "profile_url": m["html_url"],
104        }
105        for m in members
106    ]
107
108
109def outside_collaborators(org, cfg, repo_collabs):
110    """
111    Non-org members with direct repo access.
112    Accepts pre-fetched repo_collabs from fetch_repo_collaborators().
113    """
114    outside = {
115        m["login"]
116        for m in paginate(
117            f"https://api.github.com/orgs/{org}/outside_collaborators", cfg
118        )
119    }
120    rows = []
121    for entry in repo_collabs:
122        for c in entry["collaborators"]:
123            if c["login"] in outside:
124                rows.append(
125                    {
126                        "login": c["login"],
127                        "repo": entry["repo"],
128                        "permission": _permission_level(c.get("permissions", {})),
129                        "repo_visibility": entry["visibility"],
130                    }
131                )
132    return rows
133
134
135def privileged_access(_org, _cfg, repo_collabs):
136    """
137    All users with admin permission on any repo.
138    Accepts pre-fetched repo_collabs from fetch_repo_collaborators().
139    """
140    rows = []
141    for entry in repo_collabs:
142        for c in entry["collaborators"]:
143            if c.get("permissions", {}).get("admin"):
144                rows.append(
145                    {
146                        "login": c["login"],
147                        "repo": entry["repo"],
148                        "permission": "admin",
149                        "repo_visibility": entry["visibility"],
150                    }
151                )
152    return rows
153
154
155def pending_invitations(org, cfg):
156    now = datetime.now(timezone.utc)
157    rows = []
158    for inv in paginate(f"https://api.github.com/orgs/{org}/invitations", cfg):
159        created = inv.get("created_at", "")
160        age_days = None
161        if created:
162            dt = datetime.fromisoformat(created.replace("Z", "+00:00"))
163            age_days = (now - dt).days
164        rows.append(
165            {
166                "login": inv.get("login") or inv.get("email", "unknown"),
167                "role": inv.get("role", ""),
168                "invited_by": inv.get("inviter", {}).get("login", ""),
169                "created_at": created,
170                "age_days": age_days,
171            }
172        )
173    return rows
174
175
176def team_permissions(org, cfg):
177    rows = []
178    for team in paginate(f"https://api.github.com/orgs/{org}/teams", cfg):
179        slug = team["slug"]
180        team_members = paginate(
181            f"https://api.github.com/orgs/{org}/teams/{slug}/members", cfg
182        )
183        team_repos = paginate(
184            f"https://api.github.com/orgs/{org}/teams/{slug}/repos", cfg
185        )
186        member_logins = ", ".join(m["login"] for m in team_members) or "(none)"
187        for repo in team_repos:
188            rows.append(
189                {
190                    "team": team["name"],
191                    "repo": repo["name"],
192                    "permission": _permission_level(repo.get("permissions", {})),
193                    "members": member_logins,
194                }
195            )
196    return rows
197
198
199def permission_matrix(_org, _cfg, repo_collabs):
200    """
201    Full per-repo/per-user permission cross-reference.
202    Accepts pre-fetched repo_collabs from fetch_repo_collaborators().
203    """
204    rows = []
205    for entry in repo_collabs:
206        for c in entry["collaborators"]:
207            rows.append(
208                {
209                    "repo": entry["repo"],
210                    "login": c["login"],
211                    "permission": _permission_level(c.get("permissions", {})),
212                    "visibility": entry["visibility"],
213                }
214            )
215    return rows