From 92d2637cb0b2bca45b61bcde8afaff0be1303e72 Mon Sep 17 00:00:00 2001 From: Justin Soderberg <723679+sodejm@users.noreply.github.com> Date: Mon, 5 Oct 2026 21:49:07 -0400 Subject: [PATCH] test(attack-path): prove bounded ingestion and search --- cops/evidence/canonical.py | 17 ++- .../attack-path-workbench/README.md | 4 +- .../_runtime/cops/evidence/canonical.py | 17 ++- .../attackpath/_runtime/source-manifest.json | 2 +- .../tests/test_ingestion.py | 114 +++++++++++++++++- .../tests/test_workbench.py | 91 ++++++++++++++ tests/test_evidence_contract.py | 24 ++++ 7 files changed, 258 insertions(+), 11 deletions(-) diff --git a/cops/evidence/canonical.py b/cops/evidence/canonical.py index 87a285f..c306c2a 100644 --- a/cops/evidence/canonical.py +++ b/cops/evidence/canonical.py @@ -30,16 +30,22 @@ def utc_now(): def canonical(value, *, max_bytes=1024 * 1024, max_depth=32): if type(max_bytes) is not int or max_bytes <= 0 or type(max_depth) is not int or not 1 <= max_depth <= 64: raise EvidenceError("invalid_limit") - pending = [(value, 1)] + # Pending depth is the number of enclosing containers. Count each container + # before inspecting children so empty and populated containers have the same + # depth, matching the streaming decoder's structural preflight. + pending = [(value, 0)] nodes = 0 minimum_bytes = 0 while pending: item, depth = pending.pop() nodes += 1 minimum_bytes += 1 - if depth > max_depth or nodes > max_bytes: + if nodes > max_bytes: raise EvidenceError("payload_limit") if type(item) is dict: + container_depth = depth + 1 + if container_depth > max_depth: + raise EvidenceError("payload_limit") minimum_bytes += 1 if nodes + len(pending) + len(item) > max_bytes: raise EvidenceError("payload_limit") @@ -49,12 +55,15 @@ def canonical(value, *, max_bytes=1024 * 1024, max_depth=32): minimum_bytes += len(key) + 3 if minimum_bytes > max_bytes: raise EvidenceError("payload_limit") - pending.extend((child, depth + 1) for child in item.values()) + pending.extend((child, container_depth) for child in item.values()) elif type(item) is list: + container_depth = depth + 1 + if container_depth > max_depth: + raise EvidenceError("payload_limit") minimum_bytes += 1 if nodes + len(pending) + len(item) > max_bytes: raise EvidenceError("payload_limit") - pending.extend((child, depth + 1) for child in item) + pending.extend((child, container_depth) for child in item) elif type(item) is float: if not math.isfinite(item): raise EvidenceError("invalid_json") diff --git a/plugins/detection-hunting/attack-path-workbench/README.md b/plugins/detection-hunting/attack-path-workbench/README.md index f82cf85..a0a7fb3 100644 --- a/plugins/detection-hunting/attack-path-workbench/README.md +++ b/plugins/detection-hunting/attack-path-workbench/README.md @@ -39,7 +39,9 @@ The analyzer generates four detailed output files in your chosen directory: > [!NOTE] > The illustrative fixture contains synthetic test data designed to demonstrate graph analysis algorithms safely. Structural routes are conditional on exploiting starting findings, and candidate routes highlight explicit evidence gaps. -The illustrative graph search considers routes of at most eight transitions. Its default hard ceilings are 50,000 expansions, 10,000 frontier entries, 1,000 complete paths, 1,000 partial paths, 100 emitted paths, and 64 MiB of serialized report JSON. The `analyze` command exposes corresponding `--max-*` options that can only tighten these ceilings. The `attackpath.report/v2` search receipt records effective limits, consumed counts, completeness within the eight-transition search, and the stop reason. The policy hash is part of the run identity. A legacy v1 report has no such receipt; audit accepts it only if replay finishes within current hard limits and does not infer historical search completeness. +The illustrative graph search considers routes of at most eight transitions. Its default hard ceilings are 50,000 expansions, 10,000 frontier entries, 1,000 complete paths, 1,000 partial paths, 100 emitted paths, and 64 MiB of serialized report JSON. The `analyze` command exposes corresponding `--max-*` options that can only tighten these ceilings. The `attackpath.report/v2` search receipt records effective limits, exact consumed counts, completeness within the eight-transition search, and the stop reason. These counters provide the reproducible work bound; elapsed time and peak memory still depend on the host runtime. The policy hash is part of the run identity. A legacy v1 report has no such receipt; audit accepts it only if replay finishes within current hard limits and does not infer historical search completeness. + +Local evidence ingestion defaults to 4 MiB per file, 64 MiB in aggregate, 64 files, 1 MiB per physical JSONL line, 50,000 admitted records, and 32 nested JSON containers. Operators may tighten or raise these values only within hard maxima of 16 MiB per file, 256 MiB in aggregate, 256 files, 4 MiB per line, 200,000 records, and 64 containers. Each limit is inclusive: the exact boundary is accepted and the next byte, file, record, line byte, or container is rejected before report completion. Aggregate bytes and admitted records accumulate across reads and batches; JSON depth counts containers consistently whether they are empty or populated. --- diff --git a/plugins/detection-hunting/attack-path-workbench/attackpath/_runtime/cops/evidence/canonical.py b/plugins/detection-hunting/attack-path-workbench/attackpath/_runtime/cops/evidence/canonical.py index 87a285f..c306c2a 100644 --- a/plugins/detection-hunting/attack-path-workbench/attackpath/_runtime/cops/evidence/canonical.py +++ b/plugins/detection-hunting/attack-path-workbench/attackpath/_runtime/cops/evidence/canonical.py @@ -30,16 +30,22 @@ def utc_now(): def canonical(value, *, max_bytes=1024 * 1024, max_depth=32): if type(max_bytes) is not int or max_bytes <= 0 or type(max_depth) is not int or not 1 <= max_depth <= 64: raise EvidenceError("invalid_limit") - pending = [(value, 1)] + # Pending depth is the number of enclosing containers. Count each container + # before inspecting children so empty and populated containers have the same + # depth, matching the streaming decoder's structural preflight. + pending = [(value, 0)] nodes = 0 minimum_bytes = 0 while pending: item, depth = pending.pop() nodes += 1 minimum_bytes += 1 - if depth > max_depth or nodes > max_bytes: + if nodes > max_bytes: raise EvidenceError("payload_limit") if type(item) is dict: + container_depth = depth + 1 + if container_depth > max_depth: + raise EvidenceError("payload_limit") minimum_bytes += 1 if nodes + len(pending) + len(item) > max_bytes: raise EvidenceError("payload_limit") @@ -49,12 +55,15 @@ def canonical(value, *, max_bytes=1024 * 1024, max_depth=32): minimum_bytes += len(key) + 3 if minimum_bytes > max_bytes: raise EvidenceError("payload_limit") - pending.extend((child, depth + 1) for child in item.values()) + pending.extend((child, container_depth) for child in item.values()) elif type(item) is list: + container_depth = depth + 1 + if container_depth > max_depth: + raise EvidenceError("payload_limit") minimum_bytes += 1 if nodes + len(pending) + len(item) > max_bytes: raise EvidenceError("payload_limit") - pending.extend((child, depth + 1) for child in item) + pending.extend((child, container_depth) for child in item) elif type(item) is float: if not math.isfinite(item): raise EvidenceError("invalid_json") diff --git a/plugins/detection-hunting/attack-path-workbench/attackpath/_runtime/source-manifest.json b/plugins/detection-hunting/attack-path-workbench/attackpath/_runtime/source-manifest.json index c9ef32b..732a2bf 100644 --- a/plugins/detection-hunting/attack-path-workbench/attackpath/_runtime/source-manifest.json +++ b/plugins/detection-hunting/attack-path-workbench/attackpath/_runtime/source-manifest.json @@ -11,7 +11,7 @@ }, { "path": "cops/evidence/canonical.py", - "sha256": "4f8a90a6738995313a2a0fc4723d15d062c18df3842d559d1a329594b68d2bae" + "sha256": "18deb4b9f2d68ddde1b2429d0d09f500facbcdef9bf16a6254c358f1005ef680" }, { "path": "cops/evidence/contract.py", diff --git a/plugins/detection-hunting/attack-path-workbench/tests/test_ingestion.py b/plugins/detection-hunting/attack-path-workbench/tests/test_ingestion.py index f611cb3..3ec0102 100644 --- a/plugins/detection-hunting/attack-path-workbench/tests/test_ingestion.py +++ b/plugins/detection-hunting/attack-path-workbench/tests/test_ingestion.py @@ -6,7 +6,8 @@ import unittest from attackpath.core import GateError, analyze -from attackpath.ingestion import IngestError, Limits, RunBudget, parse_json, read_regular +from attackpath.ingestion import (CEILINGS, IngestError, Limits, RunBudget, + iter_jsonl, parse_json, read_regular) FIXTURE = Path(__file__).resolve().parents[1] / "fixtures" / "illustrative" @@ -36,6 +37,117 @@ def test_file_and_aggregate_bounds_include_rechecks(self): with self.assertRaisesRegex(IngestError, "file_limit"): read_regular(root, "data", 3, RunBudget(limits)) + def test_configured_ingestion_ceilings_accept_below_and_at_but_reject_above(self): + for name, ceiling in CEILINGS.items(): + for offset in (-1, 0, 1): + value = ceiling + offset + with self.subTest(limit=name, value=value): + if offset > 0: + with self.assertRaisesRegex(IngestError, "invalid_limit"): + Limits.from_values({name: value}) + else: + self.assertEqual(value, getattr(Limits.from_values({name: value}), name)) + + def test_file_byte_limit_accepts_below_and_at_but_rejects_above(self): + limits = Limits.from_values({"file_bytes": 3}) + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + for size in (2, 3, 4): + (root / "data").write_bytes(b"x" * size) + with self.subTest(size=size): + if size > limits.file_bytes: + with self.assertRaisesRegex(IngestError, "file_limit"): + read_regular(root, "data", limits.file_bytes, RunBudget(limits)) + else: + result = read_regular(root, "data", limits.file_bytes, RunBudget(limits)) + self.assertEqual(size, len(result.data)) + + def test_aggregate_byte_limit_accepts_below_and_at_but_rejects_above(self): + limits = Limits.from_values({"file_bytes": 1, "total_bytes": 3}) + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + for index in range(4): + (root / f"data-{index}").write_bytes(b"x") + budget = RunBudget(limits) + for index in range(2): + self.assertEqual(b"x", read_regular( + root, f"data-{index}", limits.file_bytes, budget).data) + self.assertEqual(2, budget.bytes) + self.assertEqual(b"x", read_regular( + root, "data-2", limits.file_bytes, budget).data) + self.assertEqual(3, budget.bytes) + with self.assertRaisesRegex(IngestError, "total_byte_limit"): + read_regular(root, "data-3", limits.file_bytes, budget) + self.assertEqual(3, budget.bytes) + + def test_file_count_limit_accepts_below_and_at_but_rejects_above(self): + limits = Limits.from_values({"files": 2}) + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + for index in range(3): + (root / f"data-{index}").write_bytes(b"x") + for count in (1, 2, 3): + with self.subTest(count=count): + budget = RunBudget(limits) + if count > limits.files: + with self.assertRaisesRegex(IngestError, "file_count_limit"): + for index in range(count): + read_regular(root, f"data-{index}", limits.file_bytes, budget) + else: + for index in range(count): + read_regular(root, f"data-{index}", limits.file_bytes, budget) + self.assertEqual(count, budget.files) + + def test_record_limit_accepts_below_and_at_but_rejects_above(self): + limits = Limits.from_values({"records": 2}) + for count in (1, 2, 3): + raw = json.dumps({"records": [{}] * count}).encode() + with self.subTest(count=count): + if count > limits.records: + with self.assertRaisesRegex(IngestError, "record_limit"): + parse_json(raw, limits, max_bytes=100, record_cap=limits.records) + else: + parsed = parse_json(raw, limits, max_bytes=100, + record_cap=limits.records) + self.assertEqual(count, len(parsed["records"])) + budget = RunBudget(limits) + budget.add_records(count) + self.assertEqual(count, budget.records) + + def test_record_budget_accumulates_across_admitted_batches(self): + budget = RunBudget(Limits.from_values({"records": 3})) + budget.add_records(2) + self.assertEqual(2, budget.records) + budget.add_records(1) + self.assertEqual(3, budget.records) + with self.assertRaisesRegex(IngestError, "record_limit"): + budget.add_records(1) + self.assertEqual(3, budget.records) + + def test_physical_line_limit_accepts_below_and_at_but_rejects_above(self): + limits = Limits.from_values({"line_bytes": 3}) + for size in (2, 3, 4): + budget = RunBudget(limits) + with self.subTest(size=size): + if size > limits.line_bytes: + with self.assertRaisesRegex(IngestError, "line_limit"): + list(iter_jsonl(b"x" * size + b"\n", limits, budget)) + else: + self.assertEqual([b"x" * size], list(iter_jsonl( + b"x" * size + b"\n", limits, budget))) + self.assertEqual(1, budget.lines) + + def test_json_depth_limit_accepts_below_and_at_but_rejects_above(self): + limits = Limits.from_values({"json_depth": 2}) + for depth in (1, 2, 3): + raw = b"[" * depth + b"0" + b"]" * depth + with self.subTest(depth=depth): + if depth > limits.json_depth: + with self.assertRaisesRegex(IngestError, "json_depth_limit"): + parse_json(raw, limits, max_bytes=100) + else: + self.assertIsNotNone(parse_json(raw, limits, max_bytes=100)) + def test_reader_denies_links_and_nonregular_files(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) diff --git a/plugins/detection-hunting/attack-path-workbench/tests/test_workbench.py b/plugins/detection-hunting/attack-path-workbench/tests/test_workbench.py index 69b93fa..f0ccc0e 100644 --- a/plugins/detection-hunting/attack-path-workbench/tests/test_workbench.py +++ b/plugins/detection-hunting/attack-path-workbench/tests/test_workbench.py @@ -10,6 +10,8 @@ import subprocess import sys import tempfile +import time +import tracemalloc import unittest from attackpath.core import (GateError, analyze, audit_report, canonical, @@ -167,6 +169,95 @@ def test_frontier_complete_and_partial_limits_stop_search(self) -> None: self.assertFalse(receipt["complete"]) self.assertEqual("best_discovered", receipt["ranking_scope"]) + def test_high_branch_search_is_bounded_repeatable_and_visibly_incomplete(self) -> None: + width = 12 + depth = 5 + nodes = { + "START": {"record": {"node_type": "asset"}, "class": "observed", + "evidence_id": "N-START", "confidence": "high"} + } + for layer in range(1, depth + 1): + for index in range(width): + name = f"L{layer}-{index:02d}" + nodes[name] = {"record": {"node_type": "asset"}, "class": "observed", + "evidence_id": f"N-{name}", "confidence": "high"} + findings = [{"record": {"id": "F-START", "asset_ref": "START"}, + "class": "observed", "evidence_id": "E-F-START", + "confidence": "high", "scope": "SCOPE"}] + edges = [] + prior = ["START"] + for layer in range(1, depth + 1): + current = [f"L{layer}-{index:02d}" for index in range(width)] + for source in prior: + for target in current: + edge_id = f"E-{source}-{target}" + edges.append({ + "record": {"id": edge_id, "from": source, "to": target, + "relation": "reachable_from", + "preconditions": (["finding_on_asset"] if source == "START" + else ["network_reachability"]), + "postcondition": "network_reachability", "support": "observed"}, + "evidence_id": edge_id, "class": "observed", "confidence": "high", + }) + prior = current + crowns = [{"asset_ref": f"L{depth}-{index:02d}", "priority": index + 1} + for index in range(width)] + limits = SearchLimits.from_values({"expansions": 600, "frontier": 1_000, + "complete_paths": 1_000, + "partial_paths": 1_000, + "emitted_paths": 8}) + + # Fixture construction is intentionally outside both measurements. The peak cap + # allows one 4 KiB page for every fixed graph item and permitted expansion. That + # leaves headroom for Python object overhead while still catching retained queued + # or discovered state that grows materially beyond the bounded search workload. + memory_ceiling = 4 * 1024 * (len(nodes) + len(edges) + limits.expansions) + tracemalloc.start() + try: + started = time.perf_counter() + first = trace_paths(nodes, findings, edges, crowns, limits, max_depth=depth) + search_seconds = time.perf_counter() - started + _, peak_bytes = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + second = trace_paths(nodes, findings, edges, crowns, limits, max_depth=depth) + + paths, partial, receipt = first + self.assertEqual(first, second) + self.assertEqual( + {"expansions": 600, "max_frontier": 56, "complete_paths": 547, + "partial_paths": 0, "emitted_paths": 8}, + receipt["consumed"], + ) + self.assertEqual("expansion_limit", receipt["stop_reason"]) + self.assertFalse(receipt["complete"]) + self.assertEqual("best_discovered", receipt["ranking_scope"]) + self.assertEqual([], partial) + self.assertEqual(8, len(paths)) + self.assertLessEqual(search_seconds, 5.0) + self.assertLessEqual(peak_bytes, memory_ceiling) + self.assertTrue(all(path["target"] == f"L{depth}-00" for path in paths)) + self.assertEqual( + [ + ("P-02847eeae0883daf", ["E-START-L1-00", "E-L1-00-L2-00", + "E-L2-00-L3-00", "E-L3-00-L4-09", + "E-L4-09-L5-00"]), + ("P-1a31062c62b8f6a5", ["E-START-L1-00", "E-L1-00-L2-00", + "E-L2-00-L3-03", "E-L3-03-L4-01", + "E-L4-01-L5-00"]), + ("P-1e9ad07a7a67ca9b", ["E-START-L1-00", "E-L1-00-L2-00", + "E-L2-00-L3-00", "E-L3-00-L4-05", + "E-L4-05-L5-00"]), + ], + [(path["path_id"], [step["edge_id"] for step in path["steps"]]) + for path in paths[:3]], + ) + self.assertEqual( + [{"crown_priority": 1, "business_rating": "unrated", + "supported_step_count": depth, "total_step_count": depth}] * 3, + [path["rank_inputs"] for path in paths[:3]], + ) + def test_v1_report_audits_without_claiming_search_completeness(self) -> None: manifest = FIXTURE / "input.json" report = analyze(manifest) diff --git a/tests/test_evidence_contract.py b/tests/test_evidence_contract.py index aa4bbf8..6d41401 100644 --- a/tests/test_evidence_contract.py +++ b/tests/test_evidence_contract.py @@ -91,6 +91,30 @@ def test_bounded_json(raw): decode_json(raw, max_depth=3) +@pytest.mark.parametrize( + ("payload", "encoded", "accepted"), + [ + ([[0]], b"[[0]]", True), + ([[[0]]], b"[[[0]]]", True), + ([[[[0]]]], b"[[[[0]]]]", False), + ([[]], b"[[]]", True), + ([[[]]], b"[[[]]]", True), + ([[[[]]]], b"[[[[]]]]", False), + ], + ids=["populated-below", "populated-at", "populated-above", + "empty-below", "empty-at", "empty-above"], +) +def test_container_depth_consistent_for_canonical_and_decode(payload, encoded, accepted): + if accepted: + assert canonical(payload, max_depth=3) == encoded + assert decode_json(encoded, max_depth=3) == payload + return + with pytest.raises(EvidenceError, match="payload_limit"): + canonical(payload, max_depth=3) + with pytest.raises(EvidenceError, match="payload_limit"): + decode_json(encoded, max_depth=3) + + def test_freshness_does_not_invent_observation_time(): acquisition = receipt() assert assess(envelope(), acquisition, as_of="2026-09-28T00:10:00Z", max_age_seconds=60) == {