> ## 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.

# SQL Read-Only Policy

> Block any SQL capability call that is not a SELECT statement.

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/before-capability) policy inspects the `query` argument and denies anything that does not start with an allowed read-only prefix.

## Complete Example

```python theme={null}
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`:

```python theme={null}
@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:

```python theme={null}
@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.

## Related

* [Before Capability Policies](/policy/before-capability)
* [PolicyContext](/concepts/policy-context)
* [Refund Limit Recipe](/recipes/refund-limit)