feat(learning-harness): closed family taxonomy with a verifier per family

Task 1.1. Ten families are the only key that routes work and scores a pair;
CWE ids and mechanism ids are attributes reached through a family. Each family
names what can confirm a claim in it: a sandbox canary for injection, file path
and deserialization, static corroboration for access control, nothing for the
rest. The unknown family carries the mechanisms that span families and the
weaknesses this project deliberately does not target, so a declined row is
visible instead of silently unresolved.

Every labeled corpus weakness resolves, no weakness id is claimed twice, and
every mechanism in the closed vocabulary belongs to exactly one family. Open
redirect joins server-side request forgery under one untrusted-destination
family because both are untrusted input choosing a request destination; that
places 34 real rows instead of discarding them. The relation lookup answers
same, parent-child, lateral or fabricated, and abstention earns no partial
credit: unknown against a real family is lateral.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG

src/openultrasast/learning/__init__.py

@@ -0,0 +1,13 @@
+"""The learning harness: classified detectors improved one family at a time (learning-harness spec).
+
+Layers left to right, and nothing imports rightward: ``families`` -> ``classify`` -> ``scoring`` ->
+``split`` -> ``detectors`` and ``verifiers`` -> ``rounds`` -> ``publish``. ``pairs`` and ``benchmark``
+never import this package; the corpus stores a family label as an opaque string and the classifier
+validates it.
+"""
+
+from __future__ import annotations
+
+from .families import FAMILY_IDS, VERIFIER_KINDS, FamiliesError, Family, FamilyTaxonomy, load_families
+
+__all__ = ["FAMILY_IDS", "VERIFIER_KINDS", "FamiliesError", "Family", "FamilyTaxonomy", "load_families"]

src/openultrasast/learning/families.py

@@ -0,0 +1,177 @@
+"""The closed family taxonomy (learning-harness, Req 1).
+
+Ten families, loaded from ``ruleset/families.toml``. A family is the only key that routes work and
+scores a pair: CWE ids and mechanism ids are attributes reached *through* a family, so no caller may
+group by them. Every family names what can verify a claim in it (a sandbox canary, static
+corroboration, or nothing), which is what keeps reporting honest one layer up. An unknown field, a
+family outside the closed set, a missing verifier kind or a CWE claimed twice is a data bug and fails
+loud, naming the offender.
+"""
+
+from __future__ import annotations
+
+import tomllib
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Literal
+
+FAMILY_IDS = (
+    "injection",
+    "path",
+    "deserialization",
+    "access_control",
+    "output_encoding",
+    "untrusted_destination",
+    "prototype",
+    "config_secrets",
+    "memory",
+    "unknown",
+)
+VERIFIER_KINDS = ("canary", "static", "none")
+Relation = Literal["same", "parent_child", "lateral", "fabricated"]
+DEFAULT_FAMILIES_PATH = Path(__file__).resolve().parents[1] / "ruleset" / "families.toml"
+
+_FAMILY_FIELDS = frozenset({"id", "description", "cwes", "mechanisms", "verifier", "parent"})
+_TOP_FIELDS = frozenset({"version", "family"})
+
+
+class FamiliesError(ValueError):
+    """Raised when the taxonomy file names a family, field, verifier or CWE outside the closed sets."""
+
+
+@dataclass(frozen=True)
+class Family:
+    id: str  # FAMILY_IDS
+    description: str
+    cwes: frozenset[str]
+    mechanisms: frozenset[str]
+    verifier: str  # VERIFIER_KINDS
+    parent: str | None = None  # a family this one specializes; partial credit when one side is the other's parent
+
+
+@dataclass(frozen=True)
+class FamilyTaxonomy:
+    version: str
+    families: tuple[Family, ...]
+
+    def by_id(self, family_id: str) -> Family:
+        for family in self.families:
+            if family.id == family_id:
+                return family
+        raise FamiliesError(f"unknown family {family_id!r}; expected one of {', '.join(f.id for f in self.families)}")
+
+    def family_of_cwe(self, cwe: str) -> Family | None:
+        """The family that claims this CWE, or None when nobody has placed it yet (a data gap, not `unknown`)."""
+        token = cwe.strip().upper()
+        return next((family for family in self.families if token in family.cwes), None)
+
+    def family_of_mechanism(self, mechanism: str) -> Family | None:
+        return next((family for family in self.families if mechanism in family.mechanisms), None)
+
+    def related(self, left: str, right: str) -> Relation:
+        """How two family answers relate. Abstention is not partial credit: ``unknown`` against a real
+        family is ``lateral``, never ``parent_child``."""
+        known = {family.id: family for family in self.families}
+        if left not in known or right not in known:
+            return "fabricated"
+        if left == right:
+            return "same"
+        if known[left].parent == right or known[right].parent == left:
+            return "parent_child"
+        return "lateral"
+
+
+def load_families(path: Path | None = None, *, require_all: bool = True) -> FamilyTaxonomy:
+    """Load the taxonomy. ``require_all`` demands the full closed set in order (the shipped file);
+    a scoped fixture may declare a subset, but never a family outside ``FAMILY_IDS``."""
+    source = path if path is not None else DEFAULT_FAMILIES_PATH
+    try:
+        payload = tomllib.loads(source.read_text(encoding="utf-8"))
+    except OSError as exc:
+        raise FamiliesError(f"families file unreadable: {source}") from exc
+    except tomllib.TOMLDecodeError as exc:
+        raise FamiliesError(f"families file is not valid TOML: {source}: {exc}") from exc
+    unknown_top = sorted(set(payload) - _TOP_FIELDS)
+    if unknown_top:
+        raise FamiliesError(f"{source}: unknown top-level field(s) {', '.join(unknown_top)}")
+    version = str(payload.get("version", "")).strip()
+    if not version:
+        raise FamiliesError(f"{source}: version is required")
+    families: list[Family] = []
+    claimed: dict[str, str] = {}
+    for entry in payload.get("family", []):
+        if not isinstance(entry, dict):
+            raise FamiliesError(f"{source}: every [[family]] entry must be a table")
+        family = _family(source, entry)
+        if any(existing.id == family.id for existing in families):
+            raise FamiliesError(f"{source}: family {family.id!r} declared twice")
+        for cwe in sorted(family.cwes):
+            if cwe in claimed:
+                raise FamiliesError(f"{source}: {cwe} is claimed by both {claimed[cwe]!r} and {family.id!r}")
+            claimed[cwe] = family.id
+        families.append(family)
+    _check_parents(source, families)
+    if require_all and tuple(family.id for family in families) != FAMILY_IDS:
+        raise FamiliesError(f"{source}: expected exactly {', '.join(FAMILY_IDS)} in that order, got {', '.join(f.id for f in families)}")
+    if not families:
+        raise FamiliesError(f"{source}: no families declared")
+    return FamilyTaxonomy(version=version, families=tuple(families))
+
+
+def _family(source: Path, entry: dict[str, object]) -> Family:
+    unknown_fields = sorted(set(entry) - _FAMILY_FIELDS)
+    if unknown_fields:
+        raise FamiliesError(f"{source}: family {entry.get('id')!r} has unknown field(s) {', '.join(unknown_fields)}")
+    family_id = str(entry.get("id", "")).strip()
+    if family_id not in FAMILY_IDS:
+        raise FamiliesError(f"{source}: unknown family {family_id!r}; expected one of {', '.join(FAMILY_IDS)}")
+    verifier = str(entry.get("verifier", "")).strip()
+    if verifier not in VERIFIER_KINDS:
+        raise FamiliesError(f"{source}: family {family_id!r} has verifier {verifier!r}; expected one of {', '.join(VERIFIER_KINDS)}")
+    description = str(entry.get("description", "")).strip()
+    if not description:
+        raise FamiliesError(f"{source}: family {family_id!r} needs a description")
+    cwes = _tokens(source, family_id, entry.get("cwes", ()), "cwes", upper=True)
+    mechanisms = _tokens(source, family_id, entry.get("mechanisms", ()), "mechanisms")
+    parent = entry.get("parent")
+    return Family(
+        id=family_id,
+        description=description,
+        cwes=frozenset(cwes),
+        mechanisms=frozenset(mechanisms),
+        verifier=verifier,
+        parent=str(parent) if parent else None,
+    )
+
+
+def _tokens(source: Path, family_id: str, value: object, field: str, *, upper: bool = False) -> list[str]:
+    if not isinstance(value, list) or any(not isinstance(item, str) for item in value):
+        raise FamiliesError(f"{source}: family {family_id!r} field {field} must be a list of strings")
+    tokens = [item.strip().upper() if upper else item.strip() for item in value]
+    duplicates = sorted({token for token in tokens if tokens.count(token) > 1})
+    if duplicates:
+        raise FamiliesError(f"{source}: family {family_id!r} lists {', '.join(duplicates)} twice in {field}")
+    return tokens
+
+
+def _check_parents(source: Path, families: list[Family]) -> None:
+    ids = {family.id for family in families}
+    for family in families:
+        if family.parent is None:
+            continue
+        if family.parent not in ids:
+            raise FamiliesError(f"{source}: family {family.id!r} declares unknown parent {family.parent!r}")
+        if family.parent == family.id:
+            raise FamiliesError(f"{source}: family {family.id!r} is its own parent")
+
+
+__all__ = [
+    "DEFAULT_FAMILIES_PATH",
+    "FAMILY_IDS",
+    "VERIFIER_KINDS",
+    "FamiliesError",
+    "Family",
+    "FamilyTaxonomy",
+    "Relation",
+    "load_families",
+]

src/openultrasast/ruleset/families.toml

@@ -0,0 +1,126 @@
+# Closed family taxonomy (learning-harness, Req 1). A family is the ONLY key that routes work and
+# scores a pair; CWE ids and mechanism ids are attributes looked up through a family and are never
+# keys themselves. `verifier` says what can confirm a claim in this family: a canary oracle in the
+# sandbox, static corroboration from another checker, or nothing (the claim stays a suspicion).
+# `unknown` carries the generic mechanisms and the deliberately out-of-scope weaknesses, so a row we
+# have considered and declined is visible instead of silently unresolved. Bump `version` on any change.
+version = "1"
+
+[[family]]
+id = "injection"
+description = "Untrusted input reaches an interpreter: SQL, OS command, code, expression or server-side template."
+cwes = ["CWE-77", "CWE-78", "CWE-88", "CWE-89", "CWE-90", "CWE-91", "CWE-94", "CWE-95", "CWE-564", "CWE-917", "CWE-943", "CWE-1336"]
+mechanisms = []
+verifier = "canary"
+
+[[family]]
+id = "path"
+description = "Untrusted input reaches a filesystem path: traversal, unsafe join, file inclusion."
+cwes = ["CWE-22", "CWE-23", "CWE-36", "CWE-73", "CWE-98", "CWE-99"]
+mechanisms = ["path_join_user_input"]
+verifier = "canary"
+
+[[family]]
+id = "deserialization"
+description = "Untrusted data is deserialized into objects the process then trusts."
+cwes = ["CWE-502"]
+mechanisms = []
+verifier = "canary"
+
+[[family]]
+id = "access_control"
+description = "An operation on a protected resource is reached without the guard or ownership constraint its siblings carry."
+cwes = ["CWE-284", "CWE-285", "CWE-306", "CWE-425", "CWE-566", "CWE-639", "CWE-862", "CWE-863", "CWE-1220"]
+mechanisms = [
+    "identity_from_request_body",
+    "missing_auth_guard",
+    "unconstrained_protected_read",
+    "unconstrained_protected_write",
+    "unguarded_privileged_action",
+]
+verifier = "static"
+
+[[family]]
+id = "output_encoding"
+description = "Untrusted data reaches a rendered document or a header without encoding: cross-site scripting, response splitting."
+cwes = ["CWE-79", "CWE-80", "CWE-83", "CWE-113", "CWE-116"]
+mechanisms = []
+verifier = "none"
+
+[[family]]
+id = "untrusted_destination"
+description = "Untrusted input controls where a request or a redirect goes: server-side request forgery and open redirect."
+cwes = ["CWE-441", "CWE-601", "CWE-611", "CWE-918"]
+mechanisms = []
+verifier = "none"
+
+[[family]]
+id = "prototype"
+description = "Untrusted keys reach an object merge or attribute assignment and change behaviour other code relies on."
+cwes = ["CWE-471", "CWE-915", "CWE-1321"]
+mechanisms = ["prototype_pollution"]
+verifier = "none"
+
+[[family]]
+id = "config_secrets"
+description = "A security setting ships permissive or a credential ships in the artifact: CORS, debug, cookie flags, weak crypto, hardcoded secrets."
+cwes = [
+    "CWE-16",
+    "CWE-215",
+    "CWE-259",
+    "CWE-311",
+    "CWE-312",
+    "CWE-315",
+    "CWE-326",
+    "CWE-327",
+    "CWE-346",
+    "CWE-489",
+    "CWE-522",
+    "CWE-614",
+    "CWE-798",
+    "CWE-942",
+    "CWE-1004",
+]
+mechanisms = ["config_sink_weak_literal", "permissive_default", "secret_in_client"]
+verifier = "none"
+
+[[family]]
+id = "memory"
+description = "Memory safety in C and C++: bounds, lifetime, arithmetic. Measured on the vendored slice, never gated; specialized systems own this class."
+cwes = [
+    "CWE-119",
+    "CWE-120",
+    "CWE-121",
+    "CWE-122",
+    "CWE-125",
+    "CWE-126",
+    "CWE-131",
+    "CWE-134",
+    "CWE-190",
+    "CWE-191",
+    "CWE-401",
+    "CWE-415",
+    "CWE-416",
+    "CWE-476",
+    "CWE-680",
+    "CWE-770",
+    "CWE-787",
+    "CWE-843",
+]
+mechanisms = [
+    "fixed_buffer_capacity",
+    "null_deref_unguarded",
+    "signed_size_operand",
+    "type_confusion",
+    "uaf",
+    "unchecked_alloc_size",
+    "unchecked_length_copy",
+]
+verifier = "none"
+
+[[family]]
+id = "unknown"
+description = "No family admitted: a mechanism that spans families, or a weakness this project deliberately does not target."
+cwes = ["CWE-20", "CWE-835"]
+mechanisms = ["container_taint", "cross_artifact", "other", "redos", "source_reaches_sink", "validation_strength"]
+verifier = "none"

tests/test_learning_families.py

@@ -0,0 +1,119 @@
+"""learning-harness task 1.1: the closed family taxonomy with a verifier per family (offline)."""
+
+from __future__ import annotations
+
+import tomllib
+from pathlib import Path
+
+import pytest
+
+GOOD = (
+    'version = "1"\n\n'
+    '[[family]]\nid = "injection"\ndescription = "x"\ncwes = ["CWE-89"]\nmechanisms = ["source_reaches_sink"]\nverifier = "canary"\n'
+)
+
+
+def test_taxonomy_loads_ten_families_each_with_a_verifier() -> None:
+    from openultrasast.learning.families import FAMILY_IDS, VERIFIER_KINDS, load_families
+
+    taxonomy = load_families()
+    assert len(FAMILY_IDS) == 10 and FAMILY_IDS[-1] == "unknown"
+    assert tuple(family.id for family in taxonomy.families) == FAMILY_IDS
+    assert taxonomy.version and all(family.verifier in VERIFIER_KINDS for family in taxonomy.families)
+    assert {family.id for family in taxonomy.families if family.verifier == "canary"} == {"injection", "path", "deserialization"}
+    assert taxonomy.by_id("access_control").verifier == "static"
+    assert taxonomy.by_id("memory").verifier == "none"  # measured on the vendored slice, never proven by us
+    assert taxonomy.by_id("unknown").verifier == "none" and all(family.description for family in taxonomy.families)
+
+
+def test_every_mechanism_id_resolves_to_exactly_one_family() -> None:
+    from openultrasast.learning.families import load_families
+    from openultrasast.pairs import load_mechanisms
+
+    taxonomy = load_families()
+    _, mechanism_ids = load_mechanisms()
+    for mechanism in sorted(mechanism_ids):
+        owners = [family.id for family in taxonomy.families if mechanism in family.mechanisms]
+        assert len(owners) == 1, (mechanism, owners)
+    assert taxonomy.family_of_mechanism("missing_auth_guard").id == "access_control"
+    assert taxonomy.family_of_mechanism("unchecked_length_copy").id == "memory"
+    # a mechanism that spans families names none of them; the labeled CWE decides those rows
+    assert taxonomy.family_of_mechanism("source_reaches_sink").id == "unknown"
+    assert taxonomy.family_of_mechanism("not_a_mechanism") is None
+
+
+def test_every_labeled_corpus_cwe_resolves_and_no_cwe_is_claimed_twice() -> None:
+    from openultrasast.learning.families import load_families
+
+    taxonomy = load_families()
+    seen: dict[str, str] = {}
+    for family in taxonomy.families:
+        for cwe in family.cwes:
+            assert cwe not in seen, (cwe, seen.get(cwe), family.id)
+            seen[cwe] = family.id
+    labeled: set[str] = set()
+    for catalog in sorted(Path("benchmarks/pairs").rglob("catalog.toml")):
+        for pair in tomllib.loads(catalog.read_text()).get("pair", []):
+            labeled |= {str(row["cwe"]) for row in pair.get("expected", []) if row.get("cwe")}
+    assert sorted(cwe for cwe in labeled if taxonomy.family_of_cwe(cwe) is None) == []
+    assert taxonomy.family_of_cwe("cwe-89").id == "injection"  # case and spacing tolerant
+    assert taxonomy.family_of_cwe("CWE-601").id == "untrusted_destination"  # 34 open-redirect rows
+    assert taxonomy.family_of_cwe("CWE-835").id == "unknown"  # considered and out of scope, not unresolved
+    assert taxonomy.family_of_cwe("CWE-99999") is None
+
+
+def test_relation_answers_same_lateral_fabricated_and_parent_child_when_declared(tmp_path: Path) -> None:
+    from openultrasast.learning.families import load_families
+
+    taxonomy = load_families()
+    assert taxonomy.related("injection", "injection") == "same"
+    assert taxonomy.related("injection", "access_control") == "lateral"
+    assert taxonomy.related("injection", "not_a_family") == "fabricated"
+    assert taxonomy.related("unknown", "injection") == "lateral"  # abstention earns no partial credit
+    assert all(family.parent is None for family in taxonomy.families)  # the shipped taxonomy is flat
+    (tmp_path / "families.toml").write_text(
+        GOOD + '\n[[family]]\nid = "path"\nparent = "injection"\ndescription = "y"\ncwes = []\nmechanisms = []\nverifier = "canary"\n'
+    )
+    scoped = load_families(tmp_path / "families.toml", require_all=False)
+    assert scoped.related("path", "injection") == "parent_child" and scoped.related("injection", "path") == "parent_child"
+
+
+@pytest.mark.parametrize(
+    ("text", "needle"),
+    [
+        (GOOD + 'color = "blue"\n', "color"),
+        (GOOD.replace('verifier = "canary"\n', ""), "verifier"),
+        (GOOD.replace('verifier = "canary"', 'verifier = "maybe"'), "maybe"),
+        (GOOD.replace('id = "injection"', 'id = "sql_injection"'), "sql_injection"),
+        (GOOD.replace('cwes = ["CWE-89"]', 'cwes = ["CWE-89", "CWE-89"]'), "CWE-89"),
+        (GOOD.replace('description = "x"\n', ""), "description"),
+        (GOOD + '\n[[family]]\nid = "path"\ndescription = "y"\ncwes = ["CWE-89"]\nmechanisms = []\nverifier = "canary"\n', "CWE-89"),
+        (GOOD.replace('version = "1"\n', ""), "version"),
+        (GOOD + '\n[[family]]\nid = "path"\nparent = "nope"\ndescription = "y"\ncwes = []\nmechanisms = []\nverifier = "none"\n', "nope"),
+    ],
+    ids=[
+        "unknown_field",
+        "missing_verifier",
+        "bad_verifier",
+        "fabricated_family",
+        "duplicate_cwe",
+        "no_description",
+        "cwe_claimed_twice",
+        "no_version",
+        "unknown_parent",
+    ],
+)
+def test_loader_rejects_data_bugs_by_name(tmp_path: Path, text: str, needle: str) -> None:
+    from openultrasast.learning.families import FamiliesError, load_families
+
+    (tmp_path / "families.toml").write_text(text)
+    with pytest.raises(FamiliesError, match=needle):
+        load_families(tmp_path / "families.toml", require_all=False)
+
+
+def test_the_shipped_file_must_declare_the_whole_closed_set_in_order(tmp_path: Path) -> None:
+    from openultrasast.learning.families import FamiliesError, load_families
+
+    (tmp_path / "families.toml").write_text(GOOD)
+    with pytest.raises(FamiliesError, match="in that order"):
+        load_families(tmp_path / "families.toml")