AI agent SQL injection begins at one call site: where an agent's SQL string reaches a database driver. agent_sql_seam.py, a static ast detector, finds that seam before anything runs and classifies every DB sink as RAW_STRING_TO_DB, PARAM_OK, POLICY_MEDIATED, or UNRESOLVED. In this post's fixtures, one f-string marks a sink RAW; one bound parameter marks it PARAM_OK, same rows either way.
AI disclosure: I wrote
agent_sql_seam.pywith an AI assistant and ran it myself, offline, before publishing. Every output block below is pasted from a real local run on Python 3.13.5, standard library only, no network. I checked the exit codes (0 / 1 / 2), hashed the STDOUT of each scenario twice to confirm it is byte-for-byte deterministic, and edited every line. The external quotes and the 1034 / 23 / 0 benchmark numbers belong to Dipankar Sarkar's OrmAI writeup, not to me; I link the primary sources and keep their numbers in their own paragraphs, away from my fixture counts.
In short:
- Whether a query can be injected into is a property of how the SQL string was assembled, not of what the query returns. Two calls can pull the identical rows and only one of them ever holds a string an attacker's value can land inside.
- The tool parses Python with
astand looks at the argument each DB sink receives. An f-string, a%format, a.format(), a concatenation, or a variable assigned to one of those is a seam: RAW_STRING_TO_DB. A literal shipped with bound params is PARAM_OK. - The trap is that
%sreads two ways. Inside a string literal it is a driver placeholder and is safe. As the Python%operator on a string it builds the query at call time and is a seam. One isast.Constant; the other isast.BinOp. - The demo: one text-to-SQL node, one f-string, exit 1. Change that single line to a bound parameter and the same query over the same rows gives exit 0. The diff is one line long.
- Standard library only (
ast,os,sys). Offline, keyless, read-only, zero network, deterministic STDOUT. The tool and every fixture are in this post.
AI agent SQL injection lives in how the string is built
Here is the call I keep finding in text-to-SQL nodes. An agent reads a request, pulls out a vendor name, and drops it into a query:
cur.execute(f"SELECT * FROM invoices WHERE vendor = '{agent_vendor}'")
It works in the demo. It passes review, because reviewers read it as "look up invoices by vendor," which is exactly what it does. The rows come back correct. Nothing about the returned data is wrong. And that is the whole problem: correctness of the result tells you nothing about the safety of the assembly. The moment agent_vendor is a value the model produced from an untrusted request, that f-string is a place where a ' and a trailing OR 1=1 -- become part of the SQL text the driver parses.
Now the version that returns the same rows:
cur.execute("SELECT * FROM invoices WHERE vendor = %s", (agent_vendor,))
At the database this runs the same lookup. The vendor value still comes from the agent. The difference is that the value now rides a bound parameter, so it reaches the driver as a value in its own right, not spliced into the SQL your code assembled. The driver, not your f-string, decides how that value is encoded, so it cannot change the query's structure and there is no string for it to land inside. That single move is what closes the hole, and a static parser can see the move on either side without running a line.
What reaches the driver, exactly
ast gives you the shape of the argument, and the shape is the tell. An f-string with a substitution is an ast.JoinedStr that contains a FormattedValue. A % format is an ast.BinOp whose operator is Mod and whose left side is a string. A .format() is an ast.Call on a string. A concatenation with a variable is an ast.BinOp with Add. Each of those assembles the query at call time. A plain literal is an ast.Constant, and it assembles nothing.
The case people trip on is %s. Look at these two lines:
cur.execute("SELECT * FROM users WHERE id = %s", (uid,)) # placeholder, safe
cur.execute("SELECT * FROM invoices WHERE id = %d" % row_id) # % operator, seam
Same two characters on the page, opposite verdicts. The first %s sits inside an ast.Constant. It is a DBAPI placeholder, and the value travels in the params tuple. The second is the Python % operator applied to a string, an ast.BinOp(Mod), and it builds the query text at call time before the driver ever sees it. (That %d coerces row_id to an int, so this exact line is a seam by how the string is assembled, not a proof it can be injected. The tool flags the assembly and leaves exploitability to a separate question, which is the boundary the closing section draws.) If a detector cannot tell these apart, it either cries wolf on every parameterized query or waves through real string formatting. The whole tool rides on that one distinction, and I test it against itself below.
Run it in sixty seconds
No keys. No network. No install beyond Python. Save the file, point it at a .py file or a directory, run one command. Here is the whole thing, one file, standard library only:
#!/usr/bin/env python3
"""
agent_sql_seam.py -- a static seam detector for the point where an
agent-authored SQL string reaches a database sink, run BEFORE anything ships.
It parses one or more Python files with `ast` (it never executes them) and, for
every DB sink call it finds (`.execute`, `.executemany`, `.executescript`,
`.execute_many`, and a SQLAlchemy `text(...)` wrapper), it classifies the SQL
argument into one of four verdicts:
RAW_STRING_TO_DB -- the SQL string is BUILT AT CALL TIME: an f-string with a
substitution, a `%` format operator on a string, a
`.format()` call, string concatenation with a non-constant,
or a variable that this file assigns to one of those.
There is a string, and something got interpolated into it.
PARAM_OK -- a plain string literal shipped with a separate params
argument (bound placeholders `%s` / `?` / `:name`), or a
fully static literal with no interpolation at all.
POLICY_MEDIATED -- the sink receives a query OBJECT from an allowlisted
builder / ORM construct, not a bare string. The offline
echo of "the database never receives an agent-authored
SQL string": there is no string to inject into.
UNRESOLVED -- the SQL is a variable whose origin does not trace inside
this file (cross-file or dynamic). Counted as a failure,
fail-closed, but flagged apart from RAW so you can whitelist
it on purpose.
The distinction the whole tool turns on: `%s` INSIDE a string literal is a DB
placeholder and is PARAM_OK; the Python `%` OPERATOR applied to a string is a
runtime format and is RAW. Same two characters, opposite verdicts, because one
is `ast.Constant` and the other is `ast.BinOp(op=ast.Mod)`.
Offline. Keyless. Read-only. Zero network. Standard library only (ast, os, sys).
No subprocess, no exec, no eval, no import of the analyzed code, no model, no DB
connection. It does NOT prove a query is exploitable, does NOT run the agent,
does NOT replace parameterization / an ORM / a policy layer, and does NOT do
full cross-file taint tracking. It flags the seam; the fix is architectural.
Exit codes (usable as a CI gate):
0 no sink builds a raw SQL string and nothing is unresolved
1 >=1 RAW_STRING_TO_DB or UNRESOLVED sink
2 bad input (no path, missing path, unreadable/unparseable file, empty scan)
Usage:
python3 agent_sql_seam.py <file.py | directory>
"""
import ast
import os
import sys
# Sink method names. Edit for your driver set (asyncpg .fetch, etc.).
EXEC_ATTRS = {"execute", "executemany", "executescript", "execute_many"}
# Callables that return a query OBJECT rather than a raw string: ORM Core
# constructs plus an allowlist of intent/builder functions. Edit for your stack.
POLICY_BUILDERS = frozenset({
"select", "insert", "update", "delete",
"build_query", "safe_query", "query_builder", "allowlisted_query",
})
# Worst-case wins when a variable has several assignments (fail-closed).
SEVERITY = {"RAW": 3, "UNRESOLVED": 2, "POLICY": 1, "LITERAL": 0}
LABEL = {"RAW": "RAW_STRING_TO_DB", "UNRESOLVED": "UNRESOLVED",
"POLICY": "POLICY_MEDIATED", "LITERAL": "PARAM_OK"}
def _bad(msg):
print("ERROR: " + msg)
raise SystemExit(2)
def _worst(classes):
return max(classes, key=lambda c: SEVERITY[c])
def _call_root_name(node):
"""Unwind a call/attribute chain to its root Name id, e.g.
select(x).where(y) -> 'select'. Returns None if the root is not a Name."""
cur = node
while True:
if isinstance(cur, ast.Call):
cur = cur.func
elif isinstance(cur, ast.Attribute):
cur = cur.value
elif isinstance(cur, ast.Name):
return cur.id
else:
return None
def _is_str_cj(node):
"""A string constant or an f-string node."""
if isinstance(node, ast.Constant) and isinstance(node.value, str):
return True
return isinstance(node, ast.JoinedStr)
def _is_text_call(node):
if not isinstance(node, ast.Call):
return False
f = node.func
if isinstance(f, ast.Name):
return f.id == "text"
if isinstance(f, ast.Attribute):
return f.attr == "text"
return False
def _is_execute_call(node):
return (isinstance(node, ast.Call)
and isinstance(node.func, ast.Attribute)
and node.func.attr in EXEC_ATTRS)
def _flatten_add(node):
if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add):
return _flatten_add(node.left) + _flatten_add(node.right)
return [node]
def build_assign_map(tree):
"""name -> list of assigned value nodes (module + function scope, flattened,
one hop, shallow on purpose). Plus the set of names built by `+=`."""
assign = {}
aug_add = set()
for node in ast.walk(tree):
if isinstance(node, ast.Assign) and len(node.targets) == 1 \
and isinstance(node.targets[0], ast.Name):
assign.setdefault(node.targets[0].id, []).append(node.value)
elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name) \
and node.value is not None:
assign.setdefault(node.target.id, []).append(node.value)
elif isinstance(node, ast.AugAssign) and isinstance(node.target, ast.Name) \
and isinstance(node.op, ast.Add):
aug_add.add(node.target.id)
return assign, aug_add
def _name_is_str(name, assign):
"""One hop: does this name resolve to a string literal / f-string?"""
for val in assign.get(name, []):
if _is_str_cj(val):
return True
return False
def classify_expr_direct(node, assign):
"""Direct class of an expression node, or None if it needs a name trace.
Returns one of RAW / LITERAL / POLICY / None."""
if isinstance(node, ast.Constant) and isinstance(node.value, str):
return "LITERAL"
if isinstance(node, ast.JoinedStr):
if any(isinstance(v, ast.FormattedValue) for v in node.values):
return "RAW"
return "LITERAL"
if isinstance(node, ast.BinOp):
if isinstance(node.op, ast.Mod):
left = node.left
if _is_str_cj(left) or (isinstance(left, ast.Name)
and _name_is_str(left.id, assign)):
return "RAW"
return None
if isinstance(node.op, ast.Add):
leaves = _flatten_add(node)
has_str = any(_is_str_cj(l) for l in leaves) or any(
isinstance(l, ast.Name) and _name_is_str(l.id, assign)
for l in leaves)
has_dyn = any(not isinstance(l, ast.Constant) for l in leaves)
if has_str and has_dyn:
return "RAW"
if has_str and not has_dyn:
return "LITERAL"
return None
return None
if isinstance(node, ast.Call):
f = node.func
if isinstance(f, ast.Attribute) and f.attr == "format":
if _is_str_cj(f.value) or isinstance(f.value, ast.Name):
return "RAW"
return None
if _call_root_name(node) in POLICY_BUILDERS:
return "POLICY"
return None
return None
def _argkind(node):
if isinstance(node, ast.Constant) and isinstance(node.value, str):
return "string literal"
if isinstance(node, ast.JoinedStr):
return "f-string"
if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Mod):
return "%-format"
if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add):
return "string concat"
if isinstance(node, ast.Call):
f = node.func
if isinstance(f, ast.Attribute) and f.attr == "format":
return ".format()"
root = _call_root_name(node)
if root in POLICY_BUILDERS:
return "builder call"
return "call '%s()'" % (root or "?")
if isinstance(node, ast.Name):
return "variable '%s'" % node.id
return type(node).__name__
def _raw_reason(argkind):
table = {
"f-string": "f-string interpolation builds the SQL string at call time",
"%-format": "%-operator formatting builds the SQL string at call time",
"string concat": "string concatenation builds the SQL string at call time",
".format()": ".format() builds the SQL string at call time",
}
return table.get(argkind, "the SQL string is built at call time")
def _trace_name(name, assign, aug_add):
ak = "variable '%s'" % name
if name not in assign and name not in aug_add:
return ("UNRESOLVED", ak,
"variable '%s' has no in-file assignment "
"(cross-file or dynamic origin)" % name)
classes, kinds = set(), []
if name in aug_add:
classes.add("RAW")
kinds.append("augmented concat")
for val in assign.get(name, []):
d = classify_expr_direct(val, assign)
classes.add(d if d is not None else "UNRESOLVED")
kinds.append(_argkind(val))
worst = _worst(classes)
joined = "/".join(sorted(set(kinds)))
if worst == "RAW":
return "RAW", ak, ("variable '%s' is assigned a runtime-built SQL string "
"(%s)" % (name, joined))
if worst == "UNRESOLVED":
return "UNRESOLVED", ak, ("variable '%s' is assigned from %s the linter "
"cannot resolve" % (name, joined))
if worst == "POLICY":
return "POLICY", ak, "variable '%s' routed through an allowlisted builder" % name
return "LITERAL", ak, "variable '%s' is a static literal" % name
def classify_sink_arg(node, assign, aug_add, has_params):
"""Return (class, argkind, note) for the SQL argument of a sink."""
if _is_text_call(node):
inner = node.args[0] if node.args else None
if inner is None:
return "UNRESOLVED", "text()", "text() called with no SQL argument"
return classify_sink_arg(inner, assign, aug_add, has_params=True)
direct = classify_expr_direct(node, assign)
ak = _argkind(node)
if direct == "RAW":
return "RAW", ak, _raw_reason(ak)
if direct == "LITERAL":
return "LITERAL", ak, ("literal + bound params" if has_params else "static literal")
if direct == "POLICY":
return "POLICY", ak, "routed through allowlisted builder '%s'" % _call_root_name(node)
if isinstance(node, ast.Name):
return _trace_name(node.id, assign, aug_add)
return ("UNRESOLVED", ak,
"sink receives a %s the linter cannot resolve to a literal, a "
"bound-param call, or an allowlisted builder" % ak)
def analyze_file(path, tree):
assign, aug_add = build_assign_map(tree)
parent = {}
for node in ast.walk(tree):
for child in ast.iter_child_nodes(node):
parent[child] = node
sinks = []
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
if _is_execute_call(node):
has_params = len(node.args) > 1
arg = node.args[0] if node.args else None
if arg is None:
klass, ak, note = "UNRESOLVED", "(no arg)", "sink called with no SQL argument"
else:
klass, ak, note = classify_sink_arg(arg, assign, aug_add, has_params)
func = ast.unparse(node.func)
elif _is_text_call(node):
p = parent.get(node)
if p is not None and _is_execute_call(p) and p.args and p.args[0] is node:
continue # counted at the execute site
klass, ak, note = classify_sink_arg(node, assign, aug_add, has_params=True)
func = ast.unparse(node.func)
else:
continue
sinks.append({"file": path, "line": node.lineno, "func": func,
"argkind": ak, "klass": klass, "note": note})
return sinks
def collect_files(path):
if os.path.isdir(path):
files = []
for root, dirs, names in os.walk(path):
dirs.sort()
for name in sorted(names):
if name.endswith(".py"):
files.append(os.path.join(root, name))
return sorted(files)
return [path]
def main(argv):
if len(argv) != 2:
print("usage: agent_sql_seam.py <file.py | directory>")
raise SystemExit(2)
path = argv[1]
if not os.path.exists(path):
_bad("path does not exist: %s" % path)
files = collect_files(path)
if not files:
_bad("no .py files found under %s" % path)
all_sinks = []
for f in files:
try:
with open(f, "r") as fh:
src = fh.read()
except OSError as exc:
_bad("cannot read %s: %s" % (f, exc))
try:
tree = ast.parse(src, filename=f)
except SyntaxError as exc:
_bad("cannot parse %s: %s" % (f, exc))
all_sinks.extend(analyze_file(f, tree))
all_sinks.sort(key=lambda s: (s["file"], s["line"]))
counts = {"RAW": 0, "UNRESOLVED": 0, "POLICY": 0, "LITERAL": 0}
for s in all_sinks:
counts[s["klass"]] += 1
fails = counts["RAW"] + counts["UNRESOLVED"]
out = []
out.append("AGENT-SQL-SEAM REPORT")
out.append("files scanned: %d" % len(files))
out.append("db sinks found: %d" % len(all_sinks))
out.append(" RAW_STRING_TO_DB: %d" % counts["RAW"])
out.append(" UNRESOLVED: %d" % counts["UNRESOLVED"])
out.append(" POLICY_MEDIATED: %d" % counts["POLICY"])
out.append(" PARAM_OK: %d" % counts["LITERAL"])
out.append("sinks:")
for s in all_sinks:
if s["klass"] in ("RAW", "UNRESOLVED"):
out.append(" - %s:%d %s(%s) -> %s"
% (s["file"], s["line"], s["func"], s["argkind"],
LABEL[s["klass"]]))
out.append(" %s" % s["note"])
else:
out.append(" - %s:%d %s(%s) -> %s [%s]"
% (s["file"], s["line"], s["func"], s["argkind"],
LABEL[s["klass"]], s["note"]))
if fails:
out.append("VERDICT: FAIL: %d sink(s) receive a runtime-built or "
"unresolved SQL string" % fails)
out.append(" the same rows ship from a parameterized call with no "
"string to inject into")
code = 1
else:
out.append("VERDICT: PASS: every db sink gets a bound-param literal, a "
"static literal, or a policy-mediated query")
code = 0
print("\n".join(out))
raise SystemExit(code)
if __name__ == "__main__":
main(sys.argv)
The baseline: every sink is safe
Start with an agent tool that does nothing wrong. It reads a balance with a bound %s param, lists invoices with ? placeholders, runs a fully static count(*), and builds one query through a SQLAlchemy select(...) construct instead of a string. Four sinks, four clean assemblies.
$ python3 agent_sql_seam.py fixtures/clean.py
AGENT-SQL-SEAM REPORT
files scanned: 1
db sinks found: 4
RAW_STRING_TO_DB: 0
UNRESOLVED: 0
POLICY_MEDIATED: 1
PARAM_OK: 3
sinks:
- fixtures/clean.py:14 cur.execute(string literal) -> PARAM_OK [literal + bound params]
- fixtures/clean.py:19 cur.execute(string literal) -> PARAM_OK [literal + bound params]
- fixtures/clean.py:27 cur.execute(string literal) -> PARAM_OK [static literal]
- fixtures/clean.py:33 conn.execute(variable 'stmt') -> POLICY_MEDIATED [variable 'stmt' routed through an allowlisted builder]
VERDICT: PASS: every db sink gets a bound-param literal, a static literal, or a policy-mediated query
Exit 0. Note the last line. stmt = select(invoices).where(...) is a query object, so the sink never receives a bare string. That is the POLICY_MEDIATED class, and it is the static shadow of the architecture the OrmAI writeup argues for below: hand the driver a structured query, and there is nothing left to inject into. The tool trusts that the named builder is safe, which is a real limit I come back to later.
One line decides whether there is a string to inject into
This is the demo the post exists for. Two files, a text-to-SQL node that returns invoices for a vendor the agent extracted. They differ by one line:
$ diff fixtures/killer_raw.py fixtures/killer_fixed.py
9c9
< cur.execute(f"SELECT * FROM invoices WHERE vendor = '{vendor}'")
---
> cur.execute("SELECT * FROM invoices WHERE vendor = %s", (vendor,))
Same function, same table, same rows for any given vendor, provided the %s matches your driver. %s is the psycopg2 and mysql-connector placeholder; sqlite3 spells the same bind ? (both are PARAM_OK, see the edge fixture). The detector reads the assembly and does not care which style you use. Run the detector on each:
$ python3 agent_sql_seam.py fixtures/killer_raw.py
AGENT-SQL-SEAM REPORT
files scanned: 1
db sinks found: 1
RAW_STRING_TO_DB: 1
UNRESOLVED: 0
POLICY_MEDIATED: 0
PARAM_OK: 0
sinks:
- fixtures/killer_raw.py:9 cur.execute(f-string) -> RAW_STRING_TO_DB
f-string interpolation builds the SQL string at call time
VERDICT: FAIL: 1 sink(s) receive a runtime-built or unresolved SQL string
the same rows ship from a parameterized call with no string to inject into
Exit 1. Now the fixed twin:
$ python3 agent_sql_seam.py fixtures/killer_fixed.py
AGENT-SQL-SEAM REPORT
files scanned: 1
db sinks found: 1
RAW_STRING_TO_DB: 0
UNRESOLVED: 0
POLICY_MEDIATED: 0
PARAM_OK: 1
sinks:
- fixtures/killer_fixed.py:9 cur.execute(string literal) -> PARAM_OK [literal + bound params]
VERDICT: PASS: every db sink gets a bound-param literal, a static literal, or a policy-mediated query
Exit 0. The vendor value still comes from the agent in both files. Nothing about the data changed. The verdict flipped because the seam moved: in the first file the value lands inside the query string, in the second it rides a bound parameter and reaches the driver as data. This is the part I want you to sit with. A behavioral test cannot tell these two apart, because they behave identically on every honest input. The seam only shows up when someone sends a dishonest one, and by then the check you needed ran days ago.
Four seams in one text-to-SQL node
Real code rarely has one style. Here is a node that leaks the same way four different times: an f-string, a .format(), a concatenation, and a % operator.
def by_vendor(cur, agent_vendor):
cur.execute(f"SELECT * FROM invoices WHERE vendor = '{agent_vendor}'")
def by_region(cur, region):
cur.execute("SELECT * FROM invoices WHERE region = '{}'".format(region))
def by_status(cur, status):
cur.execute("SELECT * FROM invoices WHERE status = '" + status + "'")
def by_id(cur, row_id):
cur.execute("SELECT * FROM invoices WHERE id = %d" % row_id)
$ python3 agent_sql_seam.py fixtures/violating.py
AGENT-SQL-SEAM REPORT
files scanned: 1
db sinks found: 4
RAW_STRING_TO_DB: 4
UNRESOLVED: 0
POLICY_MEDIATED: 0
PARAM_OK: 0
sinks:
- fixtures/violating.py:11 cur.execute(f-string) -> RAW_STRING_TO_DB
f-string interpolation builds the SQL string at call time
- fixtures/violating.py:16 cur.execute(.format()) -> RAW_STRING_TO_DB
.format() builds the SQL string at call time
- fixtures/violating.py:21 cur.execute(string concat) -> RAW_STRING_TO_DB
string concatenation builds the SQL string at call time
- fixtures/violating.py:26 cur.execute(%-format) -> RAW_STRING_TO_DB
%-operator formatting builds the SQL string at call time
VERDICT: FAIL: 4 sink(s) receive a runtime-built or unresolved SQL string
the same rows ship from a parameterized call with no string to inject into
Exit 1, four seams, each with the line number and the reason. The %d on line 26 is worth a second look next to the placeholder case, because it is the same % symbol doing the opposite thing.
The falsifiability test: %s is not always a seam
If a bound %s inside a literal got flagged as RAW, the thesis would be broken, and the tool would be noise. So here is the counter-fixture: four calls that all look tempting and are all safe. A %s placeholder with params. A ? placeholder with params. A static SELECT 1. A SQLAlchemy text() around a literal with a :name bind.
def by_id(cur, uid):
cur.execute("SELECT * FROM users WHERE id = %s", (uid,))
def sqlite_by_id(cur, uid):
cur.execute("SELECT * FROM users WHERE id = ?", (uid,))
def ping(cur):
cur.execute("SELECT 1")
def by_name(conn, name):
conn.execute(text("SELECT * FROM users WHERE name = :name"), {"name": name})
$ python3 agent_sql_seam.py fixtures/edge.py
AGENT-SQL-SEAM REPORT
files scanned: 1
db sinks found: 4
RAW_STRING_TO_DB: 0
UNRESOLVED: 0
POLICY_MEDIATED: 0
PARAM_OK: 4
sinks:
- fixtures/edge.py:15 cur.execute(string literal) -> PARAM_OK [literal + bound params]
- fixtures/edge.py:20 cur.execute(string literal) -> PARAM_OK [literal + bound params]
- fixtures/edge.py:25 cur.execute(string literal) -> PARAM_OK [static literal]
- fixtures/edge.py:29 conn.execute(string literal) -> PARAM_OK [literal + bound params]
VERDICT: PASS: every db sink gets a bound-param literal, a static literal, or a policy-mediated query
Exit 0. Put this run next to the %d on line 26 of the violating file. Same character on the page, and the detector splits them: the %s living inside an ast.Constant is a placeholder, the % operator building an ast.BinOp is a seam. That split is the tool earning its keep. A grep for %s cannot do it. A grep would flag the safe line and the dangerous line the same.
When the origin will not trace, fail closed
Not every SQL argument is a literal or an obvious expression. Sometimes it is a variable filled by a function this file cannot see:
from templates import load_template
def run(cur, report_name):
sql = load_template(report_name)
cur.execute(sql)
The honest answer is "I do not know." load_template lives in another module. Statically, this file cannot say whether it returns a safe constant or a spliced string. The tool refuses to guess in your favor:
$ python3 agent_sql_seam.py fixtures/unresolved.py
AGENT-SQL-SEAM REPORT
files scanned: 1
db sinks found: 1
RAW_STRING_TO_DB: 0
UNRESOLVED: 1
POLICY_MEDIATED: 0
PARAM_OK: 0
sinks:
- fixtures/unresolved.py:14 cur.execute(variable 'sql') -> UNRESOLVED
variable 'sql' is assigned from call 'load_template()' the linter cannot resolve
VERDICT: FAIL: 1 sink(s) receive a runtime-built or unresolved SQL string
the same rows ship from a parameterized call with no string to inject into
Exit 1, but filed as UNRESOLVED, not RAW. That is deliberate. UNRESOLVED means "the string may be perfectly safe, but this tool cannot see far enough to say so." If you know load_template returns only vetted constants, you whitelist this call on purpose and move on. Fail-closed here means an unknown costs you a review, not a silent pass. A tool that guessed "probably fine" on things it cannot see would be worse than no tool, because it would teach you to trust a green run it did not earn.
How does the detector classify each sink?
Small enough to hold in your head. For each Call node, ask if it is a DB sink: an execute-family attribute call, or a text() wrapper (deduped, so execute(text("...")) is counted once, at the execute). Take the SQL argument. If it is an ast.Constant string, it is PARAM_OK, with the note distinguishing a bound-param literal from a static one. If it is a JoinedStr with a substitution, a BinOp(Mod) on a string, a .format(), or a string concatenation with a non-constant, it is RAW. If it is a call whose root name is in the builder allowlist, it is POLICY_MEDIATED. If it is a variable, trace one hop to its in-file assignments and take the worst class those produce; a variable with no traceable origin is UNRESOLVED. Every RAW or UNRESOLVED sink pushes the exit code to 1.
One design choice a reviewer will poke. Variable tracing is one hop and file-local by construction. If sql is assigned from another variable, or built across two modules, the tool reports UNRESOLVED rather than chasing it. That is not laziness disguised as caution; a shallow tracer that fails closed is honest about its reach, and a deep one that occasionally guesses wrong on a real codebase would hand you false confidence, which is the one thing a security check must never do.
The architecture crowd is drawing the same line
I am not the first to point at this seam, and the strongest recent statement of the fix is not a linter at all. On July 2, Dipankar Sarkar published a benchmark of two ways to let an agent hit a database. I checked the post and the repo myself before quoting.
Their numbers, from their run, on their bench, stated as theirs: they ran an agent against the Spider dataset, 1034 natural language queries. A text-to-SQL setup, where the model writes SQL, executed 23 unsafe operations. A policy layer that routes intent through fixed queries executed 0. Those are OrmAI's figures for OrmAI, not measurements of anyone else's stack, and none of them are mine.
The line I want to borrow is their conclusion about why the second number is 0. Their phrasing, from the writeup: "there is no string to inject into." The database in their design never receives an agent-authored SQL string, because the agent emits a structured intent and the server builds the query from a template it controls. That is the same property my POLICY_MEDIATED class checks for, from the other end. They enforce it at runtime, in their OrmAI repo. I detect, before you ship, which of your sinks already have that property and which still hold a raw string. We are pointed at one seam from two sides, and neither replaces the other: a runtime enforcer stops the call, a static detector tells you where the calls that need fixing live.
Where this sits next to the rest
This is a spoke on the pre-execution gate for AI agents cluster, and its object is the data plane: the SQL string that reaches the driver, and whether the agent authored it. The neighbors, and how this differs:
- Write-chain taint lint for agent gates traces who wrote the signals a gate authorizes on. Same provenance instinct, different object: that one asks who filled a trust table, this one asks who assembled a query string.
- A gate that compares trace against policy works the control plane: was an action allowed. This works the data plane: can the payload be injected into. Orthogonal questions about the same call.
- The lethal trifecta reachability gate asks whether a dangerous capability path exists across a tool manifest. This asks a per-call-site question about one argument, no reachability graph involved.
- The blast radius of an agent's API key is the scenario people mean when they say "the agent deleted the prod database." A raw write built from agent output is exactly how a small mistake becomes a large one; this finds the write before it ships.
- The supply-chain gate before install fires at a different point in the lifecycle, when a package arrives, not when a query runs. Same fail-closed, pre-execution shape, earlier gate.
What this is NOT
I would rather undersell this than have you deploy it as something it is not.
- It is not the first SQL injection linter, and I am not claiming a new detector for hardcoded SQL. Bandit's
B608(hardcoded_sql_expressions) has flagged string-built queries for years, and you should keep running it. What is framed differently here is the agent case: the author of the string is a model at runtime, not a developer at the keyboard; the POLICY_MEDIATED class names an architecture (route intent through a builder, ship no string) that generic SQLi linters do not model; and the whole thing hangs off distinct exit codes meant for a pre-execution gate. Bandit runs in CI too, so the gate itself is not the new part; the framing and the third class are, not the discovery that f-strings inexecuteare dangerous. - It does not prove exploitability. RAW_STRING_TO_DB means a runtime-built string reaches a sink, which is the seam. Whether an attacker can actually reach that argument with a hostile value is a separate question this tool does not answer. A string built entirely from a trusted constant still gets flagged, on purpose, because the tool reads the assembly, not the trust of the inputs.
- It is not a runtime interceptor and not OrmAI. It opens no connection, runs no SQL, blocks nothing. It is a static check you run in CI before you ship. To stop a live call you need enforcement in the driver or a policy layer.
- It is not a full taint tracker. Variable tracing is one hop, name-only, and file-local. Most of what it cannot follow (cross-file calls, multi-hop assignments,
str.join, a.format()on a call) comes back UNRESOLVED and fails closed. A few in-file rewrites slip the other way: an augmented format likesql %= vendoris not tracked, so a name first assigned a literal and then mutated in place can still read as PARAM_OK. If you lean on in-place string mutation right next to a sink, do not trust a green from this tool alone. Real taint engines reconstruct data flow across modules; this classifies what one file makes visible. - It finds sinks by method name, not by object type. The sink set is
.execute/.executemany/.executescript/.execute_manyplus atext()wrapper. An execute-style call it does not know (asyncpg's.fetch, a Django.raw()) is invisible to it and passes silently, and a non-DB.execute(a subprocess or Redis wrapper) can draw a false flag. Atext()assigned to a variable and then executed comes back UNRESOLVED rather than resolved, erring loud rather than quiet. EditEXEC_ATTRSandPOLICY_BUILDERSfor your stack; the tool fails closed only inside the names it recognizes. - It checks a builder by name, not by what it returns. POLICY_MEDIATED fires when the call's root name is on the allowlist (
select,insert,build_query, and the rest); it does not inspect the body, even one defined in the same file. So a function you happen to nameselectthat returns an f-string, or abuild_querythat interpolates agent output internally, passes as a clean policy-mediated call it did not earn. The allowlist is a promise you make; the tool only checks that you routed through a name on it. - The counts here are fixture units, not a measurement of anyone's production. The 4 seams and the 1 policy-mediated sink describe the synthetic files in this post. Run it on your own repo to get numbers that mean something about your code.
Bad input fails closed
A gate that crashes open is worse than no gate. Point it at a file that does not parse, and it refuses to analyze rather than pretend the file is clean:
$ python3 agent_sql_seam.py fixtures/bad_syntax.py
ERROR: cannot parse fixtures/bad_syntax.py: invalid syntax (bad_syntax.py, line 1)
$ echo $?
2
No path argument, a missing path, an unreadable file, an empty scan: all exit 2, distinct from exit 1 so your CI can tell "found a seam" apart from "could not read the input." I ran each fixture twice and hashed the full STDOUT, trailing newline stripped, both times: clean is e7d0ff4f..., violating is 89c00229..., edge is a0b974e8..., unresolved is 4be175f0..., killer_raw is c13cecb7..., killer_fixed is 1f7422bd.... Identical across both runs, on Python 3.13.5, offline.
The question I actually want answered
Here is the one I do not have a good number for. For teams running a text-to-SQL node or a RAG-to-database tool in production: how many of your DB sinks would come back RAW or UNRESOLVED if you ran this against the repo today? Not "do you parameterize," which everyone says yes to, but the count after the one f-string someone added in a hurry, the .format() that felt harmless, the helper two modules over that the tracer cannot follow. If you export even a rough run, tell me the split between RAW and UNRESOLVED. I suspect UNRESOLVED is the bigger pile for most stacks, and I would like to know if I am wrong.
If this was useful, follow along here for the next runnable gate in the series, and drop the strangest place you have found an agent-built SQL string reaching a driver in the comments. I read every one.
Top comments (0)