diff --git a/analysis/findings.py b/analysis/findings.py index e4cc7e9..4d00ead 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,8 +121,82 @@ 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") + + +#: 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 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. + """ + 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"): + 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 + 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, @@ -130,6 +205,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 +219,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 +238,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 +247,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/docs/finding-format.md b/docs/finding-format.md index 54c9160..24e90fe 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** | ✅ - 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 +> #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 new file mode 100644 index 0000000..d32dd1e --- /dev/null +++ b/tests/test_finding_subject.py @@ -0,0 +1,173 @@ +"""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 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`. + * 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_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} + + +@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(): + f = findings.make_finding(**BASE, subject={"dst": "10.20.0.5"}) + assert f["evidence"] == {"detail": "d", "source": "x"} + + +# --- 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", [ + {"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): + with pytest.raises(ValueError): + findings.make_finding(**BASE, subject=bad) + + +# --- 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_product_code_populates_subject_yet(): + """A-3's order: the field first, a consumer next (#238). If this fails, + 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