diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f159d2..ef4c444 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,16 @@ Streamable HTTP; the v0.3 `guard` proxy adds deterministic runtime *result* insp ## [Unreleased] +### Artifact trust foundation + +- Add opt-in Ed25519 verification for existing policy, runtime, adapter and + executable-bundle activation APIs, with an independently pinned public-key + configuration and explicit artifact-kind signer roles. +- Add `trust roots-digest`, `trust prepare` and `trust verify` for enrollment + inspection, unsigned canonical review/signing artifacts, and offline signature + checks. Human signing stays external; signature validity alone does not permit + effects. Durable receipts and live guard integration remain separate work. + ### Release engineering - Add a manual `verify-pypi` production OIDC exchange check in the existing release diff --git a/DOCUMENTATION_INDEX.md b/DOCUMENTATION_INDEX.md index ebb470e..50955c7 100644 --- a/DOCUMENTATION_INDEX.md +++ b/DOCUMENTATION_INDEX.md @@ -1,6 +1,6 @@ # Documentation Index — mcp-warden -Last Updated: 2026-10-02 +Last Updated: 2026-10-03 Master index of every document in this repository. The `docs/` files are the **security contract and source of truth** for all algorithms; the three core docs @@ -155,6 +155,7 @@ scope-honesty box and makes no compliance/regulatory claim. | Doc | Defines | |-----|---------| +| [`docs/ARTIFACT_TRUST.md`](docs/ARTIFACT_TRUST.md) | Opt-in Ed25519 implementation of DSE-716's external artifact verifier: explicit key/kind roles, independently pinned canonical roots, exact version/kind/key-bound signing frame, unsigned review preparation and offline signature verification. No private signing capability, durable receipts, live guard wiring or whole-kernel claim | | [`docs/PIN_CHECK_DEMO.md`](docs/PIN_CHECK_DEMO.md) | Full end-to-end pin/check walkthrough, archived out of `README.md` on 2026-08-24 to hold the 500-line core-doc limit | | [`docs/SPEC.md`](docs/SPEC.md) | **MCP Lock Format v1** — the vendor-neutral, self-contained format specification any tool can implement: on-disk `warden.lock` schema, RFC 8785 (JCS) canonicalization, SHA-256 `sha256:` hashing, `overall_digest` construction, the normative drift class + severity table, the optional per-tool inspection block, and a Conformance section (§12.1: passing `vectors/` **is** conformance) + worked example. `WARDEN_LOCK_SCHEMA.md` is the mcp-warden implementation of this format | | [`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 | @@ -177,6 +178,7 @@ scope-honesty box and makes no compliance/regulatory claim. | Plan | Purpose | |---|---| +| [`docs/plans/2026-10-03-checkpoint-artifact-trust.md`](docs/plans/2026-10-03-checkpoint-artifact-trust.md) | First human-checkpoint development wave: opt-in signature verification and unsigned review tooling, with unpublished DSE-717 work preserved and live DSE-1076 integration still dependent on it | | [`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` | | [`docs/plans/2026-07-19-pdp-pep-design.md`](docs/plans/2026-07-19-pdp-pep-design.md) | **Non-normative DSE-716 execution record.** Records the activated-snapshot, structural-PEP, TDD, and review plan; verified behavior is documented in `docs/POLICY_ENFORCEMENT.md` | @@ -212,6 +214,11 @@ scope-honesty box and makes no compliance/regulatory claim. ## Source layout +`src/mcp_warden/artifact_payload.py`, `artifact_trust.py`, and `cli_trust.py` implement +the strict existing-artifact decoder, opt-in public-key verifier, and unsigned +review/signature CLI. Tests are `tests/test_artifact_trust.py`, +`test_cli_trust.py`, and `test_artifact_trust_integration.py`. + | Module | Responsibility | Spec anchor | |--------|----------------|-------------| | `src/mcp_warden/hashing.py` | `canon()` (RFC 8785) + `hash()` + field hashes | WARDEN_LOCK_SCHEMA §3 | diff --git a/README.md b/README.md index 2990645..33ff2bc 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # mcp-warden -Last Updated: 2026-10-02 +Last Updated: 2026-10-03 [![CI](https://github.com/DataScience-EngineeringExperts/mcp-warden/actions/workflows/integrity-gate.yml/badge.svg)](https://github.com/DataScience-EngineeringExperts/mcp-warden/actions/workflows/integrity-gate.yml) [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE) @@ -24,6 +24,13 @@ The [upgrade plan and checkpoint proposal](docs/plans/2026-10-02-tool-integrity- maps prompts, retrieval, code execution, and serverless adapters to existing Agent Trust Kernel work. Those broader checkpoints are proposals, not shipped guarantees. +The opt-in [artifact trust foundation](docs/ARTIFACT_TRUST.md) adds public-key +verification for the existing kernel activation APIs and `trust` CLI commands +to prepare unsigned review artifacts and verify external signatures. Keys and +signer roles require an independently protected root pin. This development +feature is not in the published 2.0.0 package; live checkpoint enforcement still +depends on DSE-717 evidence and DSE-1076 guard integration. + > ⚠️ **Install `mcp-warden-cli`, not `mcp-warden`.** The PyPI name `mcp-warden` is > an **unrelated package by a different author** — it is not this project. The > correct install is `pip install mcp-warden-cli` (the CLI command is still diff --git a/SYSTEM_CONTEXT_DIAGRAM.md b/SYSTEM_CONTEXT_DIAGRAM.md index faf0e70..1690383 100644 --- a/SYSTEM_CONTEXT_DIAGRAM.md +++ b/SYSTEM_CONTEXT_DIAGRAM.md @@ -1,6 +1,6 @@ # mcp-warden — System Context Diagram -Last Updated: 2026-10-02 +Last Updated: 2026-10-03 **CLI 2.0.0 / schema level 4** extends capture/lock/check to complete tool annotations and output schemas, with structural output drift and Python/TypeScript parity. The existing @@ -38,6 +38,11 @@ logic) plus a separate informational provenance section. It never prints raw > branch), but it must still deliver durable signed evidence, fallback, rollback-resistant state, > and the recovery latch. The current `guard` path is not represented as ATK-conformant. +> The opt-in [artifact verifier](docs/ARTIFACT_TRUST.md) implements DSE-716's external +> signature port using pinned Ed25519 public keys and explicit artifact-kind roles. +> `trust prepare` emits unsigned review/signing bytes; `trust verify` checks signatures +> only. Human signing remains external and the root pin must be host-protected. + > `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 > runtime dependency and is never invoked by `pin`/`check`/`policy`. @@ -109,9 +114,11 @@ flowchart TB envelope["Content Envelope V1\nDSE-715 · implemented evidence foundation\nNOT wired to guard · grants no authority"] decision["Deterministic PDP/PEP V1\nDSE-716 · signed adapter/bundle gates + fixed corpus\nNOT wired to guard"] + artifactTrust["Opt-in Ed25519 artifact verifier\nartifact_trust.py · protected root pin + key roles\ntrust CLI: unsigned preparation / signature verification"] evidence["Durable evidence + recovery state\nDSE-717 · IN PROGRESS\ncurrent default gate denies effects"] atk -. "governs partial foundation" .-> envelope envelope -. "required input" .-> decision + artifactTrust -. "external signature verification port" .-> decision decision -. "requires production gate" .-> evidence subgraph ci["CI pipeline (GitHub Actions / local)"] diff --git a/docs-site/artifact-trust.md b/docs-site/artifact-trust.md new file mode 100644 index 0000000..52f9d2b --- /dev/null +++ b/docs-site/artifact-trust.md @@ -0,0 +1,60 @@ +# Verify externally approved artifacts + +The development branch adds public-key signature verification for Agent Trust +Kernel policy, runtime, adapter, and executable-bundle artifacts. This feature +is not included in the published **2.0.0** package. Install a reviewed Git commit +with the `artifact-trust` extra to use it. + +It prepares unsigned review artifacts and verifies external Ed25519 signatures. +Live human-checkpoint enforcement still requires the durable evidence layer and +guard integration. + +## Enroll public keys and signer roles + +Configure the public keys and the artifact kinds each key may sign. Keep the +private signing capability in an independently governed external signer, outside +agent and tool access. + +```bash +mcp-warden trust roots-digest public-roots.json +``` + +This prints a candidate root digest and signer identities for independent +enrollment review. Pin the reviewed digest through a protected host boundary. +An agent-controlled roots file and matching agent-controlled pin do not +establish trust. + +## Prepare an unsigned review artifact + +```bash +mcp-warden trust prepare policy policy-draft.json \ + --signer "$ENROLLED_SIGNER_DIGEST" \ + --canonical-out policy.canonical.json \ + --signing-out policy.signing.bin +``` + +Review the canonical artifact together with its resolved grant/lease bindings. +Then have the external signer sign the exact prepared bytes. Preparation does +not approve the artifact or execute an action. Both output paths must be new. + +## Verify the returned signature + +```bash +mcp-warden trust verify policy policy.canonical.json \ + --roots public-roots.json \ + --roots-digest "$PROTECTED_ROOTS_DIGEST" \ + --signer "$ENROLLED_SIGNER_DIGEST" \ + --signature policy.sig +``` + +Success reports `signature-verified`. The variables here illustrate values +provided by the trusted host; writable environment variables alone are not a +protected boundary. The signature must be exactly 64 raw bytes. + +Signature validity alone does not check current authority, expiry, revocation, +dependency measurements, or whether an action is allowed. Existing activation +and policy enforcement APIs remain responsible for those checks. The default +evidence gate continues to block effects pending durable evidence integration. + +See the [artifact trust contract](https://github.com/DataScience-EngineeringExperts/mcp-warden/blob/main/docs/ARTIFACT_TRUST.md) +for the roots schema, exact signing frame, SDK usage and remaining dependencies. diff --git a/docs/AGENT_TRUST_KERNEL.md b/docs/AGENT_TRUST_KERNEL.md index 2a1c4c9..e3cf5c5 100644 --- a/docs/AGENT_TRUST_KERNEL.md +++ b/docs/AGENT_TRUST_KERNEL.md @@ -452,8 +452,10 @@ foundation harness with registration evidence, multiple serialized output-channe non-optional versioned malformed-input corpus. The default evidence gate denies every otherwise allowed effect. -This is still a partial implementation. It is not wired into the historical `guard`, supplies no -built-in production verifier or live protocol adapter, and does not implement DSE-717's durable +This is still a partial implementation. It is not wired into the historical `guard`. A separate +opt-in [artifact trust verifier](ARTIFACT_TRUST.md) implements the signature port using explicit +public-key/role roots and an independently protected digest pin. It supplies no live protocol +adapter and does not implement DSE-717's durable signed receipts, independent fallback evidence, rollback-resistant state, or persistent recovery latch. A custom evidence gate becomes TCB code and does not create an ATK-conformance claim. No production effect or whole-kernel claim is valid until DSE-717 closes ATK-10 through ATK-12 and diff --git a/docs/ARTIFACT_TRUST.md b/docs/ARTIFACT_TRUST.md new file mode 100644 index 0000000..b9a1ad5 --- /dev/null +++ b/docs/ARTIFACT_TRUST.md @@ -0,0 +1,160 @@ +# Public-key governance artifact trust + +This opt-in verifier implements the existing DSE-716 `ArtifactVerifierV1` port +for externally signed policy, runtime, adapter, and executable-bundle artifacts. +It is a foundation for human checkpoints. It does not wire those checkpoints +into the historical `guard`, implement DSE-717's durable receipts or protected +state, or establish whole-kernel conformance. + +## Trust and key custody + +Install this development branch with the `artifact-trust` extra. The published +2.0.0 package does not include this feature. No signing command, private-key +loader, or key generator exists in the product. Use an independently governed +external signing service or signing station; keep its private signing capability +outside agent/tool access. The external signer must review the canonical +artifact and sign exactly the prepared frame, not merely a displayed summary. + +The consumer explicitly configures raw Ed25519 public keys and the artifact +kinds each may sign. Governance and runtime signing roles should use separate +keys. In particular, a key allowed to sign reviewed policy or executable bundles +does not acquire permission to attest to runtime health or trusted time. + +The consumer MUST obtain `expected_roots_digest` through a protected trusted +boundary independent of the roots file and artifact. The complete key AND role +configuration is pinned. Computing a digest from attacker-supplied roots and +immediately passing it as the pin does not establish trust. Replacing both roots +and pin defeats this local boundary; filesystem permissions and custody belong +to the embedding host. This verifier provides no protection against host/TCB or +authorized-admin compromise. + +## Public roots format + +```json +{ + "schema_version": 1, + "keys": [ + { + "public_key": "<64 lowercase hex characters encoding 32 raw public-key bytes>", + "artifact_kinds": ["adapter", "bundle", "policy"] + } + ] +} +``` + +Replace the illustrative public-key placeholder with an enrolled public key. +Kinds are a nonempty, sorted, duplicate-free subset of `adapter`, `bundle`, +`policy`, `runtime`. Duplicate keys, unknown fields, unsupported schemas, +non-finite numbers, and duplicate JSON object keys are rejected. Roots are +bounded to 64 keys and 64 KiB. SDK records are immutable; verification snapshots +the public keys and allowed roles. + +Let `H(domain, bytes)` be `sha256:` plus the lowercase hexadecimal SHA-256 of +`ASCII(domain) || 0x00 || bytes`. + +- Signer identity: `H("mcp-warden/artifact-signature/v1/key-id", raw_public_key)`. +- Roots digest: `H("mcp-warden/artifact-signature/v1/roots", canonical_roots)`. +- Canonical roots: RFC 8785 JSON with `schema_version: 1`, keys sorted by signer + identity, public keys encoded as lowercase hex, and kinds sorted as above. + +The signer identity identifies a key, not a verified human name. The operator's +external enrollment record binds that key to a human or service and its role. +Adding/removing a key or changing its roles changes the root pin. Root freshness, +rotation and revocation require independently governed configuration updates; +the verifier does not infer them from untrusted input. + +## Signature format + +Sign the exact byte concatenation: + +```text +ASCII("mcp-warden/artifact-signature/v1") || 0x00 || +ASCII(artifact_kind) || 0x00 || ASCII(signer_identity) || 0x00 || +canonical_artifact_payload +``` + +Use Ed25519 and a raw detached 64-byte signature. The existing external port +selector remains `VerificationAlgorithmV1.EXTERNAL_V1`; this implementation +does not negotiate algorithms. There is no base64/hex/signature-envelope +autodetection. Kind, key identity and format version are signed to prevent +cross-role/context reuse. Signing raw artifact JSON without the frame is invalid. + +The payload is the existing canonical serialization from +`canonical_policy_bytes`, `canonical_runtime_bytes`, `canonical_manifest_bytes`, +or `canonical_bundle_manifest_bytes`. Preparation accepts an unambiguous JSON +draft and validates the strict existing model. Verification requires exact +canonical bytes. Unknown artifact fields and schemas are rejected; schema +integers cannot be booleans. The global input cap is 512 KiB, with narrower +existing artifact caps retained. + +## Offline CLI workflow + +Enrollment inspection prints the candidate root digest and derived signer +identities. A trusted administrator independently reviews/enrolls the keys and +roles and pins this digest; the command itself does not approve enrollment. + +```bash +mcp-warden trust roots-digest public-roots.json +mcp-warden trust prepare policy policy-draft.json \ + --signer "$ENROLLED_SIGNER_DIGEST" \ + --canonical-out policy.canonical.json --signing-out policy.signing.bin +``` + +The second command creates an **unsigned** canonical review artifact and exact +signing frame. Outputs must be new files. Input/output collisions and clobbering +are rejected; failures clean up newly created outputs. Review all resolved +grants/leases and their subject, input, action, argument, destination, policy and +version bindings before signing externally. A policy's digest-only grants are +not a substitute for that review package. + +After the external signer returns `policy.sig`: + +```bash +mcp-warden trust verify policy policy.canonical.json \ + --roots public-roots.json --roots-digest "$PROTECTED_ROOTS_DIGEST" \ + --signer "$ENROLLED_SIGNER_DIGEST" --signature policy.sig +``` + +`$PROTECTED_ROOTS_DIGEST` and `$ENROLLED_SIGNER_DIGEST` above are illustrative +values supplied by the trusted host; environment variables writable by the +agent do not create a protected boundary. Success emits `signature-verified`; +it is not an execution authorization. Failure exits 2 with a code-only error and +no raw artifact, signature, key file, or lower-layer exception text. + +## SDK activation + +```python +from mcp_warden.artifact_trust import Ed25519ArtifactVerifierV1, parse_roots +from mcp_warden.policy_decision import activate_policy + +# The host supplies protected_roots_digest independently and bounds roots_bytes. +verifier = Ed25519ArtifactVerifierV1( + parse_roots(roots_bytes), expected_roots_digest=protected_roots_digest +) +active_policy = activate_policy(signed_policy_candidate, verifier=verifier) +``` + +The same verifier can be passed to `activate_runtime`, `activate_adapter`, and +`activate_executable_bundle`, with separately enrolled roles. Existing APIs +still require exact request/lease/policy matching and actual adapter, handler, +artifact and dependency evidence. Invalid signatures return false through the +verification port and fail activation. Missing optional crypto support fails +explicitly with `TRUST-CRYPTO-UNAVAILABLE`. + +## Limits and next dependencies + +A valid signature authenticates exact bytes under an enrolled key/role. It does +not establish that code/content is safe, authenticate every tool result, clear +taint, verify current executable measurements, or permit effects. It also does +not enforce expiry, trusted time, revocation or rollback floors by itself. +Existing PDP/PEP policy/runtime checks remain necessary; an expired artifact can +have a mathematically valid signature while being unusable for authorization. + +The default evidence gate continues to permit nothing. DSE-717 must deliver +durable signed receipts, negative-decision fallback, rollback-resistant state +and recovery before DSE-1076 can connect the kernel to live `guard`. Its recorded +unpublished implementation must be recovered rather than duplicated. + +Tests use ephemeral test-only private keys for all four real activation paths, +key/role/kind/payload substitution, invalid parsing/pins, dependency drift, and +proof that valid signatures cannot bypass the default evidence gate. diff --git a/docs/POLICY_ENFORCEMENT.md b/docs/POLICY_ENFORCEMENT.md index 8cc7a0b..8b963d5 100644 --- a/docs/POLICY_ENFORCEMENT.md +++ b/docs/POLICY_ENFORCEMENT.md @@ -235,6 +235,8 @@ module markers are outside the supported API and are TCB compromise, not an auth not silently upgraded and is not ATK-conformant. - DSE-716 APIs are importable client-agnostic foundations, not a new CLI command or deployed runtime adapter. -- No built-in production verifier, durable evidence gate, recovery store, or live protocol - adapter is selected by this ticket. +- DSE-716 itself selects no built-in verifier. The separate opt-in + [artifact trust foundation](ARTIFACT_TRUST.md) now implements its external verifier port + with pinned public keys and explicit signer roles; it supplies no durable evidence gate, + recovery store, or live protocol adapter. - DSE-717 remains required before any whole-kernel or production effect claim. diff --git a/docs/plans/2026-10-03-checkpoint-artifact-trust.md b/docs/plans/2026-10-03-checkpoint-artifact-trust.md new file mode 100644 index 0000000..72d163a --- /dev/null +++ b/docs/plans/2026-10-03-checkpoint-artifact-trust.md @@ -0,0 +1,55 @@ +# Checkpoint Artifact Trust Implementation Plan + +> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. + +**Goal:** Verify externally approved DSE-716 governance artifacts using pinned public keys and artifact-specific signer roles, without exposing signing authority to an agent. + +**Architecture:** Implement the existing `ArtifactVerifierV1` port using Ed25519. A consumer pins the complete public-key/role configuration by digest outside agent-editable state; each signature binds the exact canonical artifact, artifact kind, key identity, and signature format. Offline CLI commands prepare unsigned review artifacts and verify signatures; existing activation APIs retain all structural and policy checks. + +**Tech Stack:** Python 3.11+, existing Pydantic/RFC 8785 serializers, optional `cryptography` Ed25519 support, Typer, pytest. + +--- + +## Brief and scope + +The owner approved development of a human checkpoint for a reviewed tool version and its allowed actions. DSE-716 already defines exact policy grants, leases, adapter operations, and executable/dependency closure. This wave replaces its test-only signature verifier seam with a usable public-key verifier and review/verification tooling. + +DSE-717 is started and records unpublished work on `codex/dse-717-receipts` (design `cbf9dfe`, plan `73ab1b0`, protected-state task `c319fa7`). Those objects and that worktree are absent from the builder and GitHub. Recover them before implementing its receipt/recovery design. Do not mutate that started issue or recreate its unfinished modules. DSE-1076 runtime integration remains blocked by it. + +Done means real signatures activate the existing policy/runtime/adapter/bundle contracts; wrong keys, roles, kinds, algorithms, changed bytes, malformed artifacts, and changed root configurations fail closed. Existing evidence gate and guard behavior stay unchanged. No claim of live human-checkpoint enforcement, protected-state persistence, artifact safety, or whole-kernel conformance follows from signature verification. + +## Task 1: Canonical public-key trust and signing contract + +**Files:** Create `src/mcp_warden/artifact_trust.py`, `src/mcp_warden/artifact_payload.py`; test `tests/test_artifact_trust.py`. + +1. Write real Ed25519 fixtures and failing tests for canonical root pins, key-derived signer identity, role separation, exact signing bytes, and malformed/bounded inputs. +2. Run `.venv/bin/pytest -q tests/test_artifact_trust.py`; expect missing-module failures. +3. Implement immutable public-key/role records, bounded duplicate-rejecting JSON parsing, canonical existing-model validation, domain-separated signing bytes and `Ed25519ArtifactVerifierV1`. Require a pinned digest of the complete root configuration. Accept only `VerificationAlgorithmV1.EXTERNAL_V1`; do not introduce algorithm negotiation or private-key access. +4. Run the focused tests; expect valid signature verification and failure for every substitution case. + +## Task 2: Offline review and verification CLI + +**Files:** Create `src/mcp_warden/cli_trust.py`; modify `src/mcp_warden/cli.py`, `pyproject.toml`, `requirements-dev.lock`; test `tests/test_cli_trust.py`. + +1. Write failing CLI tests for `trust roots-digest`, `trust prepare`, and `trust verify`, including secret-safe errors and output-file collisions. +2. Add an optional `artifact-trust` extra and a dev dependency using the existing locked cryptography version as the resolution baseline. Regenerate the dev lock with its documented universal/hash command. +3. Implement bounded file reads; create review outputs exclusively, reject clobbering or input/output collisions, and clean up partial new outputs. `prepare` emits canonical artifact JSON plus the exact bytes to sign externally. `verify` requires explicit roots, pinned roots digest, signer and detached 64-byte signature. Report only signature verification, not activation or execution authorization. +4. Run `.venv/bin/pytest -q tests/test_artifact_trust.py tests/test_cli_trust.py` and `.venv/bin/ruff check` on changed files. + +## Task 3: Integration proof and documentation + +**Files:** Test `tests/test_artifact_trust_integration.py`; create `docs/ARTIFACT_TRUST.md`; modify the three core docs and documentation navigation. + +1. Prove actual signatures work through existing policy/runtime/adapter/bundle activation; use the existing instrumented adapter fixtures only in tests. +2. Demonstrate payload/kind/key substitutions fail activation and the default evidence gate still blocks execution. Test duplicate keys, unknown fields, booleans in integer fields, unsupported schemas, role changes, malformed signature lengths, over-cap data, and unavailable crypto support. +3. Document exact signing frame, public-key identity and trust digest, protected pin/key-custody requirements, SDK usage, offline CLI workflow, and remaining DSE-717/DSE-1076 dependencies. Signature verification alone does not enforce freshness, expiry, revocation, artifact/dependency measurements, or allowed actions; those checks remain in activated policy/runtime/adapter/bundle and PDP/PEP contracts. +4. Run focused integration tests, the existing DSE-716 suites, Ruff, strict docs build and the full CI-equivalent Python suite. Require independent security evaluator `APPROVE` against the final diff before release. Record any remaining release gate against a concrete PR/head. + +## Adversarial plan pass + +- **Root substitution:** require an independently pinned digest covering keys AND roles; a roots file supplied by an artifact cannot confer trust. +- **Cross-role signatures:** bind artifact kind and derived signer identity in the signing frame; every configured key must enumerate allowed kinds. +- **Ambiguous parsing:** reject duplicate keys/non-finite numbers and unknown fields; compare canonical payloads exactly in SDK verification. +- **Agent self-approval:** no private signing key, generation, or signing command exists in product code. Key enrollment/pin updates are governance operations outside the model. +- **Unfinished work:** preserve the unpublished DSE-717 implementation; no duplicate receipts, logs, or recovery stores in this wave. +- **Rollback:** revert the new opt-in CLI/verifier and its documentation; existing default-deny evidence behavior is unchanged. A local root file/pin controlled by the agent provides no defended trust boundary. diff --git a/mkdocs.yml b/mkdocs.yml index b092e18..7eb098e 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -63,6 +63,7 @@ nav: - Pin MCP servers in CI: pin-in-ci.md - MCP security checklist: checklist.md - Reference: + - External artifact signatures: artifact-trust.md - The MCP Lock Format: lock-format.md - Comparison vs scanners & gateways: comparison.md diff --git a/pyproject.toml b/pyproject.toml index a057d78..f83bed1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,6 +33,7 @@ dependencies = [ [project.optional-dependencies] dev = [ + "cryptography>=50.0.0,<51", "pytest>=9.0.3", "pytest-asyncio>=0.23.0", "pytest-cov>=6.0.0", @@ -43,6 +44,8 @@ dev = [ # install stays sigstore-free. Pinned to the 4.3.x line the implementation was # verified against (signing.py warns at runtime if installed outside this window). sigstore = ["sigstore>=4.3.0,<5"] +# Public-key governance artifact verification; never loads private signing keys. +artifact-trust = ["cryptography>=50.0.0,<51"] [project.scripts] mcp-warden = "mcp_warden.cli:app" diff --git a/requirements-dev.lock b/requirements-dev.lock index df17c9f..27b0c3c 100644 --- a/requirements-dev.lock +++ b/requirements-dev.lock @@ -276,7 +276,9 @@ cryptography==50.0.0 \ --hash=sha256:f59e38625469987d7ef6d495323c55e7db6c212eaf6112267e0d3b565a2e9c9f \ --hash=sha256:f89831ef99dd7dd169ab06d63a831adb9e20a87aac6d380266bbda5823349169 \ --hash=sha256:fd9192b7b70c573d7f214eb1ae35e00d359f6f5e4b27c7e21e30de1fc6204645 - # via pyjwt + # via + # mcp-warden-cli (pyproject.toml) + # pyjwt h11==0.16.0 ; sys_platform != 'emscripten' \ --hash=sha256:4e35b956cf45792e4caa5885e69fba00bdbc6ffafbfa020300e549b208ee5ff1 \ --hash=sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86 diff --git a/src/mcp_warden/artifact_payload.py b/src/mcp_warden/artifact_payload.py new file mode 100644 index 0000000..8d978e2 --- /dev/null +++ b/src/mcp_warden/artifact_payload.py @@ -0,0 +1,110 @@ +"""Bounded, unambiguous parsing of existing governance artifact contracts.""" + +from __future__ import annotations + +import json + +from mcp_warden.decision_models import ( + ArtifactKindV1, + PolicyBundleV1, + PolicyGrantV1, + RuntimeSnapshotV1, +) +from mcp_warden.executable_bundle import ( + ExecutableBundleManifestV1, + canonical_bundle_manifest_bytes, +) +from mcp_warden.policy_decision import canonical_policy_bytes, canonical_runtime_bytes +from mcp_warden.policy_enforcement import ( + AdapterManifestV1, + ManifestOperationV1, + canonical_manifest_bytes, +) + +MAX_ARTIFACT_BYTES = 512 * 1024 + + +class ArtifactTrustError(Exception): + """Code-only failure; never retain input or lower-layer exception text.""" + + def __repr__(self) -> str: + return str(self) + + +def _unique(pairs: list[tuple[str, object]]) -> dict[str, object]: + result: dict[str, object] = {} + for key, value in pairs: + if key in result: + raise ValueError + result[key] = value + return result + + +def _nonfinite(_: str) -> None: + raise ValueError + + +def bounded_json(payload: bytes, *, cap: int, code: str) -> object: + result = None + invalid = type(payload) is not bytes or len(payload) > cap + if not invalid: + try: + result = json.loads( + payload.decode("utf-8"), object_pairs_hook=_unique, parse_constant=_nonfinite + ) + except Exception: + invalid = True + if invalid: + raise ArtifactTrustError(code) from None + return result + + +_ARTIFACTS = { + ArtifactKindV1.POLICY: (PolicyBundleV1, canonical_policy_bytes), + ArtifactKindV1.RUNTIME: (RuntimeSnapshotV1, canonical_runtime_bytes), + ArtifactKindV1.ADAPTER: (AdapterManifestV1, canonical_manifest_bytes), + ArtifactKindV1.BUNDLE: (ExecutableBundleManifestV1, canonical_bundle_manifest_bytes), +} + + +def canonical_artifact_payload(kind: ArtifactKindV1, payload: bytes) -> bytes: + """Validate a review draft using the existing model; return exact canonical bytes.""" + if type(kind) is not ArtifactKindV1: + raise ArtifactTrustError("TRUST-ARTIFACT-MALFORMED") from None + parsed = bounded_json(payload, cap=MAX_ARTIFACT_BYTES, code="TRUST-ARTIFACT-MALFORMED") + if ( + type(parsed) is not dict + or type(parsed.get("schema_version")) is not int + or parsed["schema_version"] != 1 + ): + raise ArtifactTrustError("TRUST-ARTIFACT-MALFORMED") from None + canonical = None + try: + model, serialize = _ARTIFACTS[kind] + # Existing custom model constructors intentionally reject Python lists + # for tuple fields. Translate only the declared JSON array fields; + # all scalar, nested-object and extra-field checks stay strict. + fields = dict(parsed) + arrays = { + ArtifactKindV1.POLICY: ("grants", "revoked_lease_digests"), + ArtifactKindV1.ADAPTER: ("dependency_digests", "operations"), + ArtifactKindV1.BUNDLE: ("dependency_digests",), + ArtifactKindV1.RUNTIME: (), + } + for name in arrays[kind]: + if type(fields[name]) is not list: + raise ValueError + fields[name] = tuple(fields[name]) + if kind is ArtifactKindV1.POLICY: + fields["grants"] = tuple(PolicyGrantV1(**item) for item in fields["grants"]) + if kind is ArtifactKindV1.ADAPTER: + fields["operations"] = tuple( + ManifestOperationV1(**item) for item in fields["operations"] + ) + candidate = model(**fields) + canonical = serialize(candidate) + except Exception: + pass + if canonical is None: + raise ArtifactTrustError("TRUST-ARTIFACT-MALFORMED") from None + return canonical diff --git a/src/mcp_warden/artifact_trust.py b/src/mcp_warden/artifact_trust.py new file mode 100644 index 0000000..7d0a66a --- /dev/null +++ b/src/mcp_warden/artifact_trust.py @@ -0,0 +1,219 @@ +"""Opt-in public-key verification for DSE-716's external verifier port. + +No private-key operations live here. The configured root digest must come from +a consumer-controlled boundary outside agent-editable inputs and state. +""" + +from __future__ import annotations + +import hashlib +import re +from dataclasses import dataclass +from types import MappingProxyType + +import rfc8785 + +from mcp_warden.artifact_payload import ( + ArtifactTrustError, + bounded_json, + canonical_artifact_payload, +) +from mcp_warden.decision_models import DIGEST_RE, ArtifactKindV1, VerificationAlgorithmV1 + +MAX_ROOTS_BYTES = 64 * 1024 +MAX_TRUST_KEYS = 64 +_PUBLIC_KEY_RE = re.compile(r"^[0-9a-f]{64}$") +_FRAME = b"mcp-warden/artifact-signature/v1\x00" + + +def _hash(domain: bytes, payload: bytes) -> str: + return "sha256:" + hashlib.sha256(domain + b"\x00" + payload).hexdigest() + + +@dataclass(frozen=True, slots=True) +class TrustedArtifactKeyV1: + public_key: bytes + artifact_kinds: tuple[ArtifactKindV1, ...] + + def __post_init__(self) -> None: + if ( + type(self.public_key) is not bytes + or len(self.public_key) != 32 + or type(self.artifact_kinds) is not tuple + or not self.artifact_kinds + or len(self.artifact_kinds) > len(ArtifactKindV1) + or any(type(kind) is not ArtifactKindV1 for kind in self.artifact_kinds) + or self.artifact_kinds != tuple(sorted(set(self.artifact_kinds))) + ): + raise ArtifactTrustError("TRUST-ROOTS-MALFORMED") from None + + @property + def signer_identity(self) -> str: + return _hash(b"mcp-warden/artifact-signature/v1/key-id", self.public_key) + + +def canonical_roots_bytes(roots: tuple[TrustedArtifactKeyV1, ...]) -> bytes: + if ( + type(roots) is not tuple + or not roots + or len(roots) > MAX_TRUST_KEYS + or any(type(root) is not TrustedArtifactKeyV1 for root in roots) + ): + raise ArtifactTrustError("TRUST-ROOTS-MALFORMED") from None + # Rebuild records to reject mutated/forged dataclass instances. + validated = tuple(TrustedArtifactKeyV1(root.public_key, root.artifact_kinds) for root in roots) + if len({root.signer_identity for root in validated}) != len(validated): + raise ArtifactTrustError("TRUST-ROOTS-MALFORMED") from None + return rfc8785.dumps( + { + "schema_version": 1, + "keys": [ + { + "public_key": root.public_key.hex(), + "artifact_kinds": [kind.value for kind in root.artifact_kinds], + } + for root in sorted(validated, key=lambda item: item.signer_identity) + ], + } + ) + + +def roots_digest(roots: tuple[TrustedArtifactKeyV1, ...]) -> str: + return _hash(b"mcp-warden/artifact-signature/v1/roots", canonical_roots_bytes(roots)) + + +def parse_roots(payload: bytes) -> tuple[TrustedArtifactKeyV1, ...]: + parsed = bounded_json(payload, cap=MAX_ROOTS_BYTES, code="TRUST-ROOTS-MALFORMED") + roots = None + try: + if ( + type(parsed) is not dict + or set(parsed) != {"schema_version", "keys"} + or type(parsed["schema_version"]) is not int + or parsed["schema_version"] != 1 + or type(parsed["keys"]) is not list + or not 0 < len(parsed["keys"]) <= MAX_TRUST_KEYS + ): + raise ValueError + records = [] + for item in parsed["keys"]: + if ( + type(item) is not dict + or set(item) != {"public_key", "artifact_kinds"} + or type(item["public_key"]) is not str + or _PUBLIC_KEY_RE.fullmatch(item["public_key"]) is None + or type(item["artifact_kinds"]) is not list + or any(type(kind) is not str for kind in item["artifact_kinds"]) + ): + raise ValueError + records.append( + TrustedArtifactKeyV1( + bytes.fromhex(item["public_key"]), + tuple(ArtifactKindV1(kind) for kind in item["artifact_kinds"]), + ) + ) + candidate = tuple(sorted(records, key=lambda item: item.signer_identity)) + canonical_roots_bytes(candidate) + roots = candidate + except Exception: + pass + if roots is None: + raise ArtifactTrustError("TRUST-ROOTS-MALFORMED") from None + return roots + + +def artifact_signing_bytes(kind: ArtifactKindV1, signer_identity: str, payload: bytes) -> bytes: + if ( + type(kind) is not ArtifactKindV1 + or type(signer_identity) is not str + or DIGEST_RE.fullmatch(signer_identity) is None + ): + raise ArtifactTrustError("TRUST-ARTIFACT-MALFORMED") from None + if canonical_artifact_payload(kind, payload) != payload: + raise ArtifactTrustError("TRUST-ARTIFACT-NONCANONICAL") from None + return ( + _FRAME + + kind.value.encode("ascii") + + b"\x00" + + signer_identity.encode("ascii") + + b"\x00" + + payload + ) + + +def _public_key_type(): + key_type = None + try: + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey + + key_type = Ed25519PublicKey + except ImportError: + pass + if key_type is None: + raise ArtifactTrustError("TRUST-CRYPTO-UNAVAILABLE") from None + return key_type + + +class Ed25519ArtifactVerifierV1: + """Pinned keys and explicit kind roles; bool-only fail-closed verification. + + This implements ArtifactVerifierV1, not a policy decision or evidence gate. + A consumer must protect the pin and perform all existing activation checks. + """ + + __slots__ = ("_roots",) + + def __init__(self, roots: tuple[TrustedArtifactKeyV1, ...], *, expected_roots_digest: str): + if ( + type(expected_roots_digest) is not str + or DIGEST_RE.fullmatch(expected_roots_digest) is None + or roots_digest(roots) != expected_roots_digest + ): + raise ArtifactTrustError("TRUST-ROOT-PIN-MISMATCH") from None + key_type = _public_key_type() + records = {} + failed = False + try: + for root in roots: + records[root.signer_identity] = ( + key_type.from_public_bytes(root.public_key), + frozenset(root.artifact_kinds), + ) + except Exception: + failed = True + if failed: + raise ArtifactTrustError("TRUST-CRYPTO-UNAVAILABLE") from None + object.__setattr__(self, "_roots", MappingProxyType(records)) + + def __setattr__(self, name: str, value: object) -> None: + raise ArtifactTrustError("TRUST-IMMUTABLE") from None + + def __delattr__(self, name: str) -> None: + raise ArtifactTrustError("TRUST-IMMUTABLE") from None + + def verify( + self, + *, + artifact_kind: ArtifactKindV1, + algorithm: VerificationAlgorithmV1, + signer_identity: str, + payload: bytes, + signature: bytes, + ) -> bool: + if ( + type(artifact_kind) is not ArtifactKindV1 + or type(algorithm) is not VerificationAlgorithmV1 + or algorithm is not VerificationAlgorithmV1.EXTERNAL_V1 + or type(signer_identity) is not str + or type(signature) is not bytes + or len(signature) != 64 + ): + return False + try: + key, kinds = self._roots[signer_identity] + if artifact_kind not in kinds: + return False + key.verify(signature, artifact_signing_bytes(artifact_kind, signer_identity, payload)) + except Exception: + return False + return True diff --git a/src/mcp_warden/cli.py b/src/mcp_warden/cli.py index 5c0159a..849fd4d 100644 --- a/src/mcp_warden/cli.py +++ b/src/mcp_warden/cli.py @@ -41,6 +41,7 @@ from .cli_guard import register as register_guard_commands from .cli_lock import register as register_lock_commands from .cli_sign import sign_after_pin, verify_lock_signature +from .cli_trust import register as register_trust_commands from .corpus_coordinate import parse_explicit as parse_coordinate from .emitters import build_sarif, findings_to_jsonl, sarif_to_json from .lockfile import ( @@ -95,6 +96,7 @@ def _root( register_auth_commands(app, console, err_console) register_deploy_gate_command(app, console, err_console) register_doctor_command(app, console, err_console) +register_trust_commands(app, console, err_console) def _split_server_cmd(server_cmd: list[str]) -> tuple[str, list[str]]: diff --git a/src/mcp_warden/cli_trust.py b/src/mcp_warden/cli_trust.py new file mode 100644 index 0000000..078d790 --- /dev/null +++ b/src/mcp_warden/cli_trust.py @@ -0,0 +1,154 @@ +"""Unsigned governance review artifacts and pinned signature verification.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import typer + +from mcp_warden.artifact_payload import MAX_ARTIFACT_BYTES, canonical_artifact_payload +from mcp_warden.artifact_trust import ( + MAX_ROOTS_BYTES, + ArtifactTrustError, + Ed25519ArtifactVerifierV1, + artifact_signing_bytes, + parse_roots, + roots_digest, +) +from mcp_warden.decision_models import ArtifactKindV1, VerificationAlgorithmV1 + + +def _read(path: Path, cap: int) -> bytes: + payload = None + try: + with path.open("rb") as source: + payload = source.read(cap + 1) + except OSError: + pass + if payload is None: + raise ArtifactTrustError("TRUST-FILE-UNAVAILABLE") from None + if len(payload) > cap: + raise ArtifactTrustError("TRUST-FILE-OVER-CAP") from None + return payload + + +def _write_pair(source: Path, outputs: tuple[tuple[Path, bytes], ...]) -> None: + created: list[Path] = [] + invalid = False + try: + paths = [path.resolve() for path, _ in outputs] + if len(set(paths)) != len(paths) or source.resolve() in paths: + raise OSError + for path, payload in outputs: + with path.open("xb") as target: + created.append(path) + target.write(payload) + except (OSError, RuntimeError): + invalid = True + if invalid: + for path in created: + try: + path.unlink() + except OSError: + pass + raise ArtifactTrustError("TRUST-OUTPUT-UNAVAILABLE") from None + + +def register(app, console, err_console) -> None: + trust = typer.Typer( + add_completion=False, help="Prepare unsigned artifacts and verify external signatures." + ) + app.add_typer(trust, name="trust") + + def fail(error: ArtifactTrustError) -> None: + err_console.print(str(error), markup=False) + raise typer.Exit(code=2) + + def emit(value: dict) -> None: + typer.echo(json.dumps(value, sort_keys=True)) + + @trust.command("roots-digest") + def digest_roots( + roots: Path = typer.Argument(..., help="Public keys and explicit artifact-kind roles."), + ) -> None: + """Inspect roots for independent human enrollment; this does not establish trust.""" + try: + records = parse_roots(_read(roots, MAX_ROOTS_BYTES)) + emit( + { + "roots_digest": roots_digest(records), + "keys": [ + { + "signer_identity": item.signer_identity, + "artifact_kinds": [kind.value for kind in item.artifact_kinds], + } + for item in records + ], + } + ) + except ArtifactTrustError as error: + fail(error) + + @trust.command("prepare") + def prepare( + kind: ArtifactKindV1 = typer.Argument(...), + artifact: Path = typer.Argument(..., help="Unsigned artifact review draft."), + signer: str = typer.Option(..., "--signer", help="Enrolled public-key identity digest."), + canonical_out: Path = typer.Option( + ..., "--canonical-out", help="New canonical artifact file." + ), + signing_out: Path = typer.Option( + ..., "--signing-out", help="New exact bytes for the external signer." + ), + ) -> None: + """Validate a draft and prepare bytes for review/signing outside agent access.""" + try: + payload = canonical_artifact_payload(kind, _read(artifact, MAX_ARTIFACT_BYTES)) + frame = artifact_signing_bytes(kind, signer, payload) + _write_pair(artifact, ((canonical_out, payload), (signing_out, frame))) + emit( + { + "status": "unsigned-review-artifact", + "artifact_kind": kind.value, + "signer_identity": signer, + } + ) + except ArtifactTrustError as error: + fail(error) + + @trust.command("verify") + def verify( + kind: ArtifactKindV1 = typer.Argument(...), + artifact: Path = typer.Argument(..., help="Exact canonical artifact file."), + roots: Path = typer.Option(..., "--roots", help="Explicit public-key/role configuration."), + roots_pin: str = typer.Option( + ..., "--roots-digest", help="Root digest from an independent trusted boundary." + ), + signer: str = typer.Option(..., "--signer", help="Enrolled public-key identity digest."), + signature: Path = typer.Option( + ..., "--signature", help="Detached raw 64-byte Ed25519 signature." + ), + ) -> None: + """Verify a signature only; activation, freshness and action authorization are separate.""" + try: + verifier = Ed25519ArtifactVerifierV1( + parse_roots(_read(roots, MAX_ROOTS_BYTES)), expected_roots_digest=roots_pin + ) + if not verifier.verify( + artifact_kind=kind, + algorithm=VerificationAlgorithmV1.EXTERNAL_V1, + signer_identity=signer, + payload=_read(artifact, MAX_ARTIFACT_BYTES), + signature=_read(signature, 64), + ): + raise ArtifactTrustError("TRUST-SIGNATURE-INVALID") + emit( + { + "status": "signature-verified", + "artifact_kind": kind.value, + "signer_identity": signer, + } + ) + except ArtifactTrustError as error: + fail(error) diff --git a/tests/test_artifact_trust.py b/tests/test_artifact_trust.py new file mode 100644 index 0000000..53d8ccc --- /dev/null +++ b/tests/test_artifact_trust.py @@ -0,0 +1,190 @@ +from __future__ import annotations + +import json + +import pytest +import rfc8785 +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from mcp_warden.artifact_trust import ( + ArtifactTrustError, + Ed25519ArtifactVerifierV1, + TrustedArtifactKeyV1, + artifact_signing_bytes, + canonical_roots_bytes, + parse_roots, + roots_digest, +) +from mcp_warden.decision_models import ArtifactKindV1, VerificationAlgorithmV1 +from mcp_warden.policy_decision import canonical_policy_bytes +from tests.test_policy_decision import _policy_bundle + + +def trust_fixture(kinds=(ArtifactKindV1.POLICY,)): + private = Ed25519PrivateKey.generate() + root = TrustedArtifactKeyV1(private.public_key().public_bytes_raw(), kinds) + roots = (root,) + verifier = Ed25519ArtifactVerifierV1(roots, expected_roots_digest=roots_digest(roots)) + return private, root, verifier + + +def policy_payload(): + return canonical_policy_bytes(_policy_bundle(lease_digest=None)) + + +def check(verifier, root, base_payload, base_signature, **changes): + args = dict( + artifact_kind=ArtifactKindV1.POLICY, + algorithm=VerificationAlgorithmV1.EXTERNAL_V1, + signer_identity=root.signer_identity, + payload=base_payload, + signature=base_signature, + ) + args.update(changes) + return verifier.verify(**args) + + +def test_real_signature_and_exact_frame(): + private, root, verifier = trust_fixture() + payload = policy_payload() + frame = artifact_signing_bytes(ArtifactKindV1.POLICY, root.signer_identity, payload) + assert frame == ( + b"mcp-warden/artifact-signature/v1\x00policy\x00" + + root.signer_identity.encode("ascii") + + b"\x00" + + payload + ) + assert check(verifier, root, payload, private.sign(frame)) is True + + +@pytest.mark.parametrize( + "changes", + [ + {"artifact_kind": ArtifactKindV1.RUNTIME}, + {"artifact_kind": "policy"}, + {"algorithm": "external-v1"}, + {"signer_identity": "sha256:" + "0" * 64}, + {"signature": b""}, + {"signature": b"x" * 63}, + {"signature": b"x" * 65}, + {"payload": b"null"}, + {"payload": b"{}"}, + {"payload": b"\xff"}, + {"payload": b" " * (512 * 1024 + 1)}, + {"payload": bytearray(b"{}")}, + ], +) +def test_substitution_fails_closed(changes): + private, root, verifier = trust_fixture() + payload = policy_payload() + signature = private.sign( + artifact_signing_bytes(ArtifactKindV1.POLICY, root.signer_identity, payload) + ) + assert check(verifier, root, payload, signature, **changes) is False + + +def test_changed_payload_wrong_key_raw_signature_and_noncanonical_fail(): + private, root, verifier = trust_fixture() + payload = policy_payload() + signature = private.sign( + artifact_signing_bytes(ArtifactKindV1.POLICY, root.signer_identity, payload) + ) + changed = rfc8785.dumps({**json.loads(payload), "policy_generation": 3}) + assert not check(verifier, root, changed, signature) + assert not check(verifier, root, payload, Ed25519PrivateKey.generate().sign(payload)) + assert not check(verifier, root, payload, private.sign(payload)) + assert not check(verifier, root, b" " + payload, signature) + + +def test_role_is_checked_even_for_cryptographically_valid_signature(): + private, root, verifier = trust_fixture((ArtifactKindV1.BUNDLE,)) + payload = policy_payload() + signature = private.sign( + artifact_signing_bytes(ArtifactKindV1.POLICY, root.signer_identity, payload) + ) + assert not check(verifier, root, payload, signature) + + +def test_roots_pin_covers_roles_and_keys_and_configuration_is_immutable(): + _, root, verifier = trust_fixture() + changed = (TrustedArtifactKeyV1(root.public_key, (ArtifactKindV1.RUNTIME,)),) + assert roots_digest(changed) != roots_digest((root,)) + with pytest.raises(ArtifactTrustError, match="TRUST-ROOT-PIN-MISMATCH"): + Ed25519ArtifactVerifierV1(changed, expected_roots_digest=roots_digest((root,))) + with pytest.raises(ArtifactTrustError, match="TRUST-IMMUTABLE"): + verifier._roots = {} + assert parse_roots(canonical_roots_bytes((root,))) == (root,) + + +@pytest.mark.parametrize( + "raw", + [ + b"{}", + b"null", + b'{"schema_version":true,"keys":[]}', + b'{"schema_version":1,"schema_version":1,"keys":[]}', + b'{"schema_version":1,"keys":[],"private_key":"secret"}', + b'{"schema_version":1,"keys":[]}', + b'{"schema_version":NaN,"keys":[]}', + b"x" * (64 * 1024 + 1), + ], +) +def test_malformed_roots_are_code_only(raw): + with pytest.raises(ArtifactTrustError) as exc: + parse_roots(raw) + assert str(exc.value) == repr(exc.value) == "TRUST-ROOTS-MALFORMED" + assert exc.value.__context__ is None + + +def test_unknown_root_fields_duplicate_roles_and_duplicate_keys_rejected(): + _, root, _ = trust_fixture() + value = json.loads(canonical_roots_bytes((root,))) + variants = [] + item = value["keys"][0] + variants.append({**value, "keys": [{**item, "private_key": "planted-secret"}]}) + variants.append({**value, "keys": [{**item, "artifact_kinds": ["policy", "policy"]}]}) + variants.append({**value, "keys": [item, item]}) + variants.append({**value, "keys": [{**item, "public_key": "0" * 62}]}) + for malformed in variants: + with pytest.raises(ArtifactTrustError): + parse_roots(json.dumps(malformed).encode()) + + +def test_optional_dependency_failure_is_explicit(monkeypatch): + from mcp_warden import artifact_trust + + def unavailable(): + raise ArtifactTrustError("TRUST-CRYPTO-UNAVAILABLE") + + _, root, _ = trust_fixture() + monkeypatch.setattr(artifact_trust, "_public_key_type", unavailable) + with pytest.raises(ArtifactTrustError, match="TRUST-CRYPTO-UNAVAILABLE"): + Ed25519ArtifactVerifierV1((root,), expected_roots_digest=roots_digest((root,))) + + +@pytest.mark.parametrize( + "field,value", + [ + ("schema_version", True), + ("schema_version", 2), + ("policy_generation", True), + ("valid_from", "100"), + ("grants", {}), + ("unknown", "PLANTED-SECRET"), + ], +) +def test_artifact_parser_rejects_coercion_and_unknown_fields(field, value): + from mcp_warden.artifact_payload import canonical_artifact_payload + + body = {**json.loads(policy_payload()), field: value} + with pytest.raises(ArtifactTrustError, match="TRUST-ARTIFACT-MALFORMED") as exc: + canonical_artifact_payload(ArtifactKindV1.POLICY, json.dumps(body).encode()) + assert exc.value.__context__ is None + + +def test_roots_order_does_not_change_pin_and_duplicate_roles_cannot_expand_authority(): + _, one, _ = trust_fixture() + _, two, _ = trust_fixture((ArtifactKindV1.RUNTIME,)) + assert roots_digest((one, two)) == roots_digest((two, one)) + with pytest.raises(ArtifactTrustError): + TrustedArtifactKeyV1(one.public_key, (ArtifactKindV1.POLICY, ArtifactKindV1.POLICY)) diff --git a/tests/test_artifact_trust_integration.py b/tests/test_artifact_trust_integration.py new file mode 100644 index 0000000..330e44c --- /dev/null +++ b/tests/test_artifact_trust_integration.py @@ -0,0 +1,171 @@ +from __future__ import annotations + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from mcp_warden.artifact_payload import canonical_artifact_payload +from mcp_warden.artifact_trust import ( + Ed25519ArtifactVerifierV1, + TrustedArtifactKeyV1, + artifact_signing_bytes, + roots_digest, +) +from mcp_warden.content_models import BundleEvidenceInput +from mcp_warden.decision_models import ( + ArtifactKindV1, + DecisionError, + DecisionVerdictV1, + SignedPolicyCandidateV1, + SignedRuntimeCandidateV1, + VerificationAlgorithmV1, +) +from mcp_warden.executable_bundle import ( + BundleActivationError, + ExecutableBundleManifestV1, + SignedExecutableBundleCandidateV1, + activate_executable_bundle, + bundle_evidence_from_input, + canonical_bundle_manifest_bytes, +) +from mcp_warden.policy_decision import ( + PolicyDecisionPointV1, + activate_policy, + activate_runtime, + canonical_policy_bytes, + canonical_runtime_bytes, +) +from mcp_warden.policy_enforcement import ( + AdapterRegistryV1, + EnforcementCodeV1, + PolicyEnforcementPointV1, + SignedAdapterCandidateV1, + activate_adapter, + canonical_manifest_bytes, +) +from tests.test_policy_enforcement import _active_components, _manifest, _noop_handler + + +def real_components(): + governance = Ed25519PrivateKey.generate() + runtime_key = Ed25519PrivateKey.generate() + human = TrustedArtifactKeyV1( + governance.public_key().public_bytes_raw(), + (ArtifactKindV1.ADAPTER, ArtifactKindV1.BUNDLE, ArtifactKindV1.POLICY), + ) + runtime_root = TrustedArtifactKeyV1( + runtime_key.public_key().public_bytes_raw(), (ArtifactKindV1.RUNTIME,) + ) + roots = (human, runtime_root) + verifier = Ed25519ArtifactVerifierV1(roots, expected_roots_digest=roots_digest(roots)) + + def sign(kind, payload): + root, private = ( + (runtime_root, runtime_key) if kind is ArtifactKindV1.RUNTIME else (human, governance) + ) + assert canonical_artifact_payload(kind, payload) == payload + return dict( + algorithm=VerificationAlgorithmV1.EXTERNAL_V1, + signer_identity=root.signer_identity, + signature=private.sign(artifact_signing_bytes(kind, root.signer_identity, payload)), + ) + + effect, request, original_policy, original_runtime = _active_components() + policy = activate_policy( + SignedPolicyCandidateV1( + policy=original_policy.policy, + **sign(ArtifactKindV1.POLICY, canonical_policy_bytes(original_policy.policy)), + ), + verifier=verifier, + ) + runtime = activate_runtime( + SignedRuntimeCandidateV1( + runtime=original_runtime.runtime, + **sign(ArtifactKindV1.RUNTIME, canonical_runtime_bytes(original_runtime.runtime)), + ), + verifier=verifier, + ) + manifest = _manifest(_noop_handler, policy_id=policy.policy.policy_id) + registry = AdapterRegistryV1() + registry.register(operation_id="document.read", handler=_noop_handler) + candidate = SignedAdapterCandidateV1( + manifest=manifest, + implementation=b"fixture-adapter-binary", + dependencies=(b"dependency-a", b"dependency-b"), + **sign(ArtifactKindV1.ADAPTER, canonical_manifest_bytes(manifest)), + ) + adapter = activate_adapter(candidate, registry=registry, verifier=verifier, policy=policy) + return effect, request, policy, runtime, adapter, verifier, sign + + +def test_real_policy_runtime_and_adapter_activate_but_default_evidence_blocks(): + effect, request, policy, runtime, adapter, _, _ = real_components() + pdp = PolicyDecisionPointV1(policy) + decision = pdp.evaluate(request, runtime=runtime) + assert decision.verdict == DecisionVerdictV1.ALLOW.value + result = PolicyEnforcementPointV1(pdp, adapter).execute(request, runtime=runtime, effect=effect) + assert result.code == EnforcementCodeV1.EVIDENCE_UNAVAILABLE.value + assert not result.invoked + + +def test_policy_mutation_and_cross_role_signature_fail_activation(): + _, _, policy, runtime, _, verifier, sign = real_components() + signature = sign(ArtifactKindV1.POLICY, canonical_policy_bytes(policy.policy)) + changed = policy.policy.model_copy(update={"policy_generation": 8}) + with pytest.raises(DecisionError, match="PDP-POLICY-VERIFICATION"): + activate_policy(SignedPolicyCandidateV1(policy=changed, **signature), verifier=verifier) + signature = sign(ArtifactKindV1.POLICY, canonical_policy_bytes(policy.policy)) + with pytest.raises(DecisionError, match="PDP-RUNTIME-VERIFICATION"): + activate_runtime( + SignedRuntimeCandidateV1(runtime=runtime.runtime, **signature), verifier=verifier + ) + + +def test_real_bundle_signature_still_requires_exact_dependency_closure(): + _, _, policy, _, adapter, verifier, sign = real_components() + source = BundleEvidenceInput( + artifact=b'{"artifact":"reviewed"}', + signature_evidence=b'{"signature":"reviewed"}', + version_claims=b'{"version":"1.0.0"}', + publisher_claims=b'{"publisher":"fixture"}', + dependencies=(b'{"dependency":"fixture"}',), + policy_binding_claims=b'{"policy_generation":7}', + ) + evidence = bundle_evidence_from_input(source) + # Publisher identity is the enrolled signing-key identity, not a claim from the bundle. + signer = sign(ArtifactKindV1.POLICY, canonical_policy_bytes(policy.policy))["signer_identity"] + manifest = ExecutableBundleManifestV1( + schema_version=1, + bundle_id="fixture.bundle", + bundle_version="1.0.0", + publisher_identity=signer, + artifact_digest=evidence.artifact_digest, + signature_evidence_digest=evidence.signature_evidence_digest, + version_claims_digest=evidence.version_claims_digest, + publisher_claims_digest=evidence.publisher_claims_digest, + dependency_digests=evidence.dependency_digests, + policy_binding_claims_digest=evidence.policy_binding_claims_digest, + policy_id=policy.policy.policy_id, + policy_generation=policy.policy.policy_generation, + adapter_manifest_digest=adapter.manifest_digest, + ) + signed = sign(ArtifactKindV1.BUNDLE, canonical_bundle_manifest_bytes(manifest)) + candidate = SignedExecutableBundleCandidateV1(manifest=manifest, evidence=source, **signed) + active = activate_executable_bundle( + candidate, verifier=verifier, policy=policy, adapter_manifest_digest=adapter.manifest_digest + ) + assert active.evidence == evidence + changed = BundleEvidenceInput( + artifact=source.artifact, + signature_evidence=source.signature_evidence, + version_claims=source.version_claims, + publisher_claims=source.publisher_claims, + dependencies=(b'{"dependency":"substituted"}',), + policy_binding_claims=source.policy_binding_claims, + ) + with pytest.raises(BundleActivationError, match="PEP-BUNDLE-INTEGRITY"): + activate_executable_bundle( + SignedExecutableBundleCandidateV1(manifest=manifest, evidence=changed, **signed), + verifier=verifier, + policy=policy, + adapter_manifest_digest=adapter.manifest_digest, + ) diff --git a/tests/test_cli_trust.py b/tests/test_cli_trust.py new file mode 100644 index 0000000..4671990 --- /dev/null +++ b/tests/test_cli_trust.py @@ -0,0 +1,187 @@ +from __future__ import annotations + +import json +import subprocess +import sys + +import pytest +from typer.testing import CliRunner + +from mcp_warden.artifact_trust import ( + artifact_signing_bytes, + canonical_roots_bytes, + roots_digest, +) +from mcp_warden.cli import app +from mcp_warden.decision_models import ArtifactKindV1 +from tests.test_artifact_trust import policy_payload, trust_fixture + +runner = CliRunner() + + +def inputs(tmp_path): + private, root, _ = trust_fixture() + candidate = tmp_path / "policy.json" + candidate.write_bytes(policy_payload()) + roots = tmp_path / "roots.json" + roots.write_bytes(canonical_roots_bytes((root,))) + signature = tmp_path / "approval.sig" + signature.write_bytes( + private.sign( + artifact_signing_bytes( + ArtifactKindV1.POLICY, root.signer_identity, candidate.read_bytes() + ) + ) + ) + return root, candidate, roots, signature + + +def verify_args(root, candidate, roots, signature): + return [ + "trust", + "verify", + "policy", + str(candidate), + "--roots", + str(roots), + "--roots-digest", + roots_digest((root,)), + "--signer", + root.signer_identity, + "--signature", + str(signature), + ] + + +def test_roots_digest_and_prepare_unsigned_review_artifacts(tmp_path): + root, candidate, roots, _ = inputs(tmp_path) + result = runner.invoke(app, ["trust", "roots-digest", str(roots)]) + assert result.exit_code == 0, result.output + assert json.loads(result.stdout)["roots_digest"] == roots_digest((root,)) + canonical = tmp_path / "canonical.json" + frame = tmp_path / "signing.bin" + # Human review drafts may be pretty-printed, but ambiguous JSON is rejected. + candidate.write_text(json.dumps(json.loads(candidate.read_bytes()), indent=2)) + result = runner.invoke( + app, + [ + "trust", + "prepare", + "policy", + str(candidate), + "--signer", + root.signer_identity, + "--canonical-out", + str(canonical), + "--signing-out", + str(frame), + ], + ) + assert result.exit_code == 0, result.output + assert json.loads(result.stdout)["status"] == "unsigned-review-artifact" + assert canonical.read_bytes() == policy_payload() + assert frame.read_bytes() == artifact_signing_bytes( + ArtifactKindV1.POLICY, root.signer_identity, canonical.read_bytes() + ) + + +def test_verify_reports_signature_only_and_needs_independent_pin(tmp_path): + root, candidate, roots, signature = inputs(tmp_path) + result = runner.invoke(app, verify_args(root, candidate, roots, signature)) + assert result.exit_code == 0, result.output + assert json.loads(result.stdout)["status"] == "signature-verified" + args = verify_args(root, candidate, roots, signature) + args[args.index("--roots-digest") + 1] = "sha256:" + "0" * 64 + result = runner.invoke(app, args) + assert result.exit_code == 2 + assert "TRUST-ROOT-PIN-MISMATCH" in result.output + + +@pytest.mark.parametrize("mutation", ["payload", "signature", "oversize", "duplicate", "unknown"]) +def test_cli_tamper_and_secret_safe_failures(tmp_path, mutation): + root, candidate, roots, signature = inputs(tmp_path) + planted = "PLANTED-SECRET-DO-NOT-REFLECT" + if mutation == "payload": + body = json.loads(candidate.read_bytes()) + body["policy_generation"] += 1 + import rfc8785 + + candidate.write_bytes(rfc8785.dumps(body)) + elif mutation == "signature": + signature.write_bytes(b"x" * 64) + elif mutation == "oversize": + signature.write_bytes(planted.encode() * 100) + elif mutation == "duplicate": + candidate.write_text('{"schema_version":1,"schema_version":1,"secret":"' + planted + '"}') + else: + body = json.loads(candidate.read_bytes()) + body["secret"] = planted + candidate.write_text(json.dumps(body)) + result = runner.invoke(app, verify_args(root, candidate, roots, signature)) + assert result.exit_code == 2 + assert planted not in result.output + assert "TRUST-" in result.output + + +def test_prepare_never_clobbers_input_or_existing_outputs_and_rolls_back(tmp_path): + root, candidate, _, _ = inputs(tmp_path) + first = tmp_path / "canonical.json" + existing = tmp_path / "signing.bin" + existing.write_bytes(b"preserve") + original = candidate.read_bytes() + common = ["trust", "prepare", "policy", str(candidate), "--signer", root.signer_identity] + result = runner.invoke( + app, common + ["--canonical-out", str(first), "--signing-out", str(existing)] + ) + assert result.exit_code == 2 + assert not first.exists() + assert existing.read_bytes() == b"preserve" + result = runner.invoke( + app, common + ["--canonical-out", str(candidate), "--signing-out", str(first)] + ) + assert result.exit_code == 2 + assert candidate.read_bytes() == original + assert not first.exists() + + +def test_missing_file_and_wrong_artifact_type_are_safe(tmp_path): + root, candidate, roots, signature = inputs(tmp_path) + candidate.unlink() + result = runner.invoke(app, verify_args(root, candidate, roots, signature)) + assert result.exit_code == 2 + assert "TRUST-FILE-UNAVAILABLE" in result.output + result = runner.invoke(app, ["trust", "roots-digest", str(signature)]) + assert result.exit_code == 2 + + +def test_prepare_symlink_loop_is_code_only_and_creates_no_outputs(tmp_path): + root, candidate, _, _ = inputs(tmp_path) + loop = tmp_path / "PLANTED-SENSITIVE-PATH" + loop.symlink_to(loop.name) + other = tmp_path / "signing.bin" + args = [ + "trust", + "prepare", + "policy", + str(candidate), + "--signer", + root.signer_identity, + "--canonical-out", + str(loop), + "--signing-out", + str(other), + ] + result = runner.invoke(app, args) + assert result.exit_code == 2 + assert result.output.strip() == "TRUST-OUTPUT-UNAVAILABLE" + assert not other.exists() + process = subprocess.run( + [sys.executable, "-c", "from mcp_warden.cli import app; app()", *args], + capture_output=True, + text=True, + timeout=20, + ) + assert process.returncode == 2 + assert process.stdout == "" + assert process.stderr.strip() == "TRUST-OUTPUT-UNAVAILABLE" + assert not other.exists()