audit-labs/audit-tools

  1"""
  2Drive the GitLab audit collectors from the TUI.
  3
  4Reuses the collectors and CSV reporter under ``applications/gitlab`` unchanged.
  5Mirrors github_runner: a ``CHECKS`` registry plus ``run_audit`` that writes the
  6same package ``applications/gitlab/audit.py`` produces and reports progress
  7through a callback.
  8"""
  9
 10import os
 11import sys
 12from collections.abc import Iterable
 13from datetime import date
 14
 15from tui.common import Check, ProgressCallback, ProgressEvent
 16
 17# Namespaced import so the GitHub and GitLab collector packages (both named
 18# ``collectors`` on disk) can coexist in one process.
 19_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 20if _REPO_ROOT not in sys.path:
 21    sys.path.insert(0, _REPO_ROOT)
 22
 23from applications.gitlab.collectors import (
 24    approvals,
 25    audit_events,
 26    branch_protections,
 27    members,
 28    pipelines,
 29    projects,
 30    settings,
 31)
 32from applications.gitlab.reporters import csv_reporter
 33
 34# --- Check registry ---------------------------------------------------------
 35
 36# For GitLab, ``arg`` is "base" -> fn(group, cfg) or "projects" -> fn(group,
 37# cfg, projects), where the project cache is fetched once and shared.
 38
 39CHECKS: list[Check] = [
 40    Check("group_members", "Group members", members.group_members, "group_members.csv"),
 41    Check(
 42        "projects", "Projects", projects.project_list, "projects.csv", arg="projects"
 43    ),
 44    Check(
 45        "project_members",
 46        "Project members",
 47        members.project_members,
 48        "project_members.csv",
 49        arg="projects",
 50    ),
 51    Check(
 52        "branch_protections",
 53        "Branch protections",
 54        branch_protections.branch_protections,
 55        "branch_protections.csv",
 56        arg="projects",
 57    ),
 58    Check(
 59        "pipelines",
 60        "Pipelines",
 61        pipelines.pipelines,
 62        "pipelines.csv",
 63        arg="projects",
 64    ),
 65    Check(
 66        "approval_rules",
 67        "Approval rules",
 68        approvals.approval_rules,
 69        "approval_rules.csv",
 70        arg="projects",
 71        note="requires Premium/Ultimate",
 72    ),
 73    Check(
 74        "audit_events",
 75        "Audit events",
 76        audit_events.audit_events,
 77        "audit_events.csv",
 78        note="requires Premium/Ultimate",
 79    ),
 80    Check(
 81        "password_policy",
 82        "Password policy",
 83        settings.password_policy,
 84        "password_policy.csv",
 85        note="self-hosted, admin token",
 86    ),
 87]
 88
 89_PREMIUM = {"approval_rules", "audit_events", "password_policy"}
 90DEFAULT_SELECTION = [c.key for c in CHECKS if c.key not in _PREMIUM]
 91
 92
 93# --- Config + output helpers ------------------------------------------------
 94
 95
 96def build_cfg(token: str, base_url: str) -> dict:
 97    """Build the config dict the collectors expect (mirrors config.load())."""
 98    return {
 99        "token": token,
100        "base_url": base_url.rstrip("/"),
101        "headers": {"PRIVATE-TOKEN": token},
102        "timeout": 30,
103    }
104
105
106def default_output_dir(out: str, group: str) -> str:
107    """Match the folder naming used by applications/gitlab/audit.py."""
108    safe_group = group.replace("/", "-")
109    return os.path.join(out, f"gitlab_audit_{safe_group}_{date.today().isoformat()}")
110
111
112# --- Runner -----------------------------------------------------------------
113
114
115def run_audit(
116    *,
117    group: str,
118    token: str,
119    base_url: str,
120    output_dir: str,
121    selected_keys: Iterable[str],
122    on_event: ProgressCallback,
123) -> list[tuple[str, int]]:
124    """
125    Run the selected checks and write the audit package to ``output_dir``.
126
127    A collector that raises is reported as an error and recorded with a count
128    of 0, so one bad check never aborts the whole run.
129    """
130    cfg = build_cfg(token, base_url)
131    selected = set(selected_keys)
132    checks = [c for c in CHECKS if c.key in selected]
133
134    project_cache: list | None = None
135    if any(c.arg == "projects" for c in checks):
136        on_event(ProgressEvent("fetch", "Projects (shared cache)"))
137        try:
138            project_cache = projects.fetch_projects(group, cfg)
139        except Exception as e:
140            on_event(ProgressEvent("error", "Projects (shared cache)", message=str(e)))
141            project_cache = []
142
143    sections: list[tuple[str, int]] = []
144    for c in checks:
145        on_event(ProgressEvent("start", c.label))
146        try:
147            if c.arg == "projects":
148                rows = c.fn(group, cfg, project_cache or [])
149            else:
150                rows = c.fn(group, cfg)
151        except Exception as e:
152            on_event(ProgressEvent("error", c.label, message=str(e)))
153            sections.append((c.label, 0))
154            continue
155
156        csv_reporter.write(output_dir, c.filename, rows)
157        sections.append((c.label, len(rows)))
158        on_event(ProgressEvent("done", c.label, count=len(rows)))
159
160    csv_reporter.write_summary(output_dir, group, sections)
161    total = sum(n for _, n in sections)
162    on_event(ProgressEvent("summary", output_dir, count=total))
163    return sections