Secure agent-to-agent communication channels

Introduction

When agents hand off work across a pipeline — planner to researcher to code-writer to deployment — each hop is an opportunity for a compromised intermediate agent to inject malicious payloads, spoof identity, or replay captured messages. Transport-layer TLS secures the wire but cannot prove which agent authored a message. By the end of this lesson, you'll be able to implement signed A2A envelopes, a gateway that enforces signature verification, replay protection, topology policy, and content guards — giving every inter-agent channel cryptographic accountability at both the transport and application layers.

Key Terminology

• A2A envelope — the signed message container (A2AEnvelope) that bundles sender_agent_id, receiver_agent_id, sent_at, nonce, and payload with a signature field, so every inter-agent message carries its own proof of authorship rather than relying solely on transport-layer identity.
• Canonical serialization — the deterministic JSON encoding produced by _canonical_bytes using sort_keys=True and no extra whitespace, ensuring that every field of the envelope hashes to the same byte sequence regardless of key-insertion order, so the signature is reproducible by any verifier.
• HMAC-SHA256 signature — an application-layer message authentication code computed over the canonical envelope bytes with a shared secret key; the gateway re-derives the expected MAC and compares it using hmac.compare_digest (constant-time) to prevent timing-based forgery.
• Replay nonce — a UUID generated once per envelope in make_envelope and recorded in _seen_nonces on first acceptance; the gateway rejects any envelope that reuses a seen nonce or whose sent_at timestamp is older than _MAX_AGE_SECONDS, closing the replay window to 60 seconds.
• Topology policy — the POLICY dict that maps each sender agent to the set of receiver agents it may legitimately reach (e.g., "planner" → {"researcher", "code-writer"}), enforcing a least-privilege directed communication graph across the pipeline so no agent can contact an undeclared peer.
• Content guard — the _content_guard function that serializes the payload to JSON and scans for known prompt-injection phrases such as "ignore previous" or "override system", blocking a compromised agent from smuggling instruction-hijacking text through a gateway that would otherwise accept a correctly signed envelope.

Concepts

Why Transport Security Is Not Enough

TLS secures the wire — it prevents eavesdropping and ensures the connection reaches the intended server. What it cannot do is prove which agent authored the message. A compromised sidecar, a misconfigured service mesh, or any process that obtained a valid certificate speaks TLS correctly while delivering forged content. Transport identity says "this call came from pod X"; it says nothing about whether the agent running in that pod wrote the message or whether an attacker injected it mid-hop.

The lesson's answer is a second, independent identity layer at the application level: each agent signs its outbound envelope with sign_envelope before the message leaves the process, and the gateway verifies that signature before acting on anything. The two layers address distinct threat models — TLS guards the channel; the HMAC signature guards the message's claimed authorship. Stripping either layer leaves the other unable to compensate.

The Four-Check Gateway and Why Order Matters

The gateway_forward function runs its checks in a deliberate sequence: signature → replay → topology policy → content guard. This ordering is both an economic and a security decision.

Signature verification is the cheapest gate — one HMAC computation eliminates spoofed or corrupted traffic before any stateful lookup occurs. Replay protection comes second because it requires a cache read; a message that fails the signature check never touches the nonce store. Topology policy is third — consulting the POLICY table is pointless if the sender could not have legitimately signed the envelope. Content scanning comes last because it is the most expensive step (full payload serialization plus string search) and is only a meaningful threat from a sender that cleared identity and routing checks.

This strict ordering (see Code Walkthrough) means the gateway fails fast on the cheapest signals and defers the costliest work to the fewest surviving messages.

Topology Policy as Least-Privilege for Agent Communication

The POLICY dict encodes a directed communication graph: planner may reach researcher and code-writer; researcher may reach code-writer; code-writer may reach deployment. Any pair not listed is denied — deployment cannot call back to planner, and no agent can contact an undeclared peer.

This allowlist-of-pairs model applies to agent routing the same least-privilege principle that firewall rules apply to network traffic. Its blast-radius benefit is concrete: if researcher is compromised, it cannot reach deployment directly — every such attempt produces a "policy_violation" denial at the gateway before any payload is processed. Keeping the table narrow, and treating every new directed edge as requiring review, limits how far a single compromised agent can propagate damage across the pipeline.

Replay Protection: Why Nonces and Timestamps Must Work Together

A captured envelope with a valid signature remains exploitable indefinitely unless the gateway has a second line of defense. Two complementary controls close that window: a per-message nonce (UUID) that is recorded in _seen_nonces on first acceptance, and a timestamp age check that rejects any envelope older than _MAX_AGE_SECONDS.

Neither control alone suffices. Nonces without timestamps require the cache to grow unboundedly — every nonce from every envelope ever accepted must be retained forever. Timestamps without nonces allow a replay at any moment within the acceptance window, which is long enough for an attacker to resubmit a captured message. Together they bound the replay window to 60 seconds and keep the nonce cache bounded to messages seen within that same window (see Code Walkthrough).

Code Walkthrough

Now that you understand signed envelopes, replay protection, gateway verification, and topology policy, here is how those pieces fit together in working code.

Signing an Outbound Envelope

Each agent wraps its outbound message in a canonicalized, signed envelope. The signature covers every meaningful field so a single altered byte invalidates it.

Code snippetpython
1import base64
2import hashlib
3import hmac
4import json
5from dataclasses import dataclass, field
6from datetime import datetime, timezone
7from uuid import uuid4
8
9@dataclass
10class A2AEnvelope:
11    sender_agent_id: str
12    receiver_agent_id: str
13    sent_at: str          # ISO-8601 UTC
14    nonce: str
15    payload: dict
16    signature: str = ""
17
18def _canonical_bytes(env: A2AEnvelope) -> bytes:
19    doc = {
20        "sender_agent_id": env.sender_agent_id,
21        "receiver_agent_id": env.receiver_agent_id,
22        "sent_at": env.sent_at,
23        "nonce": env.nonce,
24        "payload": env.payload,
25    }
26    return json.dumps(doc, sort_keys=True, separators=(",", ":")).encode()
27
28def sign_envelope(env: A2AEnvelope, secret_key: bytes) -> A2AEnvelope:
29    """HMAC-SHA256 over the canonical envelope body (excludes 'signature')."""
30    mac = hmac.new(secret_key, _canonical_bytes(env), hashlib.sha256).digest()
31    env.signature = base64.b64encode(mac).decode()
32    return env
33
34def make_envelope(
35    sender: str, receiver: str, payload: dict, secret_key: bytes
36) -> A2AEnvelope:
37    env = A2AEnvelope(
38        sender_agent_id=sender,
39        receiver_agent_id=receiver,
40        sent_at=datetime.now(timezone.utc).isoformat(),
41        nonce=str(uuid4()),
42        payload=payload,
43    )
44    return sign_envelope(env, secret_key)

Gateway Verification with End-to-End Smoke Test

The gateway runs four checks in strict order: signature first (cheap, eliminates most spoofed traffic), then replay protection via a nonce cache, then the topology policy table, and finally a content guard that looks for prompt-injection patterns in the payload. The trailing smoke test exercises a happy path, a replayed envelope, and a topology violation against the same gateway_forward entry point.

Code snippetpython
1from typing import NamedTuple
2
3class Result(NamedTuple):
4    allowed: bool
5    reason: str
6
7    @staticmethod
8    def deny(reason: str) -> "Result":
9        return Result(allowed=False, reason=reason)
10
11    @staticmethod
12    def forward_ok() -> "Result":
13        return Result(allowed=True, reason="ok")
14
15# Allowed sender → receiver pairs (topology policy)
16POLICY: dict[str, set[str]] = {
17    "planner":     {"researcher", "code-writer"},
18    "researcher":  {"code-writer"},
19    "code-writer": {"deployment"},
20}
21
22# In-memory nonce cache; production uses Redis with a 60-second TTL
23_seen_nonces: set[str] = set()
24_MAX_AGE_SECONDS = 60
25
26INJECTION_PATTERNS = ["ignore previous", "override system", "disregard instructions"]
27
28def _verify_signature(env: A2AEnvelope, secret_key: bytes) -> bool:
29    expected = base64.b64encode(
30        hmac.new(secret_key, _canonical_bytes(env), hashlib.sha256).digest()
31    ).decode()
32    return hmac.compare_digest(expected, env.signature)
33
34def _check_replay(env: A2AEnvelope) -> bool:
35    sent = datetime.fromisoformat(env.sent_at)
36    age = (datetime.now(timezone.utc) - sent).total_seconds()
37    if age > _MAX_AGE_SECONDS or env.nonce in _seen_nonces:
38        return False
39    _seen_nonces.add(env.nonce)
40    return True
41
42def _policy_allows(sender: str, receiver: str) -> bool:
43    return receiver in POLICY.get(sender, set())
44
45def _content_guard(payload: dict) -> bool:
46    text = json.dumps(payload).lower()
47    return not any(p in text for p in INJECTION_PATTERNS)
48
49def gateway_forward(env: A2AEnvelope, secret_key: bytes) -> Result:
50    if not _verify_signature(env, secret_key):
51        return Result.deny("bad_signature")
52    if not _check_replay(env):
53        return Result.deny("replay")
54    if not _policy_allows(env.sender_agent_id, env.receiver_agent_id):
55        return Result.deny("policy_violation")
56    if not _content_guard(env.payload):
57        return Result.deny("content_guard")
58    # Production: write to audit log, then deliver to receiver
59    return Result.forward_ok()
60
61# --- End-to-end smoke test ---
62SHARED_KEY = b"change-me-rotate-quarterly"
63
64env = make_envelope("planner", "researcher", {"task": "summarise Q1 results"}, SHARED_KEY)
65result = gateway_forward(env, SHARED_KEY)
66assert result.allowed, f"Unexpected denial: {result.reason}"
67
68# Replay the same envelope — nonce is now in the cache
69replay_result = gateway_forward(env, SHARED_KEY)
70assert not replay_result.allowed and replay_result.reason == "replay"
71
72# Disallowed topology
73bad_env = make_envelope("deployment", "planner", {"task": "go back"}, SHARED_KEY)
74bad_result = gateway_forward(bad_env, SHARED_KEY)
75assert not bad_result.allowed and bad_result.reason == "policy_violation"

Verify by running the snippet with python3 gateway_smoke.py and confirming all three assertions pass with zero errors — a green run means signing, replay detection, and topology policy are all wired correctly.

Do's and Don'ts

Now that you have worked through the implementation, the practices below separate a durable approach from a fragile one.

Do's

1. ✓Do canonicalize the envelope with json.dumps(sort_keys=True, separators=(",", ":")) over every field except signature before computing the HMAC-SHA256 — field-ordering differences across agents would otherwise produce divergent MACs, and excluding signature itself is what makes the scheme self-consistent; a single altered byte in any covered field will invalidate the envelope.
2. ✓Do enforce gateway checks in strict order: signature → replay → topology → content guard — verifying the HMAC-SHA256 first eliminates most spoofed traffic cheaply, so the nonce cache, POLICY lookup, and injection-pattern scan in _content_guard only run on envelopes that are already cryptographically authenticated; reordering lets forged messages reach the policy table.
3. ✓Do compare the recomputed MAC against env.signature using hmac.compare_digest — the constant-time comparison prevents timing side-channels that would let an attacker probe the expected HMAC-SHA256 byte-by-byte by measuring response latency differences.

Don'ts

1. ✗Don't rely on TLS alone to verify which agent sent a message — transport-layer TLS secures the wire but cannot prove authorship; without the HMAC-signed A2AEnvelope, a compromised intermediate agent (e.g., researcher) can alter sender_agent_id, inject payloads into the payload dict, and the receiver has no way to detect the tampering.
2. ✗Don't use the in-memory _seen_nonces set in production — it resets on every gateway restart, reopening a full 60-second replay window for any nonce recorded before the crash; replace it with Redis using a 60-second TTL so nonces survive process restarts and the replay protection holds across gateway instances.
3. ✗Don't add entries to POLICY for reverse-pipeline directions such as "deployment": {"planner"} — POLICY.get(sender, set()) returns an empty set for any sender not listed, meaning unlisted senders are implicitly denied; adding a back-channel entry silently breaks the directed-graph topology and lets a compromised deployment agent inject instructions back into the pipeline head, bypassing the origin-scoping the policy is designed to enforce.