Implemented
SQL Read-Only Policy
Block any SQL capability call that is not a SELECT statement.
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 that inspects the query argument and denies anything that does not start with SELECT.
Complete Example
from brane import CapabilityDeniedError, Decision, Effect, Runtime
# 1. Create runtime
runtime = Runtime(
agent_id="support-agent",
environment="prod",
tenant_id="tenant_acme",
)
# 2. Register capability
@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):
# Real implementation connects to your database
# This stub returns an empty result
return {"rows": [], "count": 0}
# 3. Write the policy
@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")
# 4. Use the capability
try:
# Allowed
result = execute_sql("SELECT * FROM customers WHERE id = 42")
print(result) # {"rows": [], "count": 0}
result = execute_sql("WITH cte AS (SELECT ...) SELECT * FROM cte")
print(result) # {"rows": [], "count": 0}
# Denied — write operation
execute_sql("DELETE FROM customers WHERE id = 42")
except CapabilityDeniedError as e:
print(f"Blocked: {e.reason}")
# Output: Blocked: Only SELECT queries are allowed. Got: delete from customers where id = 42
print(f"Policy: {e.policy_name}")
# Output: Policy: sql_read_onlyWhat Happened
- The agent called
execute_sql("DELETE ...") - Brane intercepted the call before the function ran
- Brane created an
AgentActionforsupport-agentinprod - Brane built a
PolicyContextwithctx.arg("query") == "DELETE ..." - The
sql_read_onlypolicy matched and returneddeny CapabilityDeniedErrorwas raised. The database function never ran.
Variations
Also block EXPLAIN ANALYZE (which can expose query plans):
@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:
# Strict in production
if not query.startswith("select"):
return Decision(type="deny", reason="Only SELECT in prod")
else:
# Allow writes in dev/staging
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
priority=100(or another high number) to ensure this policy runs before any lower-priority policies that might allow the action. - The check is on the raw query string. For production use, consider also checking for SQL injection patterns and parameterized query enforcement.
- When audit sinks are available, every denied query will produce an audit record with the full action — agent, tenant, query string, and denial reason.