audit-labs/audit-tools

  1"""
  2Drive the existing GitHub audit collectors from the TUI.
  3
  4This module reuses the collectors and CSV reporter under
  5``applications/github`` unchanged. It exposes:
  6
  7- ``CHECKS``: the list of available audit checks the UI presents.
  8- ``run_audit``: run the selected checks, write the same output package
  9  ``audit.py`` produces, and report progress through a callback.
 10"""
 11
 12import os
 13import sys
 14from collections.abc import Iterable
 15from datetime import date
 16
 17from tui.common import Check, ProgressCallback, ProgressEvent
 18
 19# Import the GitHub collectors as a namespaced package so the GitHub and GitLab
 20# collector packages (both named ``collectors`` on disk) can coexist in one
 21# process. Requires the repo root on sys.path.
 22_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 23if _REPO_ROOT not in sys.path:
 24    sys.path.insert(0, _REPO_ROOT)
 25
 26from applications.github.collectors import (
 27    audit_log,
 28    branch_protections,
 29    commits,
 30    deploy_keys,
 31    members,
 32    org_settings,
 33    security_alerts,
 34    webhooks,
 35)
 36from applications.github.reporters import csv_reporter
 37
 38# --- Check registry ---------------------------------------------------------
 39
 40# For GitHub, ``arg`` is "base" -> fn(org, cfg), "collabs" -> fn(org, cfg,
 41# repo_collabs), or "branch" -> fn(org, cfg, branch).
 42
 43CHECKS: list[Check] = [
 44    Check("member_roster", "Member roster", members.member_roster, "member_roster.csv"),
 45    Check(
 46        "two_factor",
 47        "2FA disabled",
 48        members.two_factor_disabled,
 49        "two_factor_disabled.csv",
 50        note="requires org owner token",
 51    ),
 52    Check(
 53        "outside_collaborators",
 54        "Outside collaborators",
 55        members.outside_collaborators,
 56        "outside_collaborators.csv",
 57        arg="collabs",
 58    ),
 59    Check(
 60        "privileged_access",
 61        "Privileged access",
 62        members.privileged_access,
 63        "privileged_access.csv",
 64        arg="collabs",
 65    ),
 66    Check(
 67        "pending_invitations",
 68        "Pending invitations",
 69        members.pending_invitations,
 70        "pending_invitations.csv",
 71    ),
 72    Check(
 73        "team_permissions",
 74        "Team permissions",
 75        members.team_permissions,
 76        "team_permissions.csv",
 77    ),
 78    Check(
 79        "permission_matrix",
 80        "Permission matrix",
 81        members.permission_matrix,
 82        "permission_matrix.csv",
 83        arg="collabs",
 84    ),
 85    Check(
 86        "branch_protections",
 87        "Branch protections",
 88        branch_protections.branch_protections,
 89        "branch_protections.csv",
 90    ),
 91    Check("commits", "Commits", commits.commits, "commits.csv", arg="branch"),
 92    Check(
 93        "org_security",
 94        "Org security settings",
 95        org_settings.org_security,
 96        "org_security.csv",
 97    ),
 98    Check("webhooks", "Webhooks", webhooks.webhooks, "webhooks.csv"),
 99    Check("deploy_keys", "Deploy keys", deploy_keys.deploy_keys, "deploy_keys.csv"),
100    Check(
101        "secret_scanning",
102        "Secret scanning alerts",
103        security_alerts.secret_scanning,
104        "secret_scanning.csv",
105        note="requires GitHub Advanced Security",
106    ),
107    Check(
108        "dependabot_alerts",
109        "Dependabot alerts",
110        security_alerts.dependabot_alerts,
111        "dependabot_alerts.csv",
112        note="requires GitHub Advanced Security",
113    ),
114    Check(
115        "audit_log",
116        "Audit log (branch/ruleset changes)",
117        audit_log.audit_log,
118        "audit_log.csv",
119        note="requires GitHub Enterprise Cloud",
120    ),
121]
122
123# Off by default: checks needing Advanced Security or Enterprise Cloud.
124_OFF_BY_DEFAULT = {"secret_scanning", "dependabot_alerts", "audit_log"}
125DEFAULT_SELECTION = [c.key for c in CHECKS if c.key not in _OFF_BY_DEFAULT]
126
127
128# --- Config + output helpers ------------------------------------------------
129
130
131def build_cfg(token: str) -> dict:
132    """Build the config dict the collectors expect (mirrors config.load())."""
133    return {
134        "token": token,
135        "headers": {
136            "Authorization": f"token {token}",
137            "Accept": "application/vnd.github.v3+json",
138        },
139        "timeout": 30,
140    }
141
142
143def default_output_dir(out: str, org: str) -> str:
144    """Match the folder naming used by audit.py."""
145    return os.path.join(out, f"github_audit_{org}_{date.today().isoformat()}")
146
147
148# --- Runner -----------------------------------------------------------------
149
150
151def run_audit(
152    *,
153    org: str,
154    token: str,
155    output_dir: str,
156    branch: str,
157    selected_keys: Iterable[str],
158    on_event: ProgressCallback,
159) -> list[tuple[str, int]]:
160    """
161    Run the selected checks and write the audit package to ``output_dir``.
162
163    A collector that raises is reported as an error and recorded with a count
164    of 0, matching audit.py's behavior of never aborting the whole run.
165
166    Returns the list of (label, row_count) sections that was written to the
167    summary file.
168    """
169    cfg = build_cfg(token)
170    selected = set(selected_keys)
171    checks = [c for c in CHECKS if c.key in selected]
172
173    repo_collabs: list | None = None
174    if any(c.arg == "collabs" for c in checks):
175        on_event(ProgressEvent("fetch", "Repo collaborators (shared cache)"))
176        try:
177            repo_collabs = members.fetch_repo_collaborators(org, cfg)
178        except Exception as e:
179            on_event(
180                ProgressEvent(
181                    "error", "Repo collaborators (shared cache)", message=str(e)
182                )
183            )
184            repo_collabs = []
185
186    sections: list[tuple[str, int]] = []
187    for c in checks:
188        on_event(ProgressEvent("start", c.label))
189        try:
190            if c.arg == "collabs":
191                rows = c.fn(org, cfg, repo_collabs or [])
192            elif c.arg == "branch":
193                rows = c.fn(org, cfg, branch)
194            else:
195                rows = c.fn(org, cfg)
196        except Exception as e:
197            on_event(ProgressEvent("error", c.label, message=str(e)))
198            sections.append((c.label, 0))
199            continue
200
201        csv_reporter.write(output_dir, c.filename, rows)
202        sections.append((c.label, len(rows)))
203        on_event(ProgressEvent("done", c.label, count=len(rows)))
204
205    csv_reporter.write_summary(output_dir, org, sections)
206    total = sum(n for _, n in sections)
207    on_event(ProgressEvent("summary", output_dir, count=total))
208    return sections