如何配置Semgrep规则：仅将create_sql_agent初始化的AgentExecutor的invoke设为Sink

问题核心 #

需要追踪input()获取的用户输入，是否传递给了由langchain.agents.create_sql_agent初始化的AgentExecutor实例的invoke方法——这类调用存在SQL注入风险，需精准检测。

正确的Semgrep规则配置 #

以下是针对Python场景的完整规则，解决了匹配不准确、污点传播缺失等常见失效问题：

rules:
  - id: input-to-sql-agent-invoke
    mode: taint
    languages: [python]
    message: "用户输入直接传入SQL Agent的invoke方法，存在SQL注入风险"
    severity: ERROR
    metadata:
      category: security
      tech: [langchain, semgrep]
    sources:
      - patterns:
          - pattern: input(...)
          - label: SOURCE
    sinks:
      - patterns:
          - pattern: $AGENT.invoke(...)
          - pattern-either:
              - pattern-inside: |
                  $AGENT = create_sql_agent(...)
                  ...
              - pattern-inside: |
                  def $FUNC(...):
                      $AGENT = create_sql_agent(...)
                      ...
          - label: SINK
    propagators:
      - pattern: $X = $Y
        from: $Y
        to: $X
      - pattern: def $FUNC($ARG): ... return $ARG
        from: $ARG
        to: return

规则关键细节解释 #

• Sources：精准匹配input()的所有调用，标记为污点源。
• Sinks：仅匹配由create_sql_agent初始化的AgentExecutor实例的invoke方法，避免误报其他来源的AgentExecutor调用。
• Propagators：处理变量赋值、函数返回值的污点传播，确保输入经过多步传递后仍能被追踪到。

测试示例代码 #

命中场景（规则会触发告警） #

from langchain.agents import create_sql_agent
from langchain.agents.agent_toolkits import SQLDatabaseToolkit
from langchain.sql_database import SQLDatabase

db = SQLDatabase.from_uri("sqlite:///example.db")
toolkit = SQLDatabaseToolkit(db=db)
agent_executor = create_sql_agent(toolkit=toolkit, verbose=True)

# 直接传入input()，会被检测到
user_input = input("请输入查询指令：")
agent_executor.invoke({"input": user_input})

# 经过函数传递，也会被检测到
def get_user_input():
    return input("请输入查询指令：")

agent_executor.invoke({"input": get_user_input()})

非命中场景（规则不会触发） #

from langchain.agents import AgentExecutor

# 非create_sql_agent创建的AgentExecutor，不会触发
custom_agent = AgentExecutor(...)
custom_agent.invoke({"input": input("输入内容：")})

常见失效原因排查 #

1. Sink匹配范围问题：之前的规则可能未限定AgentExecutor来自create_sql_agent，导致误报或漏报。
2. 缺失传播规则：如果输入经过变量赋值、函数传递后才到invoke，没有传播规则会导致污点追踪中断。
3. 语法错误：检查规则的YAML格式是否正确，比如缩进、pattern语法是否符合Semgrep要求。

内容的提问来源于stack exchange，提问作者BenGababy