Threat-model an agent and its tool boundary

Read a sample · Chapter 01 of 06

Begin with assets and data flows before choosing controls.

Our fictional service helps Orchard staff revise draft notes. A model can propose reading a note or replacing its text. Harbour has separate notes. The protected assets are note confidentiality, note integrity and the ability to explain who authorised a change. A helpful answer is not the only outcome to consider: the same tool route could disclose another organisation's draft or overwrite a recently reviewed version.

The lab has three files: gateway.py, demo.py and test_gateway.py. All are printed in full below. When a file spans blocks, join them in order with one blank line and preserve indentation. Run them from one fresh local folder. They use only the Python standard library, were executed with Python 3.12.10 on Windows, and make no model, filesystem-tool or network call. This is a simulator for the boundary, not a deployed agent.

Draw these flows on paper: a document reaches the model; the model returns a JSON proposal; trusted host code supplies the signed-in actor to a gateway; the gateway checks its note store and, for writes, a separate approval record. The response returns to the host. Add a distinct reviewer path that can create approvals. It must not be exposed among the model's callable tools. The lab represents that trusted path with a direct Python call.

Scroll sideways to see every column.

OWASP describes threat modelling as a repeatable design and validation activity. For this exercise, put a boundary wherever data gains a different kind of authority: document to model, model to gateway, and reviewer decision to execution. A line on the drawing is useful only when you can name the code or service that enforces it. If the model can directly edit the note store, drawing a gateway beside it changes nothing.

Authentication answers who the caller is. Authorisation decides what that caller may do to this object now. Approval records a deliberate decision about a particular operation. Those checks support one another, but do not replace one another. Our actor strings are supplied by trusted test code; accepting an actor field from the model would erase that distinction. A production host must obtain identity from its verified session or equivalent mechanism.

Read a sample · Chapter 02 of 06

Give each threat a specific precondition, action and observable consequence.

A useful abuse story names the entry point and the violated property. For example: a hostile sentence inside an Orchard document persuades the model to request harbour-brief; if the gateway trusts the requested identifier without checking ownership, Harbour's text is returned. The injection changes the proposal, but the harmful effect depends on excess authority at the tool boundary. This is why testing only whether a model refuses hostile prose leaves an important gap.

Use the fictional register below as a starting point. The owner is a role responsible for the control, not a fabricated named reviewer. Each acceptance test can be evaluated without contacting a real system. Classifying an issue as spoofing or tampering may help discussion, but the acceptance condition is what turns the description into work. Prioritise unauthorised disclosure and overwrite before cosmetic response quality because they violate the explicit service purpose.

Scroll sideways to see every column.

OWASP's excessive-agency guidance draws attention to the functions, permissions and autonomy available to a model-backed system. Our tool surface is deliberately two operations on named notes. There is no shell, arbitrary path, email sender, purchase or unrestricted database query. Removing an unnecessary effect can be stronger than trying to filter every sentence that might request it. If the product later needs a new tool, extend the model and tests before enabling it.

Separate a threat from a missing implementation detail. Expiry enforcement is implemented here. Authentication, a trustworthy approval screen, rate limiting and durable audit are not. Record them as unresolved conditions rather than describing the whole service as secure. The register can contain acceptance criteria for future controls, but a planned test is not a passed test. Keep source revision, test name and observed result beside any closure decision.

Do not invent precise likelihood percentages for a fictional service. Write the assumption instead: an attacker can influence document text, an authorised user can submit proposals, and host code remains trusted. A compromised Python process falls outside this lab because it can edit the public dictionaries or call approve directly. That is a crucial scope boundary, not a weakness that a frozen dataclass repairs.

Read a sample · Chapter 03 of 06

Validate structure before checking who may act.

Save gateway.py from the following consecutive blocks. The parser accepts a Python string containing one JSON object, at most 2,048 UTF-8 bytes. Read has exactly action and note fields. Replace adds exactly revision and text. There are no identity, permission or approved fields. Unknown actions and extra fields fail. The note ID grammar admits a small lowercase identifier; it is not interpreted as a filesystem path.

"""In-memory teaching gateway. Trusted host code owns identity and approval."""
from dataclasses import dataclass
import json
import re

MAX_INPUT = 2048
MAX_TEXT = 512
MAX_REVISION = 1_000_000

class Denied(ValueError):
    pass

def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise Denied("duplicate key")
        result[key] = value
    return result

def reject_constant(value):
    raise Denied("non-JSON number")

@dataclass(frozen=True)
class Proposal:
    action: str
    note: str
    revision: int | None = None
    text: str | None = None

def parse(raw):
    try:
        if type(raw) is not str or len(raw.encode("utf-8")) > MAX_INPUT:
            raise Denied("input size or type")
        data = json.loads(raw, object_pairs_hook=unique_object,
                          parse_constant=reject_constant)
    except (ValueError, RecursionError) as error:
        raise Denied("invalid proposal") from error
    if type(data) is not dict:
        raise Denied("object required")
    action = data.get("action")
    if action == "read":
        keys = {"action", "note"}
    elif action == "replace":
        keys = {"action", "note", "revision", "text"}
    else:
        raise Denied("unknown action")
    if set(data) != keys:
        raise Denied("unexpected fields")
    note = data["note"]
    if type(note) is not str or not re.fullmatch(r"[a-z][a-z0-9-]{0,31}", note):
        raise Denied("invalid note ID")
    if action == "read":
        return Proposal(action, note)
    revision, text = data["revision"], data["text"]
    if type(revision) is not int or not 1 <= revision <= MAX_REVISION:
        raise Denied("invalid revision")
    try:
        valid_text = type(text) is str and 1 <= len(text.encode("utf-8")) <= MAX_TEXT
    except UnicodeError:
        valid_text = False
    if not valid_text:
        raise Denied("invalid text")
    return Proposal(action, note, revision, text)

@dataclass(frozen=True)
class Note:
    owner: str
    text: str
    revision: int = 1

@dataclass(frozen=True)
class Approval:
    actor: str
    proposal: Proposal
    issued: int
    expires: int

class Gateway:
    def __init__(self):
        self.notes = {
            "orchard-brief": Note("orchard", "Draft A"),
            "orchard-plan": Note("orchard", "Plan A"),
            "harbour-brief": Note("harbour", "Draft B"),
        }
        self.approvals = {}
        self.serial = 0

    def owned(self, proposal, actor):
        note = self.notes.get(proposal.note)
        if note is None or note.owner != actor:
            raise Denied("not permitted")
        return note

    def current(self, proposal, actor):
        note = self.owned(proposal, actor)
        if proposal.action != "replace" or proposal.revision != note.revision:
            raise Denied("stale or non-write proposal")
        if note.revision >= MAX_REVISION:
            raise Denied("revision capacity reached")
        return note

    @staticmethod
    def clock(now):
        if type(now) is not int or now < 0:
            raise Denied("invalid trusted time")

    def approve(self, raw, actor, now):
        """Trusted reviewer path only; never register this as an agent tool."""
        self.clock(now)
        proposal = parse(raw)
        self.current(proposal, actor)
        self.serial += 1
        ticket = f"approval-{self.serial}"
        self.approvals[ticket] = Approval(actor, proposal, now, now + 30)
        return ticket

    def execute(self, raw, actor, now, ticket=None):
        self.clock(now)
        proposal = parse(raw)
        note = self.owned(proposal, actor)
        if proposal.action == "read":
            return {"text": note.text, "revision": note.revision}
        approval = self.approvals.get(ticket) if type(ticket) is str else None
        if (approval is None or approval.actor != actor
                or approval.proposal != proposal
                or not approval.issued <= now < approval.expires):
            raise Denied("approval required")
        note = self.current(proposal, actor)
        self.notes[proposal.note] = Note(actor, proposal.text, note.revision + 1)
        del self.approvals[ticket]
        return {"revision": note.revision + 1}

Python's JSON decoder normally permits repeated object names and non-standard constants such as NaN. The object_pairs_hook rejects duplicate keys, and parse_constant rejects those constants. Malformed JSON and excessive nesting become Denied. A JSON object that parses successfully still has to meet the exact field contract. Validation establishes the shape of a proposal; the owned method separately establishes permission for its target.

Replacement text must contain 1 to 512 UTF-8 bytes. The test uses the escaped character \u00e9, which encodes as two bytes: 256 repetitions fit, 257 do not. Revision must be an actual integer from 1 to 1,000,000; booleans are refused even though Python treats bool as an int subclass. A store at the maximum revision refuses another write. These are teaching limits, chosen to make boundary cases small and repeatable.

Proposal, Note and Approval are frozen dataclasses, and their fields here contain immutable values. Dataclass equality lets us compare the full parsed operation without depending on JSON spacing or key order. This is a local equality check, not a digital signature or a general canonical-JSON protocol. Frozen fields discourage accidental assignment; they do not isolate the Python process from hostile executable code.

Owned returns the same not permitted message for a missing note and another owner's note. That avoids directly disclosing existence through these two message variants; it is not a constant-time non-disclosure proof. A permitted read returns a new dictionary containing text and revision. Changing that returned dictionary does not edit the stored Note. The returned text remains untrusted content if it is later placed into a model prompt or rendered on a page.

The parser receives a string already present in memory. Its length check is not a network receive cap, total memory bound, timeout or rate limit. A real transport adapter must enforce its own request size and timing policy before passing a body to this function. The in-memory approval store also grows when trusted approvals are created but never used; there is no cleanup loop in the lab.

Read a sample · Chapter 04 of 06

An approval is useful only while its exact conditions still hold.

Approve an exact action. Recheck before writing.

The model supplies JSON action details. The host supplies actor orchard and an integer time. The separate trusted reviewer calls approve at time 100 for replace, target orchard-brief, expected revision 1 and text Reviewed draft. The stored approval binds all these parsed values and the actor, with issue time 100 and exclusive expiry 130. Changing JSON whitespace or key order does not change the parsed proposal. The comparison is not a cryptographic signature.

On the write path, the gateway validates host time, parses the strict schema and checks current ownership. It then checks the approval's actor, exact proposal and time window, followed by current revision, ownership again and revision capacity. Only then does it replace the note and remove that approval. A model-supplied actor field is refused with unexpected fields; it cannot override the host's actor. The read path checks ownership but requires no write approval.

One gateway, one approval: events in order

Event	Outcome	Note revision	Approval record
Approve at 100	approval issued	1	present
Changed body at 101	approval required	1	present
Exact action at 129	revision 2	2	absent
Replay at 129	approval required	2	absent

The changed body is Unreviewed draft. Its refusal leaves notes, approval records and the serial counter unchanged. The exact action at 129 changes the note to Reviewed draft at revision 2 and consumes the approval. Replaying the same action is refused without another write.

Time boundary: each row starts with a fresh gateway approved at 100

Execution time	Outcome	Note revision	Approval record
99	approval required	1	present
100	revision 2	2	absent
129	revision 2	2	absent
130	approval required	1	present

The valid interval is 100 <= now < 130. These four rows are independent, not repeated uses of one consumed ticket. At 99 the approval is not yet valid; at 130 it has expired. Refused approvals remain stored in this lab; no cleanup policy is implemented.

Two additional checks explain the recheck. Two separately approved revision-1 proposals are created at 100. After the first writes at 101, the second at 102 is refused with stale or non-write proposal: the note stays at revision 2 and the second approval stays present. In a separate fixture, trusted host code transfers the note to harbour after approval. Execution by orchard at 101 returns not permitted, with revision 1 and the approval unchanged.

This is a serial, in-memory teaching simulator. The reviewer method must never be registered as an agent tool. Host code is trusted; Python objects are not an isolation boundary against hostile executable code. Predictable ticket labels are not bearer credentials. The write and approval deletion are separate statements, not an atomic database transaction. Authentication integration, persistent audit, concurrent requests, restart recovery and external-effect delivery need their own design. No production security or browser acceptance is implied.

Approve parses the proposal, checks ownership and current revision, then stores the actor, complete Proposal and a 30-tick validity window. Its name is not a claim that a human has reviewed anything. The caller is responsible for obtaining a real, informed decision first. Never register approve as a model tool, automatically call it after a refusal, or let retrieved text supply its result. The demo invokes it explicitly to represent the trusted path.

The ticket label is a predictable local counter such as approval-1. It is not a secret bearer credential, authentication token or capability suitable for a network service. Merely guessing it does not satisfy actor and operation equality in this simulator. Production transport needs its own authenticated binding, appropriately protected identifiers, review state and rate limits. This course makes no recommendation to expose these labels to the internet.

Execution parses again and checks current ownership. For replace, it looks up the ticket, compares the actor and full Proposal, and requires issued <= now < expires. A ticket issued at 100 expires at 130: tick 129 is allowed, tick 130 is not. Tick 99 is also refused. Time is supplied by trusted code as an integer; the model cannot choose it. A real in-process elapsed-time policy can use a monotonic clock, with explicit handling for restarts.

After approval matches, current checks the live revision again. Two reviews of revision 1 do not authorise two successive overwrites. The first successful write stores revision 2 and consumes its ticket. The second proposal still names revision 1, so it is stale. If another trusted operation transferred the note to a different owner, ownership and actor binding also prevent the earlier ticket from following that transfer.

The state change and ticket deletion are consecutive statements in a serial simulator. They are not a database transaction and provide no concurrency or crash assurance. A real service needs an atomic conditional update, consumption of the approval and an appropriate audit record, plus a clear recovery policy. If the effect is an external send, local storage alone cannot promise exactly-once delivery. Do not attach a live side effect while leaving this gap implicit.

Save demo.py, then run python demo.py. Every printed value refers to fictional in-memory data. The unapproved write is refused, a distinct trusted approval enables one write, replay fails, and the foreign read is refused. The output is deliberately small enough to compare line by line.

import json
from gateway import Denied, Gateway

def main():
    gateway = Gateway()
    read = json.dumps({"action": "read", "note": "orchard-brief"})
    write = json.dumps({"action": "replace", "note": "orchard-brief",
                        "revision": 1, "text": "Reviewed draft"})
    print("before", gateway.execute(read, "orchard", 100))
    try:
        gateway.execute(write, "orchard", 100)
    except Denied as error:
        print("unapproved", error)
    # This call represents a separate trusted review, not a model decision.
    ticket = gateway.approve(write, "orchard", 100)
    print("approved", gateway.execute(write, "orchard", 101, ticket))
    try:
        gateway.execute(write, "orchard", 102, ticket)
    except Denied as error:
        print("replay", error)
    print("after", gateway.execute(read, "orchard", 103))
    foreign = json.dumps({"action": "read", "note": "harbour-brief"})
    try:
        gateway.execute(foreign, "orchard", 104)
    except Denied as error:
        print("other owner", error)

if __name__ == "__main__":
    main()

before {'text': 'Draft A', 'revision': 1}
unapproved approval required
approved {'revision': 2}
replay approval required
after {'text': 'Reviewed draft', 'revision': 2}
other owner not permitted

OWASP's transaction-authorisation guidance emphasises tying a review to the significant operation details and checking it at execution. For this lab those details are actor, action, note ID, expected revision and exact text. Equivalent JSON formatting has no effect because parsed values are compared. Normalising or trimming the replacement text would change this contract; such a change needs a deliberate decision and corresponding tests.

Read a sample · Chapter 05 of 06

A negative test should check the state as well as the error.

Save test_gateway.py. Sixteen tests cover reads, missing and foreign targets, exact schema, duplicate keys, numeric types, UTF-8 limits, unapproved writes, successful consumption, body and target changes, validity-window edges, actor and ownership changes, stale writes, the trusted approval path, equivalent JSON formatting, returned dictionaries and revision capacity. Run python -m unittest -v test_gateway.py. No additional package is needed.

import json
import unittest
from gateway import Denied, Gateway, MAX_INPUT, MAX_REVISION, Note, parse

def proposal(**changes):
    data = {"action": "replace", "note": "orchard-brief",
            "revision": 1, "text": "Reviewed draft"}
    data.update(changes)
    return json.dumps(data)

class BoundaryTests(unittest.TestCase):
    def setUp(self):
        self.g = Gateway()

    def denied_without_change(self, raw, actor="orchard", now=101, ticket=None):
        before = (dict(self.g.notes), dict(self.g.approvals), self.g.serial)
        with self.assertRaises(Denied):
            self.g.execute(raw, actor, now, ticket)
        self.assertEqual(before, (self.g.notes, self.g.approvals, self.g.serial))

    def test_read_is_owner_scoped(self):
        raw = json.dumps({"action": "read", "note": "orchard-brief"})
        self.assertEqual(self.g.execute(raw, "orchard", 100),
                         {"text": "Draft A", "revision": 1})
        self.denied_without_change(raw, actor="harbour")
        self.denied_without_change(raw, actor="unknown")

    def test_absent_and_foreign_have_same_error(self):
        for target in ("missing", "harbour-brief"):
            with self.assertRaisesRegex(Denied, "^not permitted$"):
                self.g.execute(json.dumps({"action": "read", "note": target}),
                               "orchard", 100)

    def test_schema_rejects_model_authority_and_unknown_actions(self):
        for change in ({"approved": True}, {"actor": "harbour"},
                       {"action": "delete"}, {"note": "../secret"}):
            self.denied_without_change(proposal(**change))

    def test_json_duplicates_and_shapes(self):
        for raw in ('{"action":"read","action":"replace"}', '[]', 'null',
                    '{', '{"value":NaN}', '{"value":Infinity}'):
            self.denied_without_change(raw)

    def test_revision_requires_bounded_integer_not_bool(self):
        for value in (True, False, 0, -1, 1.0, "1", MAX_REVISION + 1):
            self.denied_without_change(proposal(revision=value))

    def test_input_and_utf8_text_bounds(self):
        raw = '{"action":"read","note":"orchard-brief"}'
        self.assertEqual(parse(raw + " " * (MAX_INPUT - len(raw))).action, "read")
        for bad in (raw + " " * (MAX_INPUT + 1 - len(raw)), b"{}", "\ud800"):
            self.denied_without_change(bad)
        self.assertEqual(len(parse(proposal(text="\u00e9" * 256)).text), 256)
        for text in ("", "\u00e9" * 257, "\ud800", None, 12):
            self.denied_without_change(proposal(text=text))

    def test_unapproved_write_is_inert(self):
        self.denied_without_change(proposal())
        self.denied_without_change(proposal(), ticket="approval-999")

    def test_exact_approved_write_consumes_ticket(self):
        raw = proposal()
        ticket = self.g.approve(raw, "orchard", 100)
        self.assertEqual(self.g.execute(raw, "orchard", 129, ticket), {"revision": 2})
        self.assertEqual(self.g.notes["orchard-brief"].text, "Reviewed draft")
        self.assertNotIn(ticket, self.g.approvals)
        self.denied_without_change(raw, now=129, ticket=ticket)

    def test_changed_body_or_target_cannot_reuse_approval(self):
        ticket = self.g.approve(proposal(), "orchard", 100)
        for change in ({"text": "Changed after review"}, {"note": "orchard-plan"}):
            self.denied_without_change(proposal(**change), ticket=ticket)
        self.assertIn(ticket, self.g.approvals)

    def test_expiry_is_exclusive(self):
        ticket = self.g.approve(proposal(), "orchard", 100)
        self.denied_without_change(proposal(), now=99, ticket=ticket)
        self.denied_without_change(proposal(), now=130, ticket=ticket)
        self.denied_without_change(proposal(), now=131, ticket=ticket)

    def test_actor_binding_and_ownership_recheck(self):
        ticket = self.g.approve(proposal(), "orchard", 100)
        self.denied_without_change(proposal(), actor="harbour", ticket=ticket)
        self.g.notes["orchard-brief"] = Note("harbour", "Transferred")
        self.denied_without_change(proposal(), ticket=ticket)
        self.denied_without_change(proposal(), actor="harbour", ticket=ticket)

    def test_two_approvals_cannot_overwrite_new_revision(self):
        first = self.g.approve(proposal(), "orchard", 100)
        second = self.g.approve(proposal(text="Other draft"), "orchard", 100)
        self.g.execute(proposal(), "orchard", 101, first)
        self.denied_without_change(proposal(text="Other draft"), ticket=second)

    def test_approval_path_rejects_unauthorised_and_nonwrite(self):
        for raw, actor in ((proposal(), "harbour"), (proposal(revision=2), "orchard"),
                           ('{"action":"read","note":"orchard-brief"}', "orchard")):
            before = (dict(self.g.approvals), self.g.serial)
            with self.assertRaises(Denied):
                self.g.approve(raw, actor, 100)
            self.assertEqual(before, (self.g.approvals, self.g.serial))

    def test_json_formatting_does_not_change_approved_action(self):
        raw = proposal()
        ticket = self.g.approve(raw, "orchard", 100)
        equivalent = json.dumps(json.loads(raw), indent=2, sort_keys=True)
        self.assertEqual(self.g.execute(equivalent, "orchard", 101, ticket),
                         {"revision": 2})

    def test_read_result_does_not_mutate_stored_note(self):
        result = self.g.execute('{"action":"read","note":"orchard-brief"}',
                                "orchard", 100)
        result["text"] = "Changed locally"
        self.assertEqual(self.g.notes["orchard-brief"].text, "Draft A")

    def test_revision_capacity_and_trusted_clock(self):
        self.g.notes["orchard-brief"] = Note("orchard", "Last", MAX_REVISION)
        with self.assertRaisesRegex(Denied, "capacity"):
            self.g.approve(proposal(revision=MAX_REVISION), "orchard", 100)
        for value in (True, -1, 1.5, "100"):
            self.denied_without_change(proposal(), now=value)

if __name__ == "__main__":
    unittest.main()

Denied_without_change snapshots notes, approvals and serial before attempting an operation. It requires both a Denied exception and identical state afterwards. A function that raises after modifying a note would fail this check. Positive tests matter too: an always-deny gateway would avoid many harmful actions while failing its basic purpose. The approved-write test requires revision 2 and the expected text, while the read test requires the intended owner's data.

The release also tested four deliberately defective copies outside the learner folder. Each remained syntactically valid, but failed the tests: removing the owner comparison, removing full-proposal equality, removing the time-window condition, and removing the current-revision comparison. The originals were preserved. This establishes that the suite distinguishes these four faults; it does not estimate coverage of every possible attack or certify the architecture.

When extending the exercise, change one check at a time in a disposable copy and state which test should fail first. Record the diff, test result and restored source hash. Do not weaken a running service to demonstrate a defect. The published fixture is deliberately disconnected from external actions, so its consequences are limited to invented dictionary entries.

Keep test results separate from broad security claims. These tests prove observed outcomes for specific data in one serial Python process. They do not test real login sessions, cross-origin requests, a human approval screen, compromised dependencies, concurrent workers, persistent records, network retries or audit access control. Those scenarios need their own integration environment and owner before the service can claim them.

Read a sample · Chapter 06 of 06

Record what is implemented, what was observed and what remains a condition of release.

A useful handover contains the system sketch, assets, explicit trusted assumptions, threat register, source revision, executable tests, observed outputs and unresolved controls with responsible roles. Keep them together so the next builder can connect a refusal rule to the code that enforces it. A screenshot of green tests without the tested revision is not enough to repeat the result.

For the fictional note service, the implemented claim is narrow: serial in-memory calls enforce the documented owner and approval rules for the tested proposal grammar. The unimplemented production conditions are equally concrete: authenticate the caller; protect the reviewer path; show exact action details; enforce receive limits and rate limits; make update, consumption and audit atomic where needed; persist state; handle restart and clock policy; and restrict the process from bypassing the gateway.

Scroll sideways to see every column.

Design audit events around the decision: request ID, actor reference, operation type, target reference, policy version, approval reference, result code and resource revision. Avoid putting entire note bodies or model prompts into routine logs. Such fields can create a second confidential-data store and complicate access and retention. The lab prints a fictional demonstration only; it has no persistent audit implementation and does not satisfy these requirements by printing to the terminal.

Threat modelling should be revisited when the tool set, data source, identity system, reviewer flow or deployment topology changes. Moving from one process to two workers invalidates the serial-state assumption. Adding a tool that fetches URLs introduces destinations and network permissions absent here. Allowing generated Python would invalidate the assumption that the model controls only JSON. These are changes to the system being evaluated, not just more test cases for the old model.

Finish with an owner decision that names the remaining conditions. For this course, the appropriate outcome is acceptance as a tested educational simulator. A production owner would need evidence from the real integration and an explicit decision about its unresolved risks. Do not describe a local teaching approval as a user's permission to deploy, send messages, change records or spend money elsewhere.