fix(agent-os): forgeable MCP signature via ambiguous canonical string

.cspell-repo-terms.txt

@@ -964,3 +964,7 @@ zerofrom
 zerotrie
 zerovec
 zoneinfo
+
+# --- Canonical-encoding math terms ---
+injective
+injectively

BREAKING_CHANGES.md

@@ -5,6 +5,62 @@ entries appear first.
 
 ---
 
+## MCP message signatures use a new canonical string; signers and verifiers must be upgraded together
+
+**Date:** TBD
+
+**Affected**
+
+- anything using `agent_os.MCPMessageSigner` to sign or verify MCP envelopes
+- envelopes already signed and stored, if they are verified after the upgrade
+- a deployment where signer and verifier are separate processes, upgraded
+  independently
+
+**What changed**
+
+`_build_canonical_string` joined the signed fields with a plain `|`:
+
+```python
+f"{nonce}|{timestamp_ms}|{sender_id or ''}|{payload}"
+```
+
+`|` is legal inside a payload and inside a sender id, so the encoding was not
+injective — distinct field tuples produced the same canonical string and
+therefore the same HMAC. Every field is now length-prefixed and `None` gets a
+marker distinct from the empty string.
+
+**Why it cannot be made compatible**
+
+The old format is what the forgery exploits. Accepting it as a fallback during
+verification would keep the vulnerability, because an attacker chooses which
+format their envelope claims to be. There is therefore no migration window: this
+is a correctness change to what the signature means, not a format upgrade.
+
+**What consumers need to do**
+
+- Upgrade signers and verifiers together. An envelope signed by an older version
+  does not verify against this one, and vice versa; verification returns a
+  failure result rather than raising.
+- Re-sign anything stored signed and verified later. There is no way to convert
+  an old signature without the signing key.
+- No API change. `sign_message` and `verify_message` keep their signatures and
+  their return types; only the bytes under the HMAC differ.
+
+**Security impact if not upgraded**
+
+Under the old format an attacker holding one valid envelope could forge others
+without the key by moving text across a field boundary, including prepending
+attacker-chosen text to the payload a consumer acts on:
+
+```
+signed:  sender_id="alice|INJECTED", payload="x"
+forged:  sender_id="alice",          payload="INJECTED|x"   -> verified
+```
+
+An envelope signed with `sender_id=None` also verified as one sent by `""`.
+
+---
+
 ## `HostSession.post_tool_call` and `pre_model_call` emit the adapter snapshot shape
 
 **Date:** TBD

agent-governance-python/agent-os/src/agent_os/mcp_message_signer.py

@@ -273,15 +273,49 @@ def _compute_signature(
         return base64.b64encode(digest).decode("ascii")
 
     @staticmethod
+    def _canonical_field(value: str | None) -> str:
+        """Return *value* framed so it cannot be confused with its neighbors.
+
+        Each field is prefixed with its own length, so a separator appearing
+        inside a field is just one of that field's characters and cannot be read
+        as the end of it. ``None`` gets a distinct marker rather than being
+        folded into the empty string, so an absent sender and an empty sender
+        do not sign identically.
+        """
+        if value is None:
+            return "-"
+        return f"{len(value)}:{value}"
+
+    @classmethod
     def _build_canonical_string(
+        cls,
         *,
         nonce: str,
         timestamp: datetime,
         sender_id: str | None,
         payload: str,
     ) -> str:
+        """Return the injective canonical string that the HMAC covers.
+
+        Concatenating the fields with a plain ``|`` separator was ambiguous:
+        ``|`` is legal inside a payload and inside a sender id, so distinct
+        (nonce, timestamp, sender, payload) tuples produced the same canonical
+        string and therefore the same signature. An attacker holding one valid
+        envelope could move text across a field boundary and the forgery
+        verified -- e.g. ``sender="alice"`` with ``payload="alpha|beta"`` and
+        ``sender="alice|alpha"`` with ``payload="beta"`` both canonicalize to
+        ``alice|alpha|beta`` in the trailing fields.
+
+        Length-prefixing every field makes the encoding injective: the reader of
+        each field is told how many characters it has before reading them, so no
+        boundary can move. The timestamp is rendered before framing so it is
+        covered the same way as the rest.
+        """
         timestamp_ms = int(timestamp.timestamp() * 1000)
-        return f"{nonce}|{timestamp_ms}|{sender_id or ''}|{payload}"
+        return "|".join(
+            cls._canonical_field(field)
+            for field in (nonce, str(timestamp_ms), sender_id, payload)
+        )
 
     def _maybe_cleanup_locked(self, now: datetime) -> None:
         if now - self._last_cleanup >= self.nonce_cache_cleanup_interval:

agent-governance-python/agent-os/tests/test_mcp_message_signer.py

@@ -43,6 +43,113 @@ def test_verify_detects_tampered_payload():
     assert "Invalid signature" in result.failure_reason
 
 
+def _reframe(envelope: MCPSignedEnvelope, *, payload=..., sender_id=...):
+    """Return *envelope* with fields moved but the signature kept as-is.
+
+    Models the whole of the attacker's capability: they hold one valid envelope
+    and may rewrite its fields, but cannot compute a new signature.
+    """
+    return MCPSignedEnvelope(
+        payload=envelope.payload if payload is ... else payload,
+        nonce=envelope.nonce,
+        timestamp=envelope.timestamp,
+        sender_id=envelope.sender_id if sender_id is ... else sender_id,
+        signature=envelope.signature,
+    )
+
+
+class TestCanonicalStringIsInjective:
+    """Field boundaries must not be movable inside the signed canonical string.
+
+    The canonical string was ``f"{nonce}|{timestamp_ms}|{sender_id or ''}|{payload}"``.
+    ``|`` is legal inside a payload and inside a sender id, so distinct
+    (nonce, timestamp, sender, payload) tuples collapsed to the same canonical
+    string and therefore the same HMAC. Holding one valid envelope was enough to
+    forge others: move text across a boundary and the signature still verified.
+
+    Each case uses a fresh verifier, since a real receiver has its own nonce
+    store and would not have seen the original envelope's nonce.
+    """
+
+    KEY = b"k" * 32
+
+    def _signer(self) -> MCPMessageSigner:
+        return MCPMessageSigner(self.KEY)
+
+    def test_text_moved_from_payload_into_sender_id(self):
+        # sender="alice" + payload="alpha|beta" and sender="alice|alpha" +
+        # payload="beta" both ended "...|alice|alpha|beta".
+        envelope = self._signer().sign_message("alpha|beta", sender_id="alice")
+        forged = _reframe(envelope, payload="beta", sender_id="alice|alpha")
+
+        result = self._signer().verify_message(forged)
+
+        assert result.is_valid is False
+        assert "Invalid signature" in result.failure_reason
+
+    def test_text_moved_from_sender_id_into_payload(self):
+        # The same shift in the other direction, which is the dangerous one: it
+        # prepends attacker-chosen text to the payload a consumer will act on.
+        envelope = self._signer().sign_message("x", sender_id="alice|INJECTED")
+        forged = _reframe(envelope, payload="INJECTED|x", sender_id="alice")
+
+        result = self._signer().verify_message(forged)
+
+        assert result.is_valid is False
+        assert result.payload is None
+
+    def test_absent_sender_is_not_an_empty_sender(self):
+        # ``sender_id or ''`` erased the difference, so an envelope signed with
+        # no sender verified as one sent by "".
+        envelope = self._signer().sign_message("p", sender_id=None)
+        forged = _reframe(envelope, sender_id="")
+
+        assert self._signer().verify_message(forged).is_valid is False
+
+    @pytest.mark.parametrize(
+        ("payload", "sender_id"),
+        [
+            ("plain", "agent-1"),
+            ("a|b|c", "x|y"),  # separator inside both fields
+            ("3:abc", "5:hello"),  # content shaped like the length prefix itself
+            ("-", "-"),  # content equal to the absent-value marker
+            ("{}", None),
+            ("unicode 中文 \U0001f600", "中文"),
+        ],
+    )
+    def test_round_trip_survives_framing(self, payload, sender_id):
+        # Fixing the ambiguity must not break any legitimate field content,
+        # including content that mimics the framing.
+        signer = self._signer()
+        result = signer.verify_message(signer.sign_message(payload, sender_id=sender_id))
+
+        assert result.is_valid is True
+        assert result.payload == payload
+        assert result.sender_id == sender_id
+
+    def test_distinct_field_tuples_never_share_a_canonical_string(self):
+        # The property the fix rests on: the encoding is injective, so two
+        # different tuples can never produce one signature.
+        timestamp = self._signer().sign_message("seed").timestamp
+        awkward = ["", "a", "|", "a|", "|a", "a|b", "1:a", "-", "2:ab"]
+        seen: dict[str, tuple] = {}
+        for nonce in ("n", "n|", "1:n"):
+            for sender_id in [*awkward, None]:
+                for payload in awkward:
+                    canonical = MCPMessageSigner._build_canonical_string(
+                        nonce=nonce,
+                        timestamp=timestamp,
+                        sender_id=sender_id,
+                        payload=payload,
+                    )
+                    key = (nonce, sender_id, payload)
+                    collided = seen.setdefault(canonical, key)
+                    assert collided == key, (
+                        f"collision: {collided} and {key} both encode to {canonical!r}"
+                    )
+        assert len(seen) == 3 * 10 * 9
+
+
 def test_verify_rejects_replay():
     signer = MCPMessageSigner(MCPMessageSigner.generate_key())
     envelope = signer.sign_message('{"method":"safe"}')

agent-governance-python/agent-os/tests/test_spec_mcp_gateway_conformance.py

@@ -10,7 +10,7 @@
 from __future__ import annotations
 import unittest
 import warnings
-from datetime import timedelta
+from datetime import datetime, timedelta, timezone
 from unittest.mock import MagicMock, patch
 from agent_control_specification import Decision, InterventionPointResult, Verdict
 from agent_os.mcp_gateway import ApprovalStatus, MCPGateway, ResponsePolicy
@@ -319,6 +319,51 @@ def test_envelope_fields(self):
         self.assertIsNotNone(envelope.signature)
         self.assertEqual(envelope.sender_id, 'agent-1')
 
+    def test_canonical_string_matches_the_spec(self):
+        """S7.4 -- Canonical string is the length-framed encoding in the spec.
+
+        Asserted against the literal from the spec rather than against
+        whatever the implementation produces. Nothing else here reads the
+        canonical string, which is how the spec came to document a plain
+        concatenation the code had stopped using -- other SDKs sign
+        against the spec, so a silent divergence makes them incompatible.
+        """
+        canonical = MCPMessageSigner._build_canonical_string(
+            nonce='n1',
+            timestamp=datetime.fromtimestamp(1785448158.059, tz=timezone.utc),
+            sender_id='alice',
+            payload='{"a":1}',
+        )
+        self.assertEqual(canonical, '2:n1|13:1785448158059|5:alice|7:{"a":1}')
+
+    def test_absent_and_empty_sender_sign_differently(self):
+        """S7.4 -- A null sender_id uses the marker, not the empty string."""
+        args = {
+            'nonce': 'n',
+            'timestamp': datetime(2026, 1, 1, tzinfo=timezone.utc),
+            'payload': 'p',
+        }
+        absent = MCPMessageSigner._build_canonical_string(sender_id=None, **args)
+        empty = MCPMessageSigner._build_canonical_string(sender_id='', **args)
+        self.assertEqual(absent, '1:n|13:1767225600000|-|1:p')
+        self.assertEqual(empty, '1:n|13:1767225600000|0:|1:p')
+
+    def test_signature_is_base64_encoded(self):
+        """S7.4 -- The signature is base64, not hex.
+
+        A hex-encoded 32-byte digest is also a valid base64 alphabet
+        string, so this checks the round trip and the length rather than
+        the character set.
+        """
+        import base64
+
+        signature = self.signer.sign_message('data').signature
+        self.assertEqual(
+            base64.b64encode(base64.b64decode(signature)).decode('ascii'),
+            signature,
+        )
+        self.assertEqual(len(signature), 44)
+
     def test_tampered_signature_rejected(self):
         """S7.5 -- Tampered signature is rejected."""
         envelope = self.signer.sign_message('data')

docs/specs/MCP-SECURITY-GATEWAY-1.0.md

@@ -580,23 +580,59 @@ A signed envelope MUST contain:
 | --- | --- | --- | --- | --- |
 | `payload` | string | Yes | -- | The message payload (JSON string) |
 | `nonce` | string | Yes | -- | Unique nonce for replay protection |
-| `timestamp` | string | Yes | -- | ISO 8601 UTC timestamp |
-| `signature` | string | Yes | -- | HMAC-SHA256 signature (hex-encoded) |
+| `timestamp` | timestamp | Yes | -- | Timezone-aware UTC instant; ISO 8601 when serialized |
+| `signature` | string | Yes | -- | HMAC-SHA256 signature (base64-encoded) |
 | `sender_id` | string or null | No | null | Optional sender identifier |
 
 **[Pure Specification]**
 
 ### 7.4 Signature Computation
 
 The HMAC-SHA256 signature MUST be computed over a canonical string
-constructed by concatenating the following fields with a separator:
+that encodes the fields *injectively*: no two distinct
+(nonce, timestamp, sender_id, payload) tuples may produce the same
+canonical string. Plain concatenation, with or without a separator,
+does not satisfy this -- any separator character is also legal inside a
+payload and inside a sender id, so a field boundary can be moved and
+the signature still verifies.
+
+Each field MUST therefore be framed with its own length in characters,
+and the fields MUST be joined in the order below:
+
+```
+field(v)  = "-"                    if v is null
+          = len(v) + ":" + v       otherwise
+
+canonical = field(nonce)     + "|" +
+            field(timestamp) + "|" +
+            field(sender_id) + "|" +
+            field(payload)
+
+signature = HMAC-SHA256(signing_key, UTF-8(canonical))
+```
+
+`timestamp` MUST be rendered as integer milliseconds since the Unix
+epoch before framing, so that it is covered exactly like the other
+fields.
+
+A null `sender_id` MUST use the `-` marker rather than being folded
+into the empty string, so that an absent sender and a present-but-empty
+sender do not sign identically.
+
+The signature MUST be base64-encoded. **[Pure Specification]**
+
+Example, for nonce `n1`, timestamp `1785448158059`, sender `alice` and
+payload `{"a":1}`:
 
 ```
-canonical = payload + nonce + timestamp + (sender_id or "")
-signature = HMAC-SHA256(signing_key, canonical)
+2:n1|13:1785448158059|5:alice|7:{"a":1}
 ```
 
-The signature MUST be hex-encoded. **[Pure Specification]**
+Implementations MUST NOT accept signatures computed over the earlier
+plain-concatenation form. Accepting both would preserve the forgery: an
+attacker chooses which form their envelope claims to be, so a verifier
+that still honours the old form remains exploitable. See
+`BREAKING_CHANGES.md`.
 
 ### 7.5 Sign Message