diff --git a/scripts/disk-gc-plan.py b/scripts/disk-gc-plan.py index 3d5228c..4130815 100755 --- a/scripts/disk-gc-plan.py +++ b/scripts/disk-gc-plan.py @@ -42,6 +42,7 @@ 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") +IDENTITY_TYPE_PREFIX = re.compile(r"^(?:lxc|qemu)/") UNIDENTIFIED_REASON = "unidentified target - refusing to schedule GC" @@ -64,14 +65,40 @@ def load_report_only_guests(contract_path: pathlib.Path) -> list[dict]: ) +def _canonical_number(number: float) -> str: + if float(number).is_integer(): + return str(int(number)) + return str(number).strip().lower() + + +def _normalize_identity(value: object) -> str: + """Canonicalise a guest identity so differently-encoded ids compare equal: + numeric and numeric-string ids collapse to an integer string, Proxmox + type prefixes and leading zeros are stripped, and hostnames/IPs are only + trimmed and lowercased.""" + if isinstance(value, bool): + return str(value).strip().lower() + if isinstance(value, (int, float)): + return _canonical_number(float(value)) + text = str(value).strip().lower() + text = IDENTITY_TYPE_PREFIX.sub("", text) + try: + return _canonical_number(float(text)) + except ValueError: + return text + + 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 IDENTITY_FIELDS: value = entry.get(field) - if value is not None and str(value).strip(): - out.add(str(value).strip().lower()) + if value is None: + continue + key = _normalize_identity(value) + if key: + out.add(key) return out diff --git a/tests/test_disk_gc_report_only.py b/tests/test_disk_gc_report_only.py index 208d763..bde9df2 100644 --- a/tests/test_disk_gc_report_only.py +++ b/tests/test_disk_gc_report_only.py @@ -61,6 +61,20 @@ def test_exclusion_matches_on_any_identity_key(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},