Enforcement Models
[!NOTE] New to Tenuo? A quick primer:
- Warrant: A short-lived, cryptographic token that says “this agent may call these tools with these constraints”
- Proof-of-Possession (PoP): A signature proving the requester holds the warrant’s private key (prevents stolen warrants from being reused)
- Attenuation: Delegating a warrant with narrower permissions — you can never add permissions, only remove them
- Control Plane: The trusted service that issues warrants (you build this, or use Tenuo Cloud)
See Core Concepts for a full introduction.
Overview
Tenuo provides Action-Level Security for AI Agents. But where exactly does that security live?
Unlike network firewalls (which block IPs) or IAM (which blocks identities), Tenuo blocks specific tool calls based on cryptographic warrants.
IAM Policies answer “may this identity do X?” Warrants answer “was this specific action authorized by a specific delegator?”
You can deploy Tenuo in four enforcement models, ranging from “Drop-in Safety” to “Zero Trust Infrastructure.”
| Model | Enforcement Point | Protects Against |
|---|---|---|
| In-Process | Inside your Python agent | Prompt injection (confused deputy) |
| Sidecar | Separate process, same pod | Compromised agent (RCE) |
| Gateway | Cluster ingress (Envoy/Istio) | Centralized policy |
| MCP Proxy | Between agent and MCP server | Unauthorized tool discovery |
Choose based on your threat model. They can be combined for defense in depth.
Shared Enforcement Core
All in-process integrations (LangGraph, CrewAI, OpenAI, AutoGen, Google ADK) use the same underlying enforcement logic. This ensures:
- Consistent behavior across all frameworks
- Single code path through the Rust core for security-critical checks
- Unified logging via
tenuo.enforcementlogger
┌─────────────────────────────────────────────────────────────┐
│ Your Integration │
│ (LangGraph / CrewAI / OpenAI / AutoGen / Google ADK) │
└─────────────────────────┬───────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ enforce_tool_call() │
│ • Allowlist filtering • Critical tool checks │
│ • Constraint extraction • Unified audit logging │
└─────────────────────────┬───────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Rust Core (tenuo_core) │
│ • PoP signature verification • Constraint validation │
│ • Expiration checking • Cryptographic operations │
└─────────────────────────────────────────────────────────────┘
Remote enforcement (FastAPI, A2A) uses verify_mode to validate precomputed signatures from clients, but still goes through the same Rust core for cryptographic verification.
Tool Policies (Defense in Depth)
The Python layer includes optional policy checks that run before the Rust core. These are defense-in-depth measures—they help catch mistakes early but do not replace the cryptographic guarantees.
[!IMPORTANT] These policies do NOT affect security invariants. The Rust core is the security boundary. Policy checks are UX/operational safeguards that can be bypassed if needed. An attacker who compromises the Python process can skip these checks, but they cannot bypass the Rust core’s cryptographic verification.
Risk Levels and Critical Tools
Tenuo includes built-in risk classification for common tools:
| Risk Level | Behavior | Example Tools |
|---|---|---|
critical |
Denied if no relevant constraint | delete_file, http_request, shell_command, execute_sql |
high |
Warning logged | write_file, send_email, fetch_url |
medium |
No special handling | read_file, query_db |
low |
No special handling | web_search, list_directory |
Example: A warrant granting delete_file with no path constraint will be denied by Python policy (before reaching Rust), with a helpful error:
Critical tool 'delete_file' requires at least one of: ['path']
Why this exists: Prevents accidentally issuing overly-permissive warrants. The Rust core would technically allow delete_file: Wildcard(), but that’s almost certainly a mistake.
Registering Custom Tool Schemas
from tenuo import ToolSchema, register_schema
# Mark your custom tool as critical
register_schema("drop_database", ToolSchema(
recommended_constraints=["database", "confirm"],
require_at_least_one=True,
risk_level="critical",
description="Drops entire database - requires explicit constraints",
))
Task-Scoped Allowlists (allowed_tools)
Restrict tools per-task, even when the warrant grants more:
from tenuo.langgraph import TenuoToolNode
# Warrant allows: ["search", "read_file", "write_file", "delete_file"]
# Task 1: Research phase (read-only)
node = TenuoToolNode(tools, allowed_tools=["search", "read_file"])
# Task 2: Write phase
node = TenuoToolNode(tools, allowed_tools=["write_file"])
# Task 3: Cleanup (dangerous - separate approval workflow)
node = TenuoToolNode(tools, allowed_tools=["delete_file"])
Key behavior:
allowed_toolscan only restrict, never expand- If the warrant doesn’t include a tool,
allowed_toolscannot grant it - This is a Python-side policy; the Rust core still verifies the warrant
Strict Mode
Enable strict=True to require constraints on tools marked require_at_least_one:
from tenuo.langchain import guard_tools
# Fail if read_file has no path constraint
protected = guard_tools(tools, bound, strict=True)
Invariant Guarantee
These policy checks do not weaken Tenuo’s security model:
| Check | Enforced By | Bypassable? | Security Impact |
|---|---|---|---|
| Critical tool constraints | Python | Yes (if process compromised) | None - Rust core still enforces warrant |
allowed_tools filtering |
Python | Yes | None - Rust core still enforces warrant |
| Warrant expiration | Rust core | No | Security boundary |
| Tool in warrant | Rust core | No | Security boundary |
| Constraint satisfaction | Rust core | No | Security boundary |
| PoP signature | Rust core | No | Security boundary |
The Rust core is the only security boundary. Python policies are convenience features for better DX and operational safety.
Model 1: In-Process Enforcement (The Library)
Best for: Preventing Prompt Injection in Monolithic Agents, LangChain/LangGraph, quick integration
In this model, Tenuo runs inside your agent’s process as a Python library / decorator.
- Architecture:
Agent (Python) └─ @guard decorator (Tenuo SDK) └─ Tool Implementation (Function)
How it works:
@guard(tool="delete_file")
def delete_file(path: str):
os.remove(path) # Never reached if unauthorized
with warrant_scope(warrant), key_scope(keypair):
delete_file("/etc/passwd") # Raises ScopeViolation
- LLM generates a tool call:
delete_file("/etc/passwd") - The
@guarddecorator checks:- Warrant existence
- Warrant validity (expiration)
- Tool authorization
- Argument constraints
- Proof-of-Possession signature
- If the warrant says
path: /data/*, Tenuo raisesScopeViolation. The tool code never runs.
Security Guarantee: Blocks confused deputy attacks. If prompt injection tricks the LLM into calling unauthorized tools, Tenuo stops it.
Limitation: If an attacker gets remote code execution (RCE) on the agent process, they can bypass Tenuo by calling tools directly. The agent process is the trust boundary. For RCE protection, use Model 2 (Sidecar).
Variant: Web Framework Middleware
Best for: Agents exposed as APIs (e.g., LangServe, Flask apps)
If your agent exposes tools as HTTP endpoints, you can enforce warrants globally using middleware. This is cleaner than decorating every single route.
Header Format: Clients send
X-Tenuo-Warrant(base64 CBOR) andX-Tenuo-PoP(base64 signature). The warrant payload can be a single warrant or a full chain (WarrantStack) — the format is self-describing.
FastAPI / Starlette:
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from tenuo import Authorizer, Warrant, ScopeViolation
app = FastAPI()
# Initialize with your control plane's public key
authorizer = Authorizer(trusted_roots=[control_plane_public_key])
@app.middleware("http")
async def tenuo_guard(request: Request, call_next):
# Skip health checks
if request.url.path in ["/health", "/ready"]:
return await call_next(request)
# 1. Extract Warrant and PoP Signature
warrant_b64 = request.headers.get("X-Tenuo-Warrant")
pop_b64 = request.headers.get("X-Tenuo-PoP")
if not warrant_b64:
return JSONResponse(status_code=401, content={"error": "Missing warrant"})
try:
# Decode Warrant
warrant = Warrant(warrant_b64)
# Decode Signature (if present)
pop_sig = base64.b64decode(pop_b64) if pop_b64 else None
except Exception:
return JSONResponse(status_code=400, content={"error": "Invalid warrant or signature"})
# 2. Identify the Tool (Endpoint) & Arguments
tool_name = request.url.path # e.g., "/tools/read_file"
# Parse body as JSON dict (check() requires a dict)
try:
args = await request.json() if request.method in ["POST", "PUT"] else {}
except:
args = {}
# 3. Enforce (including PoP verification)
try:
authorizer.check(warrant, tool_name, args, signature=pop_sig)
except Exception: # Authorizer raises generic exception on failure
return JSONResponse(status_code=403, content={"error": "Access denied"})
return await call_next(request)
Flask:
from flask import Flask, request, abort
from tenuo import Authorizer, Warrant
import base64
app = Flask(__name__)
authorizer = Authorizer(trusted_roots=[control_plane_public_key])
@app.before_request
def check_warrant():
# Skip health checks
if request.path in ["/health", "/ready"]:
return
# Extract headers
warrant_b64 = request.headers.get("X-Tenuo-Warrant")
pop_b64 = request.headers.get("X-Tenuo-PoP")
if not warrant_b64:
abort(401, description="Missing warrant")
try:
warrant = Warrant(warrant_b64)
pop_sig = base64.b64decode(pop_b64) if pop_b64 else None
args = request.get_json() or {}
# Verify warrant and authorize action
authorizer.check(warrant, request.path, args, signature=pop_sig)
except Exception:
abort(403, description="Access denied")
FastAPI Dependency Injection (Recommended)
For more control over which routes require warrants, use FastAPI’s dependency injection:
from fastapi import FastAPI, Depends, Request, HTTPException
from tenuo import (
Warrant, guard,
warrant_scope, key_scope,
ScopeViolation
)
app = FastAPI()
async def require_warrant(request: Request) -> Warrant:
"""Dependency that extracts and validates warrant."""
warrant_b64 = request.headers.get("X-Tenuo-Warrant")
if not warrant_b64:
raise HTTPException(status_code=401, detail="Missing warrant")
try:
return Warrant(warrant_b64)
except Exception:
raise HTTPException(status_code=400, detail="Invalid warrant")
@guard(tool="read_file")
def read_file(path: str) -> str:
return open(path).read()
@app.get("/files/{path:path}")
async def get_file(path: str, warrant: Warrant = Depends(require_warrant)):
# Context ensures @guard can access warrant in async handlers
with warrant_scope(warrant), key_scope(AGENT_KEYPAIR):
try:
return {"content": read_file(path)}
except ScopeViolation as e:
raise HTTPException(status_code=403, detail=str(e))
This pattern is preferred when:
- Only some routes need authorization
- You want per-route warrant requirements
- You need proper async context propagation
See examples/fastapi_integration.py for a complete example.
Model 2: Sidecar Enforcement
Best for: Microservices, Kubernetes, and High-Value Tools, zero-trust architectures
In this model, Tenuo runs alongside your application as a separate process (Sidecar). The Tool is not just a function; it is an API endpoint and Tenuo sits in front of it.
- Architecture:
┌─────────────────┐ Network ┌──────────────────────────┐ │ Agent (Client) │ ───────────────────► │ Tool Service Pod │ └─────────────────┘ (HTTP/gRPC) │ ┌──────────────────────┐ │ │ │ Tenuo Sidecar │ │ │ └─────────┬────────────┘ │ │ ▼ │ │ ┌──────────────────────┐ │ │ │ Actual API Logic │ │ │ └──────────────────────┘ │ └──────────────────────────┘ - The Flow:
- Agent sends HTTP request with warrant in header (
POST /api/delete?file=/etc/passwd+X-Tenuo-Warrant). - Request hits Tenuo sidecar first (via Kubernetes networking or reverse proxy)
- Sidecar validates warrant against parameters (path/body).
- If denied: returns
403 Forbidden. The request never reaches tool - If allowed: forwards request to actual tool API
- Agent sends HTTP request with warrant in header (
- Security Guarantee: Even if the agent is fully compromised (RCE), it cannot force unauthorized actions. The tool service is the trust boundary, not the agent.
Kubernetes deployment:
apiVersion: v1
kind: Pod
metadata:
name: tool-service
spec:
containers:
- name: tenuo-authorizer
image: tenuo/authorizer:0.1
ports:
- containerPort: 9090
- name: tool-api
image: your-tool:latest
# Only accepts traffic from localhost (sidecar)
Note: This model can also be deployed as a Gateway, where a single Tenuo instance protects multiple services. This simplifies management but can introduce a bottleneck.
Model 3: Gateway Enforcement
Best for: Protecting multiple services, centralized policy, API gateway patterns
Like sidecar, but one Tenuo instance protects many services.
┌─────────────────────────┐
│ Service A │
┌────▶│ (database) │
┌──────────────┐ │ └─────────────────────────┘
│ │ ┌──────────┴───────────┐
│ Agents │──▶│ Tenuo Gateway │
│ │ │ (ext_authz) │
└──────────────┘ └──────────┬───────────┘
│ ┌─────────────────────────┐
└────▶│ Service B │
│ (file storage) │
└─────────────────────────┘
Envoy integration:
Tenuo implements Envoy’s ext_authz protocol. If you already run Envoy or Istio, no sidecar container needed.
http_filters:
- name: envoy.filters.http.ext_authz
typed_config:
"@type": type.googleapis.com/envoy.extensions.filters.http.ext_authz.v3.ExtAuthz
grpc_service:
envoy_grpc:
cluster_name: tenuo-authorizer
How it works:
- Request hits Envoy proxy
- Envoy pauses and asks Tenuo: “Is this warrant valid for
POST /admin?” - Tenuo verifies (stateless, ~27μs)
- Tenuo returns allow/deny
- Envoy forwards or blocks
Security guarantee:
Same as sidecar: tool services are protected regardless of agent compromise. Centralized enforcement simplifies management but introduces a single point of configuration.
Model 4: The “MCP” Pattern (Model Context Protocol)
Best for: MCP-based tool integrations, standardized agent-tool interfaces
MCP standardizes how agents talk to tools. Tenuo acts as the “Middleware” that secures this channel.
- Architecture:
┌──────────────┐ ┌──────────────────┐ ┌──────────────┐
│ │ │ │ │ │
│ Agent │──MCP─▶│ Tenuo Proxy │──MCP─▶│ MCP Server │
│ │ │ │ │ (filesystem,│
│ │ │ Validates │ │ database) │
│ │◀──────│ warrant before │◀──────│ │
│ │ │ forwarding │ │ │
└──────────────┘ └──────────────────┘ └──────────────┘
How it works:
from tenuo.mcp import SecureMCPClient
async with SecureMCPClient("python", ["mcp_server.py"]) as client:
tools = client.tools
# Every call goes through Tenuo authorization
with warrant_scope(warrant), key_scope(keypair):
await tools["read_file"](path="/data/report.txt") # Checked
await tools["read_file"](path="/etc/passwd") # Denied
- Agent connects to Tenuo proxy (not raw MCP server)
- Agent sends MCP
call_toolrequest - Proxy extracts arguments, verifies warrant
- If valid: forwards to real MCP server
- If denied: returns error, MCP server never sees request
Security guarantee:
Protects MCP tool access. The proxy is the trust boundary.
Combining Models (Defense in Depth)
Enforcement Models aren’t mutually exclusive. Layer them:
┌─────────────────────────────────────────────────────┐
│ Agent Process │
│ │
│ @guard ──────────────────────────────────┐ │
│ (Model 1: catches confused deputy) │ │
│ │ │
└────────────────────────────────────────────────┼────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ Tenuo Sidecar │
│ (Model 2: catches compromised agent) │
└────────────────────────────────────────────────┬────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ Tool Service │
│ (Protected by both layers) │
└─────────────────────────────────────────────────────┘
- Model 1 catches prompt injection before it leaves the agent
- Model 2 catches anything that gets past a compromised agent
Belt and suspenders.
Summary
| Goal | Model |
|---|---|
| Protect LangChain/LangGraph agent from prompt injection | Model 1 (In-Process) |
| Protect internal APIs from any caller | Model 2 (Sidecar) |
| Centralized auth for multiple services | Model 3 (Gateway) |
| Secure MCP tool access | Model 4 (MCP Proxy) |
| Maximum security | Combine Model 1 + Model 2 |
See Also
- Kubernetes Deployment — Full sidecar and gateway patterns
- Proxy Configs — Envoy, Istio, nginx configurations
- Security — Threat model and best practices
- LangChain Integration — Tool protection for LangChain