diff --git a/docs/error-codes.md b/docs/error-codes.md index a16cf14..17f1448 100644 --- a/docs/error-codes.md +++ b/docs/error-codes.md @@ -18,8 +18,8 @@ All TRACE test failures emit a structured error code of the form `TR-- 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_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) + ) +