Query Design Notes

这一页讲怎么设计查询,不是怎么证明漏洞。规则的目标是找出”LLM 输出或 LLM-influenceable 数据到达危险 sink,且缺少匹配防护”的路径,再把路径交给人工 triage。

本页所有 QL / YAML 都是教育性伪代码。真实实现要按具体语言库、框架版本、CodeQL API 版本和 Semgrep 版本调整。

设计目标

ToLO 查询应保持三个边界:

source 端 ToLO-aware:显式识别 S_LLM^{direct,framework,parsed,structured,rag}。
sink 端尽量复用:七类 ToLO-{Deser,Exec,Shell,SQL,Path,SSRF,Template} 都对应传统 CWE sink family,不要重复发明 sink。
barrier 端保守:只把类型匹配的 C_SAFE^{schema,allowlist,parameterized,safe-codec,capability} 当作有效阻断。

一句话版本:

先建准 LLM source,再复用危险 sink,最后只承认匹配 sanitizer。

查询不是漏洞证明

静态分析规则只能说明代码里存在可能路径。它不等于:

可利用 exploit
CVE 编号
扫描覆盖率
真实环境中一定可达

ToLO 规则结果应输出为 triage queue。人工 review 需要继续看版本、权限、配置、部署边界、是否有实际 guard、是否存在可触达攻击通道。

规则包结构

建议把 CodeQL 规则拆成三层:

tolo/
├── sources/
│   ├── LLMClientSource.qll
│   ├── FrameworkSource.qll
│   ├── ParsedOutputStep.qll
│   ├── StructuredOutputStep.qll
│   └── RAGSource.qll
├── sinks/
│   └── wrappers.qll
├── sanitizers/
│   ├── SchemaBarrier.qll
│   ├── AllowlistBarrier.qll
│   ├── ParameterizedBarrier.qll
│   ├── SafeCodecBarrier.qll
│   └── CapabilityBarrier.qll
└── queries/
    ├── ToLODeser.ql
    ├── ToLOExec.ql
    ├── ToLOShell.ql
    ├── ToLOSQL.ql
    ├── ToLOPath.ql
    ├── ToLOSSRF.ql
    └── ToLOTemplate.ql

这样新增框架时主要扩展 sources/,新增 wrapper sink 时扩展 sinks/,不会把七类查询都改乱。新增框架通常只需要扩展 source 模块,不会扰动 sink 与 sanitizer 判定。

Semgrep 可以按同样思路拆文件:

semgrep/
├── tolo-exec.yml
├── tolo-shell.yml
├── tolo-sql.yml
├── tolo-path.yml
├── tolo-ssrf.yml
├── tolo-deser.yml
└── tolo-template.yml

第一版规则目标

第一版不追求覆盖全部框架。更合理的目标是:用少量旗舰框架和公开 CVE 验证 source/sink/sanitizer 三元组是否能稳定表达。能在已知案例上解释清楚,再逐步扩展到 long-tail 生态。

具体目标:

目标	验证 CVE
找到 LLM 输出进入 code execution sink 的路径	CVE-2023-29374, CVE-2023-36258, CVE-2023-39631, CVE-2024-5826
找到 LLM 输出进入 query construction 的路径	CVE-2024-8309, CVE-2024-5826
找到 agent tool 参数缺少 schema/allowlist 的路径	通用
找到 output parser 结果直接进入敏感操作的路径	通用

学习者可以把第一版目标记成三句话:

先找清楚 source
再复用成熟 sink
最后只承认匹配 sanitizer

查询设计顺序

先从少量明确 source 开始:chat completion content、parsed output、agent action。
再接入高置信 sink:eval、shell execution、database query、template rendering。直接 import CodeQL 标准库的对应 sink。
然后补 guard 识别:allowlist、schema validation、sandbox boundary、explicit approval。
最后根据公开案例调低误报,补充框架特定 API。

每新增一个 source model,要写测试

附带至少一个最小测试样例:

✓ source 进入 sink → 应报
✓ source 经类型匹配 sanitizer 后 → 不应报
✓ 普通用户输入 / 内部字符串 → 不应被错误归入 ToLO

测试样例不需要是真实漏洞,但要覆盖规则意图。

Source 设计

source 谓词要回答:“这个值为什么属于 LLM output family?”

Source 子集	CodeQL predicate 应识别	Semgrep source pattern 可先识别	不要这样写
`S_LLM^{direct}`	SDK 返回对象的 content / text / tool-call args 字段	OpenAI / Anthropic / Google / Cohere 等 response 字段	`isSource(n) := n is str`
`S_LLM^{framework}`	LangChain、LlamaIndex、Haystack、AutoGen 等 message / generation / agent 字段	`AIMessage.content`、`AgentAction.tool_input` 等	所有 `.content`
`S_LLM^{parsed}`	LLM source 经 `json.loads`、`yaml.safe_load`、regex、output parser 的返回值	`pattern-propagators`	把 parser 当 sanitizer
`S_LLM^{structured}`	function calling、response model、Pydantic / dataclass 字段	structured output 调用返回值和字段读取	把整个 model 当安全对象
`S_LLM^{rag}`	retriever / vector store 返回的 document content / metadata	`Document.page_content`、`node.text`	只看”是否由 LLM 生成”

S_LLM^{rag} 特别容易被漏掉。RAG 文档可能不是模型生成,但它可以被攻击者影响并进入 LLM 编排链路,因此属于 LLM-influenceable source。

CodeQL source 伪谓词

class LLMOutputSource extends DataFlow::Node {
  LLMOutputSource() {
    this instanceof DirectLLMClientOutput or
    this instanceof FrameworkMessageOutput or
    this instanceof ParsedLLMOutput or
    this instanceof StructuredLLMField or
    this instanceof RAGDocumentField
  }
}

真实实现时,每个子类都应能指回具体 API 或类型。例如 FrameworkMessageOutput 不应是”任何 .content”,而应绑定到 AIMessage、BaseMessage、AgentAction、Response、Document 等已知框架类型或调用返回值。

Parser 不是 sanitizer

示例:

data = json.loads(ai_message.content)
open(data["path"])

json.loads 只把文本解析成对象,没有证明 path 安全。因此它是 propagation step,不是 barrier。

伪 QL:

predicate parsedOutputStep(DataFlow::Node src, DataFlow::Node out) {
  exists(Call c |
    c.getFunc().hasQualifiedName("json", "loads") and
    src.asExpr() = c.getArg(0) and
    out.asExpr() = c
  )
}

Semgrep 伪 propagator:

pattern-propagators:
  - pattern: $OUT = json.loads($IN)
    from: $IN
    to: $OUT

Sink 设计

sink 谓词要精确到危险参数位置,不要只看函数名。

ToLO 子类	典型 sink 参数	伪代码例子
`ToLO-Deser`	反序列化输入	`pickle.loads(payload)`
`ToLO-Exec`	代码字符串	`eval(code)` / `exec(code)`
`ToLO-Shell`	shell 命令字符串	`subprocess.run(cmd, shell=True)`
`ToLO-SQL`	查询字符串或 query builder 的 raw SQL	`cursor.execute(sql)`
`ToLO-Path`	文件路径参数	`open(path)` / `Path(path)`
`ToLO-SSRF`	URL 参数	`requests.get(url)`
`ToLO-Template`	template 字符串或危险 render 上下文	`Environment.from_string(tpl)`

CodeQL 里建议每个 ToLO 子类一个配置,这样 barrier 更容易保持类型匹配:

class ToLOSQLConfig extends TaintTracking::Configuration {
  override predicate isSource(DataFlow::Node n) { n instanceof LLMOutputSource }
  override predicate isSink(DataFlow::Node n) { n instanceof SQLQueryArgument }
  override predicate isBarrier(DataFlow::Node n) { n instanceof SQLBarrier }
}

不要写成一个全局 isBarrier 吃掉所有 sanitizer。SQLBarrier、PathBarrier、SSRFBarrier 应该分开。

Barrier 设计

barrier 谓词要回答:“这个检查是否真的切断了当前 sink 的风险?”

`C_SAFE` 类	可作为 barrier 的条件	对应 sink 例子	不能当 barrier 的例子
`C_SAFE^{schema}`	输出被收窄到固定枚举、`Literal`、严格正则子语言或固定能力集合	shell tool 名只允许 `"status"` / `"list"`	`path: str`、`query: str`
`C_SAFE^{allowlist}`	值通过固定集合、固定 root、固定 host / scheme / table / column 检查	path 必须 resolve 到固定目录下	黑名单过滤 `".."`
`C_SAFE^{parameterized}`	downstream API 分离结构和值,LLM 不能控制语法结构	`execute("... where id = %s", [id])`	LLM 生成整条 SQL 字符串
`C_SAFE^{safe-codec}`	使用安全 codec 或受限 evaluator,不执行任意对象 / 代码语义	`yaml.safe_load` 针对 deser 风险	`json.loads` 后直接 `open`
`C_SAFE^{capability}`	执行环境有明确能力边界,只能做被授权操作	tool sandbox 只有读特定目录权限	只设置了一个变量名叫 `sandbox`

同一个处理可能对一个 sink 是 barrier,对另一个 sink 不是。yaml.safe_load 可以是 ToLO-Deser 的 C_SAFE^{safe-codec},但它的返回值仍然可能进入 ToLO-Path、ToLO-SSRF 或 ToLO-SQL sink。

Barrier 伪谓词

predicate isSQLBarrier(DataFlow::Node n) {
  n instanceof ParameterizedSQLCall or
  n instanceof SQLTableColumnAllowlist
}

predicate isPathBarrier(DataFlow::Node n) {
  n instanceof ResolvedPathUnderAllowedRoot or
  n instanceof LiteralPathEnum
}

predicate isSSRFBarrier(DataFlow::Node n) {
  n instanceof URLHostAllowlist or
  n instanceof NetworkCapabilityGate
}

Semgrep 设计

Semgrep 的价值是快速写出可读规则,把 review queue 做出来。它尤其适合:

明显近距离路径。
代码 review 中的新增危险 API。
为 CodeQL source model 收集框架字段候选。
解释 ToLO 规则思路。

Semgrep taint-mode 骨架

rules:
  - id: tolo-kind-pseudo
    mode: taint
    pattern-sources:
      - pattern: $SOURCE
        exact: true
    pattern-propagators:
      - pattern: $OUT = parse_llm_output($IN)
        from: $IN
        to: $OUT
    pattern-sinks:
      - pattern: dangerous_sink($SINK, ...)
    pattern-sanitizers:
      - pattern: type_matched_guard($SINK)
        exact: true
    message: "Possible ToLO path: LLM output reaches dangerous sink."
    languages: [python]
    severity: WARNING

pattern-sources、pattern-sinks、pattern-sanitizers 彼此是独立模式。不要假设 source 中的 metavariable 和 sink 中的同名 metavariable 天然绑定为同一个运行时值。需要这种约束时,要在规则选项或更精确的结构中显式表达。

最小 Semgrep 规则草稿

规则 1:`eval` 接近 `AIMessage.content`

rules:
  - id: tolo-exec-llm-eval-direct
    pattern-either:
      - pattern: eval($X.content)
      - pattern: exec($X.content)
      - pattern: compile($X.content, ...)
    message: |
      LLM output `.content` may flow to code execution sink.
      If $X is an AIMessage / ChatCompletion / similar LLM response,
      this is a likely ToLO-Exec.
    languages: [python]
    severity: WARNING
    metadata:
      tolo_sink: Exec
      tolo_subclass: ToLO-Exec
      cwe: CWE-94

规则 2:tool_call.arguments 进 subprocess

rules:
  - id: tolo-shell-tool-call-args
    pattern-either:
      - pattern: |
          $ARGS = json.loads($CALL.function.arguments)
          ...
          subprocess.run($ARGS[...], shell=True, ...)
      - pattern: |
          $ARGS = json.loads($CALL.function.arguments)
          ...
          os.system($ARGS[...])
    message: "OpenAI tool_call arguments may flow to shell."
    languages: [python]
    severity: ERROR
    metadata:
      tolo_subclass: ToLO-Shell
      cwe: CWE-78

规则 3:`@tool` 函数体内 `subprocess.run(..., shell=True)`

rules:
  - id: tolo-shell-in-tool-decorator
    pattern: |
      @tool
      def $F(...):
        ...
        subprocess.run(..., shell=True, ...)
        ...
    message: |
      Function decorated with @tool contains subprocess(shell=True).
      LLM-controlled args may inject shell commands.
    languages: [python]
    severity: ERROR
    metadata:
      tolo_subclass: ToLO-Shell

Semgrep 七类草稿

下面只展示 sink 侧形态。source 可以复用本页前面的 S_LLM source patterns。

ToLO 子类	Semgrep sink pattern 草稿	需要的 sanitizer 思路
`ToLO-Deser`	`pickle.loads($X)`、`yaml.load($X, ...)`	`safe_load`、允许类型集合
`ToLO-Exec`	`eval($X)`、`exec($X)`、`compile($X, ...)`	受限 evaluator、capability gate
`ToLO-Shell`	`os.system($X)`、`subprocess.run($X, shell=True, ...)`	command allowlist、argv 参数化、capability gate
`ToLO-SQL`	`$CUR.execute($SQL)`、`pd.read_sql($SQL, ...)`	parameterized call、表/列 allowlist
`ToLO-Path`	`open($PATH, ...)`、`Path($PATH)`	canonicalize 后 root allowlist
`ToLO-SSRF`	`requests.get($URL, ...)`、`httpx.post($URL, ...)`	scheme / host / IP range allowlist
`ToLO-Template`	`Template($TPL)`、`$ENV.from_string($TPL)`	固定模板集合、禁止 LLM 控制 template syntax

Semgrep 示例:ToLO-SQL

rules:
  - id: tolo-sql-llm-generated-query-pseudo
    mode: taint
    pattern-sources:
      - pattern: $MSG.content
        exact: true
      - pattern: $ACTION.tool_input
        exact: true
    pattern-propagators:
      - pattern: $OUT = json.loads($IN)
        from: $IN
        to: $OUT
    pattern-sinks:
      - pattern: $CURSOR.execute($SQL)
      - pattern: pandas.read_sql($SQL, ...)
    pattern-sanitizers:
      - pattern: execute_parameterized($SQL, ...)
        exact: true
      - pattern: allowlisted_sql_ast($SQL)
        exact: true
    message: "LLM-controlled SQL may reach a database query sink."
    languages: [python]
    severity: WARNING
    metadata:
      tolo_subclass: ToLO-SQL
      source_family: S_LLM

这个草稿表达的是设计意图。真实规则还要避免把所有 .content 当作 LLM message,并区分 raw SQL 参数与普通 value 参数。

Semgrep 常见坑

坑	后果	修正
`pattern: $X.content` 作为 source	HTTP response、HTML node、业务对象都被污染	绑定到框架类型、调用上下文或更具体字段路径
sanitizer 写成 `validate($X)`	任意命名为 validate 的函数都被当清洗	只承认可审查的 `C_SAFE` 模式
没有 propagator	`json.loads`、dict 字段、parser 返回值后漏报	为关键转换加 `pattern-propagators`
只写 sink search rule	只能说明有危险 API	改成 taint-mode,接入 source / sanitizer
把 `yaml.safe_load` 全局当 sanitizer	可能漏掉 safe_load 后进入 path / SSRF / SQL 的路径	sanitizer 要按 ToLO 子类匹配

Semgrep 限制

不能跨函数追踪。如果 tool 函数内调用 helper,helper 里调 sink,Semgrep 看不到完整链路。
不能精确判断类型。$X.content 可能是任何含 .content 属性的对象。
适合作为**“问题清单生成器”**,不适合作为最终判定。

CodeQL 设计

CodeQL 的价值是给出完整 source-to-sink 路径。设计时先写自然语言 spec,再落到 QL predicate。

CodeQL 查询骨架

伪 QL:

import python
import semmle.python.dataflow.new.TaintTracking
import semmle.python.dataflow.new.DataFlow::PathGraph

class ToLOShellConfig extends TaintTracking::Configuration {
  ToLOShellConfig() { this = "ToLOShellConfig" }

  override predicate isSource(DataFlow::Node n) {
    n instanceof LLMOutputSource
  }

  override predicate isSink(DataFlow::Node n) {
    n instanceof ShellCommandArgument
  }

  override predicate isBarrier(DataFlow::Node n) {
    n instanceof ShellAllowlistBarrier or
    n instanceof ShellParameterizedInvocation or
    n instanceof ShellCapabilityGate
  }

  override predicate isAdditionalTaintStep(DataFlow::Node from, DataFlow::Node to) {
    parsedOutputStep(from, to) or
    structuredFieldStep(from, to) or
    toolRegistryStep(from, to)
  }
}

from ToLOShellConfig cfg, DataFlow::PathNode source, DataFlow::PathNode sink
where cfg.hasFlowPath(source, sink)
select sink, source, sink,
  "LLM output reaches shell execution without a matching C_SAFE barrier."

真实查询要按当前 CodeQL modular data-flow API 改写,但设计上仍是这四个 predicate。

Source predicate 分层

class DirectLLMClientOutput extends DataFlow::Node {
  DirectLLMClientOutput() {
    openAIMessageContent(this) or
    anthropicTextBlock(this) or
    toolCallArguments(this)
  }
}

class FrameworkMessageOutput extends DataFlow::Node {
  FrameworkMessageOutput() {
    langChainMessageContent(this) or
    llamaIndexResponseText(this) or
    agentActionToolInput(this)
  }
}

class RAGDocumentField extends DataFlow::Node {
  RAGDocumentField() {
    langChainDocumentPageContent(this) or
    llamaIndexNodeText(this) or
    retrieverMetadata(this)
  }
}

每个 predicate 都应配最小 fixture。不要把”字段名像 output”作为充分条件。

`LangChainSource.qll` 示例

import python
import semmle.python.dataflow.new.DataFlow

/** AIMessage.content 等 LangChain 框架包装字段 */
class LangChainMessageContent extends DataFlow::Node {
  LangChainMessageContent() {
    exists(Attribute a |
      a.getAttr() in ["content", "tool_calls", "additional_kwargs"] and
      a.getObject().pointsTo().getClass().getName() in [
        "AIMessage", "BaseMessage", "ChatMessage",
        "HumanMessage", "ToolMessage", "AIMessageChunk"
      ] and
      this.asExpr() = a
    )
  }
}

/** LangChain AgentAction 字段 */
class LangChainAgentAction extends DataFlow::Node {
  LangChainAgentAction() {
    exists(Attribute a |
      a.getAttr() in ["tool_input", "tool", "log"] and
      a.getObject().pointsTo().getClass().getName() in ["AgentAction"] and
      this.asExpr() = a
    )
  }
}

Sink predicate 精确到参数

class ShellCommandArgument extends DataFlow::Node {
  ShellCommandArgument() {
    exists(Call c |
      isSubprocessShellTrue(c) and
      this.asExpr() = c.getArg(0)
    )
    or
    exists(Call c |
      isOsSystem(c) and
      this.asExpr() = c.getArg(0)
    )
  }
}

注意这里匹配的是命令参数,不是整个 call。对 SQL、path、SSRF、template 也应精确到对应参数。

Additional taint step

ToLO 常见漏报来自额外传播步骤:

传播类型	例子	建模动作
parser	`json.loads(x)`、`yaml.safe_load(x)`、`parser.parse(x)`	`x -> return value`
structured output	`response_model=Action`、`with_structured_output(Action)`	LLM response -> model field
container	`{"path": x}`、`list.append(x)`、`obj.field = x`	value -> field / element
tool registry	`TOOLS[name](**args)`	selected tool args -> tool function parameters
wrapper sink	`run_query(sql)` 内部调用 `cursor.execute`	wrapper parameter -> underlying sink

伪 QL:

predicate structuredFieldStep(DataFlow::Node from, DataFlow::Node to) {
  exists(Call c, Attribute field |
    isStructuredOutputParse(c) and
    from.asExpr() = c and
    field.getObject() = c and
    to.asExpr() = field
  )
}

`ToLOExecConfiguration.qll` 完整示例

import python
import semmle.python.dataflow.new.TaintTracking
import LangChainSource

class ToLOExecConfig extends TaintTracking::Configuration {
  ToLOExecConfig() { this = "ToLOExecConfig" }

  override predicate isSource(DataFlow::Node n) {
    n instanceof LangChainMessageContent or
    n instanceof LangChainAgentAction or
    n instanceof OpenAIResponseContent     // 另一个 source 模块
  }

  override predicate isSink(DataFlow::Node n) {
    // 复用 CodeQL 标准 code-injection sink
    n instanceof CodeExecutionSink
  }

  override predicate isAdditionalTaintStep(DataFlow::Node src, DataFlow::Node tgt) {
    // json.loads(x) → return value 传播
    exists(API::Node json |
      json = API::moduleImport("json").getMember("loads") and
      src = json.getACall().getArg(0) and
      tgt = json.getACall())
    or
    // ast.literal_eval(x) 也算
    exists(API::Node lit |
      lit = API::moduleImport("ast").getMember("literal_eval") and
      src = lit.getACall().getArg(0) and
      tgt = lit.getACall())
  }

  override predicate isBarrier(DataFlow::Node n) {
    // 经过 numexpr.evaluate(..., global_dict={}) 算 safe-codec
    exists(Call c |
      c.getFunc().(Attribute).getName() = "evaluate" and
      c.getFunc().(Attribute).getObject().(Name).getId() = "numexpr" and
      exists(c.getANamedArg("global_dict")) and
      n.asExpr() = c.getArg(0))
  }
}

主查询

import python
import semmle.python.dataflow.new.DataFlow::PathGraph
import ToLOExecConfiguration

from ToLOExecConfig cfg, DataFlow::PathNode src, DataFlow::PathNode sink
where cfg.hasFlowPath(src, sink)
select sink, src, sink, "LLM 输出经数据流到达代码执行 sink (ToLO-Exec)"

CodeQL 输出的告警(示意)

[ToLOExec] LLM 输出经数据流到达代码执行 sink

Source:
  app/chains.py:15
    answer = chain.invoke({"question": q})
    # answer.content matched LangChainMessageContent

Path:
  app/chains.py:15 → app/chains.py:17 (assignment)
  app/chains.py:17 → app/utils.py:42 (function call)
  app/utils.py:42 → app/utils.py:43 (assignment after json.loads)
  app/utils.py:43 → app/runtime.py:89 (passed to eval)

Sink:
  app/runtime.py:89
    return eval(code)
    # matched CodeExecutionSink (CWE-94)

Missing guard:
  No C_SAFE^{safe-codec} or C_SAFE^{capability} on this path

这条路径才是 ToLO 教学和 triage 的核心。

七类 ToLO 查询配置

Query	Source	Sink	Barrier
`ToLODeser.ql`	`LLMOutputSource`	unsafe deserialization argument	safe codec / class allowlist
`ToLOExec.ql`	`LLMOutputSource`	code execution argument	restricted evaluator / capability
`ToLOShell.ql`	`LLMOutputSource`	shell command argument	schema / allowlist / parameterized invocation / capability
`ToLOSQL.ql`	`LLMOutputSource`	SQL query argument	parameterized query / schema allowlist
`ToLOPath.ql`	`LLMOutputSource`	file path argument	root allowlist / fixed path enum / capability
`ToLOSSRF.ql`	`LLMOutputSource`	URL request argument	scheme / host / IP allowlist / network capability
`ToLOTemplate.ql`	`LLMOutputSource`	template string or dangerous render argument	fixed template set / schema / capability

每类 query 可以共享 source 和 propagation 模块,但 barrier 不要混用。

小型伪样例

这些样例只展示”应该报 / 不应报”的边界,不是 exploit。

`ToLO-Deser`

# expected: REPORT
payload = ai_message.content
obj = pickle.loads(payload)

# expected: NOT REPORT for ToLO-Deser
payload = ai_message.content
obj = yaml.safe_load(payload)
# 但 obj 后续进入 open / requests / SQL 时,仍要按对应 ToLO 子类继续追踪。

`ToLO-Exec`

# expected: REPORT
code = action.tool_input["code"]
exec(code)

# expected: REVIEW or NOT REPORT only if capability is explicit
code = action.tool_input["code"]
run_in_capability_limited_sandbox(code)

`ToLO-Shell`

# expected: REPORT
cmd = json.loads(msg.content)["cmd"]
subprocess.run(cmd, shell=True)

# expected: NOT REPORT for shell injection shape
tool = json.loads(msg.content)["tool"]
if tool not in {"status", "list"}:
    raise ValueError("blocked")
subprocess.run(["mycli", tool], shell=False)

`ToLO-SQL`

# expected: REPORT
sql = parsed_llm["query"]
cursor.execute(sql)

# expected: NOT REPORT for value injection shape
user_id = parsed_llm["user_id"]
cursor.execute("select * from users where id = %s", [user_id])

`ToLO-Path`

# expected: REPORT
path = parsed_llm["path"]
Path(path).read_text()

# expected: NOT REPORT if guard dominates sink
path = (ROOT / parsed_llm["name"]).resolve()
if not path.is_relative_to(ROOT):
    raise ValueError("blocked")
path.read_text()

`ToLO-SSRF`

# expected: REPORT
url = parsed_llm["url"]
requests.get(url)

# expected: NOT REPORT if host/scheme guard dominates sink
url = parsed_llm["url"]
parsed = urlparse(url)
if parsed.scheme != "https" or parsed.hostname not in ALLOWED_HOSTS:
    raise ValueError("blocked")
requests.get(url)

`ToLO-Template`

# expected: REPORT
tpl = ai_message.content
Environment().from_string(tpl).render(context)

# expected: NOT REPORT for template syntax control
name = parsed_llm["template_name"]
if name not in {"summary", "receipt"}:
    raise ValueError("blocked")
env.get_template(f"{name}.html").render(context)

验证工作流

1. 写自然语言 spec

先写清楚:

Source:
  S_LLM^{framework}: LangChain AIMessage.content

Sink:
  ToLO-Path: first argument of open() / Path()

Barrier:
  C_SAFE^{allowlist}: resolve() result must be under fixed ROOT before sink

Propagation:
  json.loads(source) -> dict object
  dict["path"] -> path value

如果自然语言 spec 说不清楚,QL / YAML 往往会写宽。

2. 写最小 fixture

每个 source / sink / barrier 至少配四个 fixture:

Fixture	预期
LLM source -> sink	REPORT
LLM source -> matching barrier -> sink	NOT REPORT
ordinary internal string -> same sink	NOT REPORT as ToLO
LLM source -> parser / field -> sink	REPORT

如果涉及 tool registry 或 wrapper sink,再加:

Fixture	预期
LLM args -> dynamic tool dispatch -> sink	REPORT or REVIEW
LLM args -> unmodeled framework wrapper	REVIEW,并记录 source spec 缺口

3. 跑本地规则

示意命令:

semgrep --config semgrep/tolo-path.yml fixtures/python

codeql database create db-python --language=python --source-root=fixtures/python
codeql query run queries/ToLOPath.ql --database db-python

实际项目可用 codeql test run 管理 query tests。命令是否需要 build 取决于语言和项目结构。

4. 看路径而不是只看数量

每条告警至少检查:

source 位置是否能解释为 S_LLM。
propagation 是否真实保留 LLM 影响。
sink 参数是否真是危险参数。
barrier 是否缺失,或者是否只是未被模型识别。
是否需要配置才能判断。

不要用”报了多少条”替代规则质量判断。ToLO 规则的关键是路径解释是否可审查。

5. 分级输出

建议分三档:

等级	条件	处理
High(可能是真实问题)	`S_LLM` 到高危 sink 的路径完整,未见 matching `C_SAFE`,source / sink 类型都明确	直接 triage,优先看
Medium(配置相关)	路径完整但风险依赖配置:数据库账号权限、sandbox 配置、tool enablement(`allow_dangerous_code` flag)、网络出口策略	需要查应用配置才能定结论
Review(模型不确定)	source、sink、barrier 或 propagation 有一个不确定;框架是 long-tail;sanitizer 是自定义校验函数,语义未确认	需要人工确认框架语义

分级不是漏洞严重性评级,只是 triage 优先级。这种分级比单一 true/false 更实用,也能避免把研究阶段的 survey 结果写成确定漏洞。

6. 记录局限

每条规则应附带局限说明:

支持哪些 SDK / 框架版本。
哪些 source 字段已建模。
哪些 parser / wrapper / registry 已建模。
哪些 sanitizer 会被承认为 barrier。
哪些场景会进入 Review 而不是 High。

这能避免把研究阶段的扫描结果误写成确定覆盖率。

一个完整示例:用规则命中 CVE-2023-29374

目标

写规则命中 LLMMathChain 的 ToLO 路径。

Semgrep 草稿

rules:
  - id: tolo-exec-llmmathchain
    pattern-either:
      - pattern: |
          $CHAIN = LLMMathChain.from_llm(...)
      - pattern: |
          PythonREPL().run($X)
      - pattern: |
          $REPL = PythonREPL(...)
          ...
          $REPL.run($X)
    message: |
      LangChain LLMMathChain or PythonREPL path detected.
      LLM output may flow to exec(). See CVE-2023-29374.
    languages: [python]
    severity: ERROR
    metadata:
      tolo_subclass: ToLO-Exec
      reference_cve: CVE-2023-29374

CodeQL 草稿

class LLMMathChainSink extends DataFlow::Node {
  LLMMathChainSink() {
    exists(Call c |
      c.getFunc().(Attribute).getName() = "run" and
      c.getFunc().(Attribute).getObject().pointsTo().getClass().getName()
        in ["PythonREPL", "PythonAstREPL"] and
      this.asExpr() = c.getArg(0))
  }
}

// 然后加入 ToLOExecConfig.isSink

测试样例

# expected: REPORT
from langchain.chains import LLMMathChain
chain = LLMMathChain.from_llm(llm=ChatOpenAI())
chain.run("compute 2024**2")
# → 内部 PythonREPL.run(expr) → 命中

# expected: NOT REPORT (修复后)
from langchain.chains import LLMMathChain
import numexpr
# 修复后 chain 内部用 numexpr.evaluate(expr, global_dict={})
# isBarrier 识别 numexpr.evaluate 调用

规则设计 checklist

写规则前逐项检查:

source 是否绑定到 S_LLM^{direct,framework,parsed,structured,rag} 中的具体子集?
source 是否过宽,例如所有 str、所有 .content、所有 dict?
sink 是否属于七类 ToLO-*,且精确到危险参数?
parser、structured output、container、tool registry 是否需要额外 propagation step?
sanitizer 是否属于五类 C_SAFE,并与当前 sink 类型匹配?
Pydantic / schema 是否只是验证形状,还是确实收窄了危险语言?
是否有 positive、sanitized、non-LLM、propagation 四类 fixture?
告警信息是否能说明 source、sink、缺失 guard?
结果是否标为 triage,而不是直接标为漏洞?

维护策略

框架升级

LLM SDK 和编排框架经常改字段名、返回类型和 streaming 结构(如 OpenAI v0 → v1 的 message 结构变化)。维护时:

保留旧 source spec,除非确认旧版本不再支持。保留旧字段名作为兼容(source 集合越宽越好,不会漏)。
为新字段加 fixture。
不确定的长尾框架先标 Review,不要直接进入 High。
测试在新旧版本都跑。

Sink 标准库更新

CodeQL 标准库会持续维护传统 CWE sink:

新 SDK 的危险方法
新发现的反射攻击面
新模板引擎

ToLO 规则不需要重写,自动受益于标准库更新。这是”sink 复用”设计的好处。自定义 wrapper sink 只处理框架封装层,不要重写基础 CWE 语义。

Sanitizer 扩展

新增 sanitizer 时先问:

它属于哪一类 C_SAFE?
它只对哪些 ToLO 子类有效?
它是否支配 sink 路径?
有没有 sanitized fixture 能证明不应报告?

如果答不清楚,先不要加入 barrier。过宽 barrier 比误报更危险,因为它会造成 false negative。

当发现新 ToLO 子类时

如果未来发现新 sink 语义(例如 LLM 输出影响 OS 资源限制、IPC、RPC 等),按以下步骤扩展:

验证它不能被现有七子类自然覆盖。
写至少一个公开 CVE 验证。
加 ToLO-X 命名 + 对应 CWE。
加 C_SAFE 类型匹配规则(可能新增一类 sanitizer)。

和 LLM-assisted triage 的关系

LLM 可以帮助 triage 告警,例如总结 taint path、指出可能的 guard、把告警分成 likely true positive / likely false positive / needs review。它不应在 detection runtime 任意改写 source / sink / sanitizer spec,也不应生成 exploit。

查询设计仍应以固定、可审查、可测试的 predicate 为准。LLM 输出只能作为人工 review 的辅助说明。

读完检查

判断下面两条规则哪条更接近 ToLO-aware 设计:

A. 找所有 eval(...) 调用
B. 找 AIMessage.content 经过传播后进入 eval(...),且中间没有 sandbox / capability

B 更接近。A 是传统危险 API 搜索,不能说明 source 是 LLM 输出。

下一步阅读

Sources and Sinks:source、sink、sanitizer 的 canonical 集合。
Predicate Rules:传播规则、container flow、structured output 和 DataFlow / TaintTracking 选型。
Public Case Studies:用公开案例练习人工拆 source、sink、barrier。

Query Design Notes

设计目标

查询不是漏洞证明

规则包结构

第一版规则目标

查询设计顺序

每新增一个 source model,要写测试

Source 设计

CodeQL source 伪谓词

Parser 不是 sanitizer

Sink 设计

Barrier 设计

Barrier 伪谓词

Semgrep 设计

Semgrep taint-mode 骨架

最小 Semgrep 规则草稿

规则 1:eval 接近 AIMessage.content

规则 2:tool_call.arguments 进 subprocess

规则 3:@tool 函数体内 subprocess.run(..., shell=True)

Semgrep 七类草稿

Semgrep 示例:ToLO-SQL

Semgrep 常见坑

Semgrep 限制

CodeQL 设计

CodeQL 查询骨架

Source predicate 分层

LangChainSource.qll 示例

Sink predicate 精确到参数

Additional taint step

ToLOExecConfiguration.qll 完整示例

主查询

CodeQL 输出的告警(示意)

七类 ToLO 查询配置

小型伪样例

ToLO-Deser

ToLO-Exec

ToLO-Shell

ToLO-SQL

ToLO-Path

ToLO-SSRF

ToLO-Template

验证工作流

1. 写自然语言 spec

2. 写最小 fixture

3. 跑本地规则

4. 看路径而不是只看数量

5. 分级输出

6. 记录局限

一个完整示例:用规则命中 CVE-2023-29374

目标

Semgrep 草稿

CodeQL 草稿

测试样例

规则设计 checklist

维护策略

框架升级

Sink 标准库更新

Sanitizer 扩展

当发现新 ToLO 子类时

和 LLM-assisted triage 的关系

读完检查

下一步阅读

规则 1:`eval` 接近 `AIMessage.content`

规则 3:`@tool` 函数体内 `subprocess.run(..., shell=True)`

`LangChainSource.qll` 示例

`ToLOExecConfiguration.qll` 完整示例

`ToLO-Deser`

`ToLO-Exec`

`ToLO-Shell`

`ToLO-SQL`

`ToLO-Path`

`ToLO-SSRF`

`ToLO-Template`