Course: Course 2B — Securing & Attacking Harnesses and LLMs Module: B6 — Inter-Agent Trust and Communication Security Duration: 60–75 minutes Environment: Python 3.10+ (or Node 18+ with TypeScript). No GPU, no network, no external services. This lab builds the signed-message channel, the replay-protection layer, and the orchestrator-enforced compartment boundary — then attacks them.
By the end of this lab you will have:
This is the engineering realization of the B6 teaching document. Every inter-agent message carries a signature the recipient verifies; every message is fresh; the orchestrator — not the agents — enforces trust compartments.
mkdir b6-inter-agent-lab && cd b6-inter-agent-lab
python3 -m venv .venv && source .venv/bin/activate
# No third-party deps needed — the standard library has hmac, hashlib, secrets, time.
# If using TypeScript: node 18+ has crypto built in. No npm install required.
No GPU, no network. The entire lab runs on the standard library.
Implement the sign/verify protocol with HMAC and a durable key store. This is DD-16 ZeroClaw's HMAC tool receipts extended with persistence, rotation, and key IDs.
from dataclasses import dataclass, field
from typing import Any
import time, json
@dataclass
class MessageHeader:
frm: str # sender agent id (e.g. "orchestrator") — 'from' is a keyword
to: str # recipient agent id (e.g. "deploy_agent")
task_id: str # binds the message to a task/session
intent: str # e.g. "approve_deployment", "delegate_research"
timestamp_ms: int # UTC millis — freshness window check
nonce: str # single-use random — replay defense
key_id: str # which key signed (supports rotation)
@dataclass
class SignedMessage:
header: MessageHeader
payload: Any # the message body — task-specific
signature: str # HMAC over header + payload (hex)
The key store persists keys, supports rotation, and retains rotated-out keys for a verification window. In production this is the B5 secrets vault the harness accesses — not a process-local dict. For the lab, an in-process KeyStore class is acceptable, but it must model: persistence (keys survive across KeyStore instances via a backing file), rotation, and key IDs.
import hmac, hashlib, secrets, os, json
from pathlib import Path
class KeyStore:
"""Durable HMAC key store. Extends ZeroClaw's ephemeral keys with
persistence + rotation + key IDs + a verify-retention window."""
def __init__(self, backing_file: str, verify_retention_days: int = 7):
self.backing_file = Path(backing_file)
self.verify_retention_days = verify_retention_days
self._keys: dict[str, dict] = {} # key_id -> {"secret": hex, "created_ms": int}
self._current_key_id: str | None = None
self._load()
def _load(self) -> None:
if self.backing_file.exists():
data = json.loads(self.backing_file.read_text())
self._keys = data["keys"]
self._current_key_id = data["current_key_id"]
# else: empty store — caller must call rotate() to mint the first key
def _persist(self) -> None:
self.backing_file.write_text(json.dumps({
"keys": self._keys, "current_key_id": self._current_key_id
}))
def current(self) -> tuple[str, bytes]:
"""Return (key_id, secret) for signing."""
if self._current_key_id is None:
raise RuntimeError("no current key — call rotate() first")
entry = self._keys[self._current_key_id]
return self._current_key_id, bytes.fromhex(entry["secret"])
def get(self, key_id: str) -> bytes | None:
"""Return the secret for verification, or None if unknown/rotated-out."""
entry = self._keys.get(key_id)
return bytes.fromhex(entry["secret"]) if entry else None
def rotate(self) -> str:
"""Mint a new key, set it current, retain the old for the verify window."""
new_key_id = f"k-{secrets.token_hex(4)}"
self._keys[new_key_id] = {
"secret": secrets.token_bytes(32).hex(),
"created_ms": int(time.time() * 1000),
}
self._current_key_id = new_key_id
self._purge_expired()
self._persist()
return new_key_id
def _purge_expired(self) -> None:
cutoff = int(time.time() * 1000) - self.verify_retention_days * 86_400_000
expired = [kid for kid, e in self._keys.items()
if e["created_ms"] < cutoff and kid != self._current_key_id]
for kid in expired:
del self._keys[kid]
def canonical(header: MessageHeader, payload: Any) -> bytes:
"""Stable serialization for signing. Order matters — must be identical
at sign and verify time."""
return json.dumps({
"header": {
"frm": header.frm, "to": header.to, "task_id": header.task_id,
"intent": header.intent, "timestamp_ms": header.timestamp_ms,
"nonce": header.nonce, "key_id": header.key_id,
},
"payload": payload,
}, sort_keys=True).encode()
def sign(frm: str, to: str, task_id: str, intent: str,
payload: Any, keys: KeyStore, now_ms: int | None = None) -> SignedMessage:
"""The sender signs a message before sending. The signature covers header + payload."""
ts = now_ms if now_ms is not None else int(time.time() * 1000)
key_id, secret = keys.current()
header = MessageHeader(
frm=frm, to=to, task_id=task_id, intent=intent,
timestamp_ms=ts, nonce=secrets.token_hex(16), key_id=key_id,
)
sig = hmac.new(secret, canonical(header, payload), hashlib.sha256).hexdigest()
return SignedMessage(header=header, payload=payload, signature=sig)
def verify_signature(msg: SignedMessage, keys: KeyStore) -> bool:
"""Signature gate only. Constant-time compare to prevent timing oracles."""
secret = keys.get(msg.header.key_id)
if secret is None:
return False # unknown/rotated-out key
expected = hmac.new(secret, canonical(msg.header, msg.payload), hashlib.sha256).digest()
actual = bytes.fromhex(msg.signature)
return hmac.compare_digest(expected, actual) # constant-time
KeyStore, sign, and verify_signature in signed_messages.py.msg.payload = "tampered") and confirm verify_signature returns False.keys = KeyStore("keys.json")
keys.rotate()
msg = sign("orchestrator", "deploy_agent", "task-42", "approve_deployment",
{"env": "staging"}, keys)
assert verify_signature(msg, keys) is True
msg.payload = "tampered"
assert verify_signature(msg, keys) is False
print("Phase 1 OK — sign/verify works, tamper detected")
This is the durable extension of ZeroClaw. Keys persist to disk (keys.json), survive a process restart, rotate on demand, and old keys are retained for the verify window. A message signed before rotation can still be verified; after the window, the key is purged and verification correctly fails. Course 1's ephemeral-key gap is closed.
A signature-only defense is replayable. Add the three replay controls: timestamp freshness window, single-use nonces, and task binding. Wrap them in a full verify() that runs five gates in series.
class NonceCache:
"""Tracks seen nonces for a TTL equal to (or greater than) the freshness window.
Bounded LRU so a flood of messages cannot exhaust memory."""
def __init__(self, ttl_ms: int, max_size: int = 10_000):
self.ttl_ms = ttl_ms
self.max_size = max_size
self._seen: dict[str, int] = {} # nonce -> expires_ms
def _cleanup(self, now_ms: int) -> None:
expired = [n for n, exp in self._seen.items() if exp <= now_ms]
for n in expired:
del self._seen[n]
def check_and_record(self, nonce: str, now_ms: int) -> bool:
"""Return True if the nonce is fresh (not seen). Records it on success."""
self._cleanup(now_ms)
if nonce in self._seen:
return False # replay
if len(self._seen) >= self.max_size:
# Evict the oldest by expiry (bounded LRU approximation)
oldest = min(self._seen, key=lambda n: self._seen[n])
del self._seen[oldest]
self._seen[nonce] = now_ms + self.ttl_ms
return True
from dataclasses import dataclass
@dataclass
class VerifyResult:
ok: bool
reason: str = ""
def verify(msg: SignedMessage, keys: KeyStore, nonces: NonceCache,
expected_recipient: str, expected_task_id: str,
freshness_ms: int = 60_000, now_ms: int | None = None) -> VerifyResult:
"""The recipient runs this BEFORE acting. Five gates, all must pass.
The first failure rejects."""
now = now_ms if now_ms is not None else int(time.time() * 1000)
h = msg.header
# Gate 1: recipient
if h.to != expected_recipient:
return VerifyResult(False, f"wrong recipient: {h.to}")
# Gate 2: task binding (cross-task replay defense)
if h.task_id != expected_task_id:
return VerifyResult(False, f"task_id mismatch: {h.task_id} != {expected_task_id}")
# Gate 3: freshness window (stale-message defense)
skew = abs(now - h.timestamp_ms)
if skew > freshness_ms:
return VerifyResult(False, f"stale: {skew}ms outside {freshness_ms}ms window")
# Gate 4: nonce (within-window replay defense)
if not nonces.check_and_record(h.nonce, now):
return VerifyResult(False, f"replayed nonce: {h.nonce}")
# Gate 5: signature (forgery defense)
if not verify_signature(msg, keys):
return VerifyResult(False, "signature invalid (forged or rotated-out key)")
return VerifyResult(True)
Implement NonceCache and verify(). Then run the four replay-attack scenarios below. Each must be rejected by the correct gate.
keys = KeyStore("keys.json")
keys.rotate()
nonces = NonceCache(ttl_ms=60_000)
# --- Scenario A: a fresh, valid message (baseline — should pass) ---
msg = sign("orchestrator", "deploy_agent", "task-42", "approve_deployment",
{"env": "staging"}, keys)
r = verify(msg, keys, nonces, "deploy_agent", "task-42")
assert r.ok, f"A should pass: {r.reason}"
# --- Scenario B: replay the same message (nonce gate) ---
r = verify(msg, keys, nonces, "deploy_agent", "task-42")
assert not r.ok and "replayed nonce" in r.reason, f"B: {r.reason}"
# --- Scenario C: replay across tasks (task-binding gate) ---
nonces2 = NonceCache(ttl_ms=60_000) # fresh cache so nonce isn't the blocker
r = verify(msg, keys, nonces2, "deploy_agent", "task-99")
assert not r.ok and "task_id mismatch" in r.reason, f"C: {r.reason}"
# --- Scenario D: replay after the freshness window (freshness gate) ---
nonces3 = NonceCache(ttl_ms=60_000)
future = msg.header.timestamp_ms + 120_000 # 2 minutes later
r = verify(msg, keys, nonces3, "deploy_agent", "task-42", now_ms=future)
assert not r.ok and "stale" in r.reason, f"D: {r.reason}"
print("Phase 2 OK — all four replay scenarios rejected at the correct gate")
Five gates, series. Drop any one and a replay path opens — and you have now demonstrated each of the four replay scenarios failing at exactly the gate that catches it. The nonce cache TTL must be ≥ the freshness window; verify this by setting ttl_ms=30_000 and freshness_ms=60_000 and finding the replay window that opens (it is the subject of exam Q11 — reproduce it here as a stretch).
Now the attack. A compromised low-trust research agent forges an orchestrator message. Without signatures, the deploy agent executes. With signatures, the forge fails at gate 5.
First, simulate the mesh default — no signatures, the recipient trusts the from field:
def naive_handle(msg_header: dict, payload: Any) -> str:
"""The mesh default: trust the 'from' field. No verification."""
if msg_header["frm"] == "orchestrator" and msg_header["intent"] == "approve_deployment":
return f"DEPLOYED {payload} (trusted 'from' field, no verification)"
return "ignored"
# A compromised research agent forges an orchestrator message:
forged_header = {"frm": "orchestrator", "intent": "approve_deployment", "task_id": "task-42"}
result = naive_handle(forged_header, {"env": "production"})
print("UNAUTHENTICATED:", result)
# → DEPLOYED {'env': 'production'} (trusted 'from' field, no verification)
# A read-only research agent just triggered a production deployment.
Now the research agent tries to forge a signed orchestrator message. It does not hold the orchestrator's key, so it either (a) sends an unsigned message, or (b) signs with its own key. Both fail at gate 5.
# The research agent does NOT have the orchestrator's key (B5 credential isolation —
# the key is in the vault, the agent never sees it). It tries to sign with... nothing
# it can use. Best case: it constructs a SignedMessage with a garbage signature.
forged = SignedMessage(
header=MessageHeader(
frm="orchestrator", to="deploy_agent", task_id="task-42",
intent="approve_deployment", timestamp_ms=int(time.time()*1000),
nonce=secrets.token_hex(16), key_id="k-orchestrator-01",
),
payload={"env": "production"},
signature="0" * 64, # garbage — the research agent cannot compute the real HMAC
)
nonces_defense = NonceCache(ttl_ms=60_000)
r = verify(forged, keys, nonces_defense, "deploy_agent", "task-42")
print("AUTHENTICATED:", "DEPLOY BLOCKED —", r.reason if not r.ok else "SHOULD NOT PASS")
# → DEPLOY BLOCKED — signature invalid (forged or rotated-out key)
assert not r.ok and "signature" in r.reason
print("Phase 3 OK — forged orchestrator message rejected at the signature gate")
This is the entire module in one demo. The unauthenticated channel lets a read-only agent trigger a production deployment (Microsoft Failure Mode Taxonomy #3). The signed channel rejects the forge at the signature gate because the research agent does not hold the orchestrator's key — and cannot, because B5's credential isolation keeps the key in the vault, away from the agent process. The from field is verified by the signature, not read from the payload.
Signed messages authenticate the channel. Now add the structural control: a low-trust agent cannot direct a high-trust agent without an orchestrator-mediated, authorized crossing.
from enum import IntEnum
class TrustLevel(IntEnum):
LOW = 1 # read-only, no commit (e.g. research agent)
MED = 2 # writes staging (e.g. code agent)
HIGH = 3 # writes production (e.g. deploy agent)
@dataclass
class AgentPolicy:
agent_id: str
trust_level: TrustLevel
@dataclass
class SessionIntent:
task_id: str
authorized_intent: set[str] # e.g. {"research", "summarize"} — NOT "deploy"
action_budget: int # max state-changing actions this session
actions_taken: int = 0
def orchestrator_authorize_crossing(
requester: AgentPolicy,
target: AgentPolicy,
intent: str,
session: SessionIntent,
) -> VerifyResult:
"""The orchestrator mediates every cross-compartment request.
The agents do NOT talk agent-to-agent for privileged actions."""
# Check 1: intent must be in the session's authorized set (session intent tracking)
if intent not in session.authorized_intents:
return VerifyResult(False, f"intent '{intent}' not authorized for session {session.task_id}")
# Check 2: a lower-trust agent cannot DIRECT a higher-trust agent.
# (A lower-trust agent may INFORM via data, but the action must be independently authorized.)
if requester.trust_level < target.trust_level and intent in {"deploy", "approve_deployment"}:
return VerifyResult(
False,
f"low-trust ({requester.agent_id}) cannot direct high-trust ({target.agent_id}) "
f"to {intent} — requires independent orchestrator authorization"
)
# Check 3: session action budget (blast-radius cap)
if intent in {"deploy", "approve_deployment", "commit"}:
if session.actions_taken >= session.action_budget:
return VerifyResult(False, f"session action budget exhausted ({session.action_budget})")
session.actions_taken += 1
return VerifyResult(True)
Implement the compartments and the orchestrator check. Then run the scenario: the research agent (LOW) tries to direct the deploy agent (HIGH) to deploy. The orchestrator must deny it.
research = AgentPolicy("research_agent", TrustLevel.LOW)
deploy = AgentPolicy("deploy_agent", TrustLevel.HIGH)
# A research-only session — deploy is NOT in the authorized intents
session = SessionIntent(
task_id="task-42",
authorized_intents={"research", "summarize"},
action_budget=3,
)
# The research agent tries to direct a deployment:
r = orchestrator_authorize_crossing(research, deploy, "deploy", session)
print("CROSS-COMPARTMENT:", "DENIED —", r.reason if not r.ok else "ALLOWED")
assert not r.ok
assert "cannot direct" in r.reason or "not authorized" in r.reason
print("Phase 4 OK — low-trust agent cannot direct high-trust agent")
Even a compromised orchestrator (which CAN sign valid messages and CAN pass the intent check) is bounded by the action budget. Simulate: a compromised orchestrator attempts 5 deploys in a session with action_budget=3. The first 3 pass; the 4th is denied by the budget cap.
orchestrator = AgentPolicy("orchestrator", TrustLevel.HIGH)
session2 = SessionIntent("task-43", authorized_intents={"deploy"}, action_budget=3)
results = [orchestrator_authorize_crossing(orchestrator, deploy, "deploy", session2).ok for _ in range(5)]
print("BUDGET CAP:", results) # [True, True, True, False, False]
assert results == [True, True, True, False, False]
print("Stretch OK — blast-radius cap stops the cascade at action 3")
The compartment boundary is enforced in the orchestrator, not the agents — because the agents can be coerced. The session intent tracking makes the policy a compiled boolean the agent cannot talk past. And the action budget is the structural circuit breaker that bounds damage even when the orchestrator itself is compromised: B8 detects the anomaly, the cap stops it.
signed_messages.py — KeyStore (durable, rotated), sign(), verify_signature() (Phase 1)replay_protection.py — NonceCache, full five-gate verify() (Phase 2)attack_demo.py — the unauthenticated baseline + the signed-channel rejection (Phase 3)compartments.py — TrustLevel, AgentPolicy, SessionIntent, orchestrator_authorize_crossing() (Phase 4)keys.json — the persisted key store (evidence the keys are durable, not ephemeral)KeyStore persists keys to disk (keys.json); survives a process restart; supports rotate() with verify-retention.sign() / verify_signature() work; tampering the payload invalidates the signature (constant-time compare).verify() rejects all four replay scenarios (within-session nonce, cross-task binding, stale timestamp) at the correct gate.# Lab Specification — Module B6: Inter-Agent Trust and Communication Security
**Course**: Course 2B — Securing & Attacking Harnesses and LLMs
**Module**: B6 — Inter-Agent Trust and Communication Security
**Duration**: 60–75 minutes
**Environment**: Python 3.10+ (or Node 18+ with TypeScript). No GPU, no network, no external services. This lab builds the signed-message channel, the replay-protection layer, and the orchestrator-enforced compartment boundary — then attacks them.
---
## Learning objectives
By the end of this lab you will have:
1. **Implemented a signed inter-agent message protocol** with HMAC and durable keys (a persistent key store with rotation and key IDs) — extending DD-16 ZeroClaw's HMAC tool receipts to the inter-agent channel and closing the ephemeral-key gap Course 1 flagged.
2. **Added replay protection** — nonces (single-use, bounded LRU cache), a timestamp freshness window, and task binding — so a captured validly-signed message cannot be re-injected out of context.
3. **Built an orchestrator-enforced compartment boundary** — a low-trust research agent cannot direct a high-trust deploy agent without an explicit, authorized, audited crossing mediated by the orchestrator.
4. **Run a trust-escalation attack demo** — a compromised low-trust agent forges an orchestrator message, and you watch the verification layer reject it at the signature gate.
5. **Run a replay attack demo** — capture a valid message, attempt to replay it, and observe which gate (task binding, freshness, nonce) rejects it.
This is the engineering realization of the B6 teaching document. Every inter-agent message carries a signature the recipient verifies; every message is fresh; the orchestrator — not the agents — enforces trust compartments.
---
## Phase 0 — Setup (3 min)
```bash
mkdir b6-inter-agent-lab && cd b6-inter-agent-lab
python3 -m venv .venv && source .venv/bin/activate
# No third-party deps needed — the standard library has hmac, hashlib, secrets, time.
# If using TypeScript: node 18+ has crypto built in. No npm install required.
```
No GPU, no network. The entire lab runs on the standard library.
---
## Phase 1 — The signed inter-agent message protocol (20 min)
Implement the sign/verify protocol with HMAC and a durable key store. This is DD-16 ZeroClaw's HMAC tool receipts extended with persistence, rotation, and key IDs.
### 1.1 The message type
```python
from dataclasses import dataclass, field
from typing import Any
import time, json
@dataclass
class MessageHeader:
frm: str # sender agent id (e.g. "orchestrator") — 'from' is a keyword
to: str # recipient agent id (e.g. "deploy_agent")
task_id: str # binds the message to a task/session
intent: str # e.g. "approve_deployment", "delegate_research"
timestamp_ms: int # UTC millis — freshness window check
nonce: str # single-use random — replay defense
key_id: str # which key signed (supports rotation)
@dataclass
class SignedMessage:
header: MessageHeader
payload: Any # the message body — task-specific
signature: str # HMAC over header + payload (hex)
```
### 1.2 The durable key store
The key store persists keys, supports rotation, and retains rotated-out keys for a verification window. **In production this is the B5 secrets vault the harness accesses — not a process-local dict.** For the lab, an in-process `KeyStore` class is acceptable, but it must model: persistence (keys survive across `KeyStore` instances via a backing file), rotation, and key IDs.
```python
import hmac, hashlib, secrets, os, json
from pathlib import Path
class KeyStore:
"""Durable HMAC key store. Extends ZeroClaw's ephemeral keys with
persistence + rotation + key IDs + a verify-retention window."""
def __init__(self, backing_file: str, verify_retention_days: int = 7):
self.backing_file = Path(backing_file)
self.verify_retention_days = verify_retention_days
self._keys: dict[str, dict] = {} # key_id -> {"secret": hex, "created_ms": int}
self._current_key_id: str | None = None
self._load()
def _load(self) -> None:
if self.backing_file.exists():
data = json.loads(self.backing_file.read_text())
self._keys = data["keys"]
self._current_key_id = data["current_key_id"]
# else: empty store — caller must call rotate() to mint the first key
def _persist(self) -> None:
self.backing_file.write_text(json.dumps({
"keys": self._keys, "current_key_id": self._current_key_id
}))
def current(self) -> tuple[str, bytes]:
"""Return (key_id, secret) for signing."""
if self._current_key_id is None:
raise RuntimeError("no current key — call rotate() first")
entry = self._keys[self._current_key_id]
return self._current_key_id, bytes.fromhex(entry["secret"])
def get(self, key_id: str) -> bytes | None:
"""Return the secret for verification, or None if unknown/rotated-out."""
entry = self._keys.get(key_id)
return bytes.fromhex(entry["secret"]) if entry else None
def rotate(self) -> str:
"""Mint a new key, set it current, retain the old for the verify window."""
new_key_id = f"k-{secrets.token_hex(4)}"
self._keys[new_key_id] = {
"secret": secrets.token_bytes(32).hex(),
"created_ms": int(time.time() * 1000),
}
self._current_key_id = new_key_id
self._purge_expired()
self._persist()
return new_key_id
def _purge_expired(self) -> None:
cutoff = int(time.time() * 1000) - self.verify_retention_days * 86_400_000
expired = [kid for kid, e in self._keys.items()
if e["created_ms"] < cutoff and kid != self._current_key_id]
for kid in expired:
del self._keys[kid]
```
### 1.3 Sign and verify (signature only — replay protection comes in Phase 2)
```python
def canonical(header: MessageHeader, payload: Any) -> bytes:
"""Stable serialization for signing. Order matters — must be identical
at sign and verify time."""
return json.dumps({
"header": {
"frm": header.frm, "to": header.to, "task_id": header.task_id,
"intent": header.intent, "timestamp_ms": header.timestamp_ms,
"nonce": header.nonce, "key_id": header.key_id,
},
"payload": payload,
}, sort_keys=True).encode()
def sign(frm: str, to: str, task_id: str, intent: str,
payload: Any, keys: KeyStore, now_ms: int | None = None) -> SignedMessage:
"""The sender signs a message before sending. The signature covers header + payload."""
ts = now_ms if now_ms is not None else int(time.time() * 1000)
key_id, secret = keys.current()
header = MessageHeader(
frm=frm, to=to, task_id=task_id, intent=intent,
timestamp_ms=ts, nonce=secrets.token_hex(16), key_id=key_id,
)
sig = hmac.new(secret, canonical(header, payload), hashlib.sha256).hexdigest()
return SignedMessage(header=header, payload=payload, signature=sig)
def verify_signature(msg: SignedMessage, keys: KeyStore) -> bool:
"""Signature gate only. Constant-time compare to prevent timing oracles."""
secret = keys.get(msg.header.key_id)
if secret is None:
return False # unknown/rotated-out key
expected = hmac.new(secret, canonical(msg.header, msg.payload), hashlib.sha256).digest()
actual = bytes.fromhex(msg.signature)
return hmac.compare_digest(expected, actual) # constant-time
```
### 1.4 Your task
- Implement `KeyStore`, `sign`, and `verify_signature` in `signed_messages.py`.
- Run the smoke test below. The signature must verify. Tamper the payload (`msg.payload = "tampered"`) and confirm `verify_signature` returns `False`.
```python
keys = KeyStore("keys.json")
keys.rotate()
msg = sign("orchestrator", "deploy_agent", "task-42", "approve_deployment",
{"env": "staging"}, keys)
assert verify_signature(msg, keys) is True
msg.payload = "tampered"
assert verify_signature(msg, keys) is False
print("Phase 1 OK — sign/verify works, tamper detected")
```
### 1.5 The point
This is the durable extension of ZeroClaw. Keys persist to disk (`keys.json`), survive a process restart, rotate on demand, and old keys are retained for the verify window. A message signed before rotation can still be verified; after the window, the key is purged and verification correctly fails. Course 1's ephemeral-key gap is closed.
---
## Phase 2 — Replay protection (20 min)
A signature-only defense is replayable. Add the three replay controls: timestamp freshness window, single-use nonces, and task binding. Wrap them in a full `verify()` that runs five gates in series.
### 2.1 The nonce cache
```python
class NonceCache:
"""Tracks seen nonces for a TTL equal to (or greater than) the freshness window.
Bounded LRU so a flood of messages cannot exhaust memory."""
def __init__(self, ttl_ms: int, max_size: int = 10_000):
self.ttl_ms = ttl_ms
self.max_size = max_size
self._seen: dict[str, int] = {} # nonce -> expires_ms
def _cleanup(self, now_ms: int) -> None:
expired = [n for n, exp in self._seen.items() if exp <= now_ms]
for n in expired:
del self._seen[n]
def check_and_record(self, nonce: str, now_ms: int) -> bool:
"""Return True if the nonce is fresh (not seen). Records it on success."""
self._cleanup(now_ms)
if nonce in self._seen:
return False # replay
if len(self._seen) >= self.max_size:
# Evict the oldest by expiry (bounded LRU approximation)
oldest = min(self._seen, key=lambda n: self._seen[n])
del self._seen[oldest]
self._seen[nonce] = now_ms + self.ttl_ms
return True
```
### 2.2 The full verify() — five gates
```python
from dataclasses import dataclass
@dataclass
class VerifyResult:
ok: bool
reason: str = ""
def verify(msg: SignedMessage, keys: KeyStore, nonces: NonceCache,
expected_recipient: str, expected_task_id: str,
freshness_ms: int = 60_000, now_ms: int | None = None) -> VerifyResult:
"""The recipient runs this BEFORE acting. Five gates, all must pass.
The first failure rejects."""
now = now_ms if now_ms is not None else int(time.time() * 1000)
h = msg.header
# Gate 1: recipient
if h.to != expected_recipient:
return VerifyResult(False, f"wrong recipient: {h.to}")
# Gate 2: task binding (cross-task replay defense)
if h.task_id != expected_task_id:
return VerifyResult(False, f"task_id mismatch: {h.task_id} != {expected_task_id}")
# Gate 3: freshness window (stale-message defense)
skew = abs(now - h.timestamp_ms)
if skew > freshness_ms:
return VerifyResult(False, f"stale: {skew}ms outside {freshness_ms}ms window")
# Gate 4: nonce (within-window replay defense)
if not nonces.check_and_record(h.nonce, now):
return VerifyResult(False, f"replayed nonce: {h.nonce}")
# Gate 5: signature (forgery defense)
if not verify_signature(msg, keys):
return VerifyResult(False, "signature invalid (forged or rotated-out key)")
return VerifyResult(True)
```
### 2.3 Your task
Implement `NonceCache` and `verify()`. Then run the four replay-attack scenarios below. Each must be rejected by the correct gate.
```python
keys = KeyStore("keys.json")
keys.rotate()
nonces = NonceCache(ttl_ms=60_000)
# --- Scenario A: a fresh, valid message (baseline — should pass) ---
msg = sign("orchestrator", "deploy_agent", "task-42", "approve_deployment",
{"env": "staging"}, keys)
r = verify(msg, keys, nonces, "deploy_agent", "task-42")
assert r.ok, f"A should pass: {r.reason}"
# --- Scenario B: replay the same message (nonce gate) ---
r = verify(msg, keys, nonces, "deploy_agent", "task-42")
assert not r.ok and "replayed nonce" in r.reason, f"B: {r.reason}"
# --- Scenario C: replay across tasks (task-binding gate) ---
nonces2 = NonceCache(ttl_ms=60_000) # fresh cache so nonce isn't the blocker
r = verify(msg, keys, nonces2, "deploy_agent", "task-99")
assert not r.ok and "task_id mismatch" in r.reason, f"C: {r.reason}"
# --- Scenario D: replay after the freshness window (freshness gate) ---
nonces3 = NonceCache(ttl_ms=60_000)
future = msg.header.timestamp_ms + 120_000 # 2 minutes later
r = verify(msg, keys, nonces3, "deploy_agent", "task-42", now_ms=future)
assert not r.ok and "stale" in r.reason, f"D: {r.reason}"
print("Phase 2 OK — all four replay scenarios rejected at the correct gate")
```
### 2.4 The point
Five gates, series. Drop any one and a replay path opens — and you have now demonstrated each of the four replay scenarios failing at exactly the gate that catches it. The nonce cache TTL must be ≥ the freshness window; verify this by setting `ttl_ms=30_000` and `freshness_ms=60_000` and finding the replay window that opens (it is the subject of exam Q11 — reproduce it here as a stretch).
---
## Phase 3 — The trust-escalation attack demo (10 min)
Now the attack. A compromised low-trust research agent forges an orchestrator message. Without signatures, the deploy agent executes. With signatures, the forge fails at gate 5.
### 3.1 The unauthenticated baseline (the vulnerability)
First, simulate the mesh default — no signatures, the recipient trusts the `from` field:
```python
def naive_handle(msg_header: dict, payload: Any) -> str:
"""The mesh default: trust the 'from' field. No verification."""
if msg_header["frm"] == "orchestrator" and msg_header["intent"] == "approve_deployment":
return f"DEPLOYED {payload} (trusted 'from' field, no verification)"
return "ignored"
# A compromised research agent forges an orchestrator message:
forged_header = {"frm": "orchestrator", "intent": "approve_deployment", "task_id": "task-42"}
result = naive_handle(forged_header, {"env": "production"})
print("UNAUTHENTICATED:", result)
# → DEPLOYED {'env': 'production'} (trusted 'from' field, no verification)
# A read-only research agent just triggered a production deployment.
```
### 3.2 The defense rejects the forge
Now the research agent tries to forge a *signed* orchestrator message. It does not hold the orchestrator's key, so it either (a) sends an unsigned message, or (b) signs with its own key. Both fail at gate 5.
```python
# The research agent does NOT have the orchestrator's key (B5 credential isolation —
# the key is in the vault, the agent never sees it). It tries to sign with... nothing
# it can use. Best case: it constructs a SignedMessage with a garbage signature.
forged = SignedMessage(
header=MessageHeader(
frm="orchestrator", to="deploy_agent", task_id="task-42",
intent="approve_deployment", timestamp_ms=int(time.time()*1000),
nonce=secrets.token_hex(16), key_id="k-orchestrator-01",
),
payload={"env": "production"},
signature="0" * 64, # garbage — the research agent cannot compute the real HMAC
)
nonces_defense = NonceCache(ttl_ms=60_000)
r = verify(forged, keys, nonces_defense, "deploy_agent", "task-42")
print("AUTHENTICATED:", "DEPLOY BLOCKED —", r.reason if not r.ok else "SHOULD NOT PASS")
# → DEPLOY BLOCKED — signature invalid (forged or rotated-out key)
assert not r.ok and "signature" in r.reason
print("Phase 3 OK — forged orchestrator message rejected at the signature gate")
```
### 3.3 The point
This is the entire module in one demo. The unauthenticated channel lets a read-only agent trigger a production deployment (Microsoft Failure Mode Taxonomy #3). The signed channel rejects the forge at the signature gate because the research agent does not hold the orchestrator's key — and cannot, because B5's credential isolation keeps the key in the vault, away from the agent process. The `from` field is verified by the signature, not read from the payload.
---
## Phase 4 — The orchestrator-enforced compartment boundary (15 min)
Signed messages authenticate the channel. Now add the structural control: a low-trust agent cannot direct a high-trust agent without an orchestrator-mediated, authorized crossing.
### 4.1 The trust compartments
```python
from enum import IntEnum
class TrustLevel(IntEnum):
LOW = 1 # read-only, no commit (e.g. research agent)
MED = 2 # writes staging (e.g. code agent)
HIGH = 3 # writes production (e.g. deploy agent)
@dataclass
class AgentPolicy:
agent_id: str
trust_level: TrustLevel
@dataclass
class SessionIntent:
task_id: str
authorized_intent: set[str] # e.g. {"research", "summarize"} — NOT "deploy"
action_budget: int # max state-changing actions this session
actions_taken: int = 0
```
### 4.2 The orchestrator's cross-compartment check
```python
def orchestrator_authorize_crossing(
requester: AgentPolicy,
target: AgentPolicy,
intent: str,
session: SessionIntent,
) -> VerifyResult:
"""The orchestrator mediates every cross-compartment request.
The agents do NOT talk agent-to-agent for privileged actions."""
# Check 1: intent must be in the session's authorized set (session intent tracking)
if intent not in session.authorized_intents:
return VerifyResult(False, f"intent '{intent}' not authorized for session {session.task_id}")
# Check 2: a lower-trust agent cannot DIRECT a higher-trust agent.
# (A lower-trust agent may INFORM via data, but the action must be independently authorized.)
if requester.trust_level < target.trust_level and intent in {"deploy", "approve_deployment"}:
return VerifyResult(
False,
f"low-trust ({requester.agent_id}) cannot direct high-trust ({target.agent_id}) "
f"to {intent} — requires independent orchestrator authorization"
)
# Check 3: session action budget (blast-radius cap)
if intent in {"deploy", "approve_deployment", "commit"}:
if session.actions_taken >= session.action_budget:
return VerifyResult(False, f"session action budget exhausted ({session.action_budget})")
session.actions_taken += 1
return VerifyResult(True)
```
### 4.3 Your task
Implement the compartments and the orchestrator check. Then run the scenario: the research agent (LOW) tries to direct the deploy agent (HIGH) to deploy. The orchestrator must deny it.
```python
research = AgentPolicy("research_agent", TrustLevel.LOW)
deploy = AgentPolicy("deploy_agent", TrustLevel.HIGH)
# A research-only session — deploy is NOT in the authorized intents
session = SessionIntent(
task_id="task-42",
authorized_intents={"research", "summarize"},
action_budget=3,
)
# The research agent tries to direct a deployment:
r = orchestrator_authorize_crossing(research, deploy, "deploy", session)
print("CROSS-COMPARTMENT:", "DENIED —", r.reason if not r.ok else "ALLOWED")
assert not r.ok
assert "cannot direct" in r.reason or "not authorized" in r.reason
print("Phase 4 OK — low-trust agent cannot direct high-trust agent")
```
### 4.4 Stretch — the compromised-orchestrator blast-radius cap
Even a compromised orchestrator (which CAN sign valid messages and CAN pass the intent check) is bounded by the action budget. Simulate: a compromised orchestrator attempts 5 deploys in a session with `action_budget=3`. The first 3 pass; the 4th is denied by the budget cap.
```python
orchestrator = AgentPolicy("orchestrator", TrustLevel.HIGH)
session2 = SessionIntent("task-43", authorized_intents={"deploy"}, action_budget=3)
results = [orchestrator_authorize_crossing(orchestrator, deploy, "deploy", session2).ok for _ in range(5)]
print("BUDGET CAP:", results) # [True, True, True, False, False]
assert results == [True, True, True, False, False]
print("Stretch OK — blast-radius cap stops the cascade at action 3")
```
### 4.5 The point
The compartment boundary is enforced in the orchestrator, not the agents — because the agents can be coerced. The session intent tracking makes the policy a compiled boolean the agent cannot talk past. And the action budget is the structural circuit breaker that bounds damage even when the orchestrator itself is compromised: B8 detects the anomaly, the cap stops it.
---
## Deliverables
- `signed_messages.py` — `KeyStore` (durable, rotated), `sign()`, `verify_signature()` (Phase 1)
- `replay_protection.py` — `NonceCache`, full five-gate `verify()` (Phase 2)
- `attack_demo.py` — the unauthenticated baseline + the signed-channel rejection (Phase 3)
- `compartments.py` — `TrustLevel`, `AgentPolicy`, `SessionIntent`, `orchestrator_authorize_crossing()` (Phase 4)
- `keys.json` — the persisted key store (evidence the keys are durable, not ephemeral)
## Success criteria
- [ ] `KeyStore` persists keys to disk (`keys.json`); survives a process restart; supports `rotate()` with verify-retention.
- [ ] `sign()` / `verify_signature()` work; tampering the payload invalidates the signature (constant-time compare).
- [ ] Full `verify()` rejects all four replay scenarios (within-session nonce, cross-task binding, stale timestamp) at the correct gate.
- [ ] The trust-escalation attack demo shows the unauthenticated channel executing the forged deploy, AND the signed channel rejecting it at the signature gate.
- [ ] The orchestrator denies a low-trust agent's attempt to direct a high-trust agent, and the action budget caps a compromised orchestrator at the configured limit.
- [ ] The nonce cache TTL is set ≥ the freshness window (verify the replay-window bug from exam Q11 does not occur with correct settings).
- [ ] Every component ties back to a specific claim from the teaching document (durable keys close the ZeroClaw gap; five gates defeat forgery + replay; orchestrator-enforced compartments because agents can be coerced; blast-radius caps bound the mesh).