audit-labs/audit-tools
A collection of scripts, queries, and other goodies you can use in an audit.
clone: git clone https://gitbay.org/audit-labs/audit-tools.git
99ace2ae84e0158b4443b4022c38be7d321f285e
verified · cmc
author: Christian Cleberg <hello@cleberg.net> · 2026-07-04T17:36:37Z
.github/workflows/ruff.yml | 22 ++--- applications/github/audit.py | 63 ++++++++++-- applications/github/collectors/api.py | 4 +- applications/github/collectors/audit_log.py | 24 ++--- .../github/collectors/branch_protections.py | 56 ++++++----- applications/github/collectors/commits.py | 24 ++--- applications/github/collectors/members.py | 109 +++++++++++++-------- applications/github/reporters/csv_reporter.py | 6 +- 8 files changed, 191 insertions(+), 117 deletions(-) @@ -12,26 +12,16 @@ jobs: matrix: python-version: ["3.x"] steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v5 - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v5 + uses: actions/setup-python@v6 with: python-version: ${{ matrix.python-version }} - ref: ${{ github.event.pull_request.head.ref }} - name: Install dependencies run: | python -m pip install --upgrade pip - pip install pandas dash plotly.express - - name: Install Ruff - uses: astral-sh/ruff-action@v3.2.2 - - name: Ruff Actions + pip install pandas dash plotly.express ruff + - name: Ruff run: | - ruff check --fix - ruff format - - name: Add and Commit - uses: EndBug/add-and-commit@v9 - env: - GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} - with: - default_author: github_actions - pathspec_error_handling: ignore + ruff check . + ruff format --check . @@ -90,8 +90,10 @@ def run(): sections.append((label, len(rows))) return rows - collect("Member roster", members.member_roster, "member_roster.csv", org, cfg) - collect("2FA disabled", members.two_factor_disabled, "two_factor_disabled.csv", org, cfg) + collect("Member roster", members.member_roster, "member_roster.csv", org, cfg) + collect( + "2FA disabled", members.two_factor_disabled, "two_factor_disabled.csv", org, cfg + ) print("Fetching repo collaborators (shared cache)...") try: @@ -100,14 +102,55 @@ def run(): print(f" Error fetching collaborators: {e}", file=sys.stderr) repo_collabs = [] - collect("Outside collaborators", members.outside_collaborators, "outside_collaborators.csv", org, cfg, repo_collabs) - collect("Privileged access", members.privileged_access, "privileged_access.csv", org, cfg, repo_collabs) - collect("Pending invitations", members.pending_invitations, "pending_invitations.csv", org, cfg) - collect("Team permissions", members.team_permissions, "team_permissions.csv", org, cfg) - collect("Permission matrix", members.permission_matrix, "permission_matrix.csv", org, cfg, repo_collabs) - collect("Branch protections", branch_protections.branch_protections, "branch_protections.csv", org, cfg) - collect("Commits", commits.commits, "commits.csv", org, cfg, args.branch) - collect("Audit log branch/ruleset changes", audit_log.audit_log, "audit_log.csv", org, cfg) + collect( + "Outside collaborators", + members.outside_collaborators, + "outside_collaborators.csv", + org, + cfg, + repo_collabs, + ) + collect( + "Privileged access", + members.privileged_access, + "privileged_access.csv", + org, + cfg, + repo_collabs, + ) + collect( + "Pending invitations", + members.pending_invitations, + "pending_invitations.csv", + org, + cfg, + ) + collect( + "Team permissions", members.team_permissions, "team_permissions.csv", org, cfg + ) + collect( + "Permission matrix", + members.permission_matrix, + "permission_matrix.csv", + org, + cfg, + repo_collabs, + ) + collect( + "Branch protections", + branch_protections.branch_protections, + "branch_protections.csv", + org, + cfg, + ) + collect("Commits", commits.commits, "commits.csv", org, cfg, args.branch) + collect( + "Audit log branch/ruleset changes", + audit_log.audit_log, + "audit_log.csv", + org, + cfg, + ) print() csv_reporter.write_summary(output_dir, org, sections) @@ -12,7 +12,9 @@ def paginate(url, cfg, params=None): while True: p["page"] = page - resp = requests.get(url, headers=cfg["headers"], params=p, timeout=cfg["timeout"]) + resp = requests.get( + url, headers=cfg["headers"], params=p, timeout=cfg["timeout"] + ) resp.raise_for_status() data = resp.json() if not data: @@ -88,17 +88,19 @@ def audit_log(org, cfg, actions=None, lookback_days=DEFAULT_LOOKBACK_DAYS): rows = [] for e in _dedupe_events(events): - rows.append({ - "action": e.get("action", ""), - "actor": e.get("actor", ""), - "repo": e.get("repo", ""), - "branch_or_pattern": _branch_or_pattern(e), - "operation_type": e.get("operation_type", ""), - "summary": _event_summary(e), - "details": _event_details(e), - "created_at": _format_created_at(e.get("created_at", "")), - "org": e.get("org", ""), - }) + rows.append( + { + "action": e.get("action", ""), + "actor": e.get("actor", ""), + "repo": e.get("repo", ""), + "branch_or_pattern": _branch_or_pattern(e), + "operation_type": e.get("operation_type", ""), + "summary": _event_summary(e), + "details": _event_details(e), + "created_at": _format_created_at(e.get("created_at", "")), + "org": e.get("org", ""), + } + ) return rows @@ -27,7 +27,10 @@ def branch_protections(org, cfg): ) except requests.HTTPError as e: if e.response is not None and e.response.status_code == 403: - print(f" Skipping {repo_name}: branches endpoint returned 403", file=sys.stderr) + print( + f" Skipping {repo_name}: branches endpoint returned 403", + file=sys.stderr, + ) continue raise @@ -40,17 +43,19 @@ def branch_protections(org, cfg): resp = requests.get(url, headers=cfg["headers"], timeout=cfg["timeout"]) if resp.status_code in (403, 404): - rows.append({ - "repo": repo_name, - "branch": branch_name, - "protected": False, - "required_reviews": None, - "dismiss_stale_reviews": None, - "require_code_owner_reviews": None, - "required_status_checks": None, - "enforce_admins": None, - "restrictions": None, - }) + rows.append( + { + "repo": repo_name, + "branch": branch_name, + "protected": False, + "required_reviews": None, + "dismiss_stale_reviews": None, + "require_code_owner_reviews": None, + "required_status_checks": None, + "enforce_admins": None, + "restrictions": None, + } + ) continue resp.raise_for_status() @@ -58,16 +63,21 @@ def branch_protections(org, cfg): reviews = p.get("required_pull_request_reviews", {}) checks = p.get("required_status_checks", {}) - rows.append({ - "repo": repo_name, - "branch": branch_name, - "protected": True, - "required_reviews": reviews.get("required_approving_review_count"), - "dismiss_stale_reviews": reviews.get("dismiss_stale_reviews"), - "require_code_owner_reviews": reviews.get("require_code_owner_reviews"), - "required_status_checks": ", ".join(checks.get("contexts", [])) or None, - "enforce_admins": p.get("enforce_admins", {}).get("enabled"), - "restrictions": bool(p.get("restrictions")), - }) + rows.append( + { + "repo": repo_name, + "branch": branch_name, + "protected": True, + "required_reviews": reviews.get("required_approving_review_count"), + "dismiss_stale_reviews": reviews.get("dismiss_stale_reviews"), + "require_code_owner_reviews": reviews.get( + "require_code_owner_reviews" + ), + "required_status_checks": ", ".join(checks.get("contexts", [])) + or None, + "enforce_admins": p.get("enforce_admins", {}).get("enabled"), + "restrictions": bool(p.get("restrictions")), + } + ) return rows @@ -31,16 +31,18 @@ def commits(org, cfg, branch="main"): commit = c.get("commit", {}) author = commit.get("author", {}) stats = c.get("stats", {}) - rows.append({ - "repo": repo_name, - "branch": branch, - "sha": c.get("sha", "")[:12], - "author_name": author.get("name", ""), - "author_email": author.get("email", ""), - "date": author.get("date", ""), - "message": commit.get("message", "").splitlines()[0], - "additions": stats.get("additions", ""), - "deletions": stats.get("deletions", ""), - }) + rows.append( + { + "repo": repo_name, + "branch": branch, + "sha": c.get("sha", "")[:12], + "author_name": author.get("name", ""), + "author_email": author.get("email", ""), + "date": author.get("date", ""), + "message": commit.get("message", "").splitlines()[0], + "additions": stats.get("additions", ""), + "deletions": stats.get("deletions", ""), + } + ) return rows @@ -50,22 +50,31 @@ def fetch_repo_collaborators(org, cfg): ) except requests.HTTPError as e: if e.response is not None and e.response.status_code == 403: - print(f" Skipping {repo_name}: collaborators endpoint returned 403", file=sys.stderr) + print( + f" Skipping {repo_name}: collaborators endpoint returned 403", + file=sys.stderr, + ) continue raise - results.append({ - "repo": repo_name, - "visibility": repo["visibility"], - "collaborators": collabs, - }) + results.append( + { + "repo": repo_name, + "visibility": repo["visibility"], + "collaborators": collabs, + } + ) return results def member_roster(org, cfg): - members = paginate(f"https://api.github.com/orgs/{org}/members", cfg, {"role": "all"}) + members = paginate( + f"https://api.github.com/orgs/{org}/members", cfg, {"role": "all"} + ) owners = { m["login"] - for m in paginate(f"https://api.github.com/orgs/{org}/members", cfg, {"role": "owner"}) + for m in paginate( + f"https://api.github.com/orgs/{org}/members", cfg, {"role": "owner"} + ) } return [ { @@ -104,18 +113,22 @@ def outside_collaborators(org, cfg, repo_collabs): """ outside = { m["login"] - for m in paginate(f"https://api.github.com/orgs/{org}/outside_collaborators", cfg) + for m in paginate( + f"https://api.github.com/orgs/{org}/outside_collaborators", cfg + ) } rows = [] for entry in repo_collabs: for c in entry["collaborators"]: if c["login"] in outside: - rows.append({ - "login": c["login"], - "repo": entry["repo"], - "permission": _permission_level(c.get("permissions", {})), - "repo_visibility": entry["visibility"], - }) + rows.append( + { + "login": c["login"], + "repo": entry["repo"], + "permission": _permission_level(c.get("permissions", {})), + "repo_visibility": entry["visibility"], + } + ) return rows @@ -128,12 +141,14 @@ def privileged_access(org, cfg, repo_collabs): for entry in repo_collabs: for c in entry["collaborators"]: if c.get("permissions", {}).get("admin"): - rows.append({ - "login": c["login"], - "repo": entry["repo"], - "permission": "admin", - "repo_visibility": entry["visibility"], - }) + rows.append( + { + "login": c["login"], + "repo": entry["repo"], + "permission": "admin", + "repo_visibility": entry["visibility"], + } + ) return rows @@ -146,13 +161,15 @@ def pending_invitations(org, cfg): if created: dt = datetime.fromisoformat(created.replace("Z", "+00:00")) age_days = (now - dt).days - rows.append({ - "login": inv.get("login") or inv.get("email", "unknown"), - "role": inv.get("role", ""), - "invited_by": inv.get("inviter", {}).get("login", ""), - "created_at": created, - "age_days": age_days, - }) + rows.append( + { + "login": inv.get("login") or inv.get("email", "unknown"), + "role": inv.get("role", ""), + "invited_by": inv.get("inviter", {}).get("login", ""), + "created_at": created, + "age_days": age_days, + } + ) return rows @@ -160,16 +177,22 @@ def team_permissions(org, cfg): rows = [] for team in paginate(f"https://api.github.com/orgs/{org}/teams", cfg): slug = team["slug"] - team_members = paginate(f"https://api.github.com/orgs/{org}/teams/{slug}/members", cfg) - team_repos = paginate(f"https://api.github.com/orgs/{org}/teams/{slug}/repos", cfg) + team_members = paginate( + f"https://api.github.com/orgs/{org}/teams/{slug}/members", cfg + ) + team_repos = paginate( + f"https://api.github.com/orgs/{org}/teams/{slug}/repos", cfg + ) member_logins = ", ".join(m["login"] for m in team_members) or "(none)" for repo in team_repos: - rows.append({ - "team": team["name"], - "repo": repo["name"], - "permission": _permission_level(repo.get("permissions", {})), - "members": member_logins, - }) + rows.append( + { + "team": team["name"], + "repo": repo["name"], + "permission": _permission_level(repo.get("permissions", {})), + "members": member_logins, + } + ) return rows @@ -181,10 +204,12 @@ def permission_matrix(org, cfg, repo_collabs): rows = [] for entry in repo_collabs: for c in entry["collaborators"]: - rows.append({ - "repo": entry["repo"], - "login": c["login"], - "permission": _permission_level(c.get("permissions", {})), - "visibility": entry["visibility"], - }) + rows.append( + { + "repo": entry["repo"], + "login": c["login"], + "permission": _permission_level(c.get("permissions", {})), + "visibility": entry["visibility"], + } + ) return rows @@ -31,10 +31,10 @@ def write_summary(output_dir, org, sections): """ path = os.path.join(output_dir, "summary.txt") lines = [ - f"GitHub Audit Package", + "GitHub Audit Package", f"Org: {org}", - f"", - f"Section Rows", + "", + "Section Rows", f"{'─' * 40}", ] for label, count in sections: