diff --git a/src/openultrasast/learning/__init__.py b/src/openultrasast/learning/__init__.py new file mode 100644 index 0000000..12f7d6f --- /dev/null +++ b/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"] diff --git a/src/openultrasast/learning/families.py b/src/openultrasast/learning/families.py new file mode 100644 index 0000000..b077e6a --- /dev/null +++ b/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", +] diff --git a/src/openultrasast/ruleset/families.toml b/src/openultrasast/ruleset/families.toml new file mode 100644 index 0000000..1f11d8a --- /dev/null +++ b/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" diff --git a/tests/test_learning_families.py b/tests/test_learning_families.py new file mode 100644 index 0000000..c57e288 --- /dev/null +++ b/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")