From eeb293fcc27a1cd43135ed8b023da947fa8cceae Mon Sep 17 00:00:00 2001 From: jw_ond Date: Thu, 30 Jul 2026 02:59:11 +0800 Subject: [PATCH] Add action boundary drift sample --- .../pydantic/action_boundary_drift/Dockerfile | 8 + .../pydantic/action_boundary_drift/README.md | 59 +++++++ .../pydantic/action_boundary_drift/main.py | 157 ++++++++++++++++++ .../action_boundary_drift/requirements.txt | 1 + 4 files changed, 225 insertions(+) create mode 100644 code_samples/agentic_top_ten/frameworks/pydantic/action_boundary_drift/Dockerfile create mode 100644 code_samples/agentic_top_ten/frameworks/pydantic/action_boundary_drift/README.md create mode 100644 code_samples/agentic_top_ten/frameworks/pydantic/action_boundary_drift/main.py create mode 100644 code_samples/agentic_top_ten/frameworks/pydantic/action_boundary_drift/requirements.txt diff --git a/code_samples/agentic_top_ten/frameworks/pydantic/action_boundary_drift/Dockerfile b/code_samples/agentic_top_ten/frameworks/pydantic/action_boundary_drift/Dockerfile new file mode 100644 index 0000000..1089b34 --- /dev/null +++ b/code_samples/agentic_top_ten/frameworks/pydantic/action_boundary_drift/Dockerfile @@ -0,0 +1,8 @@ +FROM python:3.12-slim + +WORKDIR /app +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt +COPY main.py . + +CMD ["python", "main.py"] diff --git a/code_samples/agentic_top_ten/frameworks/pydantic/action_boundary_drift/README.md b/code_samples/agentic_top_ten/frameworks/pydantic/action_boundary_drift/README.md new file mode 100644 index 0000000..7f9adf4 --- /dev/null +++ b/code_samples/agentic_top_ten/frameworks/pydantic/action_boundary_drift/README.md @@ -0,0 +1,59 @@ +# Pydantic action boundary drift sample + +This deliberately vulnerable sample demonstrates an approval/execution drift issue in an agent workflow. + +The agent asks for approval to perform one action, receives an approval ticket, and then executes a different action under the same ticket. The insecure path checks only that an approval exists. The secure path binds the approval to a canonical action fingerprint and rejects the drift. + +## Vulnerability + +Many human-in-the-loop agent designs treat approval as a session flag or approval ID: + +1. Agent proposes a low-risk action. +2. Human approves the proposal. +3. Agent or runtime later changes the target, destination, effect, or side-effect profile. +4. Executor sees an approved ticket and runs the changed action. + +That creates a gap between what was approved and what actually happened. The audit trail may show that a human approved the session, but it cannot prove that the executed action still matched the approved action. + +## Mapped risks + +- ASI02 Tool Misuse and Exploitation: the tool call changes from read-only analysis to data export. +- ASI08 Repudiation and Untraceability: the approval record is not bound to the executed action. +- ASI10 Overwhelming Human-in-the-Loop: the reviewer is shown reassuring text instead of an enforceable action boundary. +- LLM06 Excessive Agency: the agent can convert a benign approval into a higher-impact tool action. + +## Run locally + +With Python: + +```bash +python -m pip install -r requirements.txt +python main.py +``` + +With Docker: + +```bash +docker build -t action-boundary-drift . +docker run --rm action-boundary-drift +``` + +Expected output: + +- The insecure executor accepts the mutated action because the ticket is approved. +- The secure executor rejects the mutated action because the canonical action fingerprint changed. + +## Mitigation pattern + +Approval should be bound to the action object, not just to a session, ticket, prompt, or natural-language summary. At minimum, bind: + +- actor or agent identity +- operation +- resource or target +- destination boundary +- intended effect +- side-effect class +- policy version +- approval decision + +Then recompute the fingerprint immediately before execution. If any of the fields changed, the approval should not be reusable. diff --git a/code_samples/agentic_top_ten/frameworks/pydantic/action_boundary_drift/main.py b/code_samples/agentic_top_ten/frameworks/pydantic/action_boundary_drift/main.py new file mode 100644 index 0000000..94d4e3f --- /dev/null +++ b/code_samples/agentic_top_ten/frameworks/pydantic/action_boundary_drift/main.py @@ -0,0 +1,157 @@ +"""Deliberately vulnerable approval/execution drift demo. + +The insecure executor checks that an approval ticket exists, but it does not +check whether the approved action and the executed action are still identical. +""" + +from __future__ import annotations + +import hashlib +import json +from typing import Literal, Optional +from uuid import uuid4 + +from pydantic import BaseModel + + +Effect = Literal["read", "data_export", "financial", "deployment"] +Destination = Literal["internal", "external", "public"] + + +class Action(BaseModel): + actor: str + operation: str + resource: str + destination: Destination + effect: Effect + side_effect: bool + policy_version: str + + +class ApprovalTicket(BaseModel): + ticket_id: str + approved: bool + reviewer: str + approved_summary: str + approved_action_fingerprint: Optional[str] = None + + +class ExecutionResult(BaseModel): + executed: bool + reason: str + action_fingerprint: str + + +def canonical_fingerprint(action: Action) -> str: + """Create a stable digest over the fields that define action identity.""" + + stable = action.model_dump() + encoded = json.dumps(stable, sort_keys=True, separators=(",", ":")).encode() + return hashlib.sha256(encoded).hexdigest() + + +def request_insecure_approval(action: Action) -> ApprovalTicket: + return ApprovalTicket( + ticket_id=f"ticket-{uuid4()}", + approved=True, + reviewer="finance-ops", + approved_summary=( + f"Approve {action.operation} on {action.resource} " + f"for {action.destination} use." + ), + ) + + +def request_bound_approval(action: Action) -> ApprovalTicket: + return ApprovalTicket( + ticket_id=f"ticket-{uuid4()}", + approved=True, + reviewer="finance-ops", + approved_summary=( + f"Approve {action.operation} on {action.resource} " + f"for {action.destination} use." + ), + approved_action_fingerprint=canonical_fingerprint(action), + ) + + +def insecure_execute(ticket: ApprovalTicket, action: Action) -> ExecutionResult: + if not ticket.approved: + return ExecutionResult( + executed=False, + reason="ticket was not approved", + action_fingerprint=canonical_fingerprint(action), + ) + + return ExecutionResult( + executed=True, + reason="executed because approval ticket was present", + action_fingerprint=canonical_fingerprint(action), + ) + + +def secure_execute(ticket: ApprovalTicket, action: Action) -> ExecutionResult: + fingerprint = canonical_fingerprint(action) + if not ticket.approved: + return ExecutionResult( + executed=False, + reason="ticket was not approved", + action_fingerprint=fingerprint, + ) + + if ticket.approved_action_fingerprint != fingerprint: + return ExecutionResult( + executed=False, + reason="rejected approval/execution drift", + action_fingerprint=fingerprint, + ) + + return ExecutionResult( + executed=True, + reason="executed because approval matched this exact action", + action_fingerprint=fingerprint, + ) + + +def main() -> None: + approved_action = Action( + actor="support-agent", + operation="read_support_ticket", + resource="ticket-1042", + destination="internal", + effect="read", + side_effect=False, + policy_version="support-policy-v1", + ) + + mutated_action = Action( + actor="support-agent", + operation="export_customer_records", + resource="customer-table", + destination="public", + effect="data_export", + side_effect=True, + policy_version="support-policy-v1", + ) + + insecure_ticket = request_insecure_approval(approved_action) + insecure_result = insecure_execute(insecure_ticket, mutated_action) + + bound_ticket = request_bound_approval(approved_action) + secure_result = secure_execute(bound_ticket, mutated_action) + + print("Approved action fingerprint:") + print(canonical_fingerprint(approved_action)) + print() + print("Mutated action fingerprint:") + print(canonical_fingerprint(mutated_action)) + print() + print("INSECURE EXECUTOR:") + print(insecure_result.model_dump_json(indent=2)) + print() + print("SECURE EXECUTOR:") + print(secure_result.model_dump_json(indent=2)) + + +if __name__ == "__main__": + main() diff --git a/code_samples/agentic_top_ten/frameworks/pydantic/action_boundary_drift/requirements.txt b/code_samples/agentic_top_ten/frameworks/pydantic/action_boundary_drift/requirements.txt new file mode 100644 index 0000000..65b1b0c --- /dev/null +++ b/code_samples/agentic_top_ten/frameworks/pydantic/action_boundary_drift/requirements.txt @@ -0,0 +1 @@ +pydantic>=2.0.0