no-mistakes(review): Harden GC gate key aliases, fail closed, fix baseline

This commit is contained in:
root
2026-09-12 18:42:36 +00:00
parent 4ea2d0309f
commit de1428b4ae
3 changed files with 31 additions and 8 deletions
+18 -6
View File
@@ -22,7 +22,8 @@ Usage:
cat scan.json | disk-gc-plan.py # same, via stdin
disk-gc-plan.py --scan scan.json --json # machine-readable plan
Scan entries may carry any of: id / guest / vmid, hostname / name, ip.
Scan entries may carry any of: id / guest / vmid / ct / ctid, hostname / name, ip.
A threshold-crossing entry with no recognizable identity is reported, never GC'd.
Exit codes: 0 ok, 1 usage/parse error.
"""
from __future__ import annotations
@@ -40,6 +41,9 @@ DEFAULT_CONTRACT = REPO / "disk-gc-threat-response.prose.md"
AMBER, RED, CRITICAL = 75, 85, 95
IDENTITY_FIELDS = ("id", "guest", "vmid", "ct", "ctid", "hostname", "name", "ip")
UNIDENTIFIED_REASON = "unidentified target - refusing to schedule GC"
def load_report_only_guests(contract_path: pathlib.Path) -> list[dict]:
"""Read the authoritative report_only_guests block out of the contract.
@@ -64,7 +68,7 @@ def _keys(entry: dict) -> set[str]:
"""Guest/host identity keys, shared by exclusions and scan entries so the two
sides of the gate can never key on different fields."""
out: set[str] = set()
for field in ("guest", "id", "vmid", "hostname", "name", "ip"):
for field in IDENTITY_FIELDS:
value = entry.get(field)
if value is not None and str(value).strip():
out.add(str(value).strip().lower())
@@ -92,12 +96,20 @@ def build_plan(scan: list[dict], report_only: list[dict]) -> list[dict]:
if level is None:
continue # GREEN: log only, no action
scan_keys = _keys(entry)
match = next((e for e, keys in excluded if keys & scan_keys), None)
target = next(
(entry.get(k) for k in ("id", "guest", "vmid", "hostname", "name", "ip")
if entry.get(k) not in (None, "")),
(entry.get(k) for k in IDENTITY_FIELDS if entry.get(k) not in (None, "")),
"?",
)
if not scan_keys:
plan.append({
"target": target,
"level": level,
"pct": float(pct),
"action": "report-only",
"reason": UNIDENTIFIED_REASON,
})
continue
match = next((e for e, keys in excluded if keys & scan_keys), None)
if match is not None:
plan.append({
"target": target,
@@ -119,7 +131,7 @@ def build_plan(scan: list[dict], report_only: list[dict]) -> list[dict]:
def main() -> int:
ap = argparse.ArgumentParser(description="Plan disk GC actions with the report-only gate.")
ap.add_argument("--scan", help="JSON file: list of {id|hostname|ip, usage_pct}")
ap.add_argument("--scan", help="JSON file: list of {id|ct|hostname|ip, usage_pct}")
ap.add_argument("--contract", default=str(DEFAULT_CONTRACT))
ap.add_argument("--json", action="store_true", help="emit the plan as JSON")
args = ap.parse_args()