audit-labs/audit-tools

  1"""
  2Drive the AWS audit collectors from the TUI.
  3
  4Reuses the collectors and CSV reporter under ``applications/aws`` unchanged.
  5Mirrors the other runners: a ``CHECKS`` registry plus ``run_audit`` that writes
  6the same package ``applications/aws/audit.py`` produces and reports progress
  7through a callback.
  8
  9AWS collectors take a single config dict (a boto3 session plus region/account);
 10there is no per-item cache, so every check is called as ``fn(cfg)``.
 11"""
 12
 13import os
 14import sys
 15from collections.abc import Iterable
 16from datetime import date
 17
 18from tui.common import Check, ProgressCallback, ProgressEvent
 19
 20_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 21if _REPO_ROOT not in sys.path:
 22    sys.path.insert(0, _REPO_ROOT)
 23
 24from applications.aws.collectors import (
 25    api,
 26    iam,
 27    monitoring,
 28    s3,
 29    security_groups,
 30    sso,
 31)
 32from applications.aws.reporters import csv_reporter
 33
 34# --- Check registry ---------------------------------------------------------
 35
 36CHECKS: list[Check] = [
 37    Check("iam_users", "IAM users", iam.iam_users, "iam_users.csv"),
 38    Check(
 39        "password_policy",
 40        "Password policy",
 41        iam.password_policy,
 42        "password_policy.csv",
 43    ),
 44    Check(
 45        "account_security",
 46        "Account security (root MFA)",
 47        iam.account_security,
 48        "account_security.csv",
 49    ),
 50    Check(
 51        "s3_public_access",
 52        "S3 public access",
 53        s3.s3_public_access,
 54        "s3_public_access.csv",
 55    ),
 56    Check(
 57        "security_groups",
 58        "Open security groups",
 59        security_groups.security_groups,
 60        "open_security_groups.csv",
 61        note="scans all regions",
 62    ),
 63    Check("cloudtrail", "CloudTrail", monitoring.cloudtrail, "cloudtrail.csv"),
 64    Check(
 65        "config_recorders",
 66        "AWS Config recorders",
 67        monitoring.config_recorders,
 68        "config_recorders.csv",
 69        note="scans all regions",
 70    ),
 71    Check(
 72        "sso_assignments",
 73        "SSO assignments",
 74        sso.sso_assignments,
 75        "sso_assignments.csv",
 76        note="requires Identity Center + Organizations",
 77    ),
 78]
 79
 80DEFAULT_SELECTION = [c.key for c in CHECKS if c.key != "sso_assignments"]
 81
 82
 83# --- Output helper ----------------------------------------------------------
 84
 85
 86def default_output_dir(out: str, profile: str) -> str:
 87    """Match the folder naming used by applications/aws/audit.py."""
 88    subject = profile or "default"
 89    return os.path.join(out, f"aws_audit_{subject}_{date.today().isoformat()}")
 90
 91
 92# --- Runner -----------------------------------------------------------------
 93
 94
 95def run_audit(
 96    *,
 97    profile: str,
 98    region: str,
 99    account: str,
100    output_dir: str,
101    selected_keys: Iterable[str],
102    on_event: ProgressCallback,
103) -> list[tuple[str, int]]:
104    """
105    Run the selected checks and write the audit package to ``output_dir``.
106
107    A collector that raises is reported as an error and recorded with a count
108    of 0, so one bad check never aborts the whole run. If the AWS session itself
109    can't be built (e.g. an unknown profile), that is reported and the run ends
110    cleanly.
111    """
112    subject = profile or "default"
113    selected = set(selected_keys)
114    checks = [c for c in CHECKS if c.key in selected]
115
116    try:
117        cfg = api.build_cfg(profile, region, account)
118    except Exception as e:
119        on_event(ProgressEvent("error", "AWS session", message=str(e)))
120        csv_reporter.write_summary(output_dir, subject, [])
121        on_event(ProgressEvent("summary", output_dir, count=0))
122        return []
123
124    sections: list[tuple[str, int]] = []
125    for c in checks:
126        on_event(ProgressEvent("start", c.label))
127        try:
128            rows = c.fn(cfg)
129        except Exception as e:
130            on_event(ProgressEvent("error", c.label, message=str(e)))
131            sections.append((c.label, 0))
132            continue
133
134        csv_reporter.write(output_dir, c.filename, rows)
135        sections.append((c.label, len(rows)))
136        on_event(ProgressEvent("done", c.label, count=len(rows)))
137
138    csv_reporter.write_summary(output_dir, subject, sections)
139    total = sum(n for _, n in sections)
140    on_event(ProgressEvent("summary", output_dir, count=total))
141    return sections