ToLO 谓词与传播规则

ToLO 谓词用于判断一条 LLM 影响的数据流是否跨过信任边界:当 S_LLM 到危险 sink 的路径缺少类型匹配的 C_SAFE 时,才报告信任模型失效。

这一页把前一页的集合定义写成规则设计语言。实际实现可以是 CodeQL、Semgrep 或其他 SAST,但判定逻辑应保持一致:source 端是 LLM-output family,sink 端复用传统危险 API,barrier 端只接受类型匹配的 sanitizer。

先统一几个实现词:

• source:污点起点,对应 S_LLM 的五个子集。
• sink:危险参数位置,对应七类 ToLO-* sink。
• sanitizer:能消除某类风险的处理,对应五类 C_SAFE。
• barrier / guard:数据流引擎里阻断路径的节点或条件。一个 guard 只有在它是 type-matched sanitizer 时才应被当成 barrier。
• taint path:从 source 到 sink 的完整路径。
• dataflow:值在赋值、字段、容器、参数和返回值之间的传播关系。
• interprocedural flow:跨函数或跨方法的数据流。
• false positive:误报,通常来自过宽 source、过宽 sink 或漏识别有效 barrier。
• false negative:漏报,通常来自漏建 source、parser step、container step、tool registry 或跨过程流。

这一页的结构

1. 先修概念:谓词是什么 + 触发条件
2. 触发谓词
3. 谓词骨架(伪代码)
4. 传播规则:parser / structured output / 容器 / 字符串 / 跨过程 / 框架分派
5. 一个完整路径示例
6. 解析器与 guard 是否能作为 barrier
7. CodeQL DataFlow vs TaintTracking 选型
8. 两层 query 设计:core / survey
9. 边界与盲点
10. 常见建模错误
11. 概念性测试(fixture 设计)
12. 自测
13. 设计取舍

§1 先修概念:谓词是什么

这里的”谓词”不是复杂数学。你可以把它理解成一个判断函数:输入一个程序点,回答 true 或 false。

例如:

isLLMSource(x):       x 是不是 LLM 输出?
isDangerousSink(y):   y 是不是危险操作?
isSanitizer(z, sink): z 是不是对 sink 有效的防御?

ToLO 谓词就是把这些判断组合起来:

isToLOPath(src, sink) :=
  isLLMSource(src)
  AND isDangerousSink(sink)
  AND hasFlowPath(src, sink)
  AND NOT exists guard:
      guard lies on path(src, sink)
      AND isTypeMatchedSanitizer(guard, sink)

用白话说:先找到 LLM 输出,再看它是否流到危险操作,最后看中间有没有真正匹配的保护。三件事都满足才报。

其中 isLLMSource 是 ToLO-aware 规则的核心,isDangerousSink 则尽量复用已有规则库。这样设计便于解释:这是传统危险 sink,但 source 是 LLM 输出或 LLM-influenceable 框架数据。也便于与默认 CodeQL 做对照实验:同一批 sink,不同 source spec。

§2 触发谓词(展开版)

对程序点 (p_src, p_sink),存在路径 π: p_src ↝ p_sink 且满足三项:

1. p_src ∈ S_LLM^{direct, framework, parsed, structured, rag}
2. p_sink 属于 ToLO-{Deser, Exec, Shell, SQL, Path, SSRF, Template} 对应 sink
3. 路径上没有对该 sink 有效的 sanitizer

“有效”意味着 sanitizer 与 sink 类型匹配:

• html.escape 不能清洗 SQL
• SQL parameterization 不能清洗 path
• Pydantic(str) 不能清洗 shell

详细 sanitizer-sink 匹配规则见 Sources and Sinks §三个容易混淆的判断。

§3 谓词骨架

下面是和具体工具无关的规则骨架。它不是完整 CodeQL,但每一行都应能落到某个具体 predicate:

enum ToLOKind = Deser | Exec | Shell | SQL | Path | SSRF | Template

isLLMSource(n) :=
  isDirectLLMOutput(n)
  OR isFrameworkLLMOutput(n)
  OR isParsedLLMField(n)
  OR isStructuredLLMField(n)
  OR isRAGContentOrMetadata(n)

isDangerousSink(n, kind) :=
  isDangerousArgumentPosition(n, kind)
  AND kind in ToLOKind

isSanitizer(n, kind) :=
  isSchemaSanitizer(n, kind)
  OR isAllowlistSanitizer(n, kind)
  OR isParameterizedCall(n, kind)
  OR isSafeCodec(n, kind)
  OR isCapabilityGuard(n, kind)

isBarrier(n, kind) :=
  isSanitizer(n, kind)
  AND dominatesOrGuardsPath(n)

isToLOPath(src, sink, kind) :=
  isLLMSource(src)
  AND isDangerousSink(sink, kind)
  AND hasTaintPath(src, sink)
  AND NOT exists n:
      n is on taint path(src, sink)
      AND isBarrier(n, kind)

dominatesOrGuardsPath 的意思是:代码执行到 sink 前,必须已经通过这个检查。例如 if invalid: raise 后面的 sink 可以被 guard 保护;但只在日志里打印 “invalid” 不能保护 sink。

Source 谓词

Source 谓词要精确回答”为什么这个值属于 ToLO”:

isDirectLLMOutput(n):
  n 是 OpenAI / Anthropic / Google / Cohere / Mistral / Ollama 等 SDK 的
  content / text / tool_call arguments / function_call args 字段

isFrameworkLLMOutput(n):
  n 是 LangChain / LlamaIndex / Haystack / AutoGen / CrewAI /
  Semantic Kernel / LiteLLM / MCP 等框架对象的输出字段

isParsedLLMField(n):
  exists src:
    isLLMSource(src)
    AND parserStep(src, n)

isStructuredLLMField(n):
  n 是由 LLM structured output / tool use / response_model 解析出的
  Pydantic、dataclass 或 dict 字段

isRAGContentOrMetadata(n):
  n 是 retriever 或 vector store 返回的 Document / Node 内容或 metadata

不要把 str、dict、content 这些泛化形态直接当 source。正确做法是把它们绑定到具体 SDK、框架类型、调用返回值或数据流前驱。

Sink 谓词

Sink 谓词要精确到危险参数:

isDangerousArgumentPosition(n, SQL):
  n 是 cursor.execute(query, ...) 的 query 参数
  OR n 是 sqlalchemy.text(sql) 的 sql 参数
  OR n 是 pandas.read_sql(sql, ...) 的 sql 参数

isDangerousArgumentPosition(n, Shell):
  n 是 os.system(cmd) 的 cmd
  OR n 是 subprocess.run(cmd, shell=True) 的 cmd
  OR n 是 subprocess.Popen(cmd, shell=True) 的 cmd

isDangerousArgumentPosition(n, Path):
  n 是 open(path) / Path(path) / shutil.copy(src, dst) 的路径参数

其余 Deser、Exec、SSRF、Template 同理。不要只看函数名,否则会把安全配置参数或固定模板上下文误报成 sink。

Sanitizer / barrier 谓词

Sanitizer 谓词要带 kind 参数,因为同一个操作对不同 sink 的意义不同:

isParameterizedCall(n, SQL):
  n 是 execute("... WHERE id = %s", (value,))
  AND tainted value 只进入 value 参数
  AND SQL 结构不由 tainted value 决定

isAllowlistSanitizer(n, Path):
  n 检查 resolve 后的 target 位于固定 root 内

isAllowlistSanitizer(n, SSRF):
  n 检查解析后的 scheme、hostname 或 IP range

isSchemaSanitizer(n, Shell):
  n 将命令或工具名限制在 Literal / Enum / 固定集合内

如果一个 sanitizer 没带 kind,通常会过宽。例如把 html.escape 全局加入 isBarrier 会漏掉 SQL、shell、path、SSRF 等真实 ToLO 路径。

§4 传播规则

4.1 S_LLM^{parsed} 要显式建 propagation step

data = json.loads(ai.content)            # S_LLM^{direct} → S_LLM^{parsed}
obj  = yaml.safe_load(resp.text)         # 同上
parsed_action = parser.parse(message)    # 同上
fields = re.findall(r"(\w+)=(\w+)", text)  # 字段仍 tainted

在 CodeQL TaintTracking::Configuration 中加 isAdditionalTaintStep:

override predicate isAdditionalTaintStep(DataFlow::Node src, DataFlow::Node tgt) {
  // json.loads(x) → return value
  exists(Call c |
    c.getFunc().(Attribute).getName() = "loads" and
    c.getFunc().(Attribute).getObject().(Name).getId() = "json" and
    src.asExpr() = c.getArg(0) and
    tgt.asExpr() = c)
  or
  // yaml.safe_load 同上
  ...
}

4.2 结构化输出的字段默认继承污染

结构化输出的 Pydantic / dataclass 字段默认继承污染;只有 Literal、Enum、正则约束、allowlist、参数化调用、安全 codec 或 capability gate 才能作为候选 guard。

class Action(BaseModel):
    tool: Literal["a", "b"]       # ← 是 guard 候选
    path: str                      # ← 不是,继续 tainted

action = parse_action(ai_response)
action.tool                        # ← 在 {"a","b"} 内,部分清洗
action.path                        # ← 仍然 tainted

4.3 容器传播

LLM 输出放入 dict、list、dataclass、Pydantic model 后,字段读取仍应保留污染:

tainted = ai.content

# dict
d = {"x": tainted}
d["x"]                           # tainted

# list
l = [tainted]
l[0]                             # tainted

# tuple
t = (tainted,)
t[0]                             # tainted

# dataclass / Pydantic
obj = MyModel(name=tainted)
obj.name                         # tainted

# kwargs spread
fn(**{"path": tainted})           # path 参数 tainted

CodeQL TaintTracking 库默认覆盖大部分容器传播,但自定义 wrapper 类可能需要手动加 step。

4.4 字符串变换不构成清洗

sql_part = f"WHERE id = {tainted}"        # tainted
upper = tainted.upper()                    # tainted
filled = tpl.format(x=tainted)             # tainted
joined = ",".join([safe, tainted])         # tainted
encoded = tainted.encode("utf-8")          # bytes 仍 tainted
b64 = base64.b64encode(encoded)            # tainted
fixed_padding = tainted.ljust(10)          # tainted

唯一例外:真正改变语义的转换才算 sanitizer。例如 int(tainted) 把字符串转整数,后续作为 int 使用就部分安全(但要确保下游不再当字符串)。

4.5 跨过程传播

ToLO 路径经常跨函数:

def parse(message):
    return json.loads(message.content)

def run_tool(args):
    return subprocess.run(args["cmd"], shell=True)

def handle(question):
    msg = llm.invoke(question)
    args = parse(msg)
    return run_tool(args)

这条 taint path 是:

msg.content -> parse return -> args -> run_tool parameter -> args["cmd"] -> subprocess.run

规则必须允许:

• 实参到形参:parse(msg) 中 msg -> message
• 返回值到接收变量:return json.loads(...) -> args
• 调用方到被调方:run_tool(args) 中 args -> args
• 字段读取:args["cmd"]

如果工具只支持单文件或单函数匹配,这一类路径会变成 false negative。Semgrep 可以用 pattern 捕捉局部高危形态,但完整 ToLO 路径通常需要 CodeQL 这类跨过程数据流。

4.6 框架分派(tool registry)

agent action 通过 registry 找到 tool 函数时,tool_input 到函数参数的映射需要建模,否则会漏掉真实 sink:

TOOLS = {
    "read_file": read_file_impl,
    "run_cmd":   run_cmd_impl,
}

def dispatch(action: AgentAction):
    fn = TOOLS[action.tool]               # ← 动态查找
    return fn(**action.tool_input)         # ← 参数展开

# sink 不在 dispatch 这里,在 TOOLS 的某个函数体里
def run_cmd_impl(cmd: str) -> str:
    return subprocess.run(cmd, shell=True, ...)   # ← 真实 sink

CodeQL 建模:为 TOOLS[x](**args) 这种动态分派建立流向 — 每个被注册的 tool 函数的参数都成为 sink。可以:

1. 找所有 TOOLS = {...: fn, ...} 字面字典中作为 value 的函数。
2. 把这些函数的参数标为 sink source(注意:这里 source 和 sink 都成立)。
3. 让 dataflow 追踪 action.tool_input → fn 参数 → fn 体内 sink。

LangChain @tool 装饰器和 LangChain Agent registry 都需要类似建模。

§5 一个完整路径示例

msg = chain.invoke(user_input)             # S_LLM^{framework}
action = parser.parse(msg.content)         # S_LLM^{parsed} (经 OutputParser)
tool_name = action.tool                    # 字段读取,tainted
tool_args = action.args                    # 字段读取,tainted
tools[tool_name](**tool_args)              # 动态分派 → sink 在 tools 某个函数体内

人工标注:

msg.content  →  action.args  →  tool_args  →  selected tool  →  sink inside tool
   |              |               |              |                |
   S_LLM         S_LLM           S_LLM          dispatch         dangerous call

然后再决定 CodeQL 是否需要额外 propagation step。真正的 sink 不在当前文件里,可能藏在 tools[tool_name] 指向的函数中。规则需要理解 tool registry,否则会只看到 parser,看不到执行点。

§6 解析器是否是 barrier

取决于语义:

解析器	对什么 sink 算 barrier?	对其他 sink?
json.loads	ToLO-Deser(对替代 pickle 而言)	不算 sanitizer。字段内容仍 tainted
yaml.safe_load	ToLO-Deser	不算 sanitizer 给其他 sink
ast.literal_eval	ToLO-Exec(对替代 eval 而言);ToLO-Deser	不算 给其他 sink
numexpr.evaluate(expr, global_dict={}, local_dict={...})	ToLO-Exec(数学子集)	不算 给其他 sink
pickle.loads	永远不算 sanitizer,本身是 sink
re.findall(pattern, x)	不算 sanitizer。可能配合 allowlist 用
自定义 OutputParser	默认不算。除非内部做了 schema/allowlist 校验

实用判定:

解析器是否是 sanitizer,看它是否拒绝执行 / 拒绝实例化危险对象,而不是看它”看起来比 eval 安全”。

好 guard 与坏 guard

guard 必须真的控制通往 sink 的执行路径。

# bad: 只计算了一个布尔值,没有阻断 sink
allowed = path.startswith("/srv/data")
return Path(path).read_text()

# good: 失败路径抛出,成功路径才到 sink
target = (ROOT / path).resolve()
if not target.is_relative_to(ROOT):
    raise PermissionError()
return target.read_text()

# bad: allowlist 只保护 tool name,没有保护参数
if action.tool not in ALLOWED_TOOLS:
    raise ValueError()
tools[action.tool](**action.args)      # args 仍可能流到 sink

# good: tool name 和参数分别约束
if action.tool != "search":
    raise ValueError()
query = action.args["query"]
search_api(BASE_URL, params={"q": query})

第二个例子中,action.tool 的污染被 allowlist 切断,但 query 仍是 tainted。它安全是因为 query 进入固定 base URL 的 params 值位置,而不是因为整个 action.args 被清洗了。

§7 CodeQL 选型:DataFlow vs TaintTracking

DataFlow::Configuration

值保持传播(value-preserving):

• y = x ✅
• y = obj.attr(如果 obj 含 x) ✅
• y = x.replace("a","b") ❌(值变了,默认不传)
• y = json.loads(x) ❌(对象类型变了)

误报低、漏报多。适合”高置信路径”。

TaintTracking::Configuration

污点保持传播(taint-preserving):

• y = x ✅
• y = x.replace("a","b") ✅
• y = json.loads(x) ✅(配合 isAdditionalTaintStep)
• y = some_dict[x] ✅
• y = f"prefix {x}" ✅

召回高、误报多。

ToLO 通常需要哪种

ToLO 因为常有字符串拼接、容器读写、解析派生值等非值保持传播,覆盖 ToLO 更完整需要 TaintTracking。

实际规则可以先用 DataFlow 固定高置信路径,再用 TaintTracking 扩展召回。

初学阶段可以理解成:DataFlow 更保守,适合少误报;TaintTracking 更宽,适合找更多可疑路径。

§8 两层 query 设计

工程上分两层查询:

层	名字	目标	误报容忍	适用场景
1	core queries	高 precision	低	公开 benchmark、CVE 复盘、官方安全 advisory
2	survey queries	高 recall	中	生态扫描、人工 triage、研究统计

core queries(精确)

只报高置信路径:

• source = 明确的 SDK 字段(openai.OpenAI / anthropic.Anthropic 等的 message.content / tool_calls)
• sink = 标准库的危险函数(eval / subprocess.run(shell=True) 等)
• 无任何 sanitizer

survey queries(广覆盖)

扩大 parser、container、framework callback 的传播范围:

• 包含 S_LLM^{parsed}、S_LLM^{structured}、S_LLM^{rag} 完整建模
• 跨 tool registry 分派
• 反射 / 动态 import 的保守上界
• 自定义 parser 默认不算 sanitizer

两层结果不要混在一起报。core queries 支撑 precision,survey queries 支撑覆盖面,二者面向不同问题。

§9 边界与盲点

容易做的

• 跨函数:CodeQL 全局数据流默认覆盖。
• 跨对象:field / attribute step 默认覆盖(可能要手动加 wrapper 类)。
• 跨模块:依赖库模型(需要先建 library type model)。

容易漏报的

• 反射:getattr(obj, name)(args) 取决于 name 来源,如果 name 来自 LLM,sink 不确定。保守起见全部标 sink。
• 动态 import:importlib.import_module(name) 同上。
• 框架回调:LangChain BaseCallbackHandler.on_* 钩子被框架内部调用,跨 callsite。
• agent tool registry:动态字典分派(见 §4.6)。
• 自定义 sanitizer:看似有校验逻辑但实现不正确(例如 ”..” 字符串替换),isBarrier 误判。

处理盲点的姿势

宜标为保守盲点,而不是用过宽 source / sink 规则硬凑召回。

// 例:对动态分派,把所有 callable[x] 形式标 sink (over-approx)
predicate isDynamicDispatchSink(DataFlow::Node n) {
  exists(Call c |
    c.getFunc() instanceof Subscript and
    n.asExpr() = c.getArg(_))
}

报告里要写 limitation:“动态分派情况未精确建模,部分路径可能误报。“

JS / TS 端额外注意

• 动态属性:obj[name] 同 Python 的反射。
• destructuring:const { foo } = bar 字段传播。
• async callback:Promise.then(cb) 跨 callsite。
• object spread:{ ...args } 容器展开。

Python 端额外注意

• monkey patch:module.func = my_replacement 改变函数行为。
• decorator:@tool 改变函数 metadata 但通常不改 body。
• dynamic import:__import__(name) 同 importlib。
• 框架注册表:registry[name] = fn 风格分派。

§10 常见建模错误

错误	后果	更好的规则
isSource(x) := x is str	大量 false positive	绑定到 SDK / 框架对象 / LLM 调用返回值
把所有 content 字段当 source	普通 CMS、HTTP response、文件内容被污染	限定类型或前驱调用
把 json.loads 当全局 barrier	漏报 Path / SQL / Shell / SSRF	只对替代危险反序列化这类场景作为 safe codec 候选
把 Pydantic model 当整体 sanitizer	漏掉自由 str 字段进入 sink	字段级判断 Literal / Enum / 受约束类型
只看 sink 函数名	把安全参数、固定模板也报出来	标危险参数位置
只检查 tool name allowlist	漏掉 tainted tool args	tool name 和 args 分别建模
把 shell=False 当万能 barrier	executable 仍可能由 LLM 控制	要求 executable 固定或 allowlisted
把 startswith(BASE) 当路径 barrier	可被规范化、符号链接、相对路径绕过	resolve 后检查 root jail
把 URL scheme 检查当 SSRF barrier	host 仍可能是内网或 metadata 地址	同时检查 scheme、host、IP range
忽略跨过程返回值	真实路径断在 helper 函数边界	加实参/形参、返回值、字段传播

每个错误都可以转化成一个 fixture:一段应报代码、一段不应报代码。规则改动后先跑这些 fixture,再进入真实项目。

§11 概念性测试

概念性测试不需要完整漏洞样本,只需要最小代码片段。建议为每个 ToLO 子类准备三类 fixture:

Fixture 类型	例子	期望
Positive	llm.content -> json.loads -> Path(path).read_text()	报 ToLO-Path
Negative: no source	config["path"] -> Path(path).read_text()	不作为 ToLO 报告
Negative: type-matched barrier	llm_id -> cursor.execute("... id=%s", (llm_id,))	不报 ToLO-SQL
Mismatch barrier	llm_url -> html.escape -> requests.get(url)	仍报 ToLO-SSRF
Cross-function	llm.content -> parse() return -> run_tool() param -> sink	应能跨过程追踪
Container	{"cmd": llm.content} -> args["cmd"] -> subprocess.run	应保留污染
Structured output	Pydantic str field -> f"SELECT {field}"	应报
Structured enum	Literal["users","orders"] -> table allowlist mapping	不报 SQL 结构风险

评估时记录两类失败:

• false positive:例如参数化 SQL 被报出,说明 sanitizer 或危险参数位置建模太粗。
• false negative:例如 AgentAction.tool_input 到 tool 函数参数没连上,说明 source、propagation 或 dispatch 建模不足。

这些 fixture 的目的不是证明工具发现真实漏洞,而是证明规则规格和本章定义一致。

§12 自测

下面哪条应该报告 ToLO?

A. LLM output -> print(...)
B. LLM output -> json.loads(...) -> open(path)
C. LLM output -> allowlisted tool name -> safe fixed function
D. LLM output -> Pydantic str field -> cursor.execute(sql_template, (field,))
E. LLM output -> Pydantic str field -> f"SELECT {field}" -> cursor.execute(...)
F. LLM output -> Literal["a","b"] -> if x == "a": handle_a(); else: handle_b()

选项	应该报告吗	原因
A	❌	没有危险 sink,只是展示
B	✅	典型 ToLO-Path,json.loads 不算 sanitizer
C	🟡	取决于 safe fixed function 内部是否真的安全;allowlist 切断了 tool name 选择,但参数仍可能 tainted
D	❌	SQL 参数化已经类型匹配,field 作为值进入,不算结构
E	✅	f-string 拼接进入 SQL,field 决定结构,参数化无效,ToLO-SQL
F	❌	Literal 已收窄到 2 个值,然后分派到不同 handler,没有 tainted 数据真正流到 sink

§13 设计取舍

ToLO 静态分析规则应偏向可解释。每条告警最好能展示:

source location  →  transform path  →  sink location  →  missing guard

只报告”这里有 eval”不够 —— 它无法证明 LLM 输出参与其中。 只报告”这里有 LLM output”也不够 —— 它未必触达敏感操作。

第一版规则可以刻意保守:优先覆盖高置信 source 与高危 sink,避免把普通自然语言展示路径报成漏洞。随后再扩展。

下一步阅读

• Sources and Sinks:集合定义详细展开。
• Query Design Notes:CodeQL 与 Semgrep 在 ToLO 上的具体规则实现。
• Public Case Studies:用 5 个 CVE 验证你的谓词是否能命中。