From 976b0a2ed9e40ef7478118653c59e065cf8b2f15 Mon Sep 17 00:00:00 2001 From: Val Rukhaylo Date: Tue, 13 Oct 2026 05:38:27 +0000 Subject: [PATCH 1/5] test: paired experimental runtime catalog binding with fail-closed refusals --- ...test_experimental_manifest_binding_edrg.py | 128 ++++++++++++++++++ 1 file changed, 128 insertions(+) create mode 100644 tests/unit/test_experimental_manifest_binding_edrg.py diff --git a/tests/unit/test_experimental_manifest_binding_edrg.py b/tests/unit/test_experimental_manifest_binding_edrg.py new file mode 100644 index 00000000..0bca1176 --- /dev/null +++ b/tests/unit/test_experimental_manifest_binding_edrg.py @@ -0,0 +1,128 @@ +"""Paired experimental signed-catalog admission/refusal at cMCP's real binding boundary.""" + +import copy +from datetime import UTC, datetime, timedelta + +import agent_manifest as sdk +import pytest + +from cmcp_runtime.agent_manifest import verify_agent_manifest_binding +from cmcp_runtime.catalog.loader import ToolCatalog +from cmcp_runtime.config import EnforcementMode +from cmcp_runtime.errors import ConfigError +from cmcp_runtime.manifest_catalog import manifest_catalog_binding + +SEALED = "sha256:" + "c" * 64 +POLICY = "sha256:" + "b" * 64 + + +@pytest.fixture +def binding_packet(): + now = datetime.now(UTC) + key = sdk.generate_ed25519() + catalog = ToolCatalog(entries={}, catalog_hash=SEALED) + projection = manifest_catalog_binding(catalog) + manifest = { + "manifest_id": "01993f2c-8000-7000-8000-000000000001", + "agent_id": "spiffe://example.test/agent/one", + "version": "0.2", + "issued_at": (now - timedelta(minutes=1)).isoformat(), + "expires_at": (now + timedelta(hours=2)).isoformat(), + "issuer": "spiffe://example.test/issuer", + "crypto_profile": "standard", + "profile": "evidence-requirements-experimental-v1", + "artifacts": { + "system_prompt": { + "hash": "sha256:" + "a" * 64, + "version": "1", + "classification": "public", + "bound_at": now.isoformat(), + }, + "policy_bundle": { + "hash": POLICY, + "version": "1", + "policy_language": "cedar", + "enforcement_mode": "enforce", + "bound_at": now.isoformat(), + }, + "model_identity": { + "provider": "test", + "model_id": "test", + "version": "test-model", + "deployment_type": "api", + "model_attestation_type": "provider-asserted", + "bound_at": now.isoformat(), + }, + "tool_manifest": { + **projection, + "rug_pull_policy": "deny-and-alert", + "allow_dynamic_registration": False, + "bound_at": now.isoformat(), + }, + }, + "evidence_requirements": { + "components": [ + { + "component_id": "runtime.cpu", + "component_type": "runtime", + "required": True, + "accepted_profiles": ["urn:example:snp:v1"], + "accepted_appraisal_authorities": ["https://verifier.example"], + "maximum_age_seconds": 120, + }, + { + "component_id": "served-catalog", + "component_type": "tool-catalog", + "required": True, + "artifact_ref": "artifacts.tool_manifest", + "expected_observed_digest": SEALED, + "accepted_profiles": ["test-v1"], + "accepted_appraisal_authorities": ["https://verifier.example"], + "maximum_age_seconds": 300, + }, + ], + "required_bindings": [], + "combination_policy_ref": "urn:example:policy:v1", + }, + } + return manifest, key, catalog + + +def bind(packet, *, manifest=None, catalog=None, policy=POLICY): + base_manifest, key, base_catalog = packet + actual_manifest = manifest if manifest is not None else base_manifest + actual_catalog = catalog if catalog is not None else base_catalog + signed = sdk.sign_cose_sign1(actual_manifest, key) + return verify_agent_manifest_binding( + actual_manifest, + {key.key_id: key.public_bytes}, + authenticated_subject=actual_manifest["agent_id"], + policy_bundle_hash=policy, + tool_catalog_hash=actual_catalog.catalog_hash, + runtime_catalog=actual_catalog, + envelope=signed, + enforcement_mode=EnforcementMode.ENFORCING, + ) + + +def test_signed_experimental_profile_accepts_matching_actual_catalog(binding_packet): + assert bind(binding_packet).tool_catalog_hash == SEALED + + +@pytest.mark.parametrize("case", ["wrong_signed_seal", "missing_seal", "wrong_policy"]) +def test_signed_experimental_profile_refuses_changes_after_good_twin(binding_packet, case): + assert bind(binding_packet).tool_catalog_hash == SEALED + manifest = copy.deepcopy(binding_packet[0]) + if case == "wrong_signed_seal": + manifest["evidence_requirements"]["components"][1]["expected_observed_digest"] = ( + "sha256:" + "d" * 64 + ) + with pytest.raises(ConfigError, match="does not bind the sealed runtime catalog"): + bind(binding_packet, manifest=manifest) + elif case == "missing_seal": + manifest["evidence_requirements"]["components"][1].pop("expected_observed_digest") + with pytest.raises(ConfigError, match="does not bind the sealed runtime catalog"): + bind(binding_packet, manifest=manifest) + else: + with pytest.raises(ConfigError, match="policy bundle hash does not match runtime policy"): + bind(binding_packet, policy="sha256:" + "e" * 64) From 7900a5139b24c6547a11d88209adea5449db9e78 Mon Sep 17 00:00:00 2001 From: Val Rukhaylo Date: Tue, 13 Oct 2026 05:42:10 +0000 Subject: [PATCH 2/5] test: cover experimental signed catalog at actual startup gate --- ...test_experimental_manifest_binding_edrg.py | 68 +++++++++++++++++++ 1 file changed, 68 insertions(+) diff --git a/tests/unit/test_experimental_manifest_binding_edrg.py b/tests/unit/test_experimental_manifest_binding_edrg.py index 0bca1176..5a13bb75 100644 --- a/tests/unit/test_experimental_manifest_binding_edrg.py +++ b/tests/unit/test_experimental_manifest_binding_edrg.py @@ -126,3 +126,71 @@ def test_signed_experimental_profile_refuses_changes_after_good_twin(binding_pac else: with pytest.raises(ConfigError, match="policy bundle hash does not match runtime policy"): bind(binding_packet, policy="sha256:" + "e" * 64) + + +@pytest.fixture +def experimental_startup(tmp_path, monkeypatch, binding_packet): + """Real startup configuration using the generated COSE profile fixture.""" + import base64 + import json + from pathlib import Path + + from cmcp_runtime.catalog.loader import load_catalog + from cmcp_runtime.policy.bundle import load_policy_bundle + from tests.unit.test_startup import complete_setup as setup_fixture + + config_path = Path(setup_fixture.__wrapped__(tmp_path, monkeypatch)) + runtime_catalog = load_catalog(str(tmp_path / "catalog.json")) + runtime_policy = load_policy_bundle(str(tmp_path / "policy")) + manifest, key, _ = binding_packet + manifest = copy.deepcopy(manifest) + manifest["artifacts"]["policy_bundle"]["hash"] = runtime_policy.bundle_hash + manifest["artifacts"]["tool_manifest"] = { + **manifest_catalog_binding(runtime_catalog), + "rug_pull_policy": "deny-and-alert", + "allow_dynamic_registration": False, + "bound_at": manifest["issued_at"], + } + manifest["evidence_requirements"]["components"][1]["expected_observed_digest"] = ( + runtime_catalog.catalog_hash + ) + manifest_path = tmp_path / "experimental.cose" + key_path = tmp_path / "experimental-trust.json" + key_path.write_text( + json.dumps( + { + "key_id": key.key_id, + "public_key_base64url": base64.urlsafe_b64encode(key.public_bytes) + .decode() + .rstrip("="), + } + ) + ) + config_path.write_text( + config_path.read_text() + + "\nagent_manifest:\n" + + f" path: {manifest_path}\n" + + f" trust_anchor_path: {key_path}\n" + + f" authenticated_subject: {manifest['agent_id']}\n" + ) + return config_path, manifest_path, manifest, key + + +@pytest.mark.parametrize("wrong_seal", [False, True]) +def test_experimental_signed_catalog_is_enforced_by_real_startup(experimental_startup, wrong_seal): + from cmcp_runtime.startup import run_startup + + config_path, manifest_path, manifest, key = experimental_startup + if wrong_seal: + manifest["evidence_requirements"]["components"][1]["expected_observed_digest"] = ( + "sha256:" + "d" * 64 + ) + manifest_path.write_bytes(sdk.sign_cose_sign1(manifest, key)) + if wrong_seal: + with pytest.raises(SystemExit) as exc: + run_startup(str(config_path)) + assert exc.value.code == 1 + else: + ctx = run_startup(str(config_path)) + assert ctx.agent_manifest is not None + assert ctx.agent_manifest.agent_id == manifest["agent_id"] From bfc832a288c3bc5c6b12c8e719db255a7c7792d8 Mon Sep 17 00:00:00 2001 From: Val Rukhaylo Date: Tue, 13 Oct 2026 05:45:34 +0000 Subject: [PATCH 3/5] test: refuse changed server and output schema at startup --- ...test_experimental_manifest_binding_edrg.py | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/tests/unit/test_experimental_manifest_binding_edrg.py b/tests/unit/test_experimental_manifest_binding_edrg.py index 5a13bb75..f9aa7e7c 100644 --- a/tests/unit/test_experimental_manifest_binding_edrg.py +++ b/tests/unit/test_experimental_manifest_binding_edrg.py @@ -194,3 +194,36 @@ def test_experimental_signed_catalog_is_enforced_by_real_startup(experimental_st ctx = run_startup(str(config_path)) assert ctx.agent_manifest is not None assert ctx.agent_manifest.agent_id == manifest["agent_id"] + + +@pytest.mark.parametrize("substitution", ["server_identity", "output_schema"]) +def test_startup_refuses_substituted_served_catalog(experimental_startup, substitution): + """A newly loaded catalog cannot silently supersede the signed sealed digest.""" + import json + + from cmcp_runtime.catalog.loader import _compute_definition_hash, load_catalog + from cmcp_runtime.startup import run_startup + + config_path, manifest_path, manifest, key = experimental_startup + manifest_path.write_bytes(sdk.sign_cose_sign1(manifest, key)) + + catalog_path = config_path.parent / "catalog.json" + entries = json.loads(catalog_path.read_text()) + if substitution == "server_identity": + entries[0]["server"]["url"] = "https://substituted.example.test/mcp" + else: + definition = entries[0]["approved_definition"] + definition["output_schema"] = { + "type": "object", + "properties": {"changed": {"type": "string"}}, + } + entries[0]["definition_hash"] = _compute_definition_hash(definition) + catalog_path.write_text(json.dumps(entries)) + assert ( + load_catalog(str(catalog_path)).catalog_hash + != manifest["evidence_requirements"]["components"][1]["expected_observed_digest"] + ) + + with pytest.raises(SystemExit) as exc: + run_startup(str(config_path)) + assert exc.value.code == 1 From 0581739fcf02649ad6b21cf298a74f76a1520fce Mon Sep 17 00:00:00 2001 From: Val Rukhaylo Date: Tue, 13 Oct 2026 05:48:01 +0000 Subject: [PATCH 4/5] test: verify synthetic dispatch only after signed startup admission --- ...test_experimental_manifest_binding_edrg.py | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/tests/unit/test_experimental_manifest_binding_edrg.py b/tests/unit/test_experimental_manifest_binding_edrg.py index f9aa7e7c..9366443f 100644 --- a/tests/unit/test_experimental_manifest_binding_edrg.py +++ b/tests/unit/test_experimental_manifest_binding_edrg.py @@ -227,3 +227,35 @@ def test_startup_refuses_substituted_served_catalog(experimental_startup, substi with pytest.raises(SystemExit) as exc: run_startup(str(config_path)) assert exc.value.code == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("wrong_seal", [False, True]) +async def test_synthetic_dispatch_only_after_authenticated_startup( + experimental_startup, wrong_seal +): + """One actual proxy forwarding invocation only after the startup binding succeeds.""" + from cmcp_runtime.startup import run_startup + from tests.unit.test_mcp_proxy import _make_proxy + + config_path, manifest_path, manifest, key = experimental_startup + if wrong_seal: + manifest["evidence_requirements"]["components"][1]["expected_observed_digest"] = ( + "sha256:" + "d" * 64 + ) + manifest_path.write_bytes(sdk.sign_cose_sign1(manifest, key)) + + forwarded = 0 + if wrong_seal: + with pytest.raises(SystemExit) as exc: + run_startup(str(config_path)) + assert exc.value.code == 1 + else: + ctx = run_startup(str(config_path)) + assert ctx.agent_manifest is not None + proxy, _, _ = _make_proxy(catalog=ctx.catalog) + response = await proxy.call_tool("synthetic-call-1", "test.tool", {}) + assert response.allowed is True + forwarded = proxy._forward_to_upstream.await_count + + assert forwarded == (0 if wrong_seal else 1) From c2fe10d16c692cb01397e897b6f70088c037f1b6 Mon Sep 17 00:00:00 2001 From: Val Rukhaylo Date: Tue, 13 Oct 2026 14:20:39 +0000 Subject: [PATCH 5/5] test: integrate independent CLI composition and dispatch boundary controls Based on dinakarjs/cmcp PR #2, credited to independent reviewer. --- ...test_experimental_manifest_cli_boundary.py | 113 ++++++++++++++++++ 1 file changed, 113 insertions(+) create mode 100644 tests/unit/test_experimental_manifest_cli_boundary.py diff --git a/tests/unit/test_experimental_manifest_cli_boundary.py b/tests/unit/test_experimental_manifest_cli_boundary.py new file mode 100644 index 00000000..0de63347 --- /dev/null +++ b/tests/unit/test_experimental_manifest_cli_boundary.py @@ -0,0 +1,113 @@ +"""Reviewer-only CLI composition controls; synthetic state and mocked transport.""" + +import asyncio +import json +from datetime import UTC, datetime, timedelta +from unittest.mock import AsyncMock + +import agent_manifest as sdk +import pytest +from click.testing import CliRunner + +from cmcp_runtime import cli +from cmcp_runtime.mcp.proxy import CMCPProxy +from tests.unit.conftest import wire_mock_gateway + +pytest_plugins = ("tests.unit.test_experimental_manifest_binding_edrg",) + + +@pytest.mark.parametrize( + "case", + [ + "valid", + "wrong_seal", + "missing_seal", + "wrong_policy", + "wrong_subject", + "expired", + "unknown_key", + "missing_requirements", + "duplicate_component", + "unknown_requirement_field", + "signature_tamper", + "catalog_server", + ], +) +def test_cli_composition_boundary(experimental_startup, monkeypatch, case): + config, manifest_path, manifest, key = experimental_startup + if case == "wrong_seal": + manifest["evidence_requirements"]["components"][1]["expected_observed_digest"] = ( + "sha256:" + "d" * 64 + ) + elif case == "missing_seal": + manifest["evidence_requirements"]["components"][1].pop("expected_observed_digest") + elif case == "wrong_policy": + manifest["artifacts"]["policy_bundle"]["hash"] = "sha256:" + "e" * 64 + elif case == "wrong_subject": + config.write_text( + config.read_text().replace(manifest["agent_id"], "spiffe://example.test/agent/other") + ) + elif case == "expired": + manifest["issued_at"] = (datetime.now(UTC) - timedelta(hours=4)).isoformat() + manifest["expires_at"] = (datetime.now(UTC) - timedelta(hours=2)).isoformat() + elif case == "unknown_key": + (config.parent / "experimental-trust.json").write_text("{}") + elif case == "missing_requirements": + manifest.pop("evidence_requirements") + elif case == "duplicate_component": + import copy + + manifest["evidence_requirements"]["components"].append( + copy.deepcopy(manifest["evidence_requirements"]["components"][0]) + ) + elif case == "unknown_requirement_field": + manifest["evidence_requirements"]["components"][0]["unknown_rule"] = True + elif case == "catalog_server": + p = config.parent / "catalog.json" + catalog = json.loads(p.read_text()) + catalog[0]["server"]["url"] = "https://substituted.example.test/mcp" + p.write_text(json.dumps(catalog)) + signed = sdk.sign_cose_sign1(manifest, key) + if case == "signature_tamper": + signed = signed[:-1] + bytes([signed[-1] ^ 1]) + manifest_path.write_bytes(signed) + + # Observe the production CLI's build_server call, rather than assigning a + # negative counter ourselves. Successful composition uses the real builder. + observed = [] + constructed = [] + forward = AsyncMock(return_value="synthetic response") + original = cli.build_server + original_init = CMCPProxy.__init__ + + def construct(self, *args, **kwargs): + constructed.append(self) + original_init(self, *args, **kwargs) + + monkeypatch.setattr(CMCPProxy, "__init__", construct) + monkeypatch.setattr(CMCPProxy, "_forward_to_upstream", forward) + + def build(ctx): + server = original(ctx) + wire_mock_gateway(server._proxy) + server._proxy._forward_to_upstream = forward + observed.append(server) + return server + + def serve(_app, **_kwargs): + result = asyncio.run(observed[-1]._proxy.call_tool("review-control", "test.tool", {})) + assert result.allowed + + monkeypatch.setattr(cli, "build_server", build) + monkeypatch.setattr("uvicorn.run", serve) + result = CliRunner().invoke(cli.main, ["start", "--config", str(config)]) + if case == "valid": + assert result.exit_code == 0, repr(result.exception) + result.output + assert len(observed) == 1 + assert len(constructed) == 1 + assert forward.await_count == 1 + else: + assert result.exit_code == 1, repr(result.exception) + result.output + assert observed == [] + assert constructed == [] + assert forward.await_count == 0