diff --git a/docs/error-codes.md b/docs/error-codes.md index 032bace..17f1448 100644 --- a/docs/error-codes.md +++ b/docs/error-codes.md @@ -9,7 +9,7 @@ All TRACE test failures emit a structured error code of the form `TR--/` or a `did:` URI | -| TR-ENV-004 | One or more required fields are absent | Add the missing field(s); check the [Schema Reference](https://trace.agentrust-io.com/docs/schema/) for the full required set | +| TR-ENV-004 | `cnf` is absent or not an object, `cnf.jwk` is absent or not an object, or `cnf.jwk.kty` is absent | Populate `cnf.jwk` with at least `kty`. This checks that one field, not the schema's full required set, which structural validation covers | ## TR-SIG — Signature @@ -18,22 +18,23 @@ All TRACE test failures emit a structured error code of the form `TR---` | -| TR-POL-002 | `policy.enforcement_mode` | Must be `enforce`, `advisory`, or `silent` | +| TR-POL-002 | `policy.enforcement_mode` | Must be `enforce`, `advisory`, `silent`, or `declared` | | TR-RTE-001 | `runtime.platform` | Must be a registered TEE platform enum | When a finding carries `status == Status.UNVERIFIED`, the record has no signature. This is not a benign skip at Level 1 or above. diff --git a/src/trace_tests/modules/tr_sig.py b/src/trace_tests/modules/tr_sig.py index ac40add..6e15537 100644 --- a/src/trace_tests/modules/tr_sig.py +++ b/src/trace_tests/modules/tr_sig.py @@ -75,6 +75,27 @@ def _verify_ed25519(pub_x: str, sig_b64: str, body: bytes) -> tuple[bool, str]: return False, "TR-SIG-001: signature verification failed" +def _jwk_of(container: Any) -> dict[str, Any]: + """The JWK under ``container["cnf"]["jwk"]``, or ``{}`` when it is not an object. + + A malformed record must produce a finding, not an exception. ``runner.run`` calls + every module without a ``try``, so anything raised here ends the run rather than + failing the record, and the caller sees a traceback where a verdict belongs. + + The outer ``isinstance`` is for ``check_cmcp_runtime``, which passes + ``record["trace"]`` and so can hand this anything at all. ``check`` passes the + ``trace`` it was given, which for the plain format is the record itself and is + already a dict; that function reads ``trace`` directly elsewhere and is not + hardened against a non-dict ``trace``. Whether one can reach it is a question + about ``loader.extract_trace``, not about this helper. + """ + if not isinstance(container, dict): + return {} + cnf = container.get("cnf") + jwk = cnf.get("jwk") if isinstance(cnf, dict) else None + return jwk if isinstance(jwk, dict) else {} + + def check_cmcp_runtime(record: dict[str, Any]) -> list[Finding]: """Verify the Ed25519 signature on a cmcp RuntimeClaim.""" findings: list[Finding] = [] @@ -84,7 +105,7 @@ def check_cmcp_runtime(record: dict[str, Any]) -> list[Finding]: findings.append(Finding("TR-SIG-001", Status.FAIL, "TR-SIG-001: signature field is missing or empty")) return findings - jwk = record.get("trace", {}).get("cnf", {}).get("jwk", {}) + jwk = _jwk_of(record.get("trace")) kty = jwk.get("kty") crv = jwk.get("crv") x = jwk.get("x") @@ -117,16 +138,24 @@ def check(trace: dict[str, Any], record: dict[str, Any], fmt: str, level: int = return check_cmcp_runtime(record) findings: list[Finding] = [] - jwk = trace.get("cnf", {}).get("jwk", {}) + jwk = _jwk_of(trace) kty = jwk.get("kty") crv = jwk.get("crv") x = jwk.get("x") if "d" in jwk: findings.append(Finding( - rule="TR-SIG-002", - status=Status.FAIL, - message="cnf.jwk must not contain private key material ('d' field present in JWK)", + "TR-SIG-004", Status.FAIL, + "TR-SIG-004: cnf.jwk must not contain private key material " + "('d' member present in the JWK)", + )) + # The signature is not checked against a key the record should never have + # carried. Say so rather than returning nothing: a consumer reading TR-SIG-005 + # to learn whether the signature was verified would otherwise find no finding + # at all, which is the benign-omission reading UNVERIFIED exists to prevent. + findings.append(Finding( + "TR-SIG-005", Status.UNVERIFIED, + "TR-SIG-005: signature not checked; cnf.jwk carries private key material", )) return findings diff --git a/tests/test_docs_match_the_modules.py b/tests/test_docs_match_the_modules.py new file mode 100644 index 0000000..4d19a6c --- /dev/null +++ b/tests/test_docs_match_the_modules.py @@ -0,0 +1,153 @@ +"""The published error codes and record samples must agree with the modules. + +Every check here comes from drift that was live in `main`, not from first principles. +`TR-SIG-005` was carried by every signature finding and documented nowhere. `TR-ANC-002` +was documented in three files and named by no module, with two descriptions that +disagreed. `docs/levels.md` showed an `anchor` object the closed schema does not define, +and a `runtime.platform` of `sev-snp`, which is not in the enum. + +Documentation that disagrees with the code is worse than absent documentation: it reads +as verified. Nothing checked it, which is why all of it survived. +""" +from __future__ import annotations + +import json +import pathlib +import re +from typing import Any + +import jsonschema + +REPO = pathlib.Path(__file__).resolve().parents[1] +MODULES = REPO / "src" / "trace_tests" / "modules" +ERROR_CODES = REPO / "docs" / "error-codes.md" +SCHEMA = REPO / "schemas" / "trace-claim.json" +DOCS = REPO / "docs" + +_CODE = re.compile(r"TR-[A-Z]{3}-\d{3}") +_DOCUMENTED_ROW = re.compile(r"^\| (TR-[A-Z]{3}-\d{3}) ", re.M) +_JSON_BLOCK = re.compile(r"```json\n(.*?)```", re.S) + + +def _codes(text: str) -> set[str]: + return set(_CODE.findall(text)) + + +def test_the_code_set_named_by_the_modules_matches_the_code_set_documented() -> None: + """Matched on codes *named in* module source, not on what a ``Finding`` carries. + + Those two sets differ today. ``TR-SIG-003`` appears only inside a message string + that a ``TR-SIG-005`` finding carries, so it is never a ``Finding.code``. Matching + on ``Finding.code`` would demand deleting a page entry that documents a real + condition. Reconciling code and message is a change to the module rather than to + the documentation, and is not attempted here. + + This is set membership in both directions and nothing more. It cannot tell whether + a row describes what its code reports, which was a second kind of drift and was + live here too: the ``TR-SIG-004`` row described private key material in ``cnf.jwk``, + a condition the module never reports under that code. Catching that would mean + comparing prose to behaviour, so the rows are checked by reading them. + """ + named = set() + for path in sorted(MODULES.glob("*.py")): + named |= _codes(path.read_text(encoding="utf-8")) + # A row, not a mention. A code named in passing somewhere on the page is not + # documented, and treating it as documented would let the check be satisfied + # by prose that tells a reader nothing. + documented = set(_DOCUMENTED_ROW.findall(ERROR_CODES.read_text(encoding="utf-8"))) + + assert named == documented, ( + f"named by a module, undocumented: {sorted(named - documented)}\n" + f"documented, named by no module: {sorted(documented - named)}\n" + f"Add the row to {ERROR_CODES.relative_to(REPO)}, or delete it. A code in one " + "place and not the other is a claim nobody checked." + ) + + +def _without_required(node: Any) -> Any: + """The schema with every ``required`` list dropped, except inside ``if`` and ``not``. + + The documented samples are fragments: "changes from Level 1", not whole records. + Validating them as published fails on absent fields and says nothing about the + fields that are present. Dropping ``required`` leaves every statement about a value + that *is* there: ``additionalProperties``, ``enum``, ``pattern``, ``type``. + + ``if`` and ``not`` are left intact deliberately. Stripping ``required`` from an + ``if`` makes it vacuously true, which fires the matching ``then`` against records + the condition was never meant to reach. The schema's ``origin`` rule does exactly + that: strip its ``if`` and every sample is required to be ``software-only``. + """ + if isinstance(node, dict): + return { + key: value if key in ("if", "not") else _without_required(value) + for key, value in node.items() + if key != "required" + } + if isinstance(node, list): + return [_without_required(item) for item in node] + return node + + +def _drop_elisions(node: Any) -> Any: + """Remove string values that are visibly abbreviated for the page. + + A documented sample writes a signature as ``eyJhbGciOiJFZERTQSJ9...``. That is a + reader's placeholder, not a claim about the format, and holding it to the schema's + base64url pattern would report the page style as a defect. + """ + if isinstance(node, dict): + return {k: _drop_elisions(v) for k, v in node.items() + if not (isinstance(v, str) and "..." in v)} + if isinstance(node, list): + return [_drop_elisions(v) for v in node] + return node + + +def test_every_json_sample_in_the_docs_agrees_with_the_packaged_schema() -> None: + """Two drifts lived here: an ``anchor`` object the closed schema does not define, + and a ``runtime.platform`` of ``sev-snp``, which is not in the enum. A reader + copying either sample produced a record this suite rejects. + + Every ``.md`` under ``docs/`` is scanned rather than a list of pages kept by hand, + because a hand-maintained list of what gets checked is the same defect this exists + to catch, in the one place it would not show. + + Prose lists of valid values are not checked, because checking them means reading + them. ``docs/error-codes.md``, ``docs/levels.md`` and ``docs/modules/tr-rte.md`` all + listed platform values that do not exist; only the sample was mechanically catchable. + """ + schema = json.loads(SCHEMA.read_text(encoding="utf-8")) + assert schema.get("additionalProperties") is False, ( + "This test assumes the packaged schema is closed. If that changed, an unknown " + "field in a sample is no longer necessarily an error and this needs rewriting." + ) + validator = jsonschema.Draft202012Validator(_without_required(schema)) + + failures: dict[str, list[str]] = {} + validated = 0 + unparsed = 0 + for page in sorted(DOCS.rglob("*.md")): + for block in _JSON_BLOCK.findall(page.read_text(encoding="utf-8")): + try: + sample = json.loads(block) + except json.JSONDecodeError: + unparsed += 1 + continue # prose-annotated fragment, not a record + if not isinstance(sample, dict): + continue + validated += 1 + errors = [e.message for e in validator.iter_errors(_drop_elisions(sample))] + if errors: + failures.setdefault(str(page.relative_to(REPO)), []).extend(errors) + + assert not failures, ( + f"JSON samples in the documentation disagree with {SCHEMA.name}, so a reader " + f"copying one gets a record this suite rejects: {failures}" + ) + # Without this the check degrades to nothing the moment the samples stop parsing + # or the fences change, and it degrades silently, reporting a pass over no work. + assert validated, ( + f"no JSON object sample under {DOCS.relative_to(REPO)} was validated " + f"({unparsed} block(s) did not parse). Either the samples are gone or the fence " + "this reads has changed; a check over nothing must not report a pass." + ) diff --git a/tests/test_modules_never_raise.py b/tests/test_modules_never_raise.py new file mode 100644 index 0000000..f03b830 --- /dev/null +++ b/tests/test_modules_never_raise.py @@ -0,0 +1,246 @@ +"""A malformed record must produce a finding, never an exception. + +``runner.run`` calls every module directly, with no ``try``. A module that raises on +a record it does not understand ends the whole run: the caller gets a traceback where +a verdict belongs, and the record is neither passed nor failed. + +``tr_sig`` did exactly that in five ways. ``Finding(rule=...)`` raised ``TypeError`` on +the one check meant to catch a record that embeds its own private key, and reading +``cnf`` or ``cnf.jwk`` raised ``AttributeError`` whenever either was not an object. The +packaged schema does not forbid a ``d`` member in ``cnf.jwk``, so nothing rejected such +a record before the module saw it. The other six modules already guarded their inputs +with ``isinstance``; this pins that for all seven. + +Scope: the record is a dict throughout and its *fields* are malformed. The cmcp case +below hands ``check`` an envelope whose ``trace`` is junk, but it does so directly. +``loader.extract_trace`` does return ``record["trace"]`` unchecked, so reading it in +isolation suggests a hole; there is not one on the path the tool takes, because +``load_record`` refuses a cmcp envelope whose ``trace`` is not a dict before that runs, +and ``extract_trace`` is unexported with ``runner.run`` as its only caller. A library +caller that assembles a record by hand and calls ``runner.run`` without the loader can +still reach it. +""" +from __future__ import annotations + +import copy +import inspect +import json +import pathlib +from typing import Any + +import pytest + +from trace_tests.modules import tr_anc, tr_env, tr_pol, tr_rte, tr_sca, tr_sig, tr_txn +from trace_tests.result import Finding, Status +from trace_tests.runner import run + +VECTORS = pathlib.Path(__file__).resolve().parent / "vectors" + +MODULES = { + "tr_env": tr_env, "tr_sig": tr_sig, "tr_pol": tr_pol, "tr_rte": tr_rte, + "tr_txn": tr_txn, "tr_anc": tr_anc, "tr_sca": tr_sca, +} + +#: Values a record can carry where an object or a string is expected. `True` is here +#: because `isinstance(True, int)`; `False` and `0` because a bare truthiness test +#: reads them as absent, which is a different branch from a wrong type. +JUNK: tuple[Any, ...] = ("a-string", 123, None, [1, 2], True, False, 0, {}, "") + +TOP_LEVEL = ( + "cnf", "runtime", "policy", "tool_transcript", "build_provenance", + "transparency", "appraisal", "signature", "model", "subject", "iat", +) + + +def _record() -> dict[str, Any]: + raw = json.loads((VECTORS / "signed_root.json").read_text(encoding="utf-8")) + return dict(raw.get("record", raw)) + + +def _call(module: Any, record: dict[str, Any]) -> list[Finding]: + """Invoke a module's ``check`` whatever its parameter list happens to be. + + Read from the signature rather than written down here, so a module that gains a + parameter is still exercised instead of quietly dropping out of this test. + """ + params = list(inspect.signature(module.check).parameters) + if params[:3] == ["trace", "record", "fmt"]: + return list(module.check(record, record, "trace", 0)) + if "level" in params: + return list(module.check(record, 0)) + return list(module.check(record)) + + +def _mutations() -> list[tuple[str, dict[str, Any]]]: + cases: list[tuple[str, dict[str, Any]]] = [] + for field in TOP_LEVEL: + for junk in JUNK: + record = _record() + record[field] = junk + cases.append((f"{field}={junk!r}", record)) + for junk in JUNK: + record = _record() + if isinstance(record.get("cnf"), dict): + record["cnf"]["jwk"] = junk + cases.append((f"cnf.jwk={junk!r}", record)) + record = _record() + record["cnf"]["jwk"]["d"] = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" + cases.append(("cnf.jwk carries d", record)) + # Absence is its own class. Replacing a field with junk never exercises the branch + # a module takes when the key is not there at all. + for field in TOP_LEVEL: + record = _record() + record.pop(field, None) + cases.append((f"{field} removed", record)) + record = _record() + if isinstance(record.get("cnf"), dict): + record["cnf"].pop("jwk", None) + cases.append(("cnf.jwk removed", record)) + return cases + + +@pytest.mark.parametrize("name", sorted(MODULES)) +def test_no_module_raises_on_a_record_whose_fields_are_malformed(name: str) -> None: + module = MODULES[name] + raised: list[str] = [] + for label, record in _mutations(): + try: + findings = _call(module, copy.deepcopy(record)) + except Exception as exc: # noqa: BLE001 - the point is that nothing escapes + raised.append(f"{label} -> {type(exc).__name__}: {exc}") + continue + if not findings: + raised.append(f"{label} -> returned no findings at all") + + assert not raised, ( + f"{name}.check raised or returned nothing on {len(raised)} malformed record(s). " + f"runner.run has no try, so each of these ends the run instead of failing the " + f"record:\n " + "\n ".join(raised) + ) + + +def test_a_record_embedding_its_own_private_key_fails_rather_than_raising() -> None: + """The condition the check exists for, named as its own case. + + The generic test above would pass if this raised in a module that had no such + check at all. This one asserts the verdict, so removing the check fails here + rather than going unnoticed. + """ + record = _record() + record["cnf"]["jwk"]["d"] = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" + + findings = tr_sig.check(record, record, "trace", 0) + + leak = [f for f in findings if f.status is Status.FAIL and "private key" in f.message] + assert leak, f"no finding reports the embedded private key: {findings}" + assert leak[0].code == "TR-SIG-004", ( + f"the leak is reported under {leak[0].code!r}; docs/error-codes.md documents " + "this condition under TR-SIG-004" + ) + + +def test_a_leaked_key_still_reports_whether_the_signature_was_checked() -> None: + """The leak check returns early, so nothing else in the module runs. + + Before this branch that path raised, so the state was unreachable and no consumer + had met it. Making it reachable without a TR-SIG-005 would publish a record with no + signature verdict of any kind: not pass, not fail, not unverified. One consumer in + this suite already reads that finding with a bare ``next``. + """ + record = _record() + record["cnf"]["jwk"]["d"] = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" + + findings = tr_sig.check(record, record, "trace", 0) + by_code = {f.code: f for f in findings} + + assert "TR-SIG-005" in by_code, ( + f"a leaked-key record reports no signature verdict at all: {findings}" + ) + assert by_code["TR-SIG-005"].status is Status.UNVERIFIED, ( + "the signature was not checked, so it is unverified rather than passed or failed" + ) + assert by_code["TR-SIG-004"].status is Status.FAIL + + +@pytest.mark.parametrize("level", [0, 1, 2]) +def test_the_runner_completes_on_a_record_that_embeds_its_own_private_key(level: int) -> None: + """The regression as it was actually met, one layer above the module. + + The traceback came out of ``runner.run``, which calls each module with no ``try``. + Testing ``tr_sig.check`` alone would still pass if some later change moved the same + failure into the runner, so the path that broke is exercised here as well. + """ + record = _record() + record["cnf"]["jwk"]["d"] = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" + + results = run(record, "trace", level) + + sig = results["TR-SIG"] + assert sig, "the runner produced no TR-SIG findings for a record it should reject" + assert any(f.code == "TR-SIG-004" and f.status is Status.FAIL for f in sig), sig + assert any(f.code == "TR-SIG-005" and f.status is Status.UNVERIFIED for f in sig), sig + + +def test_no_finding_from_any_module_repeats_the_key_it_found() -> None: + """A finding about a leaked private key must not carry the key. + + Findings travel: ``report.py`` publishes every message into a JSON and an HTML + artifact meant to be forwarded. A message that quoted the offending value to be + helpful would copy the private key into the thing the reader sends on. The report + itself carries only the record's digest, so a message is the only place this can + go wrong, and it can go wrong in any module rather than only the one that reports + the leak. + """ + secret = "Zm9yYmlkZGVuLXByaXZhdGUta2V5LW1hdGVyaWFs" + record = _record() + record["cnf"]["jwk"]["d"] = secret + + offenders = [] + for name, module in sorted(MODULES.items()): + for finding in _call(module, copy.deepcopy(record)): + if secret in finding.message: + offenders.append(f"{name} {finding.code}: {finding.message}") + + assert not offenders, ( + "a finding repeats the private key it is reporting:\n " + "\n ".join(offenders) + ) + + +def _cmcp_record() -> dict[str, Any]: + return json.loads((VECTORS / "valid_cmcp_runtime.json").read_text(encoding="utf-8")) + + +def test_the_cmcp_path_does_not_raise_on_a_malformed_envelope() -> None: + """The other entry point, which the parametrised test above never reaches. + + ``check`` dispatches to ``check_cmcp_runtime`` on ``fmt == "cmcp-runtime"`` and + every case above passes ``"trace"``, so the branch that reads ``record["trace"]`` + three levels deep was hardened without being exercised. It reads a value the caller + supplies rather than the extracted trace, so it can be handed anything. + """ + raised: list[str] = [] + for junk in JUNK: + for path in (("trace",), ("trace", "cnf"), ("trace", "cnf", "jwk"), ("signature",)): + record = _cmcp_record() + node: Any = record + for key in path[:-1]: + if not isinstance(node, dict) or not isinstance(node.get(key), dict): + node = None + break + node = node[key] + if node is None: + continue + node[path[-1]] = junk + label = ".".join(path) + f"={junk!r}" + try: + findings = tr_sig.check(record.get("trace", {}), record, "cmcp-runtime", 0) + except Exception as exc: # noqa: BLE001 - the point is that nothing escapes + raised.append(f"{label} -> {type(exc).__name__}: {exc}") + continue + if not findings: + raised.append(f"{label} -> returned no findings at all") + + assert not raised, ( + "tr_sig.check on a cmcp envelope raised or returned nothing:\n " + "\n ".join(raised) + ) +