diff --git a/DOCUMENTATION_INDEX.md b/DOCUMENTATION_INDEX.md index 16a603f..4e6e123 100644 --- a/DOCUMENTATION_INDEX.md +++ b/DOCUMENTATION_INDEX.md @@ -83,6 +83,7 @@ scope-honesty box and makes no compliance/regulatory claim. | [`docs/THREAT_MODEL.md`](docs/THREAT_MODEL.md) | **(v0.1)** Positioning, trust model (TOFU + `--approve`), assets/actors, the four threat classes (MCP-DRIFT / MCP-CAPSURF / MCP-SECRET / MCP-SUPPLY), explicit out-of-scope limits, deliberate cuts | | [`docs/THREAT_MODEL_V2.md`](docs/THREAT_MODEL_V2.md) | **(v0.2)** Addendum extending the v0.1 model: T-RESULT vectors, the defends (BLOCK) / monitors (fuzzy) / still-does-NOT-defend (T-BEHAVE) table, runtime trust-model notes, retained + added cuts, shadow-default positioning | | [`docs/AGENT_TRUST_KERNEL.md`](docs/AGENT_TRUST_KERNEL.md) | **(DSE-714, design contract)** Normative invariants for the future deterministic Agent Trust Kernel: trust boundaries, complete mediation, default deny, non-overridable critical classes, evidence-before-effect, offline operation, residual risks, and bindings for DSE-715 through DSE-717. MCP-Warden v1.1 is explicitly not yet ATK-conformant | +| [`docs/CONTENT_ENVELOPE.md`](docs/CONTENT_ENVELOPE.md) | **(DSE-715, implemented foundation)** Strict immutable V1 content envelope, domain-separated exact-byte digests, canonical metadata boundary, bounded one-hop lineage, monotonic taint, stable code-only errors, and secret-safe public projection. Evidence only; no authority or whole-ATK conformance claim | | [`docs/WARDEN_LOCK_SCHEMA.md`](docs/WARDEN_LOCK_SCHEMA.md) | **mcp-warden implementation of [`docs/SPEC.md`](docs/SPEC.md) (MCP Lock Format v1).** `warden.lock` format, RFC 8785 canonicalization + SHA-256 hashing, field/entry/overall digests, the normative drift definition + severities; **§5.1/§6.2 structural schema diff** (normalized per-tool `schema_skeleton`, `schema_version` 3 — skeleton added at v2, in-document `$ref` resolution at v3 (#29), granular `WRD-DRIFT-SCHEMA-*` taxonomy + severities, v1 fallback); **§8.1/§8.2 (v0.3, #19)** structured out-of-digest provenance (`pinner` / `attestations` / `rotation_count`, `PROVENANCE_VERSION`, B4 `bound_digest` format) + `lock rotate` digest-invariant semantics + the #16 signing implication; **§11 (v0.2)** optional per-tool inspection policy (`expected_output_charset` / `may_return_urls` / `secret_echo_applies`, fail-safe defaults, digest impact) | | [`docs/WARDEN_LOCK_EXAMPLE.md`](docs/WARDEN_LOCK_EXAMPLE.md) | Illustrative full `warden.lock` + a post-`lock rotate` `pin` block (archived from WARDEN_LOCK_SCHEMA §9 to keep that core doc under the line cap) | | [`docs/CHECKS.md`](docs/CHECKS.md) | The deterministic `WRD-*` static-check catalog (capability/secret/supply/robustness), the shared tokenizer, severity→SARIF mapping, redaction rule, CUT list. **Reused by v0.2** `WRD-RES-SECRET-ECHO` (the `WRD-SEC-*` patterns + redaction) | @@ -97,6 +98,7 @@ scope-honesty box and makes no compliance/regulatory claim. | Plan | Purpose | |---|---| | [`docs/plans/2026-07-18-agent-trust-kernel-design.md`](docs/plans/2026-07-18-agent-trust-kernel-design.md) | **Non-normative execution record.** Records the DSE-714 design decision and verification plan; binding requirements live in `docs/AGENT_TRUST_KERNEL.md` | +| [`docs/plans/2026-07-18-content-envelope-design.md`](docs/plans/2026-07-18-content-envelope-design.md) | **Non-normative DSE-715 execution record.** Records the reviewed strict-TDD plan; verified behavior is documented in `docs/CONTENT_ENVELOPE.md` | --- @@ -130,6 +132,7 @@ scope-honesty box and makes no compliance/regulatory claim. | Module | Responsibility | Spec anchor | |--------|----------------|-------------| | `src/mcp_warden/hashing.py` | `canon()` (RFC 8785) + `hash()` + field hashes | WARDEN_LOCK_SCHEMA §3 | +| `src/mcp_warden/content_models.py` · `content_envelope.py` | **(DSE-715)** Strict immutable evidence models; typed domain-separated byte hashing call surface; canonical constructors/parser; bounded derivation and one-hop lineage; explicit secret-safe projection | CONTENT_ENVELOPE.md / AGENT_TRUST_KERNEL ATK-01/02/05/06/09/12 | | `src/mcp_warden/tokenizer.py` | Shared tokenizer + capability derivation (single source of truth) | CHECKS §3 / WARDEN_LOCK_SCHEMA §5.4 | | `src/mcp_warden/capture.py` | Declared-surface capture over stdio (argv array, no shell) or Streamable HTTP (`--url`); shared list normalization, timeouts, and errors | THREAT_MODEL §3.3 / WARDEN_LOCK_SCHEMA §4.1 | | `src/mcp_warden/models.py` | Pydantic models for captured surface + lock (incl. `Pinner`/`Attestation` provenance) | WARDEN_LOCK_SCHEMA §2–§8 | @@ -169,6 +172,7 @@ scope-honesty box and makes no compliance/regulatory claim. | File | Covers | |------|--------| | `tests/test_hashing.py` | JCS+SHA-256 reproducibility, canonical-form pins, null handling | +| `tests/test_content_envelope.py` | **(DSE-715)** Strict/frozen models, typed digest domains, nine ingress kinds, canonical/cap boundary, deterministic lineage, parser rejection matrix, atomic digest verification, and planted-secret output/error/log scans | | `tests/test_tokenizer.py` | Segment-exact tokenization + capability derivation | | `tests/test_checks.py` | Capability/secret/supply/robustness checks + redaction | | `tests/test_drift.py` | Drift per class (added/removed/modified/server-identity/unapproved) | @@ -188,7 +192,7 @@ scope-honesty box and makes no compliance/regulatory claim. | `tests/test_guard_v3.py` | **(v0.3)** opt-out demotes to shadow (`--no-block-*`/`--allow-exfil-domain`/`--no-block-deterministic`), `tools/list_changed` gate block+shadow, policy deny block+shadow, audit-only override, cancel/progress passthrough, **server-crash → `-32002` for every pending id**, client-disconnect child reap (no orphan), truncated + oversized frame fail-open | | `tests/test_guard_strict.py` | **(#21)** `--strict` fail-CLOSED: 4 terminate sites (request-policy / result-inspect / list-gate / nested-hash re-raise) → exit `3` + one `strict_abort` stderr line + `-32003` client frame + child reaped; negatives (truncated/over-cap/unparseable/clean) do NOT abort; default `--no-strict` byte-identical fail-open regression; secret-leak redaction; CLI threading; double-emission single line; `StrictInspectionAbort` is `BaseException`-not-`Exception` + anyio `ExceptionGroup` unwrap | | `tests/test_inspect_parity.py` | **(v0.2)** guard↔inspect finding parity on the same recorded frames + inspect exit codes | -| `tests/fuzz/` (`test_fuzz_framing.py` · `test_fuzz_ansi.py` · `test_fuzz_domain.py` · `test_fuzz_redact.py`) | **(#17)** `hypothesis` property-fuzzing of the live runtime attack surface: framer XOR/never-raise + mode-equivalence + truncation→parse_error + `_parse_content_length` never-negative (Finding A fix) + read_frame never-hang + Content-Length composition consistency; ANSI construction-based liveness/soundness + completeness + idempotence + strip∘redact order-independence; exfil-domain soundness (no invented hits) + URL/bare-host liveness + anchoring/IDN/trailing-dot; redactor format-structure + leak-bound + #38 short-secret contract. `ci`/`fuzz` profiles in `tests/fuzz/conftest.py`; deep soak via `make fuzz` | +| `tests/fuzz/` (`test_fuzz_framing.py` · `test_fuzz_ansi.py` · `test_fuzz_domain.py` · `test_fuzz_redact.py` · `test_fuzz_content_envelope.py`) | **(#17 / DSE-715)** `hypothesis` property-fuzzing of the live runtime attack surface plus content-envelope repeatability, round-trip, parent permutation, taint monotonicity, mutation rejection, malformed-byte bounded termination, flat-cycle non-recursion, registry drift, and planted-secret absence. `ci`/`fuzz` profiles in `tests/fuzz/conftest.py`; deep soak via `make fuzz` | | `tests/fixtures/clean_server.py` · `mutated_server.py` | Real MCP SDK stdio fixtures | | `tests/fixtures/poison_server.py` | **(v0.2)** result-poisoning fixture server (ANSI/secret-echo/exfil/inject/clean tools) | | `tests/fixtures/crash_server.py` · `listchange_server.py` · `clean_listchange.warden.lock` | **(v0.3)** raw-stdio lifecycle fixtures: crash-mid-call (`-32002`) and `tools/list_changed` rug-pull + its pinned clean lock | diff --git a/SYSTEM_CONTEXT_DIAGRAM.md b/SYSTEM_CONTEXT_DIAGRAM.md index 3108414..0ebe8ea 100644 --- a/SYSTEM_CONTEXT_DIAGRAM.md +++ b/SYSTEM_CONTEXT_DIAGRAM.md @@ -19,12 +19,12 @@ integrity drift between two EXISTING locks by reusing `compute_drift` (no captur logic) plus a separate informational provenance section. It never prints raw `server.command`/`args` (secret-safe); default exit 0, `--exit-code` → 1 on integrity drift only. -> **Agent Trust Kernel status (DSE-714):** [`docs/AGENT_TRUST_KERNEL.md`](docs/AGENT_TRUST_KERNEL.md) -> is the normative contract for a future deterministic reference monitor. It is **design-only** -> until DSE-715 through DSE-717 implement provenance/taint, PDP/PEP complete mediation, and -> signed evidence-before-effect. The current `guard` path is not represented as ATK-conformant. -> In C1, the dashed ATK edge and `Future Agent Trust Kernel runtime` node are conceptual -> documentation scope only—not a deployed component, shipped data flow, or implementation claim. +> **Agent Trust Kernel status:** [`docs/AGENT_TRUST_KERNEL.md`](docs/AGENT_TRUST_KERNEL.md) +> is the normative contract. DSE-715 implements the isolated content-envelope foundation +> (untrusted ingress, deterministic lineage, monotonic taint, secret-safe projection), but it +> is not wired into `guard` and grants no authority. DSE-716/717 must still deliver PDP/PEP +> complete mediation and signed evidence-before-effect. The current `guard` path is not +> represented as ATK-conformant. > `conclave` (the 4-model adversarial council referenced in `docs/THREAT_MODEL.md`) > is a **dev-time design reviewer** that shaped this contract. It is **NOT** a @@ -65,8 +65,10 @@ flowchart TB conclave -. "security review" .-> atk end - future["Future Agent Trust Kernel runtime\nDSE-715–717 · NOT IMPLEMENTED"] - atk -. "governs future conformance" .-> future + envelope["Content Envelope V1\nDSE-715 · implemented evidence foundation\nNOT wired to guard · grants no authority"] + future["Future PDP/PEP + signed evidence\nDSE-716–717 · NOT IMPLEMENTED"] + atk -. "governs partial foundation" .-> envelope + envelope -. "required input" .-> future subgraph ci["CI pipeline (GitHub Actions / local)"] warden["mcp-warden CLI\npin · check · policy · lock rotate · diff"] diff --git a/docs/AGENT_TRUST_KERNEL.md b/docs/AGENT_TRUST_KERNEL.md index 494cfcf..10aa1e0 100644 --- a/docs/AGENT_TRUST_KERNEL.md +++ b/docs/AGENT_TRUST_KERNEL.md @@ -1,7 +1,7 @@ # Agent Trust Kernel — Normative Security Contract -**Status:** Design contract (DSE-714). Implementation conformance is pending DSE-715, -DSE-716, and DSE-717. +**Status:** Design contract (DSE-714). DSE-715 implements the isolated content-envelope +foundation; whole-kernel conformance remains pending DSE-716 and DSE-717. **Scope:** Deterministic reference monitor beneath agents and protocol adapters. **Document owner:** Security. Changes require threat-model review. @@ -427,6 +427,19 @@ evidence-before-effect path. DSE-717 MUST NOT be considered complete until DSE-7 mediation conformance gates pass. No downstream ticket may weaken an invariant silently. A proposed exception requires a security-contract change and review before implementation. +### DSE-715 implementation status + +`ContentEnvelopeV1` now implements and tests the DSE-715 portions of ATK-01, ATK-02, ATK-05, +ATK-06, ATK-09, and ATK-12: literal untrusted ingress, immutable deterministic lineage, +monotonic registered taint, strict bounded canonical parsing, domain-separated digests, no +authority-bearing field/API, and digest-only secret-safe public output. Its lineage verifier is +intentionally bounded and one-hop. It does not verify source truth, artifact authenticity, +transitive ancestry, policy, authorization, mediation, or evidence-before-effect. + +This partial implementation does not change the current-product boundary: no shipped runtime +may claim ATK conformance until DSE-716/717 integrate the envelope through the complete +mediation and durable evidence paths and every conformance gate below passes. + ## 11. Conformance gates A component or adapter may claim ATK conformance only through a mechanically executable suite. diff --git a/docs/CONTENT_ENVELOPE.md b/docs/CONTENT_ENVELOPE.md new file mode 100644 index 0000000..6129cc7 --- /dev/null +++ b/docs/CONTENT_ENVELOPE.md @@ -0,0 +1,97 @@ +# Deterministic Content Envelope V1 + +**Status:** Implemented foundation (DSE-715). This is evidence plumbing, not an authority or +whole Agent Trust Kernel conformance claim. + +`ContentEnvelopeV1` records exact content bytes as a domain-separated SHA-256 digest plus a +strict, immutable, RFC 8785 wire object. Every envelope is explicitly `untrusted`; every taint +set contains `core:untrusted`. The API has no operation or field that grants authority, marks +content trusted, or removes taint. + +## Implemented behavior + +- Nine ingress kinds cover tool results, web, documents, email, database text, MCP prompts, + MCP resources, bundle metadata, and agent messages. Derived envelopes use the code-owned + `derived` source kind. +- Opaque content and signature evidence are hashed as exact bytes. Structured source, + transform, and bundle metadata must arrive as exact canonical RFC 8785 bytes; malformed, + duplicate-key, noncanonical, non-finite, deep, or oversized inputs reject before use. + Every byte API requires the exact built-in `bytes` type; mutable buffers and subclasses + reject rather than silently widening the byte contract. +- Every digest is `sha256(domain + NUL + payload)` using the closed `DigestDomain` enum. +- Transform identity is code-owned. Callers choose an exact `TransformKindV1` enum, not raw + ID/version text: ingress accepts only `INGRESS_CAPTURE`, derivation accepts only + `DETERMINISTIC`, and an immutable append-only V1 registry supplies the serialized + `(id, version)` pair. An exact reverse registry binds every ingress source context to the + capture pair and every derived source context to the deterministic pair in the model and + parser—not only in constructors. Self-coherent pair swaps reject as `ENV-MALFORMED` for + roots and `ENV-LINEAGE-INVALID` for derived envelopes. Adding a registry entry is a + security-sensitive change requiring a new enum member, updated model/parser context + enforcement, and explicit cross-pair and wire-forgery tests. The implementation digest + binds the exact executing bytes to the identity; deciding whether those bytes are + pre-approved belongs to DSE-716. +- Parent references are sorted immutable `{envelope_digest, content_digest}` pairs. Derivation + recomputes every parent self-digest before accepting its refs or taints, then preserves the + complete bounded local taint union. Lineage verification is bounded and one-hop; it does not + claim transitive graph verification. Parent and taint inputs are consumed through fixed + `cap + 1` loops, bounding consumed elements and preventing unbounded logical materialization + from infinite iterables that continue yielding. The loop cannot prevent a blocking + synchronous `__next__` call from stalling its calling thread; deadline enforcement is the + caller's responsibility and is outside this boundary. +- Derived source identity and claims are recomputed from the exact sorted parent refs and + transform evidence during parsing and lineage verification; self-coherent caller-forged + source digests reject. Model-level invariants also enforce the source-context/transform-pair + binding and prevent roots with parents, derived envelopes without parents, and + bundle-metadata roots without bundle evidence. +- Every accepting public boundary completely revalidates the top-level envelope and each + nested frozen model. Pydantic instance revalidation plus explicit field-specific checks + prevent `model_copy(update=...)` from bypassing digest, media, source-kind, transform, + parent-order, taint-order, or bundle invariants. A storage-level required-field/type + preflight also rejects incomplete `model_construct(...)` objects before any field is + dereferenced or any parent digest is recomputed. Hostile-subclass rejection is enforced at + every current public accepting boundary through an exact-type check before attribute access; + this test-enforced convention is required for every future public accepting boundary. +- Parsing rejects unknown fields/versions, missing `bundle`, noncanonical bytes, invalid trust + or taint state, malformed lineage, and self-digest mismatch. Errors expose a stable `ENV-*` + code only and retain no third-party exception context. +- Public output explicitly selects digest-only V1 fields. Raw content, source claims, + implementation metadata, parameters, signatures, and bundle policy claims are discarded + after hashing and are never serialized. Pydantic validation is configured to hide input + values, including when callers instantiate strict public models directly. +- Supported direct model boundaries—construction, `model_validate`, `model_validate_json`, + `model_validate_strings`, and frozen assignment/deletion—convert validation failures to + code-only `ContentEnvelopeError` values. Pydantic `TypeAdapter`, private + `__pydantic_validator__`, core-schema APIs, and unsafe `model_construct` are not validation + boundaries and MUST NOT receive untrusted input. Public envelope APIs still reject a + constructed or copied invalid model before use. Secret-safety and stable-error guarantees + apply at these documented boundaries, not to unsupported Pydantic introspection paths. +- The legacy generic `canon()` logger is excluded from this subsystem. An AST import guard + forbids importing the `mcp_warden.hashing` module as a dynamic attribute handle and permits + only the named `hash_bytes` import, keeping raw ingress metadata off legacy canonicalization, + hashing, and logging paths. + +Hard caps are 16 MiB content, 64 KiB envelope, 16 JSON levels, 4,096 JSON nodes, 64 parents, +32 taints, and 256 bundle dependencies. Component-specific caps are defined in +`content_models.py`. The pre-parser depth scan handles JSON string quoting/escaping without +allocation proportional to nesting and rejects over-depth inputs before `json.loads`. +Bundle input requires exact bytes and an exact tuple of byte dependencies at runtime; invalid +shapes fail with a stable bundle code before iteration or hashing. +Regression tests exercise exact-cap acceptance and cap-plus-one rejection for every structured +component, bundle component/signature/dependency, JSON depth/nodes/keys/string bytes, envelope +bytes, parent count, taint input count, and dependency count. + +## API boundary + +The pure API is `create_ingress`, `derive_envelope`, `serialize_envelope`, `parse_envelope`, +`verify_envelope`, `verify_lineage`, `to_public_dict`, and `to_public_bytes`. It performs no +I/O, network, clock, environment, policy decision, or effect. Bundle fields are descriptive +evidence only; DSE-716 must verify artifacts and enforce the PDP/PEP load gate. DSE-717 must +provide durable signed receipts and evidence-before-effect. + +## ATK coverage + +Tests provide the DSE-715 foundation for ATK-01, ATK-02, and the envelope portions of ATK-05, +ATK-06, ATK-09, and ATK-12: untrusted state, monotonic taint, no authority fields, strict +fail-closed parsing, canonical explicit inputs, and secret-safe output. MCP-Warden remains +nonconformant with the whole ATK contract until DSE-716 and DSE-717 land and the complete +conformance suite passes. diff --git a/docs/plans/2026-07-18-content-envelope-design.md b/docs/plans/2026-07-18-content-envelope-design.md new file mode 100644 index 0000000..7beeb7f --- /dev/null +++ b/docs/plans/2026-07-18-content-envelope-design.md @@ -0,0 +1,223 @@ +# Deterministic Content Envelope Implementation Plan + +> **Status:** Non-normative DSE-715 execution plan. The binding security requirements live in +> `docs/AGENT_TRUST_KERNEL.md`. Do not claim whole-kernel conformance from this ticket. + +**Goal:** Add a byte-deterministic, secret-safe content envelope that preserves untrusted +ingress, source evidence, transform lineage, and monotonic taint across all ATK content kinds. + +**Architecture:** Build a strict new Pydantic v2 subsystem and separate canonical codec. Reuse +RFC 8785 and SHA-256, add a domain-separated raw-byte hashing primitive, derive only from +verified parent envelopes, and expose no V1 taint-removal or authority operation. + +`content_models.py` and `content_envelope.py` must not import or call legacy `canon()` or +`hash_value()` on any path. Structured ingress may use only the subsystem's secret-safe +canonical-metadata prevalidator followed by `hash_bytes(DigestDomain)`; opaque ingress goes +directly to that typed raw-byte hasher. An AST/import guard forbids a module handle to +`mcp_warden.hashing` and permits only the named `hash_bytes` import, closing alias, `getattr`, +and dynamic attribute paths to the legacy helpers. The generic `canon()` logger remains +legacy-only, is excluded from this subsystem, and never receives ingress data. + +## Fixed contract + +- Source kinds: tool result, web, document, email, database text, MCP prompt, MCP resource, + bundle metadata, and agent message; derived content has a constructor-owned `derived` source. +- Every envelope is schema v1 and explicitly `untrusted`. +- Content and signature evidence are opaque raw bytes: never decoded, parsed, or normalized. + Every structured metadata input must be exact RFC 8785 bytes, must pass + `canon(parsed) == input`, and is then domain-hashed exactly. There is no `json.dumps` path; + structured metadata is discarded, never serialized, and receives no Unicode normalization. +- The wire schema has no absent fields. `bundle` is the only nullable field and must be present + as null or a complete object. It has no Pydantic default; the parser checks explicit key + presence before model construction. Omitted bundle rejects and golden vectors cover both. +- Parent identity is exactly `{envelope_digest, content_digest}`. Refs sort lexicographically + by the full lowercase ASCII envelope digest then content digest and are unique by envelope + digest. Revalidating + the parent self-digest commits source kind, source claims, taints, and content digest, so + repeating source kind in the flat ref would be redundant. +- Parent refs and taints are tuples in frozen nested models. Derived taints are every parent + taint plus added registered taints plus `core:untrusted`; no removal API exists. Mutation + tests cover the top-level envelope and every nested object/collection. +- Strict unknown-field/version rejection, exact RFC 8785 parsing, duplicate-key rejection, + domain-separated digests, stable code-only errors, and explicit public-field extraction. +- Transform IDs and versions are never caller text. Exact `TransformKindV1` values select an + immutable append-only registry pair: ingress is fixed to `INGRESS_CAPTURE`, derivation to + `DETERMINISTIC`, and the implementation digest completes exact transform identity. The + registry has an exact collision-checked reverse mapping. Model validation and parsing bind + every ingress root to the capture pair and every derived envelope to the deterministic pair, + including self-coherent wire objects whose source digests and envelope digest were recomputed + after a pair swap. Registry additions are security-sensitive and require a new enum member, + updated context enforcement, and explicit cross-pair/wire-forgery tests. The implementation + digest binds exact executing bytes; approval of that digest remains DSE-716 scope. +- Supported model construction/class-validation and frozen mutation entrypoints convert + Pydantic failures to code-only errors. `TypeAdapter`, private validators, core-schema APIs, + and unsafe construction are explicitly outside the untrusted-input boundary; every public + envelope API revalidates exact model types before use. Exact-type-before-access checks are a + required, test-enforced convention for every future public accepting boundary. +- Hard caps: content 16 MiB, envelope 64 KiB, source identity 8 KiB, source claims/parameters + 32 KiB, transform identity 8 KiB, bundle component 64 KiB, depth 16, nodes 4,096, parents + 64, taints 32, dependencies 256. +- Bundle digests are descriptive evidence only; DSE-716 owns verification and load authority. +- The taint registry is reviewed, append-only security code within V1. Unknown labels reject + the whole envelope as `ENV-TAINT-UNKNOWN` and are never dropped. Removing/changing a label or + incompatible wire evolution requires a schema bump and parallel explicit parser; V1 never + guesses or silently upgrades. The parser consumes a private immutable module `frozenset` + snapshot, not a caller-mutable registry. + +### Digest domains + +Every locally computed digest uses the exact construction +`sha256(domain + b"\x00" + payload)`, formatted as `sha256:<64 lowercase hex>`. Domains are: + +```text +mcp-warden/content-envelope/v1/content +mcp-warden/content-envelope/v1/source-identity +mcp-warden/content-envelope/v1/source-claims +mcp-warden/content-envelope/v1/derived-source-identity +mcp-warden/content-envelope/v1/derived-source-claims +mcp-warden/content-envelope/v1/transform-implementation +mcp-warden/content-envelope/v1/transform-parameters +mcp-warden/content-envelope/v1/bundle-artifact +mcp-warden/content-envelope/v1/bundle-signature +mcp-warden/content-envelope/v1/bundle-version +mcp-warden/content-envelope/v1/bundle-publisher +mcp-warden/content-envelope/v1/bundle-dependency +mcp-warden/content-envelope/v1/bundle-policy-binding +mcp-warden/content-envelope/v1/envelope +``` + +Golden tests prove same-domain repeatability and different-domain separation. +`content_models.py` owns a frozen `DigestDomain` string enum for every entry. All call sites use +enum members—never inline strings—and a programmatic pairwise test covers every distinct pair. + +## Task 1: Pin strict models and byte hashing + +**Files:** +- Create `src/mcp_warden/content_models.py` +- Modify `src/mcp_warden/hashing.py` +- Create `tests/test_content_envelope.py` + +1. Write failing golden tests for strict schema/version/trust state, explicit null versus + omitted bundle/key presence, frozen nested models/every primitive and collection mutation, + frozen domain enum, digest syntax, exact raw-byte digest, programmatic pairwise domain + separation, and an AST/import guard forbidding `canon`/`hash_value` in both modules. +2. Run the targeted tests and record the expected failures in the AI-SDLC log. +3. Add strict frozen models, enums/registries/caps, `ContentEnvelopeError`, and + `hash_bytes(payload, domain=...)`. +4. Rerun targeted tests to green; refactor common digest validation only after green. + +## Task 2: Ingress construction + +**Files:** +- Create `src/mcp_warden/content_envelope.py` +- Extend `tests/test_content_envelope.py` + +1. Write failing tests for all nine source kinds, exact non-UTF-8 content, canonical metadata, + opaque signature bytes, exact UTF-8/no-normalization/no-`json.dumps` behavior, media + registry, bundle requirement, sorted taints, and every cap boundary. Safely monkeypatch a + parser-owned immutable registry snapshot to prove drift/unknown taints reject rather than + disappear. Exercise exact cap and cap+1 boundary arithmetic. + The boundary matrix covers source identity/claims, transform implementation/parameters, + every bundle component, dependency count/component size, JSON depth/nodes/keys/string, + serialized envelope bytes, parents, and taint input count. +2. Implement `create_ingress()` and canonical-metadata validation. +3. Prove raw source identity/claims, transform implementation/parameters, bundle metadata, + and content never appear in envelope serialization or error text. +4. Before stdlib JSON parsing, run one allocation-free pass over byte-capped, UTF-8-validated + bytes with only integer `depth`, boolean `in_string`, and boolean `escape`. Outside strings, + ASCII `{`/`[` increments, `}`/`]` decrements, and `"` enters a string. Inside strings, + backslash escapes exactly the next byte and an unescaped quote exits; `\uXXXX`, bracket + bytes in strings, and multibyte UTF-8 are never structural. Reject depth >16, negative or + nonzero final depth, unterminated strings, and dangling escapes. Then use `json.loads` with + duplicate-key-rejecting `object_pairs_hook` and an iterative structural-cap pass. A + 10,000-level input under 64 KiB must reject before `json.loads`; monkeypatch it to prove no + call and no `RecursionError`. + +## Task 3: Deterministic transforms and lineage + +**Files:** +- Modify `src/mcp_warden/content_envelope.py` +- Extend `tests/test_content_envelope.py` + +1. Write failing tests for every parent ordering, exact ASCII lexicographic dual-digest refs, + deterministic derived source, atomic parent self-digest tamper rejection before refs/taints, + exact parent-set comparison, duplicate/missing/self parents, flat crafted cycles, and + complete bounded taint union. Cover direct model construction, public validation, and + canonical parse rejection of both source-context/transform-pair swaps, recomputing derived + source evidence and the envelope self-digest so the forged wire is otherwise self-coherent. +2. Implement `derive_envelope()` and `verify_lineage()` from verified parent objects. +3. Prove no API/schema key can remove taint, upgrade trust, or grant authority. +4. Store only flat immutable refs. Verification is iterative one-hop/exact-set, does not walk + ancestors, and rejects self-parent. Prevalidate at most 64 parents and 32 taints each, build + the complete local union over at most 2,048 parent labels plus bounded caller labels, then + perform one final 32-taint cap check. Discard the local set on error. Property tests prove + all parent orderings return the same success or stable failure code. +5. `derive_envelope()` must atomically recompute every parent's self-digest from its in-memory + constituent fields as its first step. One-hop verification commits the parent's own flat + refs but intentionally does not prove a transitive graph; multi-hop orchestration belongs + to DSE-716/717. + +## Task 4: Codec, verification, and public output + +**Files:** +- Modify `src/mcp_warden/content_envelope.py` +- Extend `tests/test_content_envelope.py` + +1. Write failing golden round-trip tests and malformed cases: unknown version/field, duplicate + key, noncanonical JSON, NaN/infinity, bad digest, self/content mismatch, wrong ordering, + exact scanner state/escape cases, deep/large structures, and attempted authority fields. +2. Implement `serialize_envelope()`, `parse_envelope()`, `verify_envelope()`, + `to_public_dict()`, and `to_public_bytes()`. + `parse_envelope()` atomically recomputes the self-digest from parsed constituent fields and + rejects mismatch before returning any envelope. +3. Public output must use explicit field extraction, never whole-model dumping. +4. Plant a unique secret in every raw input, including policy-like bundle metadata, and + byte-scan canonical/public output, `str(error)`, `repr(error)`, and formatted traceback. + Catch third-party/Pydantic/RFC/JSON errors and raise `ContentEnvelopeError(code) from None`: + no raw error logging/chaining/context, and tests assert cause/context absent. The generic + `canon()` logger must never receive raw untrusted metadata; body canon sees digests only. + A `caplog` planted-secret test proves the subsystem never logs raw exception/input text. + +## Task 5: Property and fuzz tests + +**Files:** +- Create `tests/fuzz/test_fuzz_content_envelope.py` + +Add Hypothesis properties for repeated byte identity, parent permutation invariance, sorted +uniqueness, taint monotonicity, canonical round-trip, single-byte mutation rejection, bounded +termination on malformed/deep bytes, flat-cycle non-recursion, registry drift rejection, and +planted-secret absence. Prefer deterministic structural bounds; do not use flaky wall-clock +deadlines as a correctness assertion. + +## Task 6: Document verified behavior + +**Files:** +- Create `docs/CONTENT_ENVELOPE.md` +- Modify `DOCUMENTATION_INDEX.md` +- Modify `SYSTEM_CONTEXT_DIAGRAM.md` +- Modify `docs/AGENT_TRUST_KERNEL.md` + +Document only the behavior proved by tests. Map ATK-01/02/05/06/09/12 and keep DSE-716/717 +bindings and current whole-kernel nonconformance explicit. + +## Verification + +Run after each red/green slice: + +```bash +PYTHONPATH=src .venv/bin/python -m pytest tests/test_content_envelope.py -q +PYTHONPATH=src .venv/bin/python -m pytest tests/fuzz/test_fuzz_content_envelope.py -q +``` + +Final local gate: + +```bash +.venv/bin/ruff check src/mcp_warden/content_models.py \ + src/mcp_warden/content_envelope.py src/mcp_warden/hashing.py \ + tests/test_content_envelope.py tests/fuzz/test_fuzz_content_envelope.py +PYTHONPATH=src .venv/bin/python -m pytest -q +git diff --check +``` + +Then run link/docs checks, adversarial Conclave review, independent security review, and CI. +Only verified, reviewed work proceeds to admin merge. diff --git a/src/mcp_warden/content_envelope.py b/src/mcp_warden/content_envelope.py new file mode 100644 index 0000000..b03cc24 --- /dev/null +++ b/src/mcp_warden/content_envelope.py @@ -0,0 +1,1038 @@ +"""Deterministic constructors and codec for untrusted content envelopes.""" + +from __future__ import annotations + +import json +from collections.abc import Iterable, Sequence +from typing import Any, NoReturn + +import rfc8785 +from pydantic import ValidationError + +from mcp_warden.content_models import ( + MAX_BUNDLE_COMPONENT_BYTES, + MAX_CONTENT_BYTES, + MAX_DEPENDENCIES, + MAX_ENVELOPE_BYTES, + MAX_JSON_DEPTH, + MAX_JSON_NODES, + MAX_JSON_OBJECT_KEYS, + MAX_JSON_STRING_BYTES, + MAX_PARENTS, + MAX_SOURCE_CLAIMS_BYTES, + MAX_SOURCE_IDENTITY_BYTES, + MAX_TAINTS, + MAX_TRANSFORM_IDENTITY_BYTES, + MAX_TRANSFORM_PARAMETERS_BYTES, + TAINT_REGISTRY_V1, + TRANSFORM_REGISTRY_V1, + BundleEvidenceInput, + BundleEvidenceV1, + ContentEnvelopeError, + ContentEnvelopeV1, + ContentEvidenceV1, + DigestDomain, + IngressKindV1, + MediaTypeV1, + ParentRefV1, + SourceEvidenceV1, + TaintV1, + TransformEvidenceV1, + TransformKindV1, +) +from mcp_warden.hashing import hash_bytes + +_TAINT_REGISTRY = frozenset(TAINT_REGISTRY_V1) + + +def _fail(code: str) -> NoReturn: + raise ContentEnvelopeError(code) from None + + +def _scan_json_structure(data: bytes) -> None: + depth = 0 + in_string = False + escape = False + for byte in data: + if in_string: + if escape: + escape = False + elif byte == 0x5C: # backslash + escape = True + elif byte == 0x22: # quote + in_string = False + continue + if byte == 0x22: + in_string = True + elif byte in (0x7B, 0x5B): # { [ + depth += 1 + if depth > MAX_JSON_DEPTH: + _fail("ENV-OVER-CAP") + elif byte in (0x7D, 0x5D): # } ] + depth -= 1 + if depth < 0: + _fail("ENV-MALFORMED") + if depth != 0 or in_string or escape: + _fail("ENV-MALFORMED") + + +class _DuplicateKeyError(Exception): + pass + + +def _unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise _DuplicateKeyError + result[key] = value + return result + + +def _reject_constant(_: str) -> NoReturn: + raise ValueError + + +def _check_structural_caps(value: object) -> None: + nodes = 0 + stack: list[object] = [value] + while stack: + item = stack.pop() + nodes += 1 + if nodes > MAX_JSON_NODES: + _fail("ENV-OVER-CAP") + if isinstance(item, dict): + if len(item) > MAX_JSON_OBJECT_KEYS: + _fail("ENV-OVER-CAP") + for key, child in item.items(): + if len(key.encode("utf-8")) > MAX_JSON_STRING_BYTES: + _fail("ENV-OVER-CAP") + stack.append(child) + elif isinstance(item, list): + stack.extend(item) + elif isinstance(item, str) and len(item.encode("utf-8")) > MAX_JSON_STRING_BYTES: + _fail("ENV-OVER-CAP") + + +def _canonical_metadata(data: bytes, *, cap: int, domain: DigestDomain) -> str: + if type(data) is not bytes: + _fail("ENV-MALFORMED") + if len(data) > cap: + _fail("ENV-OVER-CAP") + try: + text = data.decode("utf-8", errors="strict") + except UnicodeDecodeError: + text = "" + invalid_utf8 = True + else: + invalid_utf8 = False + if invalid_utf8: + _fail("ENV-MALFORMED") + _scan_json_structure(data) + + parsed: object | None = None + duplicate = False + malformed = False + try: + parsed = json.loads( + text, + object_pairs_hook=_unique_object, + parse_constant=_reject_constant, + ) + except _DuplicateKeyError: + duplicate = True + except (ValueError, TypeError, RecursionError): + malformed = True + if duplicate: + _fail("ENV-NONCANONICAL") + if malformed: + _fail("ENV-MALFORMED") + _check_structural_caps(parsed) + + canonical: bytes | None = None + canonical_failed = False + try: + canonical = rfc8785.dumps(parsed) + except Exception: + canonical_failed = True + if canonical_failed or canonical is None: + _fail("ENV-MALFORMED") + if canonical != data: + _fail("ENV-NONCANONICAL") + return hash_bytes(data, domain=domain) + + +def _hash_bundle(bundle: BundleEvidenceInput | None) -> BundleEvidenceV1 | None: + if bundle is None: + return None + if type(bundle) is not BundleEvidenceInput: + _fail("ENV-BUNDLE-INVALID") + components = ( + bundle.artifact, + bundle.version_claims, + bundle.publisher_claims, + bundle.policy_binding_claims, + ) + if any(type(value) is not bytes for value in components): + _fail("ENV-BUNDLE-INVALID") + if bundle.signature_evidence is not None and type(bundle.signature_evidence) is not bytes: + _fail("ENV-BUNDLE-INVALID") + if type(bundle.dependencies) is not tuple: + _fail("ENV-BUNDLE-INVALID") + if len(bundle.dependencies) > MAX_DEPENDENCIES: + _fail("ENV-OVER-CAP") + if any(type(value) is not bytes for value in bundle.dependencies): + _fail("ENV-BUNDLE-INVALID") + bounded = components + bundle.dependencies + if bundle.signature_evidence is not None: + bounded += (bundle.signature_evidence,) + if any(len(value) > MAX_BUNDLE_COMPONENT_BYTES for value in bounded): + _fail("ENV-OVER-CAP") + artifact = _canonical_metadata( + bundle.artifact, cap=MAX_BUNDLE_COMPONENT_BYTES, domain=DigestDomain.BUNDLE_ARTIFACT + ) + signature: str | None = None + if bundle.signature_evidence is not None: + if not isinstance(bundle.signature_evidence, bytes): + _fail("ENV-BUNDLE-INVALID") + if len(bundle.signature_evidence) > MAX_BUNDLE_COMPONENT_BYTES: + _fail("ENV-OVER-CAP") + signature = hash_bytes( + bundle.signature_evidence, + domain=DigestDomain.BUNDLE_SIGNATURE, + ) + dependency_digests = tuple( + sorted( + { + _canonical_metadata( + dependency, + cap=MAX_BUNDLE_COMPONENT_BYTES, + domain=DigestDomain.BUNDLE_DEPENDENCY, + ) + for dependency in bundle.dependencies + } + ) + ) + try: + return BundleEvidenceV1( + artifact_digest=artifact, + signature_evidence_digest=signature, + version_claims_digest=_canonical_metadata( + bundle.version_claims, + cap=MAX_BUNDLE_COMPONENT_BYTES, + domain=DigestDomain.BUNDLE_VERSION, + ), + publisher_claims_digest=_canonical_metadata( + bundle.publisher_claims, + cap=MAX_BUNDLE_COMPONENT_BYTES, + domain=DigestDomain.BUNDLE_PUBLISHER, + ), + dependency_digests=dependency_digests, + policy_binding_claims_digest=_canonical_metadata( + bundle.policy_binding_claims, + cap=MAX_BUNDLE_COMPONENT_BYTES, + domain=DigestDomain.BUNDLE_POLICY_BINDING, + ), + ) + except (ValidationError, TypeError, ValueError): + invalid = True + else: + invalid = False + if invalid: + _fail("ENV-BUNDLE-INVALID") + + +def _taints(added_taints: Iterable[TaintV1 | str]) -> tuple[str, ...]: + values_list: list[str] = [] + iteration_failed = False + invalid_item = False + try: + iterator = iter(added_taints) + for _ in range(MAX_TAINTS + 1): + try: + item = next(iterator) + except StopIteration: + break + if type(item) is TaintV1: + values_list.append(item.value) + elif type(item) is str: + values_list.append(item) + else: + invalid_item = True + break + except Exception: + iteration_failed = True + if iteration_failed or invalid_item: + _fail("ENV-TAINT-UNKNOWN") + if len(values_list) > MAX_TAINTS: + _fail("ENV-OVER-CAP") + values = tuple(values_list) + if len(values) > MAX_TAINTS: + _fail("ENV-OVER-CAP") + result = tuple(sorted(set(values) | {TaintV1.UNTRUSTED.value})) + if any(not isinstance(item, str) or item not in _TAINT_REGISTRY for item in result): + _fail("ENV-TAINT-UNKNOWN") + if len(result) > MAX_TAINTS: + _fail("ENV-OVER-CAP") + return result + + +def _bundle_dict(bundle: BundleEvidenceV1 | None) -> dict[str, object] | None: + if bundle is None: + return None + return { + "artifact_digest": bundle.artifact_digest, + "signature_evidence_digest": bundle.signature_evidence_digest, + "version_claims_digest": bundle.version_claims_digest, + "publisher_claims_digest": bundle.publisher_claims_digest, + "dependency_digests": list(bundle.dependency_digests), + "policy_binding_claims_digest": bundle.policy_binding_claims_digest, + } + + +def _body_dict( + *, + content: ContentEvidenceV1, + source: SourceEvidenceV1, + parents: tuple[ParentRefV1, ...], + transform: TransformEvidenceV1, + taints: tuple[str, ...], + bundle: BundleEvidenceV1 | None, +) -> dict[str, object]: + return { + "schema_version": 1, + "trust_state": "untrusted", + "content": { + "digest": content.digest, + "length": content.length, + "media_type": content.media_type, + }, + "source": { + "kind": source.kind, + "identity_digest": source.identity_digest, + "claims_digest": source.claims_digest, + }, + "parents": [ + {"envelope_digest": item.envelope_digest, "content_digest": item.content_digest} + for item in parents + ], + "transform": { + "id": transform.id, + "version": transform.version, + "implementation_digest": transform.implementation_digest, + "parameters_digest": transform.parameters_digest, + }, + "taints": list(taints), + "bundle": _bundle_dict(bundle), + } + + +def _make_envelope( + *, + content: ContentEvidenceV1, + source: SourceEvidenceV1, + parents: tuple[ParentRefV1, ...], + transform: TransformEvidenceV1, + taints: tuple[str, ...], + bundle: BundleEvidenceV1 | None, +) -> ContentEnvelopeV1: + body = _body_dict( + content=content, + source=source, + parents=parents, + transform=transform, + taints=taints, + bundle=bundle, + ) + body_bytes = rfc8785.dumps(body) + envelope_digest = hash_bytes(body_bytes, domain=DigestDomain.ENVELOPE) + try: + return ContentEnvelopeV1( + schema_version=1, + trust_state="untrusted", + content=content, + source=source, + parents=parents, + transform=transform, + taints=taints, + bundle=bundle, + envelope_digest=envelope_digest, + ) + except (ValidationError, TypeError, ValueError): + invalid = True + else: + invalid = False + if invalid: + _fail("ENV-MALFORMED") + + +def create_ingress( + *, + content: bytes, + media_type: MediaTypeV1, + source_kind: IngressKindV1, + source_identity: bytes, + source_claims: bytes, + capture_kind: TransformKindV1, + capture_implementation: bytes, + capture_parameters: bytes, + added_taints: Iterable[TaintV1 | str] = (), + bundle: BundleEvidenceInput | None = None, +) -> ContentEnvelopeV1: + """Create a root envelope from exact, explicitly untrusted ingress bytes.""" + if type(content) is not bytes or type(media_type) is not MediaTypeV1: + _fail("ENV-MALFORMED") + if type(source_kind) is not IngressKindV1: + _fail("ENV-MALFORMED") + if capture_kind is not TransformKindV1.INGRESS_CAPTURE: + _fail("ENV-MALFORMED") + if len(content) > MAX_CONTENT_BYTES: + _fail("ENV-OVER-CAP") + if source_kind is IngressKindV1.BUNDLE_METADATA and bundle is None: + _fail("ENV-BUNDLE-INVALID") + + try: + content_evidence = ContentEvidenceV1( + digest=hash_bytes(content, domain=DigestDomain.CONTENT), + length=len(content), + media_type=media_type.value, + ) + source = SourceEvidenceV1( + kind=source_kind.value, + identity_digest=_canonical_metadata( + source_identity, + cap=MAX_SOURCE_IDENTITY_BYTES, + domain=DigestDomain.SOURCE_IDENTITY, + ), + claims_digest=_canonical_metadata( + source_claims, + cap=MAX_SOURCE_CLAIMS_BYTES, + domain=DigestDomain.SOURCE_CLAIMS, + ), + ) + transform_id, transform_version = TRANSFORM_REGISTRY_V1[capture_kind] + transform = TransformEvidenceV1( + id=transform_id, + version=transform_version, + implementation_digest=_canonical_metadata( + capture_implementation, + cap=MAX_TRANSFORM_IDENTITY_BYTES, + domain=DigestDomain.TRANSFORM_IMPLEMENTATION, + ), + parameters_digest=_canonical_metadata( + capture_parameters, + cap=MAX_TRANSFORM_PARAMETERS_BYTES, + domain=DigestDomain.TRANSFORM_PARAMETERS, + ), + ) + except ContentEnvelopeError: + raise + except (ValidationError, TypeError, ValueError): + invalid = True + else: + invalid = False + if invalid: + _fail("ENV-MALFORMED") + return _make_envelope( + content=content_evidence, + source=source, + parents=(), + transform=transform, + taints=_taints(added_taints), + bundle=_hash_bundle(bundle), + ) + + +def _computed_envelope_digest(envelope: ContentEnvelopeV1) -> str: + try: + body = _body_dict( + content=envelope.content, + source=envelope.source, + parents=envelope.parents, + transform=envelope.transform, + taints=envelope.taints, + bundle=envelope.bundle, + ) + body_bytes = rfc8785.dumps(body) + return hash_bytes(body_bytes, domain=DigestDomain.ENVELOPE) + except Exception: + invalid = True + else: # pragma: no cover - return above; retained for explicit exception boundary + invalid = False + if invalid: + _fail("ENV-DIGEST-MISMATCH") + + +def _replace_self_digest(envelope: ContentEnvelopeV1) -> ContentEnvelopeV1: + """Test/support helper that replaces only the deterministically computed self digest.""" + return envelope.model_copy(update={"envelope_digest": _computed_envelope_digest(envelope)}) + + +def _preflight_envelope(envelope: ContentEnvelopeV1) -> None: + """Reject structurally incomplete constructed models before field dereference.""" + storage_failed = False + try: + storage = object.__getattribute__(envelope, "__dict__") + except (AttributeError, TypeError): + storage_failed = True + storage = {} + required = frozenset(ContentEnvelopeV1.model_fields) + if storage_failed or type(storage) is not dict or not required.issubset(storage): + _fail("ENV-MALFORMED") + if type(storage["content"]) is not ContentEvidenceV1: + _fail("ENV-MALFORMED") + if type(storage["source"]) is not SourceEvidenceV1: + _fail("ENV-MALFORMED") + if type(storage["transform"]) is not TransformEvidenceV1: + _fail("ENV-MALFORMED") + if type(storage["parents"]) is not tuple or len(storage["parents"]) > MAX_PARENTS: + _fail("ENV-LINEAGE-INVALID") + if any(type(parent) is not ParentRefV1 for parent in storage["parents"]): + _fail("ENV-LINEAGE-INVALID") + if type(storage["taints"]) is not tuple or len(storage["taints"]) > MAX_TAINTS: + _fail("ENV-LINEAGE-INVALID") + if storage["bundle"] is not None and type(storage["bundle"]) is not BundleEvidenceV1: + _fail("ENV-BUNDLE-INVALID") + + +def _verified_parents( + parents: Sequence[ContentEnvelopeV1], *, allow_empty: bool = False +) -> tuple[ContentEnvelopeV1, ...]: + if isinstance(parents, (bytes, bytearray, str)) or not isinstance(parents, Sequence): + _fail("ENV-LINEAGE-INVALID") + values_list: list[ContentEnvelopeV1] = [] + iteration_failed = False + try: + iterator = iter(parents) + for _ in range(MAX_PARENTS + 1): + try: + parent = next(iterator) + except StopIteration: + break + values_list.append(parent) + except Exception: + iteration_failed = True + if iteration_failed: + _fail("ENV-LINEAGE-INVALID") + if len(values_list) > MAX_PARENTS: + _fail("ENV-OVER-CAP") + values = tuple(values_list) + if not values and not allow_empty: + _fail("ENV-LINEAGE-INVALID") + # Recompute every complete parent self-digest before accepting refs or taints. + for parent in values: + if type(parent) is not ContentEnvelopeV1: + _fail("ENV-LINEAGE-INVALID") + _preflight_envelope(parent) + if _computed_envelope_digest(parent) != parent.envelope_digest: + _fail("ENV-DIGEST-MISMATCH") + for parent in values: + _validate_envelope_shape(parent) + if len({parent.envelope_digest for parent in values}) != len(values): + _fail("ENV-LINEAGE-INVALID") + return values + + +def _derived_source( + refs: tuple[ParentRefV1, ...], transform: TransformEvidenceV1 +) -> SourceEvidenceV1: + ref_payload = rfc8785.dumps( + [ + {"envelope_digest": ref.envelope_digest, "content_digest": ref.content_digest} + for ref in refs + ] + ) + identity_digest = hash_bytes(ref_payload, domain=DigestDomain.DERIVED_SOURCE_IDENTITY) + claims_payload = rfc8785.dumps( + { + "parent_refs_digest": identity_digest, + "transform_id": transform.id, + "transform_version": transform.version, + "implementation_digest": transform.implementation_digest, + "parameters_digest": transform.parameters_digest, + } + ) + return SourceEvidenceV1( + kind="derived", + identity_digest=identity_digest, + claims_digest=hash_bytes(claims_payload, domain=DigestDomain.DERIVED_SOURCE_CLAIMS), + ) + + +def _validate_derived_source(envelope: ContentEnvelopeV1) -> None: + if envelope.source.kind == "derived" and envelope.source != _derived_source( + envelope.parents, envelope.transform + ): + _fail("ENV-LINEAGE-INVALID") + + +def _revalidate_model(value: object, model_type: type, code: str) -> None: + invalid = False + try: + model_type.model_validate(value) + except (ValidationError, TypeError, ValueError, AttributeError): + invalid = True + if invalid: + _fail(code) + + +def _validate_envelope_shape(envelope: ContentEnvelopeV1) -> None: + _preflight_envelope(envelope) + if type(envelope.schema_version) is not int or envelope.schema_version != 1: + _fail("ENV-SCHEMA-UNKNOWN") + if type(envelope.trust_state) is not str or envelope.trust_state != "untrusted": + _fail("ENV-TRUST-INVALID") + if type(envelope.content) is not ContentEvidenceV1: + _fail("ENV-MALFORMED") + if type(envelope.source) is not SourceEvidenceV1: + _fail("ENV-MALFORMED") + if type(envelope.transform) is not TransformEvidenceV1: + _fail("ENV-MALFORMED") + _revalidate_model(envelope.content, ContentEvidenceV1, "ENV-MALFORMED") + _revalidate_model(envelope.source, SourceEvidenceV1, "ENV-MALFORMED") + _revalidate_model(envelope.transform, TransformEvidenceV1, "ENV-MALFORMED") + if type(envelope.parents) is not tuple or len(envelope.parents) > MAX_PARENTS: + _fail("ENV-LINEAGE-INVALID") + for parent in envelope.parents: + if type(parent) is not ParentRefV1: + _fail("ENV-LINEAGE-INVALID") + _revalidate_model(parent, ParentRefV1, "ENV-LINEAGE-INVALID") + if envelope.parents != tuple( + sorted( + envelope.parents, + key=lambda item: (item.envelope_digest, item.content_digest), + ) + ) or len({item.envelope_digest for item in envelope.parents}) != len(envelope.parents): + _fail("ENV-LINEAGE-INVALID") + if type(envelope.taints) is not tuple or len(envelope.taints) > MAX_TAINTS: + _fail("ENV-LINEAGE-INVALID") + if any(type(label) is not str for label in envelope.taints): + _fail("ENV-TAINT-UNKNOWN") + if "core:untrusted" not in envelope.taints: + _fail("ENV-TAINT-MISSING") + if any(label not in _TAINT_REGISTRY for label in envelope.taints): + _fail("ENV-TAINT-UNKNOWN") + if envelope.taints != tuple(sorted(set(envelope.taints))): + _fail("ENV-LINEAGE-INVALID") + if envelope.bundle is not None: + if type(envelope.bundle) is not BundleEvidenceV1: + _fail("ENV-BUNDLE-INVALID") + _revalidate_model(envelope.bundle, BundleEvidenceV1, "ENV-BUNDLE-INVALID") + if envelope.source.kind == "derived": + if not envelope.parents: + _fail("ENV-LINEAGE-INVALID") + _validate_derived_source(envelope) + elif envelope.parents: + _fail("ENV-LINEAGE-INVALID") + if envelope.source.kind == IngressKindV1.BUNDLE_METADATA.value and envelope.bundle is None: + _fail("ENV-BUNDLE-INVALID") + _revalidate_model(envelope, ContentEnvelopeV1, "ENV-MALFORMED") + + +def derive_envelope( + *, + content: bytes, + media_type: MediaTypeV1, + parents: Sequence[ContentEnvelopeV1], + transform_kind: TransformKindV1, + transform_implementation: bytes, + transform_parameters: bytes, + added_taints: Iterable[TaintV1 | str] = (), + bundle: BundleEvidenceInput | None = None, +) -> ContentEnvelopeV1: + """Derive one envelope from a bounded, atomically verified parent set.""" + verified = _verified_parents(parents) + if type(content) is not bytes or type(media_type) is not MediaTypeV1: + _fail("ENV-MALFORMED") + if transform_kind is not TransformKindV1.DETERMINISTIC: + _fail("ENV-MALFORMED") + if len(content) > MAX_CONTENT_BYTES: + _fail("ENV-OVER-CAP") + + refs = tuple( + sorted( + ( + ParentRefV1( + envelope_digest=parent.envelope_digest, + content_digest=parent.content.digest, + ) + for parent in verified + ), + key=lambda item: (item.envelope_digest, item.content_digest), + ) + ) + local_taints: set[str] = set() + for parent in verified: + if len(parent.taints) > MAX_TAINTS: + _fail("ENV-OVER-CAP") + if any(label not in _TAINT_REGISTRY for label in parent.taints): + _fail("ENV-TAINT-UNKNOWN") + local_taints.update(parent.taints) + caller_taints = _taints(added_taints) + local_taints.update(caller_taints) + if len(local_taints) > MAX_TAINTS: + _fail("ENV-OVER-CAP") + taints = tuple(sorted(local_taints)) + + try: + transform_id, transform_version = TRANSFORM_REGISTRY_V1[transform_kind] + transform = TransformEvidenceV1( + id=transform_id, + version=transform_version, + implementation_digest=_canonical_metadata( + transform_implementation, + cap=MAX_TRANSFORM_IDENTITY_BYTES, + domain=DigestDomain.TRANSFORM_IMPLEMENTATION, + ), + parameters_digest=_canonical_metadata( + transform_parameters, + cap=MAX_TRANSFORM_PARAMETERS_BYTES, + domain=DigestDomain.TRANSFORM_PARAMETERS, + ), + ) + content_evidence = ContentEvidenceV1( + digest=hash_bytes(content, domain=DigestDomain.CONTENT), + length=len(content), + media_type=media_type.value, + ) + except ContentEnvelopeError: + raise + except (ValidationError, TypeError, ValueError): + invalid = True + else: + invalid = False + if invalid: + _fail("ENV-MALFORMED") + + source = _derived_source(refs, transform) + return _make_envelope( + content=content_evidence, + source=source, + parents=refs, + transform=transform, + taints=taints, + bundle=_hash_bundle(bundle), + ) + + +def verify_lineage( + envelope: ContentEnvelopeV1, *, parents: Sequence[ContentEnvelopeV1] +) -> None: + """Verify the exact one-hop parent set and monotonic local taint union.""" + if type(envelope) is not ContentEnvelopeV1: + _fail("ENV-LINEAGE-INVALID") + _validate_envelope_shape(envelope) + if _computed_envelope_digest(envelope) != envelope.envelope_digest: + _fail("ENV-DIGEST-MISMATCH") + is_derived = envelope.source.kind == "derived" + verified = _verified_parents(parents, allow_empty=not is_derived) + if is_derived != bool(envelope.parents): + _fail("ENV-LINEAGE-INVALID") + if not is_derived: + if verified: + _fail("ENV-LINEAGE-INVALID") + return + + expected = tuple( + sorted( + ( + ParentRefV1( + envelope_digest=parent.envelope_digest, + content_digest=parent.content.digest, + ) + for parent in verified + ), + key=lambda item: (item.envelope_digest, item.content_digest), + ) + ) + if expected != envelope.parents: + _fail("ENV-LINEAGE-INVALID") + if any(ref.envelope_digest == envelope.envelope_digest for ref in envelope.parents): + _fail("ENV-LINEAGE-INVALID") + required_taints = {TaintV1.UNTRUSTED.value} + for parent in verified: + required_taints.update(parent.taints) + if not required_taints.issubset(envelope.taints): + _fail("ENV-TAINT-MISSING") + + +def verify_envelope(envelope: ContentEnvelopeV1, *, content: bytes) -> None: + """Verify envelope self-integrity and exact supplied content bytes.""" + if type(envelope) is not ContentEnvelopeV1 or type(content) is not bytes: + _fail("ENV-MALFORMED") + _validate_envelope_shape(envelope) + if len(content) > MAX_CONTENT_BYTES: + _fail("ENV-OVER-CAP") + if _computed_envelope_digest(envelope) != envelope.envelope_digest: + _fail("ENV-DIGEST-MISMATCH") + if envelope.content.length != len(content): + _fail("ENV-DIGEST-MISMATCH") + if envelope.content.digest != hash_bytes(content, domain=DigestDomain.CONTENT): + _fail("ENV-DIGEST-MISMATCH") + if envelope.source.kind == "derived": + if not envelope.parents: + _fail("ENV-LINEAGE-INVALID") + elif envelope.parents: + _fail("ENV-LINEAGE-INVALID") + if envelope.source.kind == IngressKindV1.BUNDLE_METADATA.value and envelope.bundle is None: + _fail("ENV-BUNDLE-INVALID") + + +def to_public_dict(envelope: ContentEnvelopeV1) -> dict[str, object]: + """Return the explicit digest-only V1 wire projection.""" + if type(envelope) is not ContentEnvelopeV1: + _fail("ENV-MALFORMED") + _validate_envelope_shape(envelope) + if _computed_envelope_digest(envelope) != envelope.envelope_digest: + _fail("ENV-DIGEST-MISMATCH") + body = _body_dict( + content=envelope.content, + source=envelope.source, + parents=envelope.parents, + transform=envelope.transform, + taints=envelope.taints, + bundle=envelope.bundle, + ) + body["envelope_digest"] = envelope.envelope_digest + return body + + +def to_public_bytes(envelope: ContentEnvelopeV1) -> bytes: + encoded = rfc8785.dumps(to_public_dict(envelope)) + if len(encoded) > MAX_ENVELOPE_BYTES: + _fail("ENV-OVER-CAP") + return encoded + + +def serialize_envelope(envelope: ContentEnvelopeV1) -> bytes: + """Serialize an envelope to exact RFC 8785 wire bytes.""" + return to_public_bytes(envelope) + + +def _parse_wire_json(data: bytes) -> object: + if type(data) is not bytes: + _fail("ENV-MALFORMED") + if len(data) > MAX_ENVELOPE_BYTES: + _fail("ENV-OVER-CAP") + try: + text = data.decode("utf-8", errors="strict") + except UnicodeDecodeError: + invalid_utf8 = True + text = "" + else: + invalid_utf8 = False + if invalid_utf8: + _fail("ENV-MALFORMED") + _scan_json_structure(data) + + parsed: object | None = None + duplicate = False + malformed = False + try: + parsed = json.loads( + text, + object_pairs_hook=_unique_object, + parse_constant=_reject_constant, + ) + except _DuplicateKeyError: + duplicate = True + except (ValueError, TypeError, RecursionError): + malformed = True + if duplicate: + _fail("ENV-NONCANONICAL") + if malformed: + _fail("ENV-MALFORMED") + _check_structural_caps(parsed) + + canonical: bytes | None = None + canonical_failed = False + try: + canonical = rfc8785.dumps(parsed) + except Exception: + canonical_failed = True + if canonical_failed or canonical is None: + _fail("ENV-MALFORMED") + if canonical != data: + _fail("ENV-NONCANONICAL") + return parsed + + +def _object_with_fields( + value: object, expected: frozenset[str], *, missing_code: str = "ENV-MALFORMED" +) -> dict[str, object]: + if not isinstance(value, dict) or any(not isinstance(key, str) for key in value): + _fail("ENV-MALFORMED") + keys = frozenset(value) + if keys - expected: + _fail("ENV-UNKNOWN-FIELD") + if expected - keys: + _fail(missing_code) + return value + + +def parse_envelope(data: bytes) -> ContentEnvelopeV1: + """Parse strict V1 bytes and atomically verify their self digest.""" + parsed = _parse_wire_json(data) + top_expected = frozenset( + { + "schema_version", + "trust_state", + "content", + "source", + "parents", + "transform", + "taints", + "bundle", + "envelope_digest", + } + ) + if not isinstance(parsed, dict): + _fail("ENV-MALFORMED") + if "schema_version" not in parsed: + _fail("ENV-SCHEMA-UNKNOWN") + if "trust_state" not in parsed: + _fail("ENV-TRUST-INVALID") + if "taints" not in parsed: + _fail("ENV-TAINT-MISSING") + if "bundle" not in parsed: + _fail("ENV-BUNDLE-INVALID") + top = _object_with_fields(parsed, top_expected) + if top["schema_version"] != 1 or isinstance(top["schema_version"], bool): + _fail("ENV-SCHEMA-UNKNOWN") + if top["trust_state"] != "untrusted": + _fail("ENV-TRUST-INVALID") + + content_data = _object_with_fields( + top["content"], frozenset({"digest", "length", "media_type"}) + ) + source_data = _object_with_fields( + top["source"], frozenset({"kind", "identity_digest", "claims_digest"}) + ) + transform_data = _object_with_fields( + top["transform"], + frozenset({"id", "version", "implementation_digest", "parameters_digest"}), + ) + if not isinstance(top["parents"], list) or not isinstance(top["taints"], list): + _fail("ENV-MALFORMED") + if len(top["parents"]) > MAX_PARENTS or len(top["taints"]) > MAX_TAINTS: + _fail("ENV-OVER-CAP") + + parent_dicts_list: list[dict[str, object]] = [] + for item in top["parents"]: + if not isinstance(item, dict): + _fail("ENV-LINEAGE-INVALID") + parent_dicts_list.append( + _object_with_fields( + item, + frozenset({"envelope_digest", "content_digest"}), + missing_code="ENV-LINEAGE-INVALID", + ) + ) + parent_dicts = tuple(parent_dicts_list) + try: + parents = tuple(ParentRefV1(**item) for item in parent_dicts) + except (ValidationError, TypeError, ValueError): + invalid_parents = True + parents = () + else: + invalid_parents = False + if invalid_parents: + _fail("ENV-LINEAGE-INVALID") + if parents != tuple( + sorted(parents, key=lambda item: (item.envelope_digest, item.content_digest)) + ) or len({item.envelope_digest for item in parents}) != len(parents): + _fail("ENV-LINEAGE-INVALID") + + bundle_data: dict[str, object] | None = None + if top["bundle"] is not None: + bundle_data = _object_with_fields( + top["bundle"], + frozenset( + { + "artifact_digest", + "signature_evidence_digest", + "version_claims_digest", + "publisher_claims_digest", + "dependency_digests", + "policy_binding_claims_digest", + } + ), + missing_code="ENV-BUNDLE-INVALID", + ) + dependencies = bundle_data["dependency_digests"] + if not isinstance(dependencies, list): + _fail("ENV-BUNDLE-INVALID") + if len(dependencies) > MAX_DEPENDENCIES: + _fail("ENV-OVER-CAP") + + bundle: BundleEvidenceV1 | None = None + invalid_bundle = False + if bundle_data is not None: + try: + bundle = BundleEvidenceV1( + artifact_digest=bundle_data["artifact_digest"], + signature_evidence_digest=bundle_data["signature_evidence_digest"], + version_claims_digest=bundle_data["version_claims_digest"], + publisher_claims_digest=bundle_data["publisher_claims_digest"], + dependency_digests=tuple(bundle_data["dependency_digests"]), + policy_binding_claims_digest=bundle_data["policy_binding_claims_digest"], + ) + except (ValidationError, TypeError, ValueError): + invalid_bundle = True + if invalid_bundle: + _fail("ENV-BUNDLE-INVALID") + + taint_values = top["taints"] + if any(not isinstance(item, str) or item not in _TAINT_REGISTRY for item in taint_values): + _fail("ENV-TAINT-UNKNOWN") + if "core:untrusted" not in taint_values: + _fail("ENV-TAINT-MISSING") + if taint_values != sorted(set(taint_values)): + _fail("ENV-LINEAGE-INVALID") + + try: + content = ContentEvidenceV1(**content_data) + source = SourceEvidenceV1(**source_data) + transform = TransformEvidenceV1(**transform_data) + except (ValidationError, TypeError, ValueError): + invalid = True + content = None + source = None + transform = None + else: + invalid = False + if invalid or content is None or source is None or transform is None: + _fail("ENV-MALFORMED") + + if source.kind == "derived": + if not parents: + _fail("ENV-LINEAGE-INVALID") + elif parents: + _fail("ENV-LINEAGE-INVALID") + if source.kind == IngressKindV1.BUNDLE_METADATA.value and bundle is None: + _fail("ENV-BUNDLE-INVALID") + try: + envelope = ContentEnvelopeV1( + schema_version=top["schema_version"], + trust_state=top["trust_state"], + content=content, + source=source, + parents=parents, + transform=transform, + taints=tuple(taint_values), + bundle=bundle, + envelope_digest=top["envelope_digest"], + ) + except (ValidationError, TypeError, ValueError): + invalid_envelope = True + envelope = None + else: + invalid_envelope = False + if invalid_envelope or envelope is None: + _fail("ENV-MALFORMED") + if any(ref.envelope_digest == envelope.envelope_digest for ref in parents): + _fail("ENV-LINEAGE-INVALID") + if _computed_envelope_digest(envelope) != envelope.envelope_digest: + _fail("ENV-DIGEST-MISMATCH") + _validate_envelope_shape(envelope) + return envelope diff --git a/src/mcp_warden/content_models.py b/src/mcp_warden/content_models.py new file mode 100644 index 0000000..87f89c5 --- /dev/null +++ b/src/mcp_warden/content_models.py @@ -0,0 +1,415 @@ +"""Strict immutable models for deterministic untrusted content envelopes. + +This module intentionally has no dependency on the legacy lockfile models or +canonicalization helpers. It defines evidence only; no field grants authority. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from enum import StrEnum +from types import MappingProxyType +from typing import Any, ClassVar, Literal + +from pydantic import ( + BaseModel, + ConfigDict, + StrictInt, + ValidationError, + field_validator, + model_validator, +) + +DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$") +TRANSFORM_ID_RE = re.compile(r"^[a-z0-9](?:[a-z0-9._-]{0,126}[a-z0-9])?$") +TRANSFORM_VERSION_RE = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9._+-]{0,62}[A-Za-z0-9])?$") + +MAX_CONTENT_BYTES = 16 * 1024 * 1024 +MAX_ENVELOPE_BYTES = 64 * 1024 +MAX_SOURCE_IDENTITY_BYTES = 8 * 1024 +MAX_SOURCE_CLAIMS_BYTES = 32 * 1024 +MAX_TRANSFORM_IDENTITY_BYTES = 8 * 1024 +MAX_TRANSFORM_PARAMETERS_BYTES = 32 * 1024 +MAX_BUNDLE_COMPONENT_BYTES = 64 * 1024 +MAX_JSON_DEPTH = 16 +MAX_JSON_NODES = 4_096 +MAX_JSON_OBJECT_KEYS = 128 +MAX_JSON_STRING_BYTES = 8 * 1024 +MAX_PARENTS = 64 +MAX_TAINTS = 32 +MAX_DEPENDENCIES = 256 + + +class DigestDomain(StrEnum): + """Closed, typed namespace for every V1 digest.""" + + CONTENT = "mcp-warden/content-envelope/v1/content" + SOURCE_IDENTITY = "mcp-warden/content-envelope/v1/source-identity" + SOURCE_CLAIMS = "mcp-warden/content-envelope/v1/source-claims" + DERIVED_SOURCE_IDENTITY = "mcp-warden/content-envelope/v1/derived-source-identity" + DERIVED_SOURCE_CLAIMS = "mcp-warden/content-envelope/v1/derived-source-claims" + TRANSFORM_IMPLEMENTATION = "mcp-warden/content-envelope/v1/transform-implementation" + TRANSFORM_PARAMETERS = "mcp-warden/content-envelope/v1/transform-parameters" + BUNDLE_ARTIFACT = "mcp-warden/content-envelope/v1/bundle-artifact" + BUNDLE_SIGNATURE = "mcp-warden/content-envelope/v1/bundle-signature" + BUNDLE_VERSION = "mcp-warden/content-envelope/v1/bundle-version" + BUNDLE_PUBLISHER = "mcp-warden/content-envelope/v1/bundle-publisher" + BUNDLE_DEPENDENCY = "mcp-warden/content-envelope/v1/bundle-dependency" + BUNDLE_POLICY_BINDING = "mcp-warden/content-envelope/v1/bundle-policy-binding" + ENVELOPE = "mcp-warden/content-envelope/v1/envelope" + + +class IngressKindV1(StrEnum): + TOOL_RESULT = "tool_result" + WEB = "web" + DOCUMENT = "document" + EMAIL = "email" + DATABASE_TEXT = "database_text" + MCP_PROMPT = "mcp_prompt" + MCP_RESOURCE = "mcp_resource" + BUNDLE_METADATA = "bundle_metadata" + AGENT_MESSAGE = "agent_message" + + +class MediaTypeV1(StrEnum): + APPLICATION_JSON = "application/json" + APPLICATION_OCTET_STREAM = "application/octet-stream" + APPLICATION_PDF = "application/pdf" + APPLICATION_XML = "application/xml" + APPLICATION_ZIP = "application/zip" + MESSAGE_RFC822 = "message/rfc822" + TEXT_CSV = "text/csv" + TEXT_HTML = "text/html" + TEXT_MARKDOWN = "text/markdown" + TEXT_PLAIN = "text/plain" + TEXT_XML = "text/xml" + + +class TaintV1(StrEnum): + UNTRUSTED = "core:untrusted" + MALFORMED = "core:malformed" + UNINSPECTABLE = "core:uninspectable" + SENSITIVE = "core:sensitive" + EXECUTABLE = "core:executable" + AUTHORITY_INJECTION = "core:authority-injection" + SECRET = "core:secret" + PRIVATE_NETWORK = "core:private-network" + CRITICAL = "core:critical" + + +class TransformKindV1(StrEnum): + INGRESS_CAPTURE = "ingress_capture" + DETERMINISTIC = "deterministic" + + +TAINT_REGISTRY_V1 = frozenset(item.value for item in TaintV1) +MEDIA_TYPE_REGISTRY_V1 = frozenset(item.value for item in MediaTypeV1) +INGRESS_KIND_REGISTRY_V1 = frozenset(item.value for item in IngressKindV1) +TRANSFORM_REGISTRY_V1 = MappingProxyType( + { + TransformKindV1.INGRESS_CAPTURE: ("mcp-warden.capture", "1"), + TransformKindV1.DETERMINISTIC: ("mcp-warden.transform.deterministic", "1"), + } +) +TRANSFORM_IDENTITIES_V1 = frozenset(TRANSFORM_REGISTRY_V1.values()) +TRANSFORM_KIND_BY_IDENTITY_V1 = MappingProxyType( + {identity: kind for kind, identity in TRANSFORM_REGISTRY_V1.items()} +) +if len(TRANSFORM_KIND_BY_IDENTITY_V1) != len(TRANSFORM_REGISTRY_V1): + raise RuntimeError("duplicate V1 transform identity") +TRANSFORM_ID_REGISTRY_V1 = frozenset(item[0] for item in TRANSFORM_IDENTITIES_V1) +TRANSFORM_VERSION_REGISTRY_V1 = frozenset(item[1] for item in TRANSFORM_IDENTITIES_V1) + + +class ContentEnvelopeError(Exception): + """Stable code-only envelope failure with no input-bearing message.""" + + def __init__(self, code: str) -> None: + self.code = code + super().__init__(code) + + def __str__(self) -> str: + return self.code + + def __repr__(self) -> str: + return self.code + + +class _FrozenModel(BaseModel): + _validation_code: ClassVar[str] = "ENV-MALFORMED" + model_config = ConfigDict( + extra="forbid", + frozen=True, + strict=True, + hide_input_in_errors=True, + revalidate_instances="always", + ) + + def __init__(self, **data: Any) -> None: + invalid = False + try: + super().__init__(**data) + except ValidationError: + invalid = True + if invalid: + raise ContentEnvelopeError(self._validation_code) from None + + def __setattr__(self, name: str, value: Any) -> None: + invalid = False + try: + super().__setattr__(name, value) + except ValidationError: + invalid = True + if invalid: + raise ContentEnvelopeError(self._validation_code) from None + + def __delattr__(self, name: str) -> None: + invalid = False + try: + super().__delattr__(name) + except ValidationError: + invalid = True + if invalid: + raise ContentEnvelopeError(self._validation_code) from None + + @classmethod + def model_validate(cls, obj: Any, **kwargs: Any) -> Any: + invalid = False + try: + result = super().model_validate(obj, **kwargs) + except ValidationError: + invalid = True + result = None + if invalid: + raise ContentEnvelopeError(cls._validation_code) from None + return result + + @classmethod + def model_validate_json(cls, json_data: str | bytes | bytearray, **kwargs: Any) -> Any: + invalid = False + try: + result = super().model_validate_json(json_data, **kwargs) + except ValidationError: + invalid = True + result = None + if invalid: + raise ContentEnvelopeError(cls._validation_code) from None + return result + + @classmethod + def model_validate_strings(cls, obj: Any, **kwargs: Any) -> Any: + invalid = False + try: + result = super().model_validate_strings(obj, **kwargs) + except ValidationError: + invalid = True + result = None + if invalid: + raise ContentEnvelopeError(cls._validation_code) from None + return result + + +def _digest(value: str) -> str: + if DIGEST_RE.fullmatch(value) is None: + raise ValueError("invalid digest") + return value + + +class ContentEvidenceV1(_FrozenModel): + digest: str + length: StrictInt + media_type: str + + _valid_digest = field_validator("digest")(_digest) + + @field_validator("length") + @classmethod + def _valid_length(cls, value: int) -> int: + if value < 0 or value > MAX_CONTENT_BYTES: + raise ValueError("invalid length") + return value + + @field_validator("media_type") + @classmethod + def _valid_media_type(cls, value: str) -> str: + if value not in MEDIA_TYPE_REGISTRY_V1: + raise ValueError("invalid media type") + return value + + +class SourceEvidenceV1(_FrozenModel): + kind: str + identity_digest: str + claims_digest: str + + _valid_identity = field_validator("identity_digest")(_digest) + _valid_claims = field_validator("claims_digest")(_digest) + + @field_validator("kind") + @classmethod + def _valid_kind(cls, value: str) -> str: + if value not in INGRESS_KIND_REGISTRY_V1 and value != "derived": + raise ValueError("invalid source kind") + return value + + +class ParentRefV1(_FrozenModel): + _validation_code: ClassVar[str] = "ENV-LINEAGE-INVALID" + envelope_digest: str + content_digest: str + + _valid_envelope = field_validator("envelope_digest")(_digest) + _valid_content = field_validator("content_digest")(_digest) + + +class TransformEvidenceV1(_FrozenModel): + id: str + version: str + implementation_digest: str + parameters_digest: str + + _valid_implementation = field_validator("implementation_digest")(_digest) + _valid_parameters = field_validator("parameters_digest")(_digest) + + @field_validator("id") + @classmethod + def _valid_id(cls, value: str) -> str: + if value not in TRANSFORM_ID_REGISTRY_V1 or TRANSFORM_ID_RE.fullmatch(value) is None: + raise ValueError("invalid transform id") + return value + + @field_validator("version") + @classmethod + def _valid_version(cls, value: str) -> str: + if ( + value not in TRANSFORM_VERSION_REGISTRY_V1 + or TRANSFORM_VERSION_RE.fullmatch(value) is None + ): + raise ValueError("invalid transform version") + return value + + @model_validator(mode="after") + def _valid_identity_pair(self) -> TransformEvidenceV1: + if (self.id, self.version) not in TRANSFORM_IDENTITIES_V1: + raise ValueError("invalid transform identity") + return self + + +class BundleEvidenceV1(_FrozenModel): + _validation_code: ClassVar[str] = "ENV-BUNDLE-INVALID" + artifact_digest: str + signature_evidence_digest: str | None + version_claims_digest: str + publisher_claims_digest: str + dependency_digests: tuple[str, ...] + policy_binding_claims_digest: str + + _valid_artifact = field_validator("artifact_digest")(_digest) + _valid_version = field_validator("version_claims_digest")(_digest) + _valid_publisher = field_validator("publisher_claims_digest")(_digest) + _valid_policy = field_validator("policy_binding_claims_digest")(_digest) + + @field_validator("signature_evidence_digest") + @classmethod + def _valid_signature(cls, value: str | None) -> str | None: + return None if value is None else _digest(value) + + @field_validator("dependency_digests") + @classmethod + def _valid_dependencies(cls, value: tuple[str, ...]) -> tuple[str, ...]: + if len(value) > MAX_DEPENDENCIES or value != tuple(sorted(set(value))): + raise ValueError("invalid dependency digests") + for digest in value: + _digest(digest) + return value + + +class ContentEnvelopeV1(_FrozenModel): + schema_version: Literal[1] + trust_state: Literal["untrusted"] + content: ContentEvidenceV1 + source: SourceEvidenceV1 + parents: tuple[ParentRefV1, ...] + transform: TransformEvidenceV1 + taints: tuple[str, ...] + bundle: BundleEvidenceV1 | None + envelope_digest: str + + _valid_envelope = field_validator("envelope_digest")(_digest) + + @field_validator("parents") + @classmethod + def _valid_parents(cls, value: tuple[ParentRefV1, ...]) -> tuple[ParentRefV1, ...]: + if len(value) > MAX_PARENTS: + raise ValueError("too many parents") + if value != tuple(sorted(value, key=lambda item: (item.envelope_digest, item.content_digest))): + raise ValueError("parents not sorted") + if len({item.envelope_digest for item in value}) != len(value): + raise ValueError("duplicate parents") + return value + + @field_validator("taints") + @classmethod + def _valid_taints(cls, value: tuple[str, ...]) -> tuple[str, ...]: + if not value or len(value) > MAX_TAINTS or value != tuple(sorted(set(value))): + raise ValueError("invalid taints") + if "core:untrusted" not in value or any(item not in TAINT_REGISTRY_V1 for item in value): + raise ValueError("invalid taints") + return value + + @model_validator(mode="after") + def _valid_shape(self) -> ContentEnvelopeV1: + if self.source.kind == "derived": + if not self.parents: + raise ValueError("derived envelope requires parents") + if ( + TRANSFORM_KIND_BY_IDENTITY_V1.get((self.transform.id, self.transform.version)) + is not TransformKindV1.DETERMINISTIC + ): + raise ContentEnvelopeError("ENV-LINEAGE-INVALID") from None + elif self.parents: + raise ValueError("ingress envelope cannot have parents") + elif ( + TRANSFORM_KIND_BY_IDENTITY_V1.get((self.transform.id, self.transform.version)) + is not TransformKindV1.INGRESS_CAPTURE + ): + raise ContentEnvelopeError("ENV-MALFORMED") from None + if self.source.kind == IngressKindV1.BUNDLE_METADATA.value and self.bundle is None: + raise ValueError("bundle metadata source requires bundle evidence") + return self + + +@dataclass(frozen=True, slots=True) +class BundleEvidenceInput: + """Raw evidence accepted by constructors and discarded after hashing.""" + + artifact: bytes + signature_evidence: bytes | None + version_claims: bytes + publisher_claims: bytes + dependencies: tuple[bytes, ...] + policy_binding_claims: bytes + + def __post_init__(self) -> None: + components = ( + self.artifact, + self.version_claims, + self.publisher_claims, + self.policy_binding_claims, + ) + if any(type(value) is not bytes for value in components): + raise ContentEnvelopeError("ENV-BUNDLE-INVALID") from None + if self.signature_evidence is not None and type(self.signature_evidence) is not bytes: + raise ContentEnvelopeError("ENV-BUNDLE-INVALID") from None + if type(self.dependencies) is not tuple: + raise ContentEnvelopeError("ENV-BUNDLE-INVALID") from None + if len(self.dependencies) > MAX_DEPENDENCIES: + raise ContentEnvelopeError("ENV-OVER-CAP") from None + if any(type(value) is not bytes for value in self.dependencies): + raise ContentEnvelopeError("ENV-BUNDLE-INVALID") from None + bounded = components + self.dependencies + if self.signature_evidence is not None: + bounded += (self.signature_evidence,) + if any(len(value) > MAX_BUNDLE_COMPONENT_BYTES for value in bounded): + raise ContentEnvelopeError("ENV-OVER-CAP") from None diff --git a/src/mcp_warden/hashing.py b/src/mcp_warden/hashing.py index 561cec0..052c6cc 100644 --- a/src/mcp_warden/hashing.py +++ b/src/mcp_warden/hashing.py @@ -19,12 +19,22 @@ import rfc8785 +from mcp_warden.content_models import DigestDomain + logger = logging.getLogger("mcp_warden.hashing") #: Public prefix for every digest emitted by mcp-warden. SHA256_PREFIX = "sha256:" +def hash_bytes(payload: bytes, *, domain: DigestDomain) -> str: + """Hash exact bytes under a closed content-envelope digest domain.""" + if type(payload) is not bytes or type(domain) is not DigestDomain: + raise TypeError("payload and domain must be exact typed values") + digest = hashlib.sha256(domain.value.encode("ascii") + b"\x00" + payload).hexdigest() + return SHA256_PREFIX + digest + + def canon(value: Any) -> bytes: """Return the RFC 8785 (JCS) canonical byte serialization of ``value``. diff --git a/tests/fuzz/test_fuzz_content_envelope.py b/tests/fuzz/test_fuzz_content_envelope.py new file mode 100644 index 0000000..bea5a58 --- /dev/null +++ b/tests/fuzz/test_fuzz_content_envelope.py @@ -0,0 +1,147 @@ +"""Bounded properties for the V1 deterministic content-envelope boundary.""" + +from __future__ import annotations + +from itertools import permutations + +import pytest +import rfc8785 +from hypothesis import given, settings +from hypothesis import strategies as st + +import mcp_warden.content_envelope as codec +from mcp_warden.content_envelope import ( + create_ingress, + derive_envelope, + parse_envelope, + serialize_envelope, + to_public_bytes, +) +from mcp_warden.content_models import ( + ContentEnvelopeError, + IngressKindV1, + MediaTypeV1, + ParentRefV1, + TaintV1, + TransformKindV1, +) + + +def _meta(value: object) -> bytes: + return rfc8785.dumps(value) + + +def _root(content: bytes, taints: tuple[TaintV1, ...] = ()): + return create_ingress( + content=content, + media_type=MediaTypeV1.APPLICATION_OCTET_STREAM, + source_kind=IngressKindV1.AGENT_MESSAGE, + source_identity=_meta({"adapter": "property"}), + source_claims=_meta({"truth": "unverified"}), + capture_kind=TransformKindV1.INGRESS_CAPTURE, + capture_implementation=_meta({"implementation": "property"}), + capture_parameters=_meta({"mode": "strict"}), + added_taints=taints, + ) + + +def _derived(parents): + return derive_envelope( + content=b"derived", + media_type=MediaTypeV1.APPLICATION_OCTET_STREAM, + parents=parents, + transform_kind=TransformKindV1.DETERMINISTIC, + transform_implementation=_meta({"implementation": "property"}), + transform_parameters=_meta({"mode": "strict"}), + ) + + +@settings(max_examples=75) +@given(st.binary(max_size=512)) +def test_repeated_bytes_are_identical_and_round_trip(content: bytes) -> None: + first = _root(content) + second = _root(content) + assert first == second + encoded = serialize_envelope(first) + assert serialize_envelope(parse_envelope(encoded)) == encoded + + +@settings(max_examples=35) +@given(st.lists(st.binary(min_size=1, max_size=16), min_size=1, max_size=5, unique=True)) +def test_parent_permutations_have_one_identity(contents: list[bytes]) -> None: + parents = tuple(_root(content) for content in contents) + sampled = list(permutations(parents))[:24] + assert len({_derived(order).envelope_digest for order in sampled}) == 1 + + +@settings(max_examples=50) +@given(st.sets(st.sampled_from(tuple(TaintV1)), max_size=len(TaintV1))) +def test_taint_propagation_is_monotonic(taints: set[TaintV1]) -> None: + parent = _root(b"parent", tuple(taints)) + child = _derived((parent,)) + assert set(parent.taints).issubset(child.taints) + assert child.taints == tuple(sorted(set(child.taints))) + + +@settings(max_examples=75) +@given(st.binary(min_size=1, max_size=256), st.integers(min_value=0, max_value=100_000)) +def test_single_byte_wire_mutation_never_returns_a_different_envelope( + content: bytes, selector: int +) -> None: + original = _root(content) + wire = bytearray(serialize_envelope(original)) + position = selector % len(wire) + wire[position] ^= 1 + try: + parsed = parse_envelope(bytes(wire)) + except ContentEnvelopeError: + return + assert parsed == original + + +@settings(max_examples=100, deadline=None) +@given(st.binary(max_size=2_048)) +def test_arbitrary_malformed_bytes_terminate_with_stable_failure(data: bytes) -> None: + try: + envelope = parse_envelope(data) + except ContentEnvelopeError as error: + assert error.code.startswith("ENV-") + assert error.__cause__ is None + assert error.__context__ is None + else: + assert serialize_envelope(envelope) == data + + +def test_flat_crafted_cycle_ref_does_not_recurse() -> None: + root = _root(b"root") + template = _derived((root,)) + refs = ( + ParentRefV1( + envelope_digest="sha256:" + "f" * 64, + content_digest=root.content.digest, + ), + ) + crafted = template.model_copy( + update={ + "source": codec._derived_source(refs, template.transform), + "parents": refs, + } + ) + crafted = codec._replace_self_digest(crafted) + child = _derived((crafted,)) + assert child.parents[0].envelope_digest == crafted.envelope_digest + + +def test_parser_registry_drift_rejects_instead_of_dropping(monkeypatch) -> None: + encoded = serialize_envelope(_root(b"root")) + monkeypatch.setattr(codec, "_TAINT_REGISTRY", frozenset()) + with pytest.raises(ContentEnvelopeError, match="ENV-TAINT-UNKNOWN"): + parse_envelope(encoded) + + +@settings(max_examples=50) +@given(st.binary(min_size=1, max_size=128)) +def test_raw_content_never_appears_in_public_bytes(suffix: bytes) -> None: + marker = b"PLANTED-CONTENT-SECRET-715" + envelope = _root(marker + suffix) + assert marker not in to_public_bytes(envelope) diff --git a/tests/test_content_envelope.py b/tests/test_content_envelope.py new file mode 100644 index 0000000..f1ed67d --- /dev/null +++ b/tests/test_content_envelope.py @@ -0,0 +1,1296 @@ +"""Security contract tests for deterministic untrusted content envelopes.""" + +from __future__ import annotations + +import ast +import hashlib +import json +import traceback +from collections.abc import Iterator, Sequence +from itertools import permutations +from pathlib import Path + +import pytest +import rfc8785 + +import mcp_warden.content_envelope as codec +from mcp_warden.content_envelope import ( + create_ingress, + derive_envelope, + parse_envelope, + serialize_envelope, + to_public_bytes, + to_public_dict, + verify_envelope, + verify_lineage, +) +from mcp_warden.content_models import ( + MAX_BUNDLE_COMPONENT_BYTES, + MAX_CONTENT_BYTES, + MAX_DEPENDENCIES, + MAX_ENVELOPE_BYTES, + MAX_JSON_DEPTH, + MAX_JSON_NODES, + MAX_JSON_OBJECT_KEYS, + MAX_JSON_STRING_BYTES, + MAX_PARENTS, + MAX_SOURCE_CLAIMS_BYTES, + MAX_SOURCE_IDENTITY_BYTES, + MAX_TAINTS, + MAX_TRANSFORM_IDENTITY_BYTES, + MAX_TRANSFORM_PARAMETERS_BYTES, + TRANSFORM_KIND_BY_IDENTITY_V1, + TRANSFORM_REGISTRY_V1, + BundleEvidenceInput, + BundleEvidenceV1, + ContentEnvelopeError, + ContentEnvelopeV1, + ContentEvidenceV1, + DigestDomain, + IngressKindV1, + MediaTypeV1, + ParentRefV1, + SourceEvidenceV1, + TaintV1, + TransformEvidenceV1, + TransformKindV1, +) +from mcp_warden.hashing import hash_bytes + +DIGEST_A = "sha256:" + "a" * 64 +DIGEST_B = "sha256:" + "b" * 64 + + +def _envelope(*, bundle: BundleEvidenceV1 | None = None) -> ContentEnvelopeV1: + return ContentEnvelopeV1( + schema_version=1, + trust_state="untrusted", + content=ContentEvidenceV1(digest=DIGEST_A, length=1, media_type="text/plain"), + source=SourceEvidenceV1( + kind="tool_result", identity_digest=DIGEST_A, claims_digest=DIGEST_B + ), + parents=(), + transform=TransformEvidenceV1( + id="mcp-warden.capture", + version="1", + implementation_digest=DIGEST_A, + parameters_digest=DIGEST_B, + ), + taints=("core:untrusted",), + bundle=bundle, + envelope_digest=DIGEST_B, + ) + + +def test_models_are_strict_and_bundle_is_explicit() -> None: + envelope = _envelope() + assert envelope.schema_version == 1 + assert envelope.bundle is None + + body = { + "schema_version": 1, + "trust_state": "untrusted", + "content": envelope.content, + "source": envelope.source, + "parents": (), + "transform": envelope.transform, + "taints": ("core:untrusted",), + "envelope_digest": DIGEST_B, + } + with pytest.raises(ContentEnvelopeError): + ContentEnvelopeV1(**body) + with pytest.raises(ContentEnvelopeError): + ContentEnvelopeV1(**(body | {"bundle": None, "surprise": True})) + with pytest.raises(ContentEnvelopeError): + ContentEnvelopeV1(**(body | {"bundle": None, "schema_version": "1"})) + with pytest.raises(ContentEnvelopeError): + ContentEnvelopeV1(**(body | {"bundle": None, "trust_state": "trusted"})) + + +def test_nested_models_and_collections_are_frozen() -> None: + bundle = BundleEvidenceV1( + artifact_digest=DIGEST_A, + signature_evidence_digest=None, + version_claims_digest=DIGEST_A, + publisher_claims_digest=DIGEST_B, + dependency_digests=(DIGEST_A, DIGEST_B), + policy_binding_claims_digest=DIGEST_B, + ) + envelope = _envelope(bundle=bundle) + + for obj, field, value in ( + (envelope, "trust_state", "trusted"), + (envelope.content, "length", 2), + (envelope.source, "kind", "web"), + (envelope.transform, "version", "2"), + (envelope.bundle, "artifact_digest", DIGEST_B), + ): + with pytest.raises(ContentEnvelopeError): + setattr(obj, field, value) + with pytest.raises(AttributeError): + envelope.parents.append(ParentRefV1(envelope_digest=DIGEST_A, content_digest=DIGEST_B)) + with pytest.raises(AttributeError): + envelope.taints.append("core:secret") + with pytest.raises(AttributeError): + envelope.bundle.dependency_digests.append(DIGEST_A) + + +def test_digest_domain_enum_is_complete_and_immutable() -> None: + expected = { + "content", + "source-identity", + "source-claims", + "derived-source-identity", + "derived-source-claims", + "transform-implementation", + "transform-parameters", + "bundle-artifact", + "bundle-signature", + "bundle-version", + "bundle-publisher", + "bundle-dependency", + "bundle-policy-binding", + "envelope", + } + assert {domain.value.rsplit("/", 1)[-1] for domain in DigestDomain} == expected + with pytest.raises((AttributeError, TypeError)): + DigestDomain.CONTENT.value = "changed" + + +def test_transform_registry_reverse_mapping_is_exact() -> None: + assert len(TRANSFORM_KIND_BY_IDENTITY_V1) == len(TRANSFORM_REGISTRY_V1) + assert { + identity: kind for kind, identity in TRANSFORM_REGISTRY_V1.items() + } == TRANSFORM_KIND_BY_IDENTITY_V1 + for kind, identity in TRANSFORM_REGISTRY_V1.items(): + assert TRANSFORM_KIND_BY_IDENTITY_V1[identity] is kind + + +def test_hash_bytes_exact_golden_and_pairwise_domain_separation() -> None: + payload = b"\x00raw\xffbytes" + for domain in DigestDomain: + expected = "sha256:" + hashlib.sha256( + domain.value.encode("ascii") + b"\x00" + payload + ).hexdigest() + assert hash_bytes(payload, domain=domain) == expected + assert hash_bytes(payload, domain=domain) == expected + assert len(expected) == 71 + + values = {hash_bytes(payload, domain=domain) for domain in DigestDomain} + assert len(values) == len(DigestDomain) + with pytest.raises(TypeError): + hash_bytes(payload, domain=DigestDomain.CONTENT.value) + with pytest.raises(TypeError): + hash_bytes("not-bytes", domain=DigestDomain.CONTENT) + + +def test_digest_fields_require_lowercase_sha256_syntax() -> None: + with pytest.raises(ContentEnvelopeError): + ContentEvidenceV1(digest="sha256:" + "A" * 64, length=1, media_type="text/plain") + with pytest.raises(ContentEnvelopeError): + ParentRefV1(envelope_digest="bad", content_digest=DIGEST_A) + + +def test_new_subsystem_does_not_import_or_call_legacy_hash_helpers() -> None: + root = Path(__file__).parents[1] / "src" / "mcp_warden" + for name in ("content_models.py", "content_envelope.py"): + path = root / name + tree = ast.parse(path.read_text(encoding="utf-8")) + forbidden: list[str] = [] + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom): + forbidden.extend(alias.name for alias in node.names if alias.name in {"canon", "hash_value"}) + elif isinstance(node, ast.Call) and isinstance(node.func, ast.Name): + if node.func.id in {"canon", "hash_value"}: + forbidden.append(node.func.id) + assert forbidden == [] + + +def _metadata(value: object) -> bytes: + return rfc8785.dumps(value) + + +def _ingress(**changes: object) -> ContentEnvelopeV1: + values: dict[str, object] = { + "content": b"payload", + "media_type": MediaTypeV1.TEXT_PLAIN, + "source_kind": IngressKindV1.TOOL_RESULT, + "source_identity": _metadata({"adapter": "test"}), + "source_claims": _metadata({"claim": "untrusted"}), + "capture_kind": TransformKindV1.INGRESS_CAPTURE, + "capture_implementation": _metadata({"implementation": "test"}), + "capture_parameters": _metadata({"mode": "strict"}), + } + values.update(changes) + return create_ingress(**values) + + +@pytest.mark.parametrize("kind", tuple(IngressKindV1)) +def test_create_ingress_covers_all_source_kinds(kind: IngressKindV1) -> None: + bundle = _bundle_input() if kind is IngressKindV1.BUNDLE_METADATA else None + envelope = _ingress(source_kind=kind, bundle=bundle) + assert envelope.source.kind == kind.value + assert envelope.parents == () + assert envelope.trust_state == "untrusted" + assert "core:untrusted" in envelope.taints + + +def _bundle_input(*, signature: bytes | None = b"\xffopaque-signature") -> BundleEvidenceInput: + return BundleEvidenceInput( + artifact=_metadata({"artifact": "sha256:claim"}), + signature_evidence=signature, + version_claims=_metadata({"version": "1"}), + publisher_claims=_metadata({"publisher": "unverified"}), + dependencies=(_metadata({"dependency": "b"}), _metadata({"dependency": "a"})), + policy_binding_claims=_metadata({"policy": "claim-only"}), + ) + + +def test_ingress_hashes_exact_opaque_content_and_signature() -> None: + raw = b"\xff\x00not-utf8" + signature = b"\xfe\x00opaque" + envelope = _ingress(content=raw, bundle=_bundle_input(signature=signature)) + assert envelope.content.digest == hash_bytes(raw, domain=DigestDomain.CONTENT) + assert envelope.bundle.signature_evidence_digest == hash_bytes( + signature, domain=DigestDomain.BUNDLE_SIGNATURE + ) + assert raw not in repr(envelope).encode() + assert signature not in repr(envelope).encode() + + +def test_ingress_requires_exact_canonical_metadata_and_never_normalizes() -> None: + composed = _metadata({"name": "é"}) + decomposed = _metadata({"name": "e\u0301"}) + assert _ingress(source_identity=composed).source.identity_digest != _ingress( + source_identity=decomposed + ).source.identity_digest + + for invalid in (b'{"b":1, "a":2}', b'{"a":1,"a":2}', b"{\xff}", b'{"x":NaN}'): + with pytest.raises(ContentEnvelopeError) as caught: + _ingress(source_identity=invalid) + assert caught.value.code in {"ENV-MALFORMED", "ENV-NONCANONICAL"} + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + + +def test_bundle_metadata_source_requires_bundle_and_taints_fail_closed(monkeypatch) -> None: + with pytest.raises(ContentEnvelopeError, match="ENV-BUNDLE-INVALID"): + _ingress(source_kind=IngressKindV1.BUNDLE_METADATA) + with pytest.raises(ContentEnvelopeError, match="ENV-TAINT-UNKNOWN"): + _ingress(added_taints=("future:unknown",)) + + monkeypatch.setattr(codec, "_TAINT_REGISTRY", frozenset({"core:secret"})) + with pytest.raises(ContentEnvelopeError, match="ENV-TAINT-UNKNOWN"): + _ingress(added_taints=(TaintV1.SECRET,)) + + +def test_ingress_cap_boundaries_are_exact() -> None: + at_identity_cap = b'"' + b"x" * (MAX_SOURCE_IDENTITY_BYTES - 2) + b'"' + assert len(at_identity_cap) == MAX_SOURCE_IDENTITY_BYTES + _ingress(source_identity=at_identity_cap) + with pytest.raises(ContentEnvelopeError, match="ENV-OVER-CAP"): + _ingress(source_identity=at_identity_cap + b" ") + + at_content_cap = b"x" * MAX_CONTENT_BYTES + assert _ingress(content=at_content_cap).content.length == MAX_CONTENT_BYTES + with pytest.raises(ContentEnvelopeError, match="ENV-OVER-CAP"): + _ingress(content=at_content_cap + b"x") + + +def test_depth_prescan_rejects_before_json_loads(monkeypatch) -> None: + called = False + + def forbidden_loads(*args: object, **kwargs: object) -> object: + nonlocal called + called = True + raise AssertionError("json.loads must not be reached") + + monkeypatch.setattr(json, "loads", forbidden_loads) + deep = b"[" * 10_000 + b"]" * 10_000 + with pytest.raises(ContentEnvelopeError, match="ENV-OVER-CAP"): + codec._canonical_metadata( + deep, cap=len(deep), domain=DigestDomain.SOURCE_CLAIMS + ) + assert called is False + + +@pytest.mark.parametrize( + "value", + ( + {"quoted": 'a\\"[still-string]'}, + {"slash": "\\\\"}, + {"unicode": "[é]"}, + {"escaped": "\\u005b"}, + ), +) +def test_depth_scanner_handles_string_escapes(value: object) -> None: + _ingress(source_claims=_metadata(value)) + + +def _derive(parents: tuple[ContentEnvelopeV1, ...], **changes: object) -> ContentEnvelopeV1: + values: dict[str, object] = { + "content": b"derived-content", + "media_type": MediaTypeV1.TEXT_PLAIN, + "parents": parents, + "transform_kind": TransformKindV1.DETERMINISTIC, + "transform_implementation": _metadata({"implementation": "transform"}), + "transform_parameters": _metadata({"mode": "deterministic"}), + } + values.update(changes) + return derive_envelope(**values) + + +def test_derivation_is_parent_permutation_invariant_and_refs_sort_ascii() -> None: + parents = ( + _ingress(content=b"a", added_taints=(TaintV1.SECRET,)), + _ingress(content=b"b", added_taints=(TaintV1.EXECUTABLE,)), + _ingress(content=b"c"), + ) + outputs = [_derive(tuple(order)) for order in permutations(parents)] + assert {item.envelope_digest for item in outputs} == {outputs[0].envelope_digest} + assert outputs[0].parents == tuple( + sorted(outputs[0].parents, key=lambda ref: (ref.envelope_digest, ref.content_digest)) + ) + assert outputs[0].source.kind == "derived" + assert set(outputs[0].taints) >= { + "core:untrusted", + "core:secret", + "core:executable", + } + for order in permutations(parents): + verify_lineage(outputs[0], parents=order) + + +def test_derivation_recomputes_parent_digest_before_acceptance() -> None: + parent = _ingress() + tampered_content = parent.content.model_copy(update={"digest": DIGEST_B}) + tampered = parent.model_copy(update={"content": tampered_content}) + with pytest.raises(ContentEnvelopeError, match="ENV-DIGEST-MISMATCH"): + _derive((tampered,)) + + +def test_lineage_rejects_missing_duplicate_extra_and_taint_loss() -> None: + first = _ingress(content=b"a", added_taints=(TaintV1.SECRET,)) + second = _ingress(content=b"b") + child = _derive((first, second)) + for supplied in ((first,), (first, first), (first, second, _ingress(content=b"c"))): + with pytest.raises(ContentEnvelopeError, match="ENV-LINEAGE-INVALID"): + verify_lineage(child, parents=supplied) + + lost = child.model_copy(update={"taints": ("core:untrusted",)}) + # Keep the self digest coherent so the failure specifically proves monotonic taint. + lost = codec._replace_self_digest(lost) + with pytest.raises(ContentEnvelopeError, match="ENV-TAINT-MISSING"): + verify_lineage(lost, parents=(first, second)) + + +def test_derive_rejects_duplicate_empty_and_complete_union_over_cap() -> None: + parent = _ingress() + with pytest.raises(ContentEnvelopeError, match="ENV-LINEAGE-INVALID"): + _derive(()) + with pytest.raises(ContentEnvelopeError, match="ENV-LINEAGE-INVALID"): + _derive((parent, parent)) + + taints = tuple(item for item in TaintV1 if item is not TaintV1.UNTRUSTED) + heavily_tainted = _ingress(added_taints=taints) + assert set(_derive((heavily_tainted,)).taints) == set(item.value for item in TaintV1) + + +def test_no_schema_or_api_surface_removes_taint_or_grants_authority() -> None: + fields = set(ContentEnvelopeV1.model_fields) + forbidden = {"authority", "trusted", "remove_taints", "sanitized", "capabilities"} + assert fields.isdisjoint(forbidden) + assert not hasattr(codec, "remove_taint") + assert not hasattr(codec, "grant_authority") + + +def test_codec_golden_round_trip_and_explicit_bundle_null() -> None: + envelope = _ingress() + encoded = serialize_envelope(envelope) + assert encoded == rfc8785.dumps(to_public_dict(envelope)) + assert b'"bundle":null' in encoded + assert parse_envelope(encoded) == envelope + assert serialize_envelope(parse_envelope(encoded)) == encoded + assert to_public_bytes(envelope) == encoded + verify_envelope(envelope, content=b"payload") + + +def test_parser_rejects_missing_bundle_unknown_fields_and_versions() -> None: + public = to_public_dict(_ingress()) + missing_bundle = dict(public) + missing_bundle.pop("bundle") + with pytest.raises(ContentEnvelopeError, match="ENV-BUNDLE-INVALID"): + parse_envelope(rfc8785.dumps(missing_bundle)) + + for path, code in ( + ({**public, "authority": "admin"}, "ENV-UNKNOWN-FIELD"), + ({**public, "schema_version": 2}, "ENV-SCHEMA-UNKNOWN"), + ({**public, "trust_state": "trusted"}, "ENV-TRUST-INVALID"), + ): + with pytest.raises(ContentEnvelopeError, match=code): + parse_envelope(rfc8785.dumps(path)) + + nested = dict(public) + nested["content"] = {**public["content"], "secret_raw": "no"} + with pytest.raises(ContentEnvelopeError, match="ENV-UNKNOWN-FIELD"): + parse_envelope(rfc8785.dumps(nested)) + + +@pytest.mark.parametrize( + "malformed,code", + ( + (b'{"a":1,"a":2}', "ENV-NONCANONICAL"), + (b'{"a":1, "b":2}', "ENV-NONCANONICAL"), + (b'{"a":NaN}', "ENV-MALFORMED"), + (b"]", "ENV-MALFORMED"), + (b'{"x":"unterminated}', "ENV-MALFORMED"), + ), +) +def test_parser_rejects_malformed_and_noncanonical_bytes(malformed: bytes, code: str) -> None: + with pytest.raises(ContentEnvelopeError, match=code): + parse_envelope(malformed) + + +def test_parser_atomically_recomputes_self_digest_and_ordering() -> None: + child = _derive((_ingress(content=b"a"), _ingress(content=b"b"))) + public = to_public_dict(child) + public["envelope_digest"] = DIGEST_A + with pytest.raises(ContentEnvelopeError, match="ENV-DIGEST-MISMATCH"): + parse_envelope(rfc8785.dumps(public)) + + public = to_public_dict(child) + public["parents"] = list(reversed(public["parents"])) + with pytest.raises(ContentEnvelopeError, match="ENV-LINEAGE-INVALID"): + parse_envelope(rfc8785.dumps(public)) + + +def test_verify_rejects_wrong_content_and_tampered_self() -> None: + envelope = _ingress(content=b"correct") + for content in (b"wrong", b"correct\x00"): + with pytest.raises(ContentEnvelopeError, match="ENV-DIGEST-MISMATCH"): + verify_envelope(envelope, content=content) + tampered = envelope.model_copy(update={"envelope_digest": DIGEST_A}) + with pytest.raises(ContentEnvelopeError, match="ENV-DIGEST-MISMATCH"): + verify_envelope(tampered, content=b"correct") + + +def test_planted_secrets_never_escape_output_error_traceback_or_logs(caplog) -> None: + secrets = { + "content": b"content-SECRET-715", + "identity": b'{"secret":"identity-SECRET-715"}', + "claims": b'{"secret":"claims-SECRET-715"}', + "implementation": b'{"secret":"implementation-SECRET-715"}', + "parameters": b'{"secret":"parameters-SECRET-715"}', + "signature": b"signature-SECRET-715", + "policy": b'{"secret":"policy-SECRET-715"}', + } + bundle = _bundle_input(signature=secrets["signature"]) + bundle = BundleEvidenceInput( + artifact=bundle.artifact, + signature_evidence=bundle.signature_evidence, + version_claims=bundle.version_claims, + publisher_claims=bundle.publisher_claims, + dependencies=bundle.dependencies, + policy_binding_claims=secrets["policy"], + ) + envelope = _ingress( + content=secrets["content"], + source_identity=secrets["identity"], + source_claims=secrets["claims"], + capture_implementation=secrets["implementation"], + capture_parameters=secrets["parameters"], + bundle=bundle, + ) + outputs = [serialize_envelope(envelope), to_public_bytes(envelope), repr(to_public_dict(envelope)).encode()] + for secret in secrets.values(): + assert all(secret not in output for output in outputs) + + invalid = b'{"secret":"trace-SECRET-715",}' + try: + parse_envelope(invalid) + except ContentEnvelopeError as error: + error_views = ( + str(error).encode(), + repr(error).encode(), + "".join(traceback.format_exception(error)).encode(), + ) + assert error.__cause__ is None + assert error.__context__ is None + assert all(b"trace-SECRET-715" not in view for view in error_views) + else: # pragma: no cover + pytest.fail("invalid bytes accepted") + assert "SECRET-715" not in caplog.text + + +def test_public_projection_is_explicit_not_whole_model_dump() -> None: + source = (Path(__file__).parents[1] / "src/mcp_warden/content_envelope.py").read_text() + public_source = source[source.index("def to_public_dict") :] + assert "model_dump" not in public_source + assert "__dict__" not in public_source + + +class _OverlongTaints(Iterator[TaintV1]): + def __init__(self) -> None: + self.calls = 0 + + def __next__(self) -> TaintV1: + self.calls += 1 + if self.calls > MAX_TAINTS + 1: + raise AssertionError("taint iterator was consumed past the security cap") + return TaintV1.UNTRUSTED + + +def test_taint_iterables_are_bounded_and_iterator_failures_are_stable() -> None: + overlong = _OverlongTaints() + with pytest.raises(ContentEnvelopeError, match="ENV-OVER-CAP"): + _ingress(added_taints=overlong) + assert overlong.calls == MAX_TAINTS + 1 + + def broken() -> Iterator[TaintV1]: + yield TaintV1.SECRET + raise RuntimeError("PLANTED-TAINT-ITERATOR-SECRET") + + with pytest.raises(ContentEnvelopeError, match="ENV-TAINT-UNKNOWN") as caught: + _ingress(added_taints=broken()) + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + assert "PLANTED" not in traceback.format_exception_only(caught.value)[0] + + +class _HostileParents(Sequence[ContentEnvelopeV1]): + def __init__(self, parent: ContentEnvelopeV1) -> None: + self.parent = parent + self.calls = 0 + + def __len__(self) -> int: + return 1 + + def __getitem__(self, index: int) -> ContentEnvelopeV1: + self.calls += 1 + if self.calls > MAX_PARENTS + 1: + raise AssertionError("parent sequence was consumed past the security cap") + return self.parent + + +def test_hostile_parent_sequences_are_consumed_to_a_hard_bound() -> None: + hostile = _HostileParents(_ingress()) + with pytest.raises(ContentEnvelopeError, match="ENV-OVER-CAP"): + _derive(hostile) # type: ignore[arg-type] + assert hostile.calls == MAX_PARENTS + 1 + + +def test_parent_iteration_errors_are_stable() -> None: + class BrokenParents(_HostileParents): + def __getitem__(self, index: int) -> ContentEnvelopeV1: + raise RuntimeError("PLANTED-PARENT-ITERATOR-SECRET") + + with pytest.raises(ContentEnvelopeError, match="ENV-LINEAGE-INVALID") as caught: + _derive(BrokenParents(_ingress())) # type: ignore[arg-type] + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + + +def test_bundle_input_enforces_exact_tuple_bytes_and_stable_api_errors() -> None: + with pytest.raises(ContentEnvelopeError, match="ENV-BUNDLE-INVALID"): + BundleEvidenceInput( + artifact="PLANTED-BUNDLE-SECRET", # type: ignore[arg-type] + signature_evidence=None, + version_claims=b"{}", + publisher_claims=b"{}", + dependencies=(), + policy_binding_claims=b"{}", + ) + with pytest.raises(ContentEnvelopeError, match="ENV-BUNDLE-INVALID"): + BundleEvidenceInput( + artifact=b"{}", + signature_evidence=None, + version_claims=b"{}", + publisher_claims=b"{}", + dependencies=[b"{}"], # type: ignore[arg-type] + policy_binding_claims=b"{}", + ) + + +def test_bundle_dependency_iteration_is_bounded_even_if_model_is_bypassed() -> None: + valid = _bundle_input() + object.__setattr__(valid, "dependencies", _OverlongDependencies()) + with pytest.raises(ContentEnvelopeError, match="ENV-BUNDLE-INVALID"): + _ingress(bundle=valid) + + +class _OverlongDependencies: + def __iter__(self): + for _ in range(258): + yield b"{}" + raise AssertionError("dependency iterable consumed past cap") + + +def test_parse_malformed_parent_types_are_lineage_errors() -> None: + child = _derive((_ingress(),)) + public = to_public_dict(child) + public["parents"][0]["envelope_digest"] = [] + with pytest.raises(ContentEnvelopeError, match="ENV-LINEAGE-INVALID"): + parse_envelope(rfc8785.dumps(public)) + + +def test_derived_source_digests_are_recomputed_not_self_asserted() -> None: + parents = (_ingress(content=b"parent"),) + child = _derive(parents) + forged_source = child.source.model_copy(update={"identity_digest": DIGEST_A}) + forged = codec._replace_self_digest(child.model_copy(update={"source": forged_source})) + forged_public = codec._body_dict( + content=forged.content, + source=forged.source, + parents=forged.parents, + transform=forged.transform, + taints=forged.taints, + bundle=forged.bundle, + ) + forged_public["envelope_digest"] = forged.envelope_digest + forged_wire = rfc8785.dumps(forged_public) + with pytest.raises(ContentEnvelopeError, match="ENV-LINEAGE-INVALID"): + parse_envelope(forged_wire) + with pytest.raises(ContentEnvelopeError, match="ENV-LINEAGE-INVALID"): + verify_lineage(forged, parents=parents) + with pytest.raises(ContentEnvelopeError, match="ENV-LINEAGE-INVALID"): + _derive((forged,)) + + +def test_direct_validation_errors_hide_planted_input() -> None: + marker = "PLANTED-PYDANTIC-INPUT-SECRET-715" + with pytest.raises(ContentEnvelopeError) as caught: + ContentEvidenceV1(digest=marker, length=1, media_type="text/plain") + assert marker not in str(caught.value) + assert marker not in repr(caught.value) + + +def test_models_enforce_root_derived_and_bundle_cross_field_invariants() -> None: + root = _ingress() + ref = ParentRefV1(envelope_digest=root.envelope_digest, content_digest=root.content.digest) + common = dict( + schema_version=1, + trust_state="untrusted", + content=root.content, + transform=root.transform, + taints=root.taints, + envelope_digest=root.envelope_digest, + ) + with pytest.raises(ContentEnvelopeError): + ContentEnvelopeV1(**common, source=root.source, parents=(ref,), bundle=None) + with pytest.raises(ContentEnvelopeError): + ContentEnvelopeV1( + **common, + source=root.source.model_copy(update={"kind": "derived"}), + parents=(), + bundle=None, + ) + with pytest.raises(ContentEnvelopeError): + ContentEnvelopeV1( + **common, + source=root.source.model_copy(update={"kind": "bundle_metadata"}), + parents=(), + bundle=None, + ) + + bypassed = codec._replace_self_digest(root.model_copy(update={"parents": (ref,)})) + with pytest.raises(ContentEnvelopeError, match="ENV-LINEAGE-INVALID"): + serialize_envelope(bypassed) + + +def _legacy_helper_violations(source: str) -> list[str]: + tree = ast.parse(source) + violations: list[str] = [] + aliases: set[str] = set() + modules: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom): + for alias in node.names: + if node.module == "mcp_warden.hashing" and alias.name != "hash_bytes": + violations.append(f"import-from:{alias.name}") + if node.module == "mcp_warden" and alias.name == "hashing": + violations.append("import-from:mcp_warden.hashing") + if alias.name in {"canon", "hash_value"}: + aliases.add(alias.asname or alias.name) + elif isinstance(node, ast.Import): + for alias in node.names: + if alias.name.endswith("hashing"): + modules.add(alias.asname or alias.name.rsplit(".", 1)[-1]) + if alias.name == "mcp_warden.hashing": + rendered = alias.name if alias.asname is None else f"{alias.name} as {alias.asname}" + violations.append(f"import:{rendered}") + for node in ast.walk(tree): + if isinstance(node, ast.Call): + if isinstance(node.func, ast.Name) and node.func.id in aliases | {"canon", "hash_value"}: + violations.append(node.func.id) + elif ( + isinstance(node.func, ast.Attribute) + and isinstance(node.func.value, ast.Name) + and node.func.value.id in modules + and node.func.attr in {"canon", "hash_value"} + ): + violations.append(f"{node.func.value.id}.{node.func.attr}") + return violations + + +def test_ast_guard_detects_direct_alias_and_qualified_legacy_calls() -> None: + synthetic = """ +from mcp_warden.hashing import canon as c, hash_value +import mcp_warden.hashing as hashing_alias +c({}) +hash_value({}) +hashing_alias.canon({}) +""" + assert set(_legacy_helper_violations(synthetic)) == { + "c", + "hash_value", + "hashing_alias.canon", + "import-from:canon", + "import-from:hash_value", + "import:mcp_warden.hashing as hashing_alias", + } + root = Path(__file__).parents[1] / "src" / "mcp_warden" + for name in ("content_models.py", "content_envelope.py"): + assert _legacy_helper_violations((root / name).read_text(encoding="utf-8")) == [] + + +def test_ast_guard_forbids_hashing_module_handles_and_non_hash_bytes_imports() -> None: + synthetic = """ +import mcp_warden.hashing +import mcp_warden.hashing as hashing_alias +from mcp_warden.hashing import hash_bytes, SHA256_PREFIX +""" + assert set(_legacy_helper_violations(synthetic)) == { + "import:mcp_warden.hashing", + "import:mcp_warden.hashing as hashing_alias", + "import-from:SHA256_PREFIX", + } + + +def test_parser_maps_nested_validation_failures_to_documented_codes() -> None: + child_public = to_public_dict(_derive((_ingress(),))) + child_public["parents"][0]["envelope_digest"] = "bad" + with pytest.raises(ContentEnvelopeError, match="ENV-LINEAGE-INVALID"): + parse_envelope(rfc8785.dumps(child_public)) + + bundle_public = to_public_dict(_ingress(bundle=_bundle_input())) + bundle_public["bundle"]["artifact_digest"] = "bad" + with pytest.raises(ContentEnvelopeError, match="ENV-BUNDLE-INVALID"): + parse_envelope(rfc8785.dumps(bundle_public)) + + root_public = to_public_dict(_ingress()) + for missing, code in ( + ("schema_version", "ENV-SCHEMA-UNKNOWN"), + ("trust_state", "ENV-TRUST-INVALID"), + ("taints", "ENV-TAINT-MISSING"), + ): + malformed = dict(root_public) + malformed.pop(missing) + with pytest.raises(ContentEnvelopeError, match=code): + parse_envelope(rfc8785.dumps(malformed)) + + +def test_public_api_malformed_inputs_return_only_stable_codes(caplog) -> None: + calls = ( + lambda: serialize_envelope("PLANTED-API-SECRET"), # type: ignore[arg-type] + lambda: verify_envelope(_ingress(), content="PLANTED-API-SECRET"), # type: ignore[arg-type] + lambda: verify_lineage(_ingress(), parents="PLANTED-API-SECRET"), # type: ignore[arg-type] + ) + for call in calls: + with pytest.raises(ContentEnvelopeError) as caught: + call() + assert caught.value.code.startswith("ENV-") + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + assert "PLANTED" not in str(caught.value) + assert "PLANTED" not in repr(caught.value) + assert "PLANTED" not in caplog.text + + +@pytest.mark.parametrize( + "values", + ( + (1,), + ([],), + (TaintV1.SECRET, []), + ("PLANTED-TAINT-VALUE-SECRET-715", 1), + ), +) +def test_non_string_taints_fail_with_code_only_error(values: tuple[object, ...]) -> None: + with pytest.raises(ContentEnvelopeError, match="ENV-TAINT-UNKNOWN") as caught: + _ingress(added_taints=values) + views = ( + str(caught.value), + repr(caught.value), + "".join(traceback.format_exception(caught.value)), + ) + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + assert all("PLANTED" not in view for view in views) + + +class _BytesSubclass(bytes): + pass + + +def test_exact_byte_contract_rejects_bytes_subclasses() -> None: + disguised = _BytesSubclass(b"PLANTED-BYTES-SUBCLASS-SECRET") + with pytest.raises(TypeError): + hash_bytes(disguised, domain=DigestDomain.CONTENT) + for changes in ( + {"content": disguised}, + {"source_identity": disguised}, + {"source_claims": disguised}, + {"capture_implementation": disguised}, + {"capture_parameters": disguised}, + ): + with pytest.raises(ContentEnvelopeError, match="ENV-MALFORMED") as caught: + _ingress(**changes) + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + + +def _self_coherent(candidate: ContentEnvelopeV1) -> ContentEnvelopeV1: + return codec._replace_self_digest(candidate) + + +def _assert_rejected_at_all_accepting_boundaries( + forged: ContentEnvelopeV1, *, content: bytes, code: str +) -> None: + calls = ( + lambda: serialize_envelope(forged), + lambda: verify_envelope(forged, content=content), + lambda: _derive((forged,)), + ) + for call in calls: + with pytest.raises(ContentEnvelopeError, match=code) as caught: + call() + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + + +def test_model_copy_bypasses_are_completely_revalidated() -> None: + root = _ingress() + bundled = _ingress(bundle=_bundle_input()) + cases = ( + ( + root.model_copy(update={"taints": ("core:untrusted", "core:untrusted")}), + "ENV-LINEAGE-INVALID", + ), + ( + root.model_copy(update={"taints": ("core:untrusted", "core:secret")}), + "ENV-LINEAGE-INVALID", + ), + ( + root.model_copy(update={"source": root.source.model_copy(update={"kind": "bad"})}), + "ENV-MALFORMED", + ), + ( + root.model_copy( + update={"content": root.content.model_copy(update={"media_type": "bad/type"})} + ), + "ENV-MALFORMED", + ), + ( + root.model_copy( + update={"content": root.content.model_copy(update={"digest": "bad"})} + ), + "ENV-MALFORMED", + ), + ( + root.model_copy( + update={"content": root.content.model_copy(update={"length": -1})} + ), + "ENV-MALFORMED", + ), + ( + root.model_copy( + update={"transform": root.transform.model_copy(update={"id": "BAD ID"})} + ), + "ENV-MALFORMED", + ), + ( + bundled.model_copy( + update={"bundle": bundled.bundle.model_copy(update={"artifact_digest": "bad"})} + ), + "ENV-BUNDLE-INVALID", + ), + ) + for candidate, code in cases: + forged = _self_coherent(candidate) + _assert_rejected_at_all_accepting_boundaries(forged, content=b"payload", code=code) + + +def test_model_copy_bypassed_parent_refs_and_order_are_revalidated() -> None: + child = _derive((_ingress(content=b"a"), _ingress(content=b"b"))) + bad_ref = child.parents[0].model_copy(update={"envelope_digest": "bad"}) + cases = ( + child.model_copy(update={"parents": (bad_ref, child.parents[1])}), + child.model_copy(update={"parents": tuple(reversed(child.parents))}), + child.model_copy(update={"parents": (child.parents[0], child.parents[0])}), + ) + for candidate in cases: + forged = _self_coherent(candidate) + _assert_rejected_at_all_accepting_boundaries( + forged, + content=b"derived-content", + code="ENV-LINEAGE-INVALID", + ) + + +def test_pydantic_revalidates_bypassed_nested_instances() -> None: + root = _ingress() + invalid_content = root.content.model_copy(update={"media_type": "bad/type"}) + with pytest.raises(ContentEnvelopeError): + ContentEvidenceV1.model_validate(invalid_content) + with pytest.raises(ContentEnvelopeError): + ContentEnvelopeV1.model_validate(root.model_copy(update={"content": invalid_content})) + + +@pytest.mark.parametrize( + "forged", + ( + ContentEnvelopeV1.model_construct(schema_version=1, trust_state="untrusted"), + ContentEnvelopeV1.model_construct( + schema_version=1, + trust_state="untrusted", + content={"secret": "PLANTED-CONSTRUCT-SECRET-715"}, + ), + ), +) +def test_model_construct_missing_or_malformed_fields_fail_code_only( + forged: ContentEnvelopeV1, +) -> None: + calls = ( + lambda: serialize_envelope(forged), + lambda: verify_envelope(forged, content=b"payload"), + lambda: _derive((forged,)), + ) + for call in calls: + with pytest.raises(ContentEnvelopeError, match="ENV-MALFORMED") as caught: + call() + views = ( + str(caught.value), + repr(caught.value), + "".join(traceback.format_exception(caught.value)), + ) + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + assert all("PLANTED" not in view for view in views) + + +def test_direct_model_validation_entrypoints_are_code_only() -> None: + marker = "PLANTED-STRUCTURED-PYDANTIC-SECRET-715" + payload = {"digest": marker, "length": 1, "media_type": "text/plain"} + calls = ( + lambda: ContentEvidenceV1(**payload), + lambda: ContentEvidenceV1.model_validate(payload), + lambda: ContentEvidenceV1.model_validate_json(json.dumps(payload)), + lambda: ContentEvidenceV1.model_validate_strings( + {"digest": marker, "length": "1", "media_type": "text/plain"} + ), + ) + for call in calls: + with pytest.raises(ContentEnvelopeError, match="ENV-MALFORMED") as caught: + call() + views = ( + str(caught.value), + repr(caught.value), + "".join(traceback.format_exception(caught.value)), + ) + assert not hasattr(caught.value, "errors") + assert not hasattr(caught.value, "json") + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + assert all(marker not in view for view in views) + + +def test_frozen_model_mutation_entrypoints_are_code_only() -> None: + marker = "PLANTED-MUTATION-SECRET-715" + source = _ingress().source + calls = ( + lambda: setattr(source, "kind", marker), + lambda: delattr(source, "kind"), + ) + for call in calls: + with pytest.raises(ContentEnvelopeError, match="ENV-MALFORMED") as caught: + call() + views = ( + str(caught.value), + repr(caught.value), + "".join(traceback.format_exception(caught.value)), + ) + assert not hasattr(caught.value, "errors") + assert not hasattr(caught.value, "json") + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + assert all(marker not in view for view in views) + + +def test_direct_transform_model_rejects_registered_shape_but_unknown_cleartext() -> None: + with pytest.raises(ContentEnvelopeError, match="ENV-MALFORMED"): + TransformEvidenceV1( + id="secret.exfil.channel", + version="1", + implementation_digest=DIGEST_A, + parameters_digest=DIGEST_B, + ) + ingress = _ingress() + derived = _derive((ingress,)) + assert (ingress.transform.id, ingress.transform.version) == ("mcp-warden.capture", "1") + assert (derived.transform.id, derived.transform.version) == ( + "mcp-warden.transform.deterministic", + "1", + ) + with pytest.raises(ContentEnvelopeError, match="ENV-MALFORMED"): + _ingress(capture_kind="PLANTED-TRANSFORM-SECRET") + with pytest.raises(ContentEnvelopeError, match="ENV-MALFORMED"): + _derive((ingress,), transform_kind="PLANTED-TRANSFORM-SECRET") + with pytest.raises(ContentEnvelopeError, match="ENV-MALFORMED"): + _ingress(capture_kind=TransformKindV1.DETERMINISTIC) + with pytest.raises(ContentEnvelopeError, match="ENV-MALFORMED"): + _derive((ingress,), transform_kind=TransformKindV1.INGRESS_CAPTURE) + forged_wire = to_public_dict(ingress) + forged_wire["transform"]["id"] = "secret.exfil.channel" + with pytest.raises(ContentEnvelopeError, match="ENV-MALFORMED"): + parse_envelope(rfc8785.dumps(forged_wire)) + + +class _HostileEnvelope(ContentEnvelopeV1): + def __getattribute__(self, name: str): + if name.startswith("__"): + return super().__getattribute__(name) + raise RuntimeError("PLANTED-HOSTILE-ENVELOPE-SECRET-715") + + +class _HostileBundle(BundleEvidenceInput): + def __getattribute__(self, name: str): + if name.startswith("__"): + return super().__getattribute__(name) + raise RuntimeError("PLANTED-HOSTILE-BUNDLE-SECRET-715") + + +def test_public_boundaries_reject_hostile_subclasses_before_attribute_access() -> None: + valid = _ingress() + hostile_envelope = _HostileEnvelope.model_construct( + **object.__getattribute__(valid, "__dict__") + ) + envelope_calls = ( + lambda: serialize_envelope(hostile_envelope), + lambda: verify_envelope(hostile_envelope, content=b"payload"), + lambda: verify_lineage(hostile_envelope, parents=()), + lambda: _derive((hostile_envelope,)), + ) + for call in envelope_calls: + with pytest.raises(ContentEnvelopeError) as caught: + call() + assert caught.value.code in {"ENV-MALFORMED", "ENV-LINEAGE-INVALID"} + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + assert "PLANTED" not in "".join(traceback.format_exception(caught.value)) + + hostile_bundle = object.__new__(_HostileBundle) + with pytest.raises(ContentEnvelopeError, match="ENV-BUNDLE-INVALID") as caught: + _ingress(bundle=hostile_bundle) + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + + +def _canonical_json_size(size: int) -> bytes: + for count in range(1, MAX_JSON_NODES): + payload_bytes = size - (1 + 3 * count) + if 0 <= payload_bytes <= MAX_JSON_STRING_BYTES * count: + base, extra = divmod(payload_bytes, count) + values = ["x" * (base + (1 if index < extra else 0)) for index in range(count)] + encoded = rfc8785.dumps(values) + assert len(encoded) == size + return encoded + raise AssertionError("requested canonical size is not representable within V1 caps") + + +@pytest.mark.parametrize( + "field,cap", + ( + ("source_claims", MAX_SOURCE_CLAIMS_BYTES), + ("capture_implementation", MAX_TRANSFORM_IDENTITY_BYTES), + ("capture_parameters", MAX_TRANSFORM_PARAMETERS_BYTES), + ), +) +def test_structured_component_caps_accept_exact_and_reject_plus_one(field: str, cap: int) -> None: + _ingress(**{field: _canonical_json_size(cap)}) + with pytest.raises(ContentEnvelopeError, match="ENV-OVER-CAP"): + _ingress(**{field: _canonical_json_size(cap + 1)}) + + +@pytest.mark.parametrize( + "field", + ("artifact", "version_claims", "publisher_claims", "policy_binding_claims"), +) +def test_bundle_structured_component_caps(field: str) -> None: + values = { + "artifact": b"{}", + "signature_evidence": b"signature", + "version_claims": b"{}", + "publisher_claims": b"{}", + "dependencies": (b"{}",), + "policy_binding_claims": b"{}", + } + values[field] = _canonical_json_size(MAX_BUNDLE_COMPONENT_BYTES) + _ingress(bundle=BundleEvidenceInput(**values)) + values[field] = _canonical_json_size(MAX_BUNDLE_COMPONENT_BYTES + 1) + with pytest.raises(ContentEnvelopeError, match="ENV-OVER-CAP"): + BundleEvidenceInput(**values) + + +def test_bundle_signature_and_dependency_component_caps() -> None: + _ingress(bundle=_bundle_input(signature=b"x" * MAX_BUNDLE_COMPONENT_BYTES)) + with pytest.raises(ContentEnvelopeError, match="ENV-OVER-CAP"): + _bundle_input(signature=b"x" * (MAX_BUNDLE_COMPONENT_BYTES + 1)) + + at_cap = _canonical_json_size(MAX_BUNDLE_COMPONENT_BYTES) + _ingress( + bundle=BundleEvidenceInput( + artifact=b"{}", + signature_evidence=None, + version_claims=b"{}", + publisher_claims=b"{}", + dependencies=(at_cap,), + policy_binding_claims=b"{}", + ) + ) + with pytest.raises(ContentEnvelopeError, match="ENV-OVER-CAP"): + BundleEvidenceInput( + artifact=b"{}", + signature_evidence=None, + version_claims=b"{}", + publisher_claims=b"{}", + dependencies=(_canonical_json_size(MAX_BUNDLE_COMPONENT_BYTES + 1),), + policy_binding_claims=b"{}", + ) + + +def test_json_structural_caps_exact_and_plus_one() -> None: + exact_depth = b"[" * MAX_JSON_DEPTH + b"0" + b"]" * MAX_JSON_DEPTH + _ingress(source_claims=exact_depth) + with pytest.raises(ContentEnvelopeError, match="ENV-OVER-CAP"): + _ingress( + source_claims=b"[" * (MAX_JSON_DEPTH + 1) + + b"0" + + b"]" * (MAX_JSON_DEPTH + 1) + ) + + _ingress(source_claims=rfc8785.dumps([0] * (MAX_JSON_NODES - 1))) + with pytest.raises(ContentEnvelopeError, match="ENV-OVER-CAP"): + _ingress(source_claims=rfc8785.dumps([0] * MAX_JSON_NODES)) + + _ingress(source_claims=rfc8785.dumps({f"k{index:03d}": 0 for index in range(MAX_JSON_OBJECT_KEYS)})) + with pytest.raises(ContentEnvelopeError, match="ENV-OVER-CAP"): + _ingress( + source_claims=rfc8785.dumps( + {f"k{index:03d}": 0 for index in range(MAX_JSON_OBJECT_KEYS + 1)} + ) + ) + + _ingress(source_claims=rfc8785.dumps("x" * MAX_JSON_STRING_BYTES)) + with pytest.raises(ContentEnvelopeError, match="ENV-OVER-CAP"): + _ingress(source_claims=rfc8785.dumps("x" * (MAX_JSON_STRING_BYTES + 1))) + + +def test_envelope_parent_taint_and_dependency_count_caps() -> None: + assert codec._parse_wire_json(_canonical_json_size(MAX_ENVELOPE_BYTES)) is not None + with pytest.raises(ContentEnvelopeError, match="ENV-OVER-CAP"): + codec._parse_wire_json(_canonical_json_size(MAX_ENVELOPE_BYTES + 1)) + + parents = tuple(_ingress(content=index.to_bytes(2, "big")) for index in range(MAX_PARENTS)) + _derive(parents) + with pytest.raises(ContentEnvelopeError, match="ENV-OVER-CAP"): + _derive(parents + (_ingress(content=b"overflow-parent"),)) + + _ingress(added_taints=(TaintV1.UNTRUSTED,) * MAX_TAINTS) + with pytest.raises(ContentEnvelopeError, match="ENV-OVER-CAP"): + _ingress(added_taints=(TaintV1.UNTRUSTED,) * (MAX_TAINTS + 1)) + + dependency = b"{}" + exact_dependencies = BundleEvidenceInput( + artifact=b"{}", + signature_evidence=None, + version_claims=b"{}", + publisher_claims=b"{}", + dependencies=(dependency,) * MAX_DEPENDENCIES, + policy_binding_claims=b"{}", + ) + _ingress(bundle=exact_dependencies) + with pytest.raises(ContentEnvelopeError, match="ENV-OVER-CAP"): + BundleEvidenceInput( + artifact=b"{}", + signature_evidence=None, + version_claims=b"{}", + publisher_claims=b"{}", + dependencies=(dependency,) * (MAX_DEPENDENCIES + 1), + policy_binding_claims=b"{}", + ) + + +def _swapped_transform_candidate(context: str) -> ContentEnvelopeV1: + root = _ingress(source_kind=IngressKindV1.WEB) + if context == "root": + deterministic = _derive((root,)).transform + candidate = root.model_copy(update={"transform": deterministic}) + else: + derived = _derive((root,)) + capture = root.transform + forged_source = codec._derived_source(derived.parents, capture) + candidate = derived.model_copy(update={"source": forged_source, "transform": capture}) + return codec._replace_self_digest(candidate) + + +def _candidate_wire(candidate: ContentEnvelopeV1) -> bytes: + body = codec._body_dict( + content=candidate.content, + source=candidate.source, + parents=candidate.parents, + transform=candidate.transform, + taints=candidate.taints, + bundle=candidate.bundle, + ) + body["envelope_digest"] = candidate.envelope_digest + return rfc8785.dumps(body) + + +@pytest.mark.parametrize( + "context,code", + (("root", "ENV-MALFORMED"), ("derived", "ENV-LINEAGE-INVALID")), +) +def test_direct_model_rejects_source_context_transform_swap(context: str, code: str) -> None: + candidate = _swapped_transform_candidate(context) + with pytest.raises(ContentEnvelopeError, match=code): + ContentEnvelopeV1( + schema_version=candidate.schema_version, + trust_state=candidate.trust_state, + content=candidate.content, + source=candidate.source, + parents=candidate.parents, + transform=candidate.transform, + taints=candidate.taints, + bundle=candidate.bundle, + envelope_digest=candidate.envelope_digest, + ) + + +@pytest.mark.parametrize( + "context,code", + (("root", "ENV-MALFORMED"), ("derived", "ENV-LINEAGE-INVALID")), +) +def test_public_validation_rejects_source_context_transform_swap(context: str, code: str) -> None: + candidate = _swapped_transform_candidate(context) + with pytest.raises(ContentEnvelopeError, match=code): + serialize_envelope(candidate) + + +@pytest.mark.parametrize( + "context,code", + (("root", "ENV-MALFORMED"), ("derived", "ENV-LINEAGE-INVALID")), +) +def test_parser_rejects_self_coherent_source_context_transform_swap(context: str, code: str) -> None: + candidate = _swapped_transform_candidate(context) + with pytest.raises(ContentEnvelopeError, match=code): + parse_envelope(_candidate_wire(candidate))