feat: optional independent signed catalog authority receipt

docs/catalog-authority-bridge.md

@@ -0,0 +1,40 @@
+# Agent Manifest / cMCP Catalog Authority Bridge (v1)
+
+## Purpose and trust boundary
+
+The Agent Manifest SDK verifies a Merkle root of its tool entries. cMCP independently hashes its complete approved catalog as canonical JSON. These are **different measurements**. Never substitute one for the other or insert custom fields into the Agent Manifest schema.
+
+When both `agent_manifest.catalog_bridge_path` and `agent_manifest.catalog_bridge_trust_anchor_path` are configured, startup additionally requires a signed catalog authority receipt. Without those optional settings, upstream native Agent Manifest tool binding remains mandatory. The bridge is **not** hardware attestation. It is an operator-issued software signature binding the Agent Manifest, approved catalog, policy and tool Merkle root. The Agent Manifest signature and SDK validation remain mandatory. The bridge is not a substitute for those checks.
+
+## Deployment sequence
+
+1. Validate and freeze the approved cMCP catalog and policy bundle. Record the exact cMCP runtime catalog hash and policy bundle hash.
+2. Translate the approved tools using upstream `cmcp_runtime.manifest_catalog.manifest_catalog_binding(catalog)`. This binds tool IDs, RFC 8785-canonicalized input schemas, and UTF-8 descriptions. **Do not** use the cMCP catalog hash as the Agent Manifest Merkle root.
+3. Issue an Agent Manifest with exactly those tool entries and the calculated Merkle root. Sign it with the Agent Manifest issuer key using the supported Agent Manifest signing procedure; verify it independently with the installed SDK.
+4. Using a **separate authorized bridge issuer key**, construct a receipt payload with the exact fields `version` (integer `1`), `manifest_id`, `manifest_digest`, `agent_id`, `policy_hash`, `runtime_catalog_hash`, `manifest_catalog_root`, `not_before`, `expires_at`, and `key_id`. All digest values use lowercase `sha256:` plus 64 hexadecimal digits; `key_id` is the SHA-256 hex fingerprint of the trusted 32-byte Ed25519 public key. The `manifest_digest` is SHA-256 of the signed Agent Manifest canonical signing preimage for JSON manifests, or of the COSE envelope bytes for COSE manifests, as read by cMCP.
+5. Sign this payload with `sign_bridge(payload, issuer_private_key)` in a controlled issuance environment. This emits `{"payload": ..., "signature": "<base64url-no-padding>"}`. Do **not** store the private signing key on the gateway. Write the receipt as JSON and provision its public key in the same trust-anchor JSON format accepted by `load_agent_manifest_trust_anchor`.
+6. Optionally add the bridge settings alongside existing `agent_manifest.path`, `agent_manifest.trust_anchor_path`, and `agent_manifest.authenticated_subject`:
+
+```yaml
+agent_manifest:
+  path: /etc/cmcp/agent-manifest.json
+  trust_anchor_path: /etc/cmcp/agent-manifest-issuer.json
+  authenticated_subject: spiffe://example/agent/production
+  catalog_bridge_path: /etc/cmcp/catalog-authority-receipt.json
+  catalog_bridge_trust_anchor_path: /etc/cmcp/catalog-authority-issuer.json
+  revocation_list_path: /etc/cmcp/manifest-revocations.jsonl
+```
+
+7. Restart in a staging environment and verify both acceptance of the exact approved configuration and fail-closed behavior when any measured artifact is modified. When a bridge is configured, treat missing or invalid receipts as deployment failures, not warnings.
+
+## Rotation and recovery
+
+Rotate the manifest, policy or catalog by generating a new manifest **and** a newly signed bridge receipt from the new frozen measurements. Deploy the corresponding trust anchors atomically with the receipts. To revoke a bridge signing key, remove its public key from the gateway trust-anchor file and restart; old receipts must then fail. Maintain the Agent Manifest revocation list separately: it does not revoke bridge keys. Keep prior signed artifacts in a restricted audit archive, not in the active configuration.
+
+A receipt expires at `expires_at` and cannot be used before `not_before`. There is no remote revocation lookup or online freshness guarantee. Gateway startup checks the local clock, not an independently attested clock; operational controls must protect time synchronization and trusted configuration paths. The current bridge trust-anchor format and startup checks are software-based, not hardware-backed attestation.
+
+## Limitations and assurance evidence
+
+The deterministic mapping binds tool identity, description and input/output schemas. The full cMCP runtime catalog hash separately binds additional catalog metadata, including server configuration. The bridge is a correspondence claim between two *different* hash domains, not proof that a server actually runs the approved implementation. Existing runtime discovery and drift controls remain necessary.
+
+Security validation includes receipt-signature tampering, signer substitution, stale receipts, malformed payloads, catalog changes, policy changes, and Agent Manifest substitutions. Before production use, run the full unit and integration suites and review signing-preimage/COSE compatibility against the installed SDK version. Never label a software-only deployment as TEE/TPM-attested.

src/cmcp_runtime/catalog/authority_bridge.py

@@ -0,0 +1,93 @@
+"""Explicit, signed correspondence between cMCP catalog and Agent Manifest tool root.
+
+This is a separate protocol: it does not extend the Agent Manifest schema.
+The caller must independently verify the Agent Manifest and runtime catalog.
+"""
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+import re
+from datetime import UTC, datetime
+from typing import Any
+
+from cryptography.exceptions import InvalidSignature
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey
+
+from cmcp_runtime.errors import ConfigError
+
+DOMAIN = b"cmcp.catalog-authority-bridge.v1\x00"
+FIELDS = frozenset({"version", "manifest_id", "manifest_digest", "agent_id", "policy_hash", "runtime_catalog_hash", "manifest_catalog_root", "not_before", "expires_at", "key_id"})
+
+
+def _canonical(obj: dict[str, Any]) -> bytes:
+    return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False).encode("ascii")
+
+
+def _validate_payload(payload: dict[str, Any]) -> None:
+    if not isinstance(payload, dict) or set(payload) != FIELDS or type(payload["version"]) is not int or payload["version"] != 1:
+        raise ConfigError("Unsupported catalog bridge payload")
+    for field in FIELDS - {"version"}:
+        if not isinstance(payload[field], str) or not payload[field]:
+            raise ConfigError(f"Invalid catalog bridge field: {field}")
+    for field in ("manifest_digest", "policy_hash", "runtime_catalog_hash", "manifest_catalog_root"):
+        if re.fullmatch(r"sha256:[0-9a-f]{64}", payload[field]) is None:
+            raise ConfigError(f"Invalid catalog bridge digest: {field}")
+    if re.fullmatch(r"[0-9a-f]{64}", payload["key_id"]) is None:
+        raise ConfigError("Invalid catalog bridge key identifier")
+    if _time(payload["not_before"]) >= _time(payload["expires_at"]):
+        raise ConfigError("Invalid catalog bridge validity interval")
+
+
+def _message(payload: dict[str, Any]) -> bytes:
+    _validate_payload(payload)
+    return DOMAIN + _canonical(payload)
+
+
+def _time(value: str) -> datetime:
+    try:
+        result = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except (ValueError, AttributeError) as exc:
+        raise ConfigError("Invalid catalog bridge timestamp") from exc
+    if result.tzinfo is None:
+        raise ConfigError("Catalog bridge timestamp lacks timezone")
+    return result.astimezone(UTC)
+
+
+def sign_bridge(payload: dict[str, Any], private_key: Ed25519PrivateKey) -> dict[str, Any]:
+    """Issuer-side signing; never called automatically during gateway startup."""
+    signature = private_key.sign(_message(payload))
+    return {"payload": payload, "signature": base64.urlsafe_b64encode(signature).rstrip(b"=").decode("ascii")}
+
+
+def verify_bridge(receipt: dict[str, Any], trusted_keys: dict[str, bytes], *,
+                  manifest_id: str, manifest_digest: str, agent_id: str,
+                  policy_hash: str, runtime_catalog_hash: str,
+                  manifest_catalog_root: str, now: datetime | None = None) -> None:
+    if not isinstance(receipt, dict) or set(receipt) != {"payload", "signature"}:
+        raise ConfigError("Invalid catalog bridge envelope")
+    p = receipt["payload"]
+    _validate_payload(p)
+    expected = {"manifest_id": manifest_id, "manifest_digest": manifest_digest,
+                "agent_id": agent_id, "policy_hash": policy_hash,
+                "runtime_catalog_hash": runtime_catalog_hash,
+                "manifest_catalog_root": manifest_catalog_root}
+    if any(p.get(k) != v for k, v in expected.items()):
+        raise ConfigError("Catalog bridge measurement or identity mismatch")
+    current = (now or datetime.now(UTC)).astimezone(UTC)
+    if not _time(p["not_before"]) <= current < _time(p["expires_at"]):
+        raise ConfigError("Catalog bridge outside validity window")
+    key = trusted_keys.get(p["key_id"])
+    if not isinstance(key, bytes) or len(key) != 32:
+        raise ConfigError("Catalog bridge signing key not trusted")
+    sig = receipt["signature"]
+    if not isinstance(sig, str) or not sig or any(c not in 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_' for c in sig):
+        raise ConfigError("Invalid catalog bridge signature encoding")
+    try:
+        raw = base64.urlsafe_b64decode(sig + '=' * (-len(sig) % 4))
+        if len(raw) != 64:
+            raise ValueError("Signature length")
+        Ed25519PublicKey.from_public_bytes(key).verify(raw, _message(p))
+    except (ValueError, InvalidSignature, TypeError) as exc:
+        raise ConfigError("Catalog bridge signature invalid") from exc

src/cmcp_runtime/config.py

@@ -121,6 +121,8 @@ class AgentManifestConfig:
     #: set, a manifest listed there is rejected at startup; a missing or
     #: malformed file aborts startup rather than being read as empty.
     revocation_list_path: str | None = None
+    catalog_bridge_path: str | None = None
+    catalog_bridge_trust_anchor_path: str | None = None
 
 
 @dataclass
@@ -211,6 +213,8 @@ class Config:
     "trust_anchor_path",
     "authenticated_subject",
     "revocation_list_path",
+    "catalog_bridge_path",
+    "catalog_bridge_trust_anchor_path",
 }
 
 
@@ -516,6 +520,8 @@ def load_config(path: str) -> Config:
     trust_anchor_path = manifest_raw.get("trust_anchor_path")
     authenticated_subject = manifest_raw.get("authenticated_subject")
     revocation_list_path = manifest_raw.get("revocation_list_path")
+    bridge_path = manifest_raw.get("catalog_bridge_path")
+    bridge_key_path = manifest_raw.get("catalog_bridge_trust_anchor_path")
     if agent_manifest_path is not None and not isinstance(agent_manifest_path, str):
         raise ConfigError("agent_manifest.path must be a string")
     if trust_anchor_path is not None and not isinstance(trust_anchor_path, str):
@@ -541,6 +547,16 @@ def load_config(path: str) -> Config:
             )
         _check_no_traversal("agent_manifest.revocation_list_path", revocation_list_path)
 
+    if bool(bridge_path) != bool(bridge_key_path):
+        raise ConfigError("Both catalog bridge receipt and trust anchor paths are required")
+    if bridge_path and not agent_manifest_path:
+        raise ConfigError("Catalog bridge requires Agent Manifest binding")
+    for label, path in (("catalog_bridge_path", bridge_path), ("catalog_bridge_trust_anchor_path", bridge_key_path)):
+        if path is not None:
+            if not isinstance(path, str) or not path:
+                raise ConfigError(f"agent_manifest.{label} must be a non-empty string")
+            _check_no_traversal(f"agent_manifest.{label}", path)
+
     profile = raw.get("conformance_profile")
     if profile is not None and (
         not isinstance(profile, str) or profile not in _KNOWN_CONFORMANCE_PROFILES
@@ -576,6 +592,8 @@ def load_config(path: str) -> Config:
             trust_anchor_path=trust_anchor_path,
             authenticated_subject=authenticated_subject,
             revocation_list_path=revocation_list_path,
+            catalog_bridge_path=bridge_path,
+            catalog_bridge_trust_anchor_path=bridge_key_path,
         ),
         kill_switch=KillSwitchConfig(
             enabled=ks_enabled,

src/cmcp_runtime/startup.py

@@ -758,6 +758,32 @@ def run_startup(config_path: str) -> RuntimeContext:
                 allow_dev_subject_from_manifest=config.dev_mode,
                 revocations=revocations,
             )
+            if config.agent_manifest.catalog_bridge_path is not None:
+                from pathlib import Path
+                from cmcp_runtime.agent_manifest import signing_pre_image
+                from cmcp_runtime.catalog.authority_bridge import verify_bridge
+                from cmcp_runtime.manifest_catalog import manifest_catalog_binding
+
+                try:
+                    receipt = json.loads(Path(config.agent_manifest.catalog_bridge_path).read_text())
+                except (OSError, ValueError) as exc:
+                    raise ConfigError("Catalog authority bridge unreadable") from exc
+                bridge_keys = load_agent_manifest_trust_anchor(
+                    config.agent_manifest.catalog_bridge_trust_anchor_path
+                )
+                manifest_digest = "sha256:" + hashlib.sha256(
+                    loaded.envelope if loaded.envelope is not None
+                    else signing_pre_image(loaded.manifest)
+                ).hexdigest()
+                verify_bridge(
+                    receipt, bridge_keys,
+                    manifest_id=loaded.manifest["manifest_id"],
+                    manifest_digest=manifest_digest,
+                    agent_id=loaded.manifest["agent_id"],
+                    policy_hash=policy_bundle.bundle_hash,
+                    runtime_catalog_hash=catalog.catalog_hash,
+                    manifest_catalog_root=manifest_catalog_binding(catalog)["catalog_hash"],
+                )
         except ConfigError as exc:
             _fatal("AGENT_MANIFEST_BINDING_FAILED", str(exc), action="startup_aborted")
             sys.exit(1)

tests/unit/test_catalog_authority_bridge.py

@@ -0,0 +1,108 @@
+"""Negative and positive cases for the independent catalog bridge receipt."""
+from __future__ import annotations
+
+import copy
+import hashlib
+from datetime import UTC, datetime
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
+
+from cmcp_runtime.catalog.authority_bridge import sign_bridge, verify_bridge
+from cmcp_runtime.errors import ConfigError
+
+
+@pytest.fixture
+def case():
+    private = Ed25519PrivateKey.generate()
+    public = private.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
+    key_id = hashlib.sha256(public).hexdigest()
+    payload = dict(version=1, manifest_id='manifest-1', manifest_digest='sha256:'+'a'*64,
+                   agent_id='spiffe://example/agent', policy_hash='sha256:'+'b'*64,
+                   runtime_catalog_hash='sha256:'+'c'*64, manifest_catalog_root='sha256:'+'d'*64,
+                   not_before='2026-01-01T00:00:00Z', expires_at='2027-01-01T00:00:00Z', key_id=key_id)
+    expected = {k:payload[k] for k in ('manifest_id','manifest_digest','agent_id','policy_hash','runtime_catalog_hash','manifest_catalog_root')}
+    return private, {key_id:public}, payload, expected
+
+
+def check(case, receipt, **overrides):
+    _, keys, _, expected = case
+    verify_bridge(receipt, keys, now=datetime(2026,10,10,tzinfo=UTC), **(expected|overrides))
+
+
+def test_valid_receipt(case):
+    private, _, payload, _ = case
+    check(case, sign_bridge(payload, private))
+
+
+@pytest.mark.parametrize('field', ['manifest_id','manifest_digest','agent_id','policy_hash','runtime_catalog_hash','manifest_catalog_root'])
+def test_substituted_context_fails(case, field):
+    private, _, payload, expected = case
+    with pytest.raises(ConfigError):
+        check(case, sign_bridge(payload, private), **{field:expected[field]+'-changed'})
+
+
+@pytest.mark.parametrize('field', ['manifest_id','manifest_digest','agent_id','policy_hash','runtime_catalog_hash','manifest_catalog_root','not_before','expires_at','key_id'])
+def test_modified_signed_payload_fails(case, field):
+    private, _, payload, _ = case
+    receipt = sign_bridge(payload, private)
+    receipt['payload'][field] = 'changed'
+    with pytest.raises(ConfigError):
+        check(case, receipt)
+
+
+def test_untrusted_signer_fails(case):
+    private, keys, payload, expected = case
+    with pytest.raises(ConfigError):
+        verify_bridge(sign_bridge(payload, private), {}, now=datetime(2026,10,10,tzinfo=UTC), **expected)
+
+
+def test_expired_fails(case):
+    private, _, payload, expected = case
+    with pytest.raises(ConfigError):
+        verify_bridge(sign_bridge(payload, private), case[1], now=datetime(2028,1,1,tzinfo=UTC), **expected)
+
+
+def test_extra_fields_fail(case):
+    private, _, payload, _ = case
+    with pytest.raises(ConfigError):
+        sign_bridge(payload|{'unauthorized':True}, private)
+
+
+def test_signature_corruption_fails(case):
+    private, _, payload, _ = case
+    receipt = sign_bridge(payload, private)
+    receipt['signature'] = 'A' * 86
+    with pytest.raises(ConfigError):
+        check(case, receipt)
+
+
+@pytest.mark.parametrize('field,value', [
+    ('version', True), ('version', '1'), ('key_id', 'wrong'),
+    ('manifest_digest', 'sha256:invalid'), ('policy_hash', 'sha256:invalid'),
+    ('runtime_catalog_hash', 'sha256:invalid'), ('manifest_catalog_root', 'sha256:invalid'),
+    ('not_before', 'not-a-time'), ('expires_at', '2025-01-01T00:00:00Z'),
+    ('agent_id', None), ('manifest_id', []),
+])
+def test_malformed_signed_payload_rejected(case, field, value):
+    private, _, payload, _ = case
+    with pytest.raises(ConfigError):
+        sign_bridge(payload | {field: value}, private)
+
+
+def test_receipt_cannot_be_replayed_across_signers(case):
+    private, _, payload, _ = case
+    receipt = sign_bridge(payload, private)
+    different = Ed25519PrivateKey.generate()
+    wrong_public = different.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
+    with pytest.raises(ConfigError):
+        verify_bridge(receipt, {payload['key_id']: wrong_public},
+                      now=datetime(2026,10,10,tzinfo=UTC), **case[3])
+
+
+def test_not_yet_valid_receipt_rejected(case):
+    private, _, payload, _ = case
+    receipt = sign_bridge(payload, private)
+    with pytest.raises(ConfigError):
+        verify_bridge(receipt, case[1], now=datetime(2025,10,10,tzinfo=UTC), **case[3])