186 lines
7.4 KiB
Python
186 lines
7.4 KiB
Python
"""Regression tests for the disk-gc report-only gate (CT 111 / tdunna / .129).
|
|
|
|
WHY THIS FILE EXISTS: `disk-gc-threat-response.prose.md` defined AMBER as "GC scheduled
|
|
for next run" and its Execution loop called `gc-executor` for EVERY threat, with no
|
|
guest-level exclusion. CT 111 (tdunna, 192.168.68.129) belongs to Theo and is
|
|
report-only per the captain (2026-08-17, re-confirmed 2026-09-10) — so a single AMBER
|
|
reading on that guest would have scheduled GC commands (apt clean, journal vacuum,
|
|
log/tmp deletion, snap removal) against someone else's box. The only marker was
|
|
frontmatter `report_only_agents`, which names an AGENT while the scan unit is a GUEST.
|
|
|
|
These tests execute the real planner (`scripts/disk-gc-plan.py`) and assert observable
|
|
behaviour: an excluded guest never produces a `gc-executor` action at any level, while
|
|
our own guests still do.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import pathlib
|
|
import subprocess
|
|
import sys
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parent.parent
|
|
PLAN = ROOT / "scripts" / "disk-gc-plan.py"
|
|
|
|
|
|
def _plan(scan, tmp_path):
|
|
scan_file = tmp_path / "scan.json"
|
|
scan_file.write_text(json.dumps(scan))
|
|
proc = subprocess.run(
|
|
[sys.executable, str(PLAN), "--scan", str(scan_file), "--json"],
|
|
capture_output=True, text=True,
|
|
)
|
|
assert proc.returncode == 0, proc.stderr
|
|
return json.loads(proc.stdout)
|
|
|
|
|
|
def _actions_for(plan, target):
|
|
return [row for row in plan if str(row["target"]) == str(target)]
|
|
|
|
|
|
def test_excluded_guest_never_gets_gc_at_any_level(tmp_path):
|
|
"""CT 111 at AMBER, RED and CRITICAL — always report-only, never gc-executor."""
|
|
for pct, level in ((84, "AMBER"), (90, "RED"), (97, "CRITICAL")):
|
|
plan = _plan([{"id": 111, "hostname": "tdunna", "ip": "192.168.68.129",
|
|
"usage_pct": pct}], tmp_path)
|
|
rows = _actions_for(plan, 111)
|
|
assert rows, f"CT 111 must still be reported at {level}"
|
|
assert rows[0]["level"] == level
|
|
assert rows[0]["action"] == "report-only", rows
|
|
assert not any(r["action"] == "gc-executor" for r in rows)
|
|
|
|
|
|
def test_exclusion_matches_on_any_identity_key(tmp_path):
|
|
"""The gate is keyed on guest/host, so id, ct, ctid, hostname or IP all match."""
|
|
for entry in ({"id": 111, "usage_pct": 95},
|
|
{"ct": 111, "usage_pct": 95},
|
|
{"ctid": 111, "usage_pct": 95},
|
|
{"hostname": "tdunna", "usage_pct": 95},
|
|
{"ip": "192.168.68.129", "usage_pct": 95}):
|
|
plan = _plan([entry], tmp_path)
|
|
assert all(r["action"] == "report-only" for r in plan), (entry, plan)
|
|
|
|
|
|
def test_exclusion_matches_encoded_identities(tmp_path):
|
|
"""A differently-encoded CT 111 id must not slip past the gate to gc-executor."""
|
|
encodings = ({"id": 111.0},
|
|
{"id": "111.0"},
|
|
{"id": "lxc/111"},
|
|
{"id": "qemu/111"},
|
|
{"id": "0111"},
|
|
{"id": " 111 "})
|
|
for alias in encodings:
|
|
plan = _plan([{**alias, "usage_pct": 97}], tmp_path)
|
|
assert plan, alias
|
|
assert plan[0]["action"] == "report-only", (alias, plan)
|
|
|
|
|
|
def test_our_own_guests_still_get_gc(tmp_path):
|
|
"""acerpve .9 and amdpve .15 are ours — they must still be acted on."""
|
|
plan = _plan([{"hostname": "acerpve", "ip": "192.168.68.9", "usage_pct": 77},
|
|
{"hostname": "amdpve", "ip": "192.168.68.15", "usage_pct": 76}], tmp_path)
|
|
assert len(plan) == 2
|
|
assert all(r["action"] == "gc-executor" for r in plan), plan
|
|
|
|
|
|
def test_below_threshold_emits_nothing(tmp_path):
|
|
"""GREEN guests produce no action at all."""
|
|
assert _plan([{"id": 111, "usage_pct": 40}], tmp_path) == []
|
|
|
|
|
|
def test_agent_name_alone_does_not_gate_a_guest(tmp_path):
|
|
"""An agent-name marker must not be the gate: an unrelated guest still gets GC."""
|
|
plan = _plan([{"id": 999, "hostname": "koby", "usage_pct": 95}], tmp_path)
|
|
assert plan and plan[0]["action"] == "gc-executor"
|
|
|
|
|
|
def test_unidentified_threat_fails_closed(tmp_path):
|
|
"""A threshold-crossing entry with no recognized identity must not schedule GC."""
|
|
plan = _plan([{"usage_pct": 97}], tmp_path)
|
|
assert plan, "an unidentified threat must still be reported"
|
|
assert plan[0]["action"] == "report-only", plan
|
|
assert plan[0]["reason"], plan
|
|
|
|
|
|
def test_exclusion_entry_without_identity_fails_closed(tmp_path):
|
|
"""A mis-typed exclusion entry must break the run, never silently disable the gate."""
|
|
contract = tmp_path / "broken.prose.md"
|
|
contract.write_text(
|
|
"```yaml\n"
|
|
"report_only_guests:\n"
|
|
" - node: storepve\n"
|
|
" reason: \"typo - no guest identity\"\n"
|
|
"```\n"
|
|
)
|
|
scan_file = tmp_path / "scan.json"
|
|
scan_file.write_text(json.dumps([{"ct": 111, "usage_pct": 97}]))
|
|
proc = subprocess.run(
|
|
[sys.executable, str(PLAN), "--scan", str(scan_file),
|
|
"--contract", str(contract), "--json"],
|
|
capture_output=True, text=True,
|
|
)
|
|
assert proc.returncode != 0, proc.stdout
|
|
assert "gc-executor" not in proc.stdout
|
|
assert "identity" in proc.stderr.lower(), proc.stderr
|
|
|
|
|
|
def test_empty_exclusion_block_fails_closed(tmp_path):
|
|
"""An emptied report_only_guests list must break the run, not disable the gate."""
|
|
contract = tmp_path / "empty.prose.md"
|
|
contract.write_text("```yaml\nreport_only_guests: []\n```\n")
|
|
scan_file = tmp_path / "scan.json"
|
|
scan_file.write_text(json.dumps([{"ct": 111, "usage_pct": 97}]))
|
|
proc = subprocess.run(
|
|
[sys.executable, str(PLAN), "--scan", str(scan_file),
|
|
"--contract", str(contract), "--json"],
|
|
capture_output=True, text=True,
|
|
)
|
|
assert proc.returncode != 0, proc.stdout
|
|
assert "gc-executor" not in proc.stdout
|
|
assert "report-only gate" in proc.stderr.lower(), proc.stderr
|
|
|
|
|
|
def _plan_with_contract(scan, contract_text, tmp_path, name):
|
|
contract = tmp_path / name
|
|
contract.write_text(contract_text)
|
|
scan_file = tmp_path / f"scan-{name}.json"
|
|
scan_file.write_text(json.dumps(scan))
|
|
proc = subprocess.run(
|
|
[sys.executable, str(PLAN), "--scan", str(scan_file),
|
|
"--contract", str(contract), "--json"],
|
|
capture_output=True, text=True,
|
|
)
|
|
assert proc.returncode == 0, proc.stderr
|
|
return json.loads(proc.stdout)
|
|
|
|
|
|
def test_gate_is_read_from_the_contract_block(tmp_path):
|
|
"""The gate is data-driven by the contract block: the planner excludes the guest
|
|
when the block names it and acts on it when the block does not. Executes the real
|
|
planner interface against both fixtures so the behaviour change is observable."""
|
|
scan = [{"id": 111, "hostname": "tdunna", "ip": "192.168.68.129", "usage_pct": 95}]
|
|
with_gate = (
|
|
"```yaml\n"
|
|
"report_only_guests:\n"
|
|
" - guest: 111\n"
|
|
" hostname: tdunna\n"
|
|
" ip: 192.168.68.129\n"
|
|
" reason: \"fixture reason\"\n"
|
|
"```\n"
|
|
)
|
|
without_gate = (
|
|
"```yaml\n"
|
|
"report_only_guests:\n"
|
|
" - guest: 999\n"
|
|
" reason: \"fixture excludes a different guest\"\n"
|
|
"```\n"
|
|
)
|
|
|
|
gated = _plan_with_contract(scan, with_gate, tmp_path, "gated.prose.md")
|
|
assert gated[0]["action"] == "report-only", gated
|
|
assert gated[0]["reason"] == "fixture reason", gated
|
|
|
|
ungated = _plan_with_contract(scan, without_gate, tmp_path, "ungated.prose.md")
|
|
assert ungated[0]["action"] == "gc-executor", ungated
|
|
assert ungated[0]["action"] != gated[0]["action"]
|