Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -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"]
Original file line number Diff line number Diff line change
@@ -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.
Original file line number Diff line number Diff line change
@@ -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()
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
pydantic>=2.0.0