From fab2103f7f2e258a19ff4f6a4a70826a43133e23 Mon Sep 17 00:00:00 2001 From: Shubham Date: Mon, 12 Oct 2026 14:05:09 +1300 Subject: [PATCH 1/3] Let a finding carry A-3's optional subject (first step of #238) A-3 is ratified four of four, so `subject` may now exist. This adds it to make_finding() and does nothing else: no check populates it here. A-3's own order is the field first and a consumer next, and #238's chaining is that consumer. As promised on #300, this lands on its own, ahead of the chaining work, so a change to analysis/findings.py is reviewed on its own terms rather than arriving inside a feature PR. WHAT A-3 ASKS FOR, AND HOW IT IS HELD optional make_finding(..., subject=None) is the default absent is absent no key at all, never "subject": None -- every existing finding keeps exactly its seven F-1 keys every key optional a check fills what it actually knows no new vocabulary a CLOSED key set: src, dst, srcPort, dstPort, protocol never replaces prose evidence.detail is untouched The closed key set is the decision worth reviewing. A free-form dict would let "dest" and "dst" both appear, and two findings about the same address would then fail to join, silently -- the failure A-3 was raised to remove from regex parsing, returning through a typo. Values are validated the way the other fields are, and fail loudly: an address or CIDR for src/dst, an integer 0-65535 for ports, a non-empty name for protocol (lower-cased, so "TCP" and "tcp" join). An empty subject is refused; A-3 says a check with nothing to put in it leaves it out. Mutation-tested: absent becoming subject=None, the key set opened, and protocol not normalised each fail the new tests. A guard test fails if any check starts populating subject before its consumer exists. 1865 passed, 13 skipped with Batfish running. ruff clean on touched files. --- analysis/findings.py | 64 ++++++++++++++++++++++++++- tests/test_finding_subject.py | 82 +++++++++++++++++++++++++++++++++++ 2 files changed, 144 insertions(+), 2 deletions(-) create mode 100644 tests/test_finding_subject.py diff --git a/analysis/findings.py b/analysis/findings.py index e4cc7e9..cd5ea44 100644 --- a/analysis/findings.py +++ b/analysis/findings.py @@ -33,7 +33,8 @@ dashboard shows the client. See the function for the measurements. """ -from typing import Any, Dict +import ipaddress +from typing import Any, Dict, Optional # --- The allowed values, taken verbatim from docs/finding-format.md --------- @@ -120,6 +121,53 @@ def describe_error(error: Exception, limit: int = 200) -> str: return text[: limit - 1] + "…" +#: A-3's `subject` -- the traffic a finding is about, as facts rather than +#: prose (#238). Every key optional; a check fills what it actually knows. +#: +#: A CLOSED set, deliberately. A-3 says `subject` carries "no new vocabulary", +#: and a free-form dict would let `"dest"` and `"dst"` both appear -- which +#: would make two findings about the same address fail to join, silently. +#: That is exactly the failure A-3 was raised to remove from regex parsing, +#: and it would come back through a typo. Add a key here only with a consumer +#: that needs it. +SUBJECT_KEYS = ("src", "dst", "srcPort", "dstPort", "protocol") + + +def _validate_subject(subject: Any) -> Dict[str, Any]: + """Return a checked copy of `subject`, or raise ValueError. + + Fails loudly for the same reason make_finding() does: a malformed join key + is a chain that never forms, and nothing downstream would say so. + """ + if not isinstance(subject, dict): + raise ValueError(f"subject must be a dict, got {type(subject).__name__}") + if not subject: + raise ValueError( + "subject must not be empty -- a check with nothing to put in it " + "leaves it out (A-3)" + ) + unknown = sorted(set(subject) - set(SUBJECT_KEYS)) + if unknown: + raise ValueError(f"subject has unknown key(s) {unknown}; allowed: {SUBJECT_KEYS}") + checked: Dict[str, Any] = {} + for key, value in subject.items(): + if key in ("src", "dst"): + try: + ipaddress.ip_network(str(value), strict=False) + except ValueError: + raise ValueError(f"subject[{key!r}] must be an address or CIDR, got {value!r}") + checked[key] = str(value) + elif key in ("srcPort", "dstPort"): + if isinstance(value, bool) or not isinstance(value, int) or not 0 <= value <= 65535: + raise ValueError(f"subject[{key!r}] must be a port from 0 to 65535, got {value!r}") + checked[key] = value + else: # protocol + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"subject['protocol'] must be a protocol name, got {value!r}") + checked[key] = value.strip().lower() + return checked + + def make_finding( *, # keyword-only: with eight fields, positional arguments are a trap check: str, @@ -130,6 +178,7 @@ def make_finding( source: str, status: str, number: int, + subject: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """Build ONE finding in the F-1 shape, validating every field. @@ -143,6 +192,10 @@ def make_finding( source: where the proof came from, "filename:line" where possible status: "found" | "none" | "error" number: sequence number within this check; 0 for sentinels + subject: OPTIONAL (A-3). The traffic this finding is about, as + machine-readable facts -- see SUBJECT_KEYS. Omitted from the + finding entirely when not given, so no existing finding + changes shape. Never a replacement for `detail`. Raises: ValueError: if any field is outside the agreed vocabulary. This is @@ -158,7 +211,7 @@ def make_finding( if not summary: raise ValueError("summary is required -- it is what the user reads first") - return { + finding = { "id": f"{PREFIX_BY_CHECK[check]}-{number:03d}", "check": check, "severity": severity, @@ -167,6 +220,13 @@ def make_finding( "evidence": {"detail": detail, "source": source}, "status": status, } + # ABSENT, not None. A-3 says a check that cannot fill `subject` truthfully + # leaves it out, and a consumer must handle its absence -- so "no subject" + # is spelled as no key. A `"subject": None` on every finding would change + # the shape of all of them for a field nobody has populated yet. + if subject is not None: + finding["subject"] = _validate_subject(subject) + return finding def no_issues_finding( diff --git a/tests/test_finding_subject.py b/tests/test_finding_subject.py new file mode 100644 index 0000000..9483ebe --- /dev/null +++ b/tests/test_finding_subject.py @@ -0,0 +1,82 @@ +"""A-3's optional `subject` on a finding (#238, first step). + +WHAT THESE PIN + `make_finding()` accepts `subject` and validates it, and does NOTHING else. + No check populates it in this change -- A-3 says an optional field nobody + populates is inert, and the chaining that consumes it (#238) comes next. + + The two properties most likely to be broken later: + * ABSENT IS ABSENT. A finding built without `subject` has exactly the + seven F-1 keys it always had, never `"subject": None`. + * THE KEY SET IS CLOSED. A misspelt key would make two findings about + the same address fail to join, silently -- the failure A-3 was raised + to remove from regex parsing, returning through a typo. +""" + +import pytest + +from analysis import findings + +BASE = dict(check="policy_compliance", severity="high", device="rtr-us5", + summary="s", detail="d", source="x", status="found", number=1) +F1_KEYS = {"id", "check", "severity", "device", "summary", "evidence", "status"} + + +def test_a_finding_without_subject_is_unchanged(): + f = findings.make_finding(**BASE) + assert set(f) == F1_KEYS + assert "subject" not in f, "absent must be no key, not subject=None" + + +def test_a_finding_with_subject_carries_it(): + f = findings.make_finding(**BASE, subject={"dst": "10.20.0.5", "dstPort": 443, "protocol": "tcp"}) + assert set(f) == F1_KEYS | {"subject"} + assert f["subject"] == {"dst": "10.20.0.5", "dstPort": 443, "protocol": "tcp"} + + +def test_every_key_inside_subject_is_optional(): + f = findings.make_finding(**BASE, subject={"dst": "10.20.0.0/16"}) + assert f["subject"] == {"dst": "10.20.0.0/16"} + + +def test_subject_does_not_replace_the_evidence_sentence(): + f = findings.make_finding(**BASE, subject={"dst": "10.20.0.5"}) + assert f["evidence"] == {"detail": "d", "source": "x"} + + +def test_protocol_is_normalised_so_tcp_and_TCP_join(): + f = findings.make_finding(**BASE, subject={"protocol": " TCP "}) + assert f["subject"]["protocol"] == "tcp" + + +@pytest.mark.parametrize("bad,why", [ + ({"dest": "10.20.0.5"}, "unknown key"), + ({}, "empty"), + ("10.20.0.5", "not a dict"), + ({"dst": "not-an-address"}, "bad address"), + ({"dstPort": 70000}, "port out of range"), + ({"dstPort": "443"}, "port as string"), + ({"dstPort": True}, "bool is not a port"), + ({"protocol": ""}, "empty protocol"), +]) +def test_a_malformed_subject_fails_loudly(bad, why): + with pytest.raises(ValueError): + findings.make_finding(**BASE, subject=bad) + + +def test_the_sentinel_helpers_still_produce_the_plain_shape(): + """A sentinel is about a check, not a flow, so it never carries subject.""" + assert set(findings.no_issues_finding(check="policy_compliance", device="r", + summary="s", detail="d", source="x")) == F1_KEYS + assert set(findings.error_finding(check="policy_compliance", device="r", + summary="s", detail="d", source="x")) == F1_KEYS + + +def test_no_check_populates_subject_yet(): + """A-3's order: the field first, a consumer next (#238). If this fails, + a check started emitting `subject` -- update it in the same change that + adds the consumer, so the field is never half-used.""" + import pathlib + producers = [p for p in pathlib.Path("analysis").rglob("*.py") + if p.name != "findings.py" and "subject=" in p.read_text(encoding="utf-8")] + assert producers == [], producers From f88e7ee3cc29894141daf3dd3ff31d056ce95af4 Mon Sep 17 00:00:00 2001 From: Shubham Date: Sun, 18 Oct 2026 15:12:57 +1300 Subject: [PATCH 2/3] Address #390 review: close the protocol values, canonical addresses, key set to A-3 From @ARSH871-bot's review of the first version, each item reproduced first. GOVERNANCE. A-3 ratified an optional `subject`; it never listed the keys, and a closed key set is field vocabulary under all four (CLAUDE.md 7a). The five keys and their value forms are now written into A-3's section of docs/finding-format.md with their own sign-off table -- one of four, my row only. This PR should not merge until that table is full. No check populates `subject`, so changing the set before then costs nothing. CLOSED AT BOTH LEVELS. Closing the keys alone was half the argument. * protocol VALUES are now closed: tcp, udp, icmp, or 6, 17, 1, stored as the lower-case name. "!tcp" -- policy_compliance's own syntax for NOT tcp -- was accepted, and means the opposite. So were "6", "any", "banana". * addresses are canonical: stored as the network in Python's canonical form (10.20.0.5 -> 10.20.0.5/32, 2001:DB8::1 -> 2001:db8::1/128), so two spellings of one host join. Host bits below the prefix (10.20.0.1/16) are refused rather than silently masked. ipaddress objects store as strings. TESTS. Six of thirteen mutants survived the first version. Now killed: src and srcPort each dropped from their branch, 65536 and -1 accepted, the dict check deleted (a list like ["dst"] now separates it from the key check). Also killed: addresses stored as given, host bits masked, the protocol value set reopened. THE TRIPWIRE was relative to the cwd -- from any directory but the repo root it scanned nothing and passed. It now anchors on __file__, asserts it scanned more than twenty files, walks the AST of analysis/, web/ and ai/ for a `subject=` keyword or a `["subject"] =` assignment, and a second test plants both shapes to prove the detector is not blind. Nits: "eight fields" -> "nine"; the address error raises `from None`. 1929 passed, 13 skipped with Batfish running. ruff clean. --- analysis/findings.py | 47 +++++++--- docs/finding-format.md | 41 +++++++++ tests/test_finding_subject.py | 165 ++++++++++++++++++++++++++-------- 3 files changed, 206 insertions(+), 47 deletions(-) diff --git a/analysis/findings.py b/analysis/findings.py index cd5ea44..4d00ead 100644 --- a/analysis/findings.py +++ b/analysis/findings.py @@ -133,8 +133,35 @@ def describe_error(error: Exception, limit: int = 200) -> str: SUBJECT_KEYS = ("src", "dst", "srcPort", "dstPort", "protocol") +#: The protocol VALUES a subject may carry, by name and by IANA number, each +#: mapped to one canonical name (#390 review). Closing the keys alone was half +#: the argument: `tcp` and `6` would never join, and `!tcp` -- which is +#: policy_compliance's own syntax for "not TCP" -- means the opposite of `tcp` +#: and was accepted. A value outside this map is refused, not stored. +SUBJECT_PROTOCOLS = {"tcp": "tcp", "udp": "udp", "icmp": "icmp", + "6": "tcp", "17": "udp", "1": "icmp"} + + +def _canonical_address(key: str, value: Any) -> str: + """One spelling per address, so two findings about it always join. + + `10.20.0.5` and `10.20.0.5/32` are the same host; `2001:DB8::1` and + `2001:db8::1` the same address. Both are stored as the network in + Python's canonical form -- `10.20.0.5/32`, `2001:db8::1/128`. Host bits + set below the prefix (`10.20.0.1/16`) are REFUSED rather than silently + masked, because the caller meant something and we cannot tell what. + """ + try: + return str(ipaddress.ip_network(str(value), strict=True)) + except ValueError: + raise ValueError( + f"subject[{key!r}] must be an address or a CIDR with no host bits " + f"set, got {value!r}" + ) from None + + def _validate_subject(subject: Any) -> Dict[str, Any]: - """Return a checked copy of `subject`, or raise ValueError. + """Return a canonical, checked copy of `subject`, or raise ValueError. Fails loudly for the same reason make_finding() does: a malformed join key is a chain that never forms, and nothing downstream would say so. @@ -152,24 +179,24 @@ def _validate_subject(subject: Any) -> Dict[str, Any]: checked: Dict[str, Any] = {} for key, value in subject.items(): if key in ("src", "dst"): - try: - ipaddress.ip_network(str(value), strict=False) - except ValueError: - raise ValueError(f"subject[{key!r}] must be an address or CIDR, got {value!r}") - checked[key] = str(value) + checked[key] = _canonical_address(key, value) elif key in ("srcPort", "dstPort"): if isinstance(value, bool) or not isinstance(value, int) or not 0 <= value <= 65535: raise ValueError(f"subject[{key!r}] must be a port from 0 to 65535, got {value!r}") checked[key] = value else: # protocol - if not isinstance(value, str) or not value.strip(): - raise ValueError(f"subject['protocol'] must be a protocol name, got {value!r}") - checked[key] = value.strip().lower() + name = SUBJECT_PROTOCOLS.get(str(value).strip().lower()) if not isinstance(value, bool) else None + if name is None: + raise ValueError( + f"subject['protocol'] must be one of tcp, udp, icmp (or 6, 17, 1), " + f"got {value!r}" + ) + checked[key] = name return checked def make_finding( - *, # keyword-only: with eight fields, positional arguments are a trap + *, # keyword-only: with nine fields, positional arguments are a trap check: str, severity: str, device: str, diff --git a/docs/finding-format.md b/docs/finding-format.md index 54c9160..62396d7 100644 --- a/docs/finding-format.md +++ b/docs/finding-format.md @@ -357,6 +357,47 @@ Parse `evidence.detail`. Cheaper today, no amendment, no signatures — and it puts a regex between two findings whose combination is the whole point. The two incidents above are the argument, and both were in code with tests. +#### The key set — NOT covered by the signatures below (#390) + +The table below ratified an optional `subject`. It did **not** ratify which +keys `subject` may hold: this section showed one example (`dst`, `dstPort`, +`protocol`) and never listed them. A closed set of keys and value forms is +field vocabulary, and §7a of `CLAUDE.md` puts that under all four — so it +gets its own signatures here rather than being inferred from the example. +Raised by @ARSH871-bot reviewing #390. + +| key | value | stored as | +|---|---|---| +| `src` | an address or CIDR, no host bits below the prefix | canonical network: `10.20.0.5` → `10.20.0.5/32`, `2001:DB8::1` → `2001:db8::1/128` | +| `dst` | the same | the same | +| `srcPort` | an integer 0–65535 (not a bool) | the integer | +| `dstPort` | the same | the same | +| `protocol` | `tcp`, `udp`, `icmp`, or `6`, `17`, `1` | the name, lower case | + +**Why closed, at both levels.** A free-form dict lets `dest` and `dst` both +appear, and two findings about the same address then fail to join, silently — +the failure A-3 exists to remove from regex parsing, back through a typo. The +same argument applies one level down: `tcp` and `6` never join, `10.20.0.5` +and `10.20.0.5/32` never join, and `!tcp` (policy_compliance's syntax for +*not* TCP) means the opposite of `tcp`. So keys are closed, protocol values +are closed, and addresses are stored in one canonical spelling. Anything else +is refused at `make_finding()`. + +**Adding a key** needs a consumer that needs it, and a row here. + +| Member | Agreed to this key set | +|---|---| +| **Shubham** | ✅ — proposed it in #390 | +| **Arsh** | ⬜ | +| **Ankeet** | ⬜ | +| **Samika** | ⬜ | + +> **NOT RATIFIED — one of four.** `analysis/findings.py` enforces this set in +> #390, and #390 should not merge until this table is full. No check +> populates `subject` yet — `tests/test_finding_subject.py` fails if one +> does — so changing the set before then costs nothing. Changing it after the +> first producer lands is a migration. + | Member | Why it touches them | Agreed | |---|---|---| | **Shubham** | Raised it; #238 is blocked on it and `policy_compliance` would populate it first | ✅ | diff --git a/tests/test_finding_subject.py b/tests/test_finding_subject.py index 9483ebe..d32dd1e 100644 --- a/tests/test_finding_subject.py +++ b/tests/test_finding_subject.py @@ -5,38 +5,81 @@ No check populates it in this change -- A-3 says an optional field nobody populates is inert, and the chaining that consumes it (#238) comes next. - The two properties most likely to be broken later: + The properties most likely to be broken later: * ABSENT IS ABSENT. A finding built without `subject` has exactly the seven F-1 keys it always had, never `"subject": None`. - * THE KEY SET IS CLOSED. A misspelt key would make two findings about - the same address fail to join, silently -- the failure A-3 was raised - to remove from regex parsing, returning through a typo. + * ONE SPELLING PER VALUE. Keys are a closed set, protocols a closed set, + addresses stored canonically -- because two findings about the same + traffic that spell it differently fail to join, silently. That is the + failure A-3 removed from regex parsing, and it must not come back + through a typo or a mask length. """ +import ast +import ipaddress +from pathlib import Path + import pytest from analysis import findings +REPO = Path(__file__).resolve().parents[1] BASE = dict(check="policy_compliance", severity="high", device="rtr-us5", summary="s", detail="d", source="x", status="found", number=1) F1_KEYS = {"id", "check", "severity", "device", "summary", "evidence", "status"} +def subject_of(subject): + return findings.make_finding(**BASE, subject=subject)["subject"] + + +# --- absent is absent --------------------------------------------------------- + def test_a_finding_without_subject_is_unchanged(): f = findings.make_finding(**BASE) assert set(f) == F1_KEYS assert "subject" not in f, "absent must be no key, not subject=None" -def test_a_finding_with_subject_carries_it(): - f = findings.make_finding(**BASE, subject={"dst": "10.20.0.5", "dstPort": 443, "protocol": "tcp"}) - assert set(f) == F1_KEYS | {"subject"} - assert f["subject"] == {"dst": "10.20.0.5", "dstPort": 443, "protocol": "tcp"} +def test_an_explicit_none_is_also_absent(): + assert "subject" not in findings.make_finding(**BASE, subject=None) + + +def test_the_sentinel_helpers_still_produce_the_plain_shape(): + """A sentinel is about a check, not a flow, so it never carries subject.""" + assert set(findings.no_issues_finding(check="policy_compliance", device="r", + summary="s", detail="d", source="x")) == F1_KEYS + assert set(findings.error_finding(check="policy_compliance", device="r", + summary="s", detail="d", source="x")) == F1_KEYS + + +# --- every key, accepted ------------------------------------------------------ + +def test_all_five_keys_are_carried(): + assert subject_of({"src": "10.10.10.0/24", "dst": "10.20.0.5", "srcPort": 49152, + "dstPort": 443, "protocol": "tcp"}) == { + "src": "10.10.10.0/24", "dst": "10.20.0.5/32", "srcPort": 49152, + "dstPort": 443, "protocol": "tcp"} + + +@pytest.mark.parametrize("key", ["src", "dst"]) +def test_each_address_key_is_validated(key): + assert subject_of({key: "10.20.0.5"}) == {key: "10.20.0.5/32"} + with pytest.raises(ValueError): + subject_of({key: "not-an-address"}) + + +@pytest.mark.parametrize("key", ["srcPort", "dstPort"]) +@pytest.mark.parametrize("port", [0, 65535]) +def test_the_port_range_ends_are_accepted(key, port): + assert subject_of({key: port}) == {key: port} -def test_every_key_inside_subject_is_optional(): - f = findings.make_finding(**BASE, subject={"dst": "10.20.0.0/16"}) - assert f["subject"] == {"dst": "10.20.0.0/16"} +@pytest.mark.parametrize("key", ["srcPort", "dstPort"]) +@pytest.mark.parametrize("port", [-1, 65536, "443", True]) +def test_out_of_range_and_non_integer_ports_are_refused(key, port): + with pytest.raises(ValueError): + subject_of({key: port}) def test_subject_does_not_replace_the_evidence_sentence(): @@ -44,39 +87,87 @@ def test_subject_does_not_replace_the_evidence_sentence(): assert f["evidence"] == {"detail": "d", "source": "x"} -def test_protocol_is_normalised_so_tcp_and_TCP_join(): - f = findings.make_finding(**BASE, subject={"protocol": " TCP "}) - assert f["subject"]["protocol"] == "tcp" +# --- one spelling per value --------------------------------------------------- + +@pytest.mark.parametrize("a,b", [ + ("10.20.0.5", "10.20.0.5/32"), + ("2001:DB8::1", "2001:db8::1"), + (ipaddress.ip_address("10.20.0.5"), "10.20.0.5"), +]) +def test_two_spellings_of_one_address_are_stored_identically(a, b): + """They must join. Stored as strings, so an `ipaddress` object is too.""" + assert subject_of({"dst": a}) == subject_of({"dst": b}) + assert isinstance(subject_of({"dst": a})["dst"], str) + + +def test_host_bits_below_the_prefix_are_refused_not_masked(): + with pytest.raises(ValueError): + subject_of({"dst": "10.20.0.1/16"}) + + +@pytest.mark.parametrize("value,canonical", [ + ("tcp", "tcp"), (" TCP ", "tcp"), ("6", "tcp"), (6, "tcp"), + ("udp", "udp"), ("17", "udp"), ("icmp", "icmp"), (1, "icmp"), +]) +def test_protocol_names_and_numbers_map_to_one_name(value, canonical): + assert subject_of({"protocol": value}) == {"protocol": canonical} + + +@pytest.mark.parametrize("value", ["!tcp", "any", "banana", "", True]) +def test_protocols_outside_the_closed_set_are_refused(value): + """`!tcp` is policy_compliance's own syntax for NOT tcp -- the opposite.""" + with pytest.raises(ValueError): + subject_of({"protocol": value}) + +# --- malformed subjects -------------------------------------------------------- -@pytest.mark.parametrize("bad,why", [ - ({"dest": "10.20.0.5"}, "unknown key"), - ({}, "empty"), - ("10.20.0.5", "not a dict"), - ({"dst": "not-an-address"}, "bad address"), - ({"dstPort": 70000}, "port out of range"), - ({"dstPort": "443"}, "port as string"), - ({"dstPort": True}, "bool is not a port"), - ({"protocol": ""}, "empty protocol"), +@pytest.mark.parametrize("bad", [ + {"dest": "10.20.0.5"}, # unknown key + {}, # empty + "10.20.0.5", # not a dict + ["dst"], # not a dict -- and NOT caught by the key check ]) -def test_a_malformed_subject_fails_loudly(bad, why): +def test_a_malformed_subject_fails_loudly(bad): with pytest.raises(ValueError): findings.make_finding(**BASE, subject=bad) -def test_the_sentinel_helpers_still_produce_the_plain_shape(): - """A sentinel is about a check, not a flow, so it never carries subject.""" - assert set(findings.no_issues_finding(check="policy_compliance", device="r", - summary="s", detail="d", source="x")) == F1_KEYS - assert set(findings.error_finding(check="policy_compliance", device="r", - summary="s", detail="d", source="x")) == F1_KEYS +# --- tripwire: nothing produces it yet ---------------------------------------- + +def _subject_writers(path): + """Calls passing `subject=` and assignments to `x["subject"]`.""" + tree = ast.parse(path.read_text(encoding="utf-8")) + for node in ast.walk(tree): + if isinstance(node, ast.keyword) and node.arg == "subject": + yield node.lineno if hasattr(node, "lineno") else 0 + if isinstance(node, (ast.Assign, ast.AugAssign)): + targets = node.targets if isinstance(node, ast.Assign) else [node.target] + for t in targets: + if (isinstance(t, ast.Subscript) and isinstance(t.slice, ast.Constant) + and t.slice.value == "subject"): + yield node.lineno -def test_no_check_populates_subject_yet(): +def test_no_product_code_populates_subject_yet(): """A-3's order: the field first, a consumer next (#238). If this fails, - a check started emitting `subject` -- update it in the same change that - adds the consumer, so the field is never half-used.""" - import pathlib - producers = [p for p in pathlib.Path("analysis").rglob("*.py") - if p.name != "findings.py" and "subject=" in p.read_text(encoding="utf-8")] - assert producers == [], producers + something started emitting `subject` -- land it in the same change that + adds the consumer, so the field is never half-used. + + Anchored on this file, not the working directory, and it asserts it + actually scanned something: a relative path from the wrong directory made + the first version pass while scanning nothing. + """ + files = [p for d in ("analysis", "web", "ai") for p in (REPO / d).rglob("*.py") + if p.name != "findings.py"] + assert len(files) > 20, f"the tripwire scanned only {len(files)} files" + writers = {str(p.relative_to(REPO)): lines for p in files + if (lines := list(_subject_writers(p)))} + assert writers == {}, writers + + +def test_the_tripwire_catches_both_shapes(tmp_path): + """Proves the detector, so the test above cannot pass by being blind.""" + planted = tmp_path / "x.py" + planted.write_text('make_finding(subject={"dst": "1.1.1.1"})\nf["subject"] = {}\n') + assert len(list(_subject_writers(planted))) == 2 From d01ca23c9936ed422debe932b7767ea3058b5c22 Mon Sep 17 00:00:00 2001 From: patelankeet2 Date: Sun, 18 Oct 2026 15:33:25 +1300 Subject: [PATCH 3/3] Sign off on #390's closed subject key set (A-3 key-set table) Agreed to the key set after checking it against the actual code rather than the table alone: SUBJECT_KEYS, SUBJECT_PROTOCOLS and _canonical_address() in analysis/findings.py match the table exactly. Ran tests/test_finding_subject.py (42 passed) and independently mutation-tested three of the claimed guards rather than trusting the PR's own table: - removed strict=True from the address canonicalisation -> the host-bits-set test failed for the right reason - removed address canonicalisation entirely -> 6 tests failed for the right reason, including the two-spellings-join test - removed the bool guard on srcPort/dstPort -> failed for the right reason (bool is a subclass of int in Python, so True would otherwise be silently accepted as port 1) One non-blocking note for the PR comment: the equivalent bool guard on protocol is defensive redundancy, not load-bearing - no string form of True/False collides with SUBJECT_PROTOCOLS, so removing it does not fail any test. Full suite on f88e7ee: 1929 passed, 13 skipped, matching the PR's own number. --- docs/finding-format.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/finding-format.md b/docs/finding-format.md index 62396d7..24e90fe 100644 --- a/docs/finding-format.md +++ b/docs/finding-format.md @@ -389,7 +389,7 @@ is refused at `make_finding()`. |---|---| | **Shubham** | ✅ — proposed it in #390 | | **Arsh** | ⬜ | -| **Ankeet** | ⬜ | +| **Ankeet** | ✅ - checked `SUBJECT_KEYS`, `SUBJECT_PROTOCOLS` and `_canonical_address()` directly against this table and mutation-tested the two checks that matter most (host-bits-set addresses, and `True` passing as a port), both killed for the right reason | | **Samika** | ⬜ | > **NOT RATIFIED — one of four.** `analysis/findings.py` enforces this set in