一、写在前面

到这一阶段，开始真正进入项目里最核心的一部分了：代码安全分析引擎。
前面的工作更多是在打通链路，比如后端框架、插件传参和接口通信；而从这一篇开始，真正关心的问题变成了：后端到底怎么开始分析代码。

代码安全分析引擎这一部分可以做很多东西。安全分析当然可以一下子铺很大，比如多语言、多漏洞类型、项目级扫描、AI 解释。但如果第一步就把目标铺满，很容易让功能乱飞、结构失控，最后看似什么都支持，却没有一条风险检测链路真正跑通。

所以，这一版我主动把目标压到最小，只做一个可以闭环验证的 MVP：

• 语言：Python
• 漏洞类型：SQL 注入
• 分析范围：选中代码 / 当前文件
• 分析方式：AST + 规则 + 最小污点传播
• 结果输出：先输出规则分析结果，不强依赖 AI

---

二、最小闭环——这一版我到底想先做成什么

如果把这一版压成一句话，那就是：

我想让后端识别出：用户输入 → 动态拼 SQL → execute 执行 这条链。

比如这段代码：

user_id = request.args.get("id")
sql = f"SELECT * FROM users WHERE id = {user_id}"
cursor.execute(sql)

我这一版的目标，不是“看见 execute 就报风险”，也不是“看见 SQL 字符串就报风险”。

我真正想识别的是：

1. Source（源头）：request.args.get(...) 是用户输入
2. Flow（传播）：用户输入被拼进了动态 SQL
3. Sink（汇聚点）：这个危险 SQL 最终进了 execute()

只有这三件事串起来，才算 一条真正的 SQL 注入候选风险链。

这也是后面所有设计的出发点。

---

三、结构设计

说实话，如果只是为了“先跑起来”，完全可以把所有东西都写进 analyze.py：

• 接请求
• 解析代码
• 查 source
• 查 sink
• 追踪变量
• 返回结果

但这样写，后面一定会炸。

因为一旦后面再加：

• JavaScript
• XSS
• 命令执行
• AI 解释
• 上下文扩展

这个文件会瞬间从“一个入口”变成“一锅粥”。

所以这一版最重要的一个决定就是：

即使功能先做小，也一定要把职责先拆开。

这部分结构我中间改过很多次，因为分析引擎的文件划分方式本身就不止一种。到目前为止，我最终定下来的后端结构如下：

backend/
  api/
    routes/
      analyze.py            # 只做请求分发
  core/
    parser/
      python_parser.py      # 将Python代码解析为 AST
    rules/           # 只负责小判断（算是分析引擎的一些工具吧） 
      python/
        ast_utils.py        # 公共 AST 名称工具
        source_rules.py     # 识别不可信输入来源
        sink_rules.py       # 识别危险执行点
        pattern_matcher.py  # 识别动态 SQL 等危险模式
    flow/           #  负责污染传播 
      python/
        taint_engine.py     # 污染传播分析，核心代码
    decision/       # 负责整理成稳定中间结果（ 接住 taint_engine 的内部结果）
      python/
        risk_evaluator.py
  schemas/
    request_schema.py        # 前端负责传递的内容
    intermediate_schema.py   # 我负责输出的中间结果
    final_result_schema.py   # 后端返回给插件的最终结果

这件事我认为还是很关键的，因为它直接决定了后面的可扩展性以及易读性。

---

四、结果分层

项目里的“结果”并不是只有一种，而是至少有两层。

1. 中间结果

分析引擎先给出：

• 风险候选类型
• 风险位置
• 证据链
• 当前判断是否需要更多上下文

它不是最终给插件看的，而是给后续 AI 调用 / 上下文增强 / 智能解释 用的。

2. 最终结果

前端真正要拿来做：

• 高亮
• Problems 列表
• 点击跳转
• 解释展示

所以我这里考虑把 schema 明确拆成了两层：

intermediate_schema.py

用于：安全分析 -> AI调用

class EvidenceChain(BaseModel):   # 证据链
    source_line: Optional[int] = None
    sql_build_line: Optional[int] = None
    sink_line: int
    source_variable: Optional[str] = None
    sql_variable: Optional[str] = None

class RiskCandidate(BaseModel):  # 交给 C 的候选风险(risk_evaluator负责给)
    risk_type_candidate: Literal["SQLInjection"]
    severity: Literal["low", "medium", "high"]
    file_path: str
    start_line: int
    end_line: int
    summary: str
    evidence_chain: EvidenceChain
    needs_more_context: bool = False

class IntermediateAnalyzeResult(BaseModel):  # 中间分析结果
    summary: str
    risk_level: Literal["none", "suspicious", "danger"]
    candidates: list[RiskCandidate] = Field(default_factory=list)

final_result_schema.py

用于：后端 -> 插件

class RiskItem(BaseModel):
    type: str
    severity: Literal["low", "medium", "high"]
    file_path: str
    start_line: int
    end_line: int
    reason: str
    attack_path: str
    fix_minimal: str
    fix_best: str

class AnalyzeResponse(BaseModel):
    summary: str
    risk_level: Literal["none", "suspicious", "danger"]
    risks: list[RiskItem] = Field(default_factory=list)

因为如果分析引擎一开始就直接“面向插件拼结果”，那后面一旦接入 AI 调用、上下文增强、解释生成，就会出现一个问题：

• 分析引擎想直接给前端结果
• AI调用又想基于分析结果继续加工
• 路由层也在拼结果

最后大家都会碰“结果对象”，边界会越来越糊。

所以我现在更明确地把它理解成：

分析引擎的正式产物，应该先是一份稳定的中间分析结果；至于最终展示结果，再由后续模块或当前的路由层继续组装（比如调试时，后面也写到了），而不是跳过这一步，直接去生成插件的展示结果。

---

五、我为什么先做 AST，而不是只做字符串匹配

一开始最容易想到的方法，其实是字符串匹配：

• 搜 request.args.get
• 搜 execute(
• 搜 SQL 拼接

这样当然能做一点事，但它有两个明显问题：

1. 不稳定
2. 很难扩展

所以我一开始就先把 Python 代码解析接上了：

import ast
import textwrap
from dataclasses import dataclass
from typing import Optional

@dataclass
class ParseResult:
    tree: Optional[ast.AST]
    error: Optional[str] = None

def parse_python_code(code: str) -> ParseResult:
    normalized_code = textwrap.dedent(code)    # 不要忘了先去掉公共缩进

    try:
        return ParseResult(tree=ast.parse(normalized_code))     # 会把 Python 代码变成语法树
    except SyntaxError as exc:
        return ParseResult(
            tree=None,
            error=f"{exc.msg} (line {exc.lineno})",
        )

1）ast.parse(...)

首先利用Python的代码分析库把代码变成 AST（语法树），这样后面看到的就不只是“字符串”了，而是能区分：

• 赋值语句
• 函数调用
• f-string
• 变量引用
• 属性访问

能“按结构看代码”，而不是“按文本猜代码”。

2）textwrap.dedent(...)

还有一个点需要注意：

因为插件以后会经常分析“用户选中的代码”，而选中的片段很多时候来自函数内部，会带公共缩进。如果不先把公共缩进去掉，ast.parse() 很容易直接报错。

---

六、规则层为什么要拆成 source / sink / pattern

我最开始写的其实是“一整个 python_sql_injection_rules.py 文件全塞进去”的版本，后来发现这样长期一定不行。因为以后一旦再加：

• XSS
• 命令执行
• 硬编码密钥

规则会越来越杂。

所以我决定按职责拆，而不是按“当前漏洞类型”硬塞。

---

1. source_rules.py：识别危险数据入口

import ast

from core.rules.python.ast_utils import get_full_name

REQUEST_SOURCE_NAMES = {
    "request.args.get",
    "request.form.get",
    "request.values.get",
}

def is_request_source_call(node: ast.AST) -> bool:  # 判断一个节点是不是用户输入来源
    if not isinstance(node, ast.Call):
        return False

    return get_full_name(node.func) in REQUEST_SOURCE_NAMES

这一层只回答：

这是不是来自请求的用户输入？

比如：

• request.args.get(...)
• request.form.get(...)
• request.values.get(...)

这几个都算 source。

---

2. sink_rules.py：识别危险汇聚点

import ast

from core.rules.python.ast_utils import get_full_name

def is_execute_sink_call(node: ast.AST) -> bool:  # 判断一个节点是不是数据库执行点
    if not isinstance(node, ast.Call):
        return False

    call_name = get_full_name(node.func)
    if call_name is None:
        return False

    return call_name == "execute" or call_name.endswith(".execute")

这个函数只回答一个问题：

这个调用是不是数据库执行点？

比如：

• execute(...)
• cursor.execute(...)
• db.execute(...)

这些都算 sink。

我还额外加了一个“保守放行”判断（对于完全安全的写法（参数化查询））：

def looks_like_parameterized_execute(node: ast.AST) -> bool:
    if not is_execute_sink_call(node):
        return False

    # 参数个数必须 ≥ 2
    if len(node.args) < 2:
        return False

    # 第一个参数必须是字符串常量
    first_arg = node.args[0]
    # 第一个参数必须是字符串常量（硬编码SQL，不是拼接的变量）
    return isinstance(first_arg, ast.Constant) and isinstance(first_arg.value, str)

---

3. pattern_matcher.py：识别动态 SQL 模式

import ast


def is_dynamic_sql_expr(node: ast.AST) -> bool:  # 判断“这是不是动态拼 SQL”
    # sql = f"SELECT * FROM users WHERE id = {user_id}"
    if isinstance(node, ast.JoinedStr):
        return True

    # 字符串加法 或 % 格式化
    if isinstance(node, ast.BinOp) and isinstance(node.op, (ast.Add, ast.Mod)):
        return True

    # sql = "SELECT * FROM users WHERE id = {}".format(user_id)
    if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
        return node.func.attr == "format"

    return False

这一层只回答：

这个表达式是不是动态 SQL？

目前支持最典型的三种：

• f-string
• + 拼接、% 格式化
• .format()

为什么要这么拆？因为以后就算做别的漏洞类型，也还是会重复用到这三类职责：

• 哪些是 source
• 哪些是 sink
• 哪些是危险模式

---

七、核心文件：taint_engine.py

先把“来自请求的值”标脏，再把“带脏数据的动态 SQL”标危险，最后检查 execute() 有没有执行这个危险 SQL。

也就是：

• 不可信输入先标记
• 污染关系先传播
• 真正的漏洞只在 sink 处判定

---

这里需要把“为什么只有 visit_Call 上报漏洞”讲清楚（其实也是核心设计思路）。

所以呢，为什么？

因为静态安全检测，尤其是 SQL 注入检测，必须满足这三个条件：

安全检测三要素

1. Source（源头）：用户输入，属于脏数据
2. Flow（传播）：脏数据被拼到 SQL 中
3. Sink（汇聚点）：危险 SQL 被真正执行

只有这三件事全满足，才叫 一条真正成立的漏洞链。

---

对应到我的代码里，就是：

代码位置	作用	会不会直接上报漏洞？
visit_Assign	标记脏变量、标记危险 SQL 变量	❌ 不会
_find_taint_origin	追查脏数据来源	❌ 不会
visit_Call	发现 execute() 执行了危险 SQL	✅ 会

也就是说：

• 光有用户输入，不是漏洞
• 光有动态拼 SQL，也不是漏洞
• 只有执行了这个危险 SQL，才是 SQL 注入风险

visit_Assign 和 _find_taint_origin 这些函数的逻辑，本质上都属于：

为最后的 visit_Call 提供判断依据

它们是“过程逻辑”，不是“上报逻辑”。

---

这里举一个例子，把这条链完整跑一遍

# 1. 用户输入（Source）
user_input = request.args.get("id")

# 2. 拼接 SQL（Flow）
sql = f"SELECT * FROM user WHERE id={user_input}"

# 3. 执行 SQL（Sink）
cursor.execute(sql)

第一步：扫描第 1 行赋值

会进入 visit_Assign

user_input = request.args.get("id")

这里会做什么？

• 右边是 request.args.get(...)
• 规则层会判断它是 source
• 所以 user_input 会被加入 tainted_variables

此时， user_input 现在被标记成了“脏变量”

---

第二步：扫描第 2 行赋值

还是进入 visit_Assign

sql = f"SELECT * FROM user WHERE id={user_input}"

这里会做什么？

• 右边是 f-string
• 规则层会判断它是 动态 SQL
• 同时里面又引用了已经脏掉的 user_input
• 所以 sql 会被加入 dangerous_sql_variables

sql 现在被标记成了“危险 SQL 变量”

---

第三步：扫描第 3 行函数调用

这次会进入 visit_Call

cursor.execute(sql)

这里会做什么？

1. 判断这是 execute()，也就是 sink
2. 判断它不是“看起来像参数化查询”的安全写法
3. 判断它的第一个参数 sql 是不是危险 SQL
4. 如果满足，就把结果加入 findings

到这一步，才真正形成：

用户输入 → 动态 SQL → execute 执行

---

八、为什么还要多一个 risk_evaluator.py

taint_engine.py 找到的结果，更像是一条“内部线索”：

• source 在哪一行
• SQL 在哪一行构造
• sink 在哪一行执行
• source 变量是谁
• SQL 变量是谁

这些东西对分析引擎来说够用了，但对后续模块来说还不够“稳定”。
所以 risk_evaluator.py 做的事情就是：把这些内部线索，统一包装成 RiskCandidate。也就是说，它把“引擎内部发现”变成了“模块之间可以稳定传递的中间结果”。

核心部分是这一段转换：

for finding in taint_result.findings:
    candidates.append(
        RiskCandidate(
            risk_type_candidate="SQLInjection",
            severity="high",
            file_path=file_path,
            start_line=finding.sink_line,
            end_line=finding.sink_line,
            summary="User-controlled input flows into a dynamically built SQL query before execute().",
            evidence_chain=EvidenceChain(
                source_line=finding.source_line,
                sql_build_line=finding.sql_build_line,
                sink_line=finding.sink_line,
                source_variable=finding.source_variable,
                sql_variable=finding.sql_variable,
            ),
            needs_more_context=False,
        )
    )

这段代码本质上只干了一件事：

把 taint_engine.py 的内部 finding，转成标准的 RiskCandidate。

所以对我来说，risk_evaluator.py 更像是这一阶段分析引擎的“结果整理层”。

---

九、为什么 analyze.py 现在还要做一次“临时转换”

当前阶段还有一个现实问题：插件侧已经需要 AnalyzeResponse，但 AI 增强链路还没接进来，而分析引擎产出的正式结果又是 RiskCandidate。

所以在这一版里，analyze.py 暂时承担了一个很明确的过渡职责：把中间分析结果桥接成插件当前可消费的最终返回格式。

# 当前阶段：先把分析的中间结果转成插件能显示的最终结果
risks = [
    RiskItem(
        type=candidate.risk_type_candidate,
        severity=candidate.severity,
        file_path=candidate.file_path,
        start_line=candidate.start_line,
        end_line=candidate.end_line,
        reason=candidate.summary,
        attack_path="stage candidate only. AI explanation has not been generated yet.",
        fix_minimal="Replace dynamic SQL string building with parameterized queries.",
        fix_best="Keep SQL templates constant and pass user input as bound parameters.",
    )
        for candidate in evaluation.candidates
    ]

response = AnalyzeResponse(
    summary=evaluation.summary,
    risk_level=evaluation.risk_level,
    risks=risks,
)
return response.model_dump()

这不是分析逻辑本身，而是 MVP 阶段为了联调整条链路做的临时适配。

---

十、一个很实际的问题：选区行号怎么映射回原文件

这个点特别容易被忽视，但对于插件联调非常重要。
如果用户在 VS Code 里只分析选中的一段代码，那么 AST 里：

• 第 1 行
• 第 2 行
• 第 3 行

不一定对应原文件的第 1、2、3 行。

比如用户是从原文件第 20 行开始选中，那选区里的第 1 行，其实应该对应原文件第 20 行。

所以我在路由层保留了：

line_offset = 0
if req.mode == "selection" and req.cursor_range is not None:
    line_offset = req.cursor_range.start_line - 1

然后在 taint engine 里通过 _real_line() 做统一转换。

---

十一、测试——从“后端能启动”到“接口能返回结果”

到这里，前面的结构设计和分析逻辑都已经搭起来了，接下来最重要的不是继续补功能，而是先验证：这条最小 SQL 注入检测链到底有没有真的跑通。

1. 启动后端服务

首先把 FastAPI 后端跑起来。

先切到后端目录：

cd backend

激活虚拟环境：

.\.venv\Scripts\Activate.ps1

后端启动：

python -m uvicorn app:app --host 127.0.0.1 --port 8000 --reload

如果启动成功，终端里通常会看到类似下面的输出：

Uvicorn running on http://127.0.0.1:8000

2. 手动测试 /analyze 接口

直接在浏览器里打开： http://127.0.0.1:8000/docs

在确认后端服务正常之后，可以选择直接在 POST /analyze 里点开 Try it out，然后输入一段最小 SQL 注入样例：

{
  "mode": "file",
  "language": "python",
  "file_path": "tests/cases/python/sql_injection_01.py",
  "code": "user_id = request.args.get(\"id\")\nsql = f\"SELECT * FROM users WHERE id = {user_id}\"\ncursor.execute(sql)",
  "workspace_root": "c:\\Users\\25395\\Desktop\\code-guard-tutor",
  "cursor_range": null
}

点击 Execute。

这一步我主要是想验证三件事：

1. 后端能不能正确接收请求
2. 分析引擎能不能真正跑起来
3. 返回结果里有没有我预期的风险字段

3. 返回字段

运行结果如下：

可以看到返回 JSON 里的这几个字段：

• risk_level 已经能返回 danger
• risks 数组里有内容
• 行号正确
• 风险类型和原因符合我当前 SQL 注入主线的预期

到这里，当前这一版“最小分析闭环”就算是真正跑通了。

上面是利用fastAPI的docs跑的，我还在vscode上直接利用插件测试了一下（可以看到输出这里也是成功的）：

---

十二、总结

这一版并不大，甚至只能算一个最小闭环。但它真正解决了一个更重要的问题：

分析引擎这一层，不再停留在概念上，而是已经开始落到具体代码和实际链路里了。

从插件传参，到 AST 解析、规则识别、最小污点传播，再到中间结果输出和最终结果桥接，这条链虽然还很短，却已经能够稳定跑通。

后面不管是 JavaScript、更多漏洞类型，还是上下文补全和 AI 解释链路，都会建立在这条基础链路之上。

未完待续……