Let a finding carry A-3's optional subject (first step of #238)

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(

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 | ✅ |

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