AI 에이전트 SQL 인젝션 (SQL injection)은 하나의 호출 지점, 즉 에이전트의 SQL 문자열이 데이터베이스 드라이버(database driver)에 도달하는 지점에서 시작됩니다. 정적 ast 탐지기인 agent_sql_seam.py는 코드가 실행되기 전에 해당 이음매(seam)를 찾아내며, 모든 DB 싱크(sink)를 RAW_STRING_TO_DB, PARAM_OK, POLICY_MEDIATED 또는 UNRESOLVED로 분류합니다. 이 포스트의 피스처(fixtures)에서는 하나의 f-string이 RAW 싱크로 표시되고, 하나의 바인드 파라미터(bound parameter)가 PARAM_OK로 표시되지만, 결과적으로 동일한 행을 대상으로 합니다.

AI 공개 사항: 저는 AI 어시스턴트와 함께 agent_sql_seam.py를 작성하였으며, 게시하기 전에 오프라인에서 직접 실행했습니다. 아래의 모든 출력 블록은 Python 3.13.5, 표준 라이브러리만 사용하고 네트워크 연결 없이 로컬에서 실제로 실행한 결과를 복사한 것입니다. 저는 종료 코드(0 / 1 / 2)를 확인하였고, 각 시나리오의 STDOUT을 두 번 해싱하여 바이트 단위로 결정론적(deterministic)임을 확인하였으며, 모든 줄을 직접 편집했습니다. 외부 인용구와 1034 / 23 / 0 벤치마크 수치는 제가 아닌 Dipankar Sarkar의 OrmAI 기술 문서에 속한 것이며, 저는 기본 출처를 링크하고 해당 수치들을 제 피스처 수치와 분리하여 각자의 단락에 유지했습니다.

요약하자면:

• 쿼리에 주입(injection)이 가능한지 여부는 쿼리가 무엇을 반환하는가가 아니라, SQL 문자열이 어떻게 조립되었는지에 따른 속성입니다. 두 번의 호출이 동일한 행을 가져오더라도, 그중 단 하나만이 공격자의 값이 들어갈 수 있는 문자열을 보유할 수 있습니다.
• 이 도구는 ast를 사용하여 Python을 파싱하고, 각 DB 싱크(sink)가 받는 인자를 살펴봅니다. f-string, % 포맷팅, .format(), 결합(concatenation), 또는 이러한 방식 중 하나에 할당된 변수는 RAW_STRING_TO_DB라는 이음새(seam)가 됩니다. 바인딩된 파라미터(bound params)와 함께 전달되는 리터럴은 PARAM_OK입니다.
• 함정은 %s를 두 가지 방식으로 읽는다는 점입니다. 문자열 리터럴 내부에서 이는 드라이버 플레이스홀더(driver placeholder)이며 안전합니다. 문자열에 대한 Python % 연산자로 사용될 때는 호출 시점에 쿼리를 생성하며 이는 이음새(seam)가 됩니다. 하나는 ast.Constant이고, 다른 하나는 ast.BinOp입니다.
• 데모: 하나의 text-to-SQL 노드, 하나의 f-string, 종료 코드 1(exit 1). 해당 한 줄을 바인딩된 파라미터(bound parameter)로 변경하면, 동일한 행에 대해 동일한 쿼리를 실행했을 때 종료 코드 0(exit 0)이 나옵니다. 차이점은 단 한 줄입니다.
• 표준 라이브러리만 사용합니다 (ast, os, sys). 오프라인, 키리스(keyless), 읽기 전용, 네트워크 제로, 결정론적 STDOUT(deterministic STDOUT). 도구와 모든 피스처(fixture)는 이 포스트에 포함되어 있습니다.

AI 에이전트 SQL 주입은 문자열이 구축되는 방식에 존재합니다

다음은 제가 text-to-SQL 노드에서 계속 발견하는 호출 방식입니다. 에이전트가 요청을 읽고, 공급업체 이름을 추출하여 쿼리에 삽입합니다:

cur.execute(f"SELECT * FROM invoices WHERE vendor = '{agent_vendor}'")

데모에서는 작동합니다. 리뷰도 통과합니다. 리뷰어들이 이를

데이터베이스에서 이 작업은 동일한 조회를 실행합니다. vendor 값은 여전히 에이전트(agent)로부터 전달됩니다. 차이점은 이제 해당 값이 바운드 파라미터 (bound parameter)를 타고 전달된다는 것입니다. 따라서 이 값은 코드가 조립한 SQL에 삽입되는 것이 아니라, 그 자체로 하나의 값으로서 드라이버 (driver)에 도달합니다. f-string이 아닌 드라이버가 해당 값을 어떻게 인코딩할지 결정하므로, 쿼리의 구조를 변경할 수 없으며 값이 들어갈 문자열 자체가 존재하지 않습니다. 이 단 한 번의 조치가 취약점 (hole)을 메우는 것이며, 정적 파서 (static parser)는 코드를 실행하지 않고도 양쪽 모두에서 이 조치를 확인할 수 있습니다.

정확히 무엇이 드라이버에 도달하는가

ast는 인자 (argument)의 형태를 제공하며, 그 형태가 핵심을 알려줍니다. 치환이 포함된 f-string은 FormattedValue를 포함하는 ast.JoinedStr입니다. % 포맷은 연산자가 Mod이고 왼쪽 항이 문자열인 ast.BinOp입니다. .format()은 문자열에 대한 ast.Call입니다. 변수를 사용한 결합 (concatenation)은 Add 연산자를 가진 ast.BinOp입니다. 이들 각각은 호출 시점에 쿼리를 조립합니다. 일반적인 리터럴 (literal)은 ast.Constant이며, 아무것도 조립하지 않습니다.

사람들이 실수하는 사례는 %s입니다. 다음 두 줄을 살펴보세요:

cur.execute("SELECT * FROM users WHERE id = %s", (uid,))   # 플레이스홀더 (placeholder), 안전함
cur.execute("SELECT * FROM invoices WHERE id = %d" % row_id)  # % 연산자, 틈새 (seam)

페이지 상의 동일한 두 글자이지만, 판결은 정반대입니다. 첫 번째 %s는 ast.Constant 내부에 위치합니다. 이것은 DBAPI 플레이스홀더 (placeholder)이며, 값은 params 튜플을 통해 전달됩니다. 두 번째는 문자열에 적용된 Python % 연산자로, ast.BinOp(Mod)이며 드라이버가 확인하기 전 호출 시점에 쿼리 텍스트를 생성합니다. (해당 %d는 row_id를 정수(int)로 강제 변환하므로, 이 정확한 라인은 문자열이 조립되는 방식에 따른 틈새 (seam)일 뿐, 주입이 가능하다는 증거는 아닙니다. 도구는 조립 과정을 플래그 (flag)로 표시하고, 취약성 여부는 별개의 문제로 남겨두며, 이는 마지막 섹션에서 다루는 경계선입니다.) 만약 탐지기가 이 둘을 구분하지 못한다면, 모든 매개변수화된 쿼리 (parameterized query)에 대해 잘못된 경보를 울리거나, 실제 문자열 포매팅 (string formatting)을 그냥 통과시켜 버릴 것입니다. 도구 전체가 바로 이 한 가지 차이점에 달려 있으며, 아래에서 이를 직접 테스트해 보겠습니다.

60초 안에 실행하기

키(key)도 필요 없습니다. 네트워크도 필요 없습니다. Python 외에 설치할 것도 없습니다. 파일을 저장하고, .py 파일이나 디렉토리를 지정한 뒤, 명령어 하나만 실행하면 됩니다. 표준 라이브러리만 사용하는 단일 파일 전체 코드는 다음과 같습니다:

#!/usr/bin/env python3
"""
agent_sql_seam.py -- 에이전트가 작성한 SQL 문자열이 데이터베이스 싱크 (sink)에 도달하는 지점을 찾아내는 정적 틈새 (static seam) 탐지기입니다. 코드가 배포되기 전(BEFORE)에 실행하십시오.

이 도구는 `ast`를 사용하여 하나 이상의 Python 파일을 파싱하며 (절대 실행하지 않음), 발견되는 모든 DB 싱크 호출(`.execute`, `.executemany`, `.executescript`, `.execute_many`, 그리고 SQLAlchemy의 `text(...)` 래퍼)에 대해 SQL 인자를 네 가지 판결 중 하나로 분류합니다:

RAW_STRING_TO_DB -- SQL 문자열이 호출 시점에 생성됨: 치환이 포함된 f-string, 문자열에 대한 % 포맷 연산자, .format() 호출, 상수가 아닌 값과의 문자열 연결(string concatenation), 또는 이 파일이 위 방식 중 하나로 할당하는 변수. 문자열이 존재하며, 그 안에 무언가가 보간(interpolated)되었습니다.
PARAM_OK -- 별도의 파라미터 인자(바인딩된 플레이스홀더 %s / ? / :name)와 함께 전달되는 일반 문자열 리터럴(plain string literal), 또는 보간이 전혀 없는 완전한 정적 리터럴(static literal).
POLICY_MEDIATED -- 싱크(sink)가 가공되지 않은 문자열이 아닌, 허용 목록에 있는 빌더(builder) 또는 ORM 구조로부터 쿼리 객체(query OBJECT)를 전달받음. "데이터베이스는 에이전트가 작성한 SQL 문자열을 절대 받지 않는다"라는 오프라인 에코(offline echo): 주입할 문자열 자체가 존재하지 않음.
UNRESOLVED -- SQL이 이 파일 내부에서 추적되지 않는 기원(파일 간 참조 또는 동적)을 가진 변수임. 실패로 간주하며, 기본적으로 차단(fail-closed)하되, 의도적으로 화이트리스트에 추가할 수 있도록 RAW와는 별도로 표시함.

도구 전체의 핵심이 되는 구분점: 문자열 리터럴 '내부'의 %s는 DB 플레이스홀더(placeholder)이므로 PARAM_OK이며, 문자열에 적용된 Python의 % 연산자(operator)는 런타임 포맷(runtime format)이므로 RAW입니다. 동일한 두 글자이지만, 하나는 ast.Constant이고 다른 하나는 ast.BinOp(op=ast.Mod)이기 때문에 판결이 정반대로 내려집니다.

오프라인. 키리스(Keyless). 읽기 전용. 네트워크 사용 없음. 표준 라이브러리만 사용 (ast, os, sys). subprocess, exec, eval 미사용. 분석 대상 코드를 임포트하지 않음. 모델이나 DB 연결을 사용하지 않음. 이 도구는 쿼리의 취약성(exploitable)을 증명하는 것이 아니며, 에이전트를 실행하는 것도 아니고, 파라미터화(parameterization) / ORM / 정책 계층을 대체하는 것도 아니며, 전체 파일 간의 오염 추적(cross-file taint tracking)을 수행하는 것도 아닙니다. 이 도구는 경계선(seam)을 표시할 뿐이며, 해결책은 아키텍처적인 측면에 있습니다.

종료 코드 (CI 게이트로 사용 가능):
0 RAW_STRING_TO_DB sink가 없으며 해결되지 않은 사항이 없음
1 1개 이상의 RAW_STRING_TO_DB 또는 UNRESOLVED sink 존재
2 잘못된 입력 (경로 없음, 경로 누락, 읽을 수 없거나 파싱할 수 없는 파일, 스캔 결과 없음)
사용법:
python3 agent_sql_seam.py <file.py | directory>
"""

import ast
import os
import sys

Sink 메서드 이름. 사용하는 드라이버 세트에 맞춰 수정하세요 (asyncpg .fetch 등).

EXEC_ATTRS = {"execute", "executemany", "executescript", "execute_many"}

원시 문자열(raw string)이 아닌 쿼리 객체(OBJECT)를 반환하는 호출 가능 객체: ORM Core

구성 요소 및 의도/빌더 함수(intent/builder functions)의 허용 목록(allowlist). 사용하는 스택에 맞춰 수정하세요.

POLICY_BUILDERS = frozenset({
"select", "insert", "update", "delete",
"build_query", "safe_query", "query_builder", "allowlisted_query",
})

변수에 여러 할당이 있는 경우 최악의 경우를 선택합니다 (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):
"""호출/속성 체인(call/attribute chain)을 루트 Name id까지 풀어냅니다. 예:
select(x).where(y) -> 'select'. 루트가 Name이 아니면 None을 반환합니다."""
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):
"""문자열 상수(string constant) 또는 f-string 노드인지 확인합니다."""
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