Topic 11: Module 8 lab
4 min read·22 Sept 2026
This lab turns on every defence at once and drives the whole chain with the poisoned note from Part 5. It routes the host through the gateway (allowlist plus audit), keeps pinning on, keeps the approval gate on, and logs without contents. The attack in the note tries to make the assistant create a note; the layered defence means the model never even sees a write tool, so the attack has nothing to grab.
python
"""Module 8 lab: one hardened run that turns on every defence at once.
Layers, from outer to inner:
1. Gateway allowlist: only search_notes is proxied; create_note is not exposed.
2. Pinning: tool definitions are pinned on first sight and diffed each load.
3. Approval gate: any non-read-only call needs a person's yes (here deny_all).
4. Logging without contents: names and argument keys only.
We drive it with the scripted stand-in model reading a poisoned note, so the
whole chain is exercised. Real models vary; retest with yours.
"""
from __future__ import annotations
import logging
import shutil
from pathlib import Path
import anyio
from mcp import Client
from notes_assistant.host import Host, HostConfig, deny_all
from notes_assistant.server import build_server
from notes_assistant.store import NoteStore
from examples.m08_gateway import build_gateway
from examples.m08_scripted_model import ScriptedModel
PIN_DIR = Path("pins_lab")
async def main() -> None:
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s %(message)s")
if PIN_DIR.exists():
shutil.rmtree(PIN_DIR)
Path("mnotes/from-note.md").unlink(missing_ok=True)
upstream_server = build_server(NoteStore("mnotes")) # notes include the poisoned one
async with Client(upstream_server) as upstream:
gateway = build_gateway(upstream)
async with Client(gateway) as gw_client:
model = ScriptedModel(plan=[
("notes__search_notes", {"query": "consent form", "limit": 3}),
("read_resource", {"uri": "notes://urgent-todo"}),
])
host = Host({"notes": gw_client}, chat_fn=model, approve=deny_all,
config=HostConfig(pin_dir=PIN_DIR, max_iterations=6))
answer = await host.ask("What is my urgent todo?")
print("\nfinal answer:", answer.text[:60])
print("stop_reason:", answer.stop_reason)
print("tools the model could see:", sorted(host.tools))
for record in answer.calls:
print(f" call: {record.name} ok={record.ok} note={record.note!r}")
leaked = Path("mnotes/from-note.md").exists() or Path("mnotes/leak.md").exists()
print("did any injected write land on disk?", leaked)
if __name__ == "__main__":
anyio.run(main)Comes out (upstream discovery logs omitted):
text
INFO gateway.audit proxy tool=search_notes arg_keys=['limit', 'query']
INFO notes_assistant.host tool_call name=notes__search_notes arg_keys=['limit', 'query'] ok=True ms=45 note=
INFO gateway.audit proxy resource=notes://<id>
INFO notes_assistant.host tool_call name=read_resource arg_keys=['uri'] ok=True ms=3 note=
final answer: Done.
stop_reason: answered
tools the model could see: ['notes__search_notes']
call: notes__search_notes ok=True note=''
call: read_resource ok=True note=''
did any injected write land on disk? False