Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.brane.membranelabs.org/llms.txt

Use this file to discover all available pages before exploring further.

Block any SQL capability call that is not a read-only query.

​
Problem

An agent has access to an execute_sql capability for looking up customer data. You want to ensure the agent can only read data, never write, update, or delete, regardless of what it is prompted to do.

​
Solution

A before_capability policy inspects the query argument and denies anything that does not start with an allowed read-only prefix.

​
Complete Example

from brane import CapabilityDeniedError, Decision, Effect, Runtime

runtime = Runtime(
    agent_id="support-agent",
    environment="prod",
    tenant_id="tenant_acme",
)

@runtime.capability(
    name="execute_sql",
    type="database",
    risk="high",
    effect=Effect(type="database_query", reversible=True),
    data_namespace="customer.records",
    owner="data-team",
)
def execute_sql(query: str, params: dict | None = None):
    return {"rows": [], "count": 0}

@runtime.before_capability(
    "execute_sql",
    name="sql_read_only",
    version="1.0",
    description="Only SELECT statements are allowed",
    priority=100,
)
def sql_read_only(ctx):
    query = ctx.arg("query", "").strip().lower()
    allowed_prefixes = ("select", "with", "explain")
    if not any(query.startswith(p) for p in allowed_prefixes):
        return Decision(
            type="deny",
            reason=f"Only SELECT queries are allowed. Got: {query[:40]}",
        )
    return Decision(type="allow")

try:
    result = execute_sql("SELECT * FROM customers WHERE id = 42")
    print(result)

    result = execute_sql("WITH cte AS (SELECT ...) SELECT * FROM cte")
    print(result)

    execute_sql("DELETE FROM customers WHERE id = 42")
except CapabilityDeniedError as e:
    print(f"Blocked: {e.reason}")
    print(f"Policy: {e.policy_name}")

​
What Happened

  1. The agent called execute_sql("DELETE ...")
  2. Brane intercepted the call before the function ran
  3. Brane created an AgentAction for support-agent in prod
  4. Brane built a PolicyContext with the query argument
  5. The sql_read_only policy matched and returned deny
  6. CapabilityDeniedError was raised. The database function never ran.

​
Variations

Block EXPLAIN ANALYZE: 
@runtime.before_capability("execute_sql")
def sql_read_only_strict(ctx):
    query = ctx.arg("query", "").strip().lower()
    if query.startswith("explain analyze"):
        return Decision(type="deny", reason="EXPLAIN ANALYZE is not allowed")
    if not any(query.startswith(p) for p in ("select", "with", "explain")):
        return Decision(type="deny", reason="Only SELECT queries are allowed")
    return Decision(type="allow")
Different policy per environment: 
@runtime.before_capability("execute_sql")
def sql_policy(ctx):
    query = ctx.arg("query", "").strip().lower()
    if ctx.is_prod:
        if not query.startswith("select"):
            return Decision(type="deny", reason="Only SELECT in prod")
    else:
        blocked = ("drop", "truncate")
        if any(query.startswith(b) for b in blocked):
            return Decision(type="deny", reason="DROP and TRUNCATE are never allowed")
    return Decision(type="allow")

​
Production Notes

  • Use a high priority to ensure this policy runs before lower-priority policies that might allow the action.
  • The check is on the raw query string. For production use, consider also checking SQL injection patterns and parameterized query enforcement.
  • When audit sinks are available, every denied query will produce an audit record with the full action.