rohitsar567 commited on
Commit
a519e2c
·
1 Parent(s): 74e87b6

feat(audit): core contract + runner + CLI

Browse files
audit/__init__.py ADDED
File without changes
audit/__main__.py ADDED
@@ -0,0 +1,16 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import argparse, sys
2
+ from audit import core
3
+
4
+ def main() -> int:
5
+ p = argparse.ArgumentParser(prog="python -m audit")
6
+ g = p.add_mutually_exclusive_group()
7
+ for t in ("static","build","functional","deploy","all"):
8
+ g.add_argument(f"--{t}", action="store_const", const=t, dest="tier")
9
+ p.add_argument("--selftest", action="store_true")
10
+ p.add_argument("--json", action="store_true")
11
+ a = p.parse_args()
12
+ if a.selftest: return core.selftest()
13
+ return core.run(core.TIER_SETS[a.tier or "all"], as_json=a.json)
14
+
15
+ if __name__ == "__main__":
16
+ sys.exit(main())
audit/core.py ADDED
@@ -0,0 +1,88 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ """Audit framework core: check contract, registry, runner, selftest."""
2
+ from __future__ import annotations
3
+ import dataclasses, enum, json, pathlib, subprocess
4
+ from typing import Callable
5
+
6
+ REPO = pathlib.Path(__file__).resolve().parent.parent
7
+
8
+ class Status(enum.Enum):
9
+ PASS = "PASS"; WARN = "WARN"; FAIL = "FAIL"; SKIP = "SKIP"
10
+
11
+ @dataclasses.dataclass
12
+ class Result:
13
+ check_id: str; status: "Status"; evidence: str; remediation: str = ""
14
+
15
+ @dataclasses.dataclass
16
+ class Check:
17
+ id: str; tier: str; title: str; fn: Callable[[], "Result"]
18
+
19
+ CHECKS: list[Check] = []
20
+
21
+ def register(id: str, tier: str, title: str):
22
+ def deco(fn):
23
+ CHECKS.append(Check(id, tier, title, fn)); return fn
24
+ return deco
25
+
26
+ TIER_SETS = {
27
+ "static": {"static"}, "build": {"static","build"},
28
+ "functional": {"static","build","functional"}, "deploy": {"deploy"},
29
+ "all": {"static","build","functional","deploy"},
30
+ }
31
+
32
+ def sh(cmd: list[str], timeout: int = 120) -> subprocess.CompletedProcess:
33
+ return subprocess.run(cmd, cwd=REPO, capture_output=True, text=True, timeout=timeout)
34
+
35
+ def git(*args: str) -> str:
36
+ return sh(["git", *args]).stdout.strip()
37
+
38
+ def _load_all_checks() -> None:
39
+ from audit import (tier1_repo, tier2_code, tier3_build, # noqa: F401
40
+ tier4_functional, tier5_deploy)
41
+
42
+ def run(selected: set[str], as_json: bool = False) -> int:
43
+ if not CHECKS:
44
+ _load_all_checks()
45
+ rows = []
46
+ for c in sorted(CHECKS, key=lambda c: c.id):
47
+ if c.tier not in selected: continue
48
+ try:
49
+ r = c.fn()
50
+ except Exception as e:
51
+ r = Result(c.id, Status.FAIL, f"check raised {type(e).__name__}: {e}",
52
+ "fix the check or the underlying issue")
53
+ rows.append((c, r))
54
+ fails = [r for _, r in rows if r.status is Status.FAIL]
55
+ if as_json:
56
+ print(json.dumps([{"id": c.id, "status": r.status.value,
57
+ "evidence": r.evidence, "remediation": r.remediation}
58
+ for c, r in rows], indent=2))
59
+ else:
60
+ for c, r in rows:
61
+ mark = {"PASS":"OK","WARN":"WARN","FAIL":"FAIL","SKIP":"SKIP"}[r.status.value]
62
+ print(f" [{mark}] {c.id} {c.title}")
63
+ if r.status in (Status.FAIL, Status.WARN):
64
+ print(f" {r.evidence}")
65
+ if r.remediation: print(f" fix: {r.remediation}")
66
+ print(f"\n {len(rows)} checks · "
67
+ f"{sum(1 for _,r in rows if r.status is Status.PASS)} pass · "
68
+ f"{sum(1 for _,r in rows if r.status is Status.WARN)} warn · "
69
+ f"{len(fails)} fail · "
70
+ f"{sum(1 for _,r in rows if r.status is Status.SKIP)} skip")
71
+ return 1 if fails else 0
72
+
73
+ def selftest() -> int:
74
+ from audit.selftest_fixtures import FIXTURES
75
+ if not CHECKS:
76
+ _load_all_checks()
77
+ bad = []
78
+ for c in CHECKS:
79
+ fx = FIXTURES.get(c.id)
80
+ if fx is None:
81
+ bad.append(f"{c.id}: NO selftest fixture"); continue
82
+ with fx():
83
+ r = c.fn()
84
+ if r.status is not Status.FAIL:
85
+ bad.append(f"{c.id}: expected FAIL on broken fixture, got {r.status.value}")
86
+ for b in bad: print(f" FAIL {b}")
87
+ print(f"\n selftest: {len(CHECKS)} checks · {len(bad)} not self-verifying")
88
+ return 1 if bad else 0
audit/selftest_fixtures.py ADDED
@@ -0,0 +1 @@
 
 
1
+ FIXTURES: dict = {}
audit/tier1_repo.py ADDED
@@ -0,0 +1 @@
 
 
1
+ # filled in a later task
audit/tier2_code.py ADDED
@@ -0,0 +1 @@
 
 
1
+ # filled in a later task
audit/tier3_build.py ADDED
@@ -0,0 +1 @@
 
 
1
+ # filled in a later task
audit/tier4_functional.py ADDED
@@ -0,0 +1 @@
 
 
1
+ # filled in a later task
audit/tier5_deploy.py ADDED
@@ -0,0 +1 @@
 
 
1
+ # filled in a later task
tests/test_audit_selftest.py ADDED
@@ -0,0 +1,20 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import subprocess, sys, pathlib
2
+ REPO = pathlib.Path(__file__).resolve().parent.parent
3
+
4
+ def test_core_runner_passes_a_trivial_pass_check():
5
+ code = subprocess.run(
6
+ [sys.executable, "-c",
7
+ "import audit.core as c; c.CHECKS.clear(); "
8
+ "c.register('X.1','static','dummy')(lambda: c.Result('X.1', c.Status.PASS, 'ok')); "
9
+ "import sys; sys.exit(c.run({'static'}))"],
10
+ cwd=REPO).returncode
11
+ assert code == 0
12
+
13
+ def test_core_runner_fails_on_a_fail_check():
14
+ code = subprocess.run(
15
+ [sys.executable, "-c",
16
+ "import audit.core as c; c.CHECKS.clear(); "
17
+ "c.register('X.2','static','dummy')(lambda: c.Result('X.2', c.Status.FAIL, 'bad')); "
18
+ "import sys; sys.exit(c.run({'static'}))"],
19
+ cwd=REPO).returncode
20
+ assert code == 1
tools/audit.sh ADDED
@@ -0,0 +1,4 @@
 
 
 
 
 
1
+ #!/usr/bin/env bash
2
+ set -euo pipefail
3
+ cd "$(dirname "$0")/.."
4
+ exec .venv/bin/python -m audit "$@"