大模型辅助的 SQL 注入检测与安全审计：从规则匹配到语义理解

一、SQL 注入的"变异对抗"：规则引擎的检测盲区

SQL 注入（SQLi）是最持久的 Web 安全威胁之一，OWASP Top 10 常年上榜。传统防御依赖 WAF（Web Application Firewall）的规则匹配——检测 OR 1=1、UNION SELECT、-- 注释符等特征模式。但攻击者不断变异绕过手段：大小写混写（oR 1=1）、编码绕过（%4F%52）、注释分割（UN/**/ION）、等价表达式（OR 'a'='a'）。规则引擎的检测率在已知模式上可达 95%+，但对零日变异的检测率骤降至 60% 以下。更关键的是，规则引擎的误报率在复杂业务 SQL 中偏高——一条包含 UNION 的合法报表查询可能被误拦截，导致业务中断。

二、SQL 注入检测的技术演进

2.1 从规则匹配到语义理解的层级提升

flowchart TB
    A[输入 SQL] --> B[规则引擎<br/>正则匹配]
    B --> C{命中规则?}
    C -->|是| D[拦截]
    C -->|否| E[AST 解析]
    E --> F{语法合法?}
    F -->|否| G[可疑：编码绕过]
    F -->|是| H[语义分析]
    H --> I[LLM 判断意图]
    I --> J{注入意图?}
    J -->|是| K[拦截 + 告警]
    J -->|否| L[放行]

    subgraph 规则引擎盲区
        M[大小写变异]
        N[编码绕过]
        O[等价表达式]
        P[注释分割]
    end

    M & N & O & P --> E

2.2 传统规则引擎的局限

import re

class RuleBasedSQLiDetector:
    """传统规则引擎：正则匹配已知注入模式"""

    RULES = [
        # 经典注入模式
        (re.compile(r"(?i)(\bOR\b\s+\d+\s*=\s*\d+)"), "OR恒真条件"),
        (re.compile(r"(?i)(\bUNION\b\s+\bSELECT\b)"), "UNION注入"),
        (re.compile(r"(?i)(--|#|/\*)"), "注释注入"),
        (re.compile(r"(?i)(\bDROP\b\s+\bTABLE\b)"), "DROP注入"),
        (re.compile(r"(?i)(;\s*\b(SELECT|INSERT|UPDATE|DELETE)\b)"), "堆叠注入"),
        (re.compile(r"(\bCHAR\s*\(|\bCONCAT\s*\()", re.I), "函数注入"),
    ]

    def detect(self, sql: str) -> dict:
        for pattern, name in self.RULES:
            if pattern.search(sql):
                return {"detected": True, "rule": name, "confidence": 0.7}

        # 无法检测的变异
        # OR 'a'='a'  → 不匹配 OR 数字=数字
        # UN/**/ION SELECT → 不匹配 UNION SELECT
        return {"detected": False, "confidence": 0.3}

三、大模型辅助的语义检测方案

3.1 基于 AST 的规范化预处理

import sqlparse
from sqlparse.sql import Identifier, Function, Parenthesis
from sqlparse.tokens import Keyword, DML, Punctuation

class SQLNormalizer:
    """SQL 规范化：消除编码和注释干扰"""

    def normalize(self, sql: str) -> str:
        # 1. URL 解码
        decoded = self._url_decode(sql)

        # 2. 移除注释
        parsed = sqlparse.parse(decoded)
        cleaned = []
        for statement in parsed:
            # 移除块注释和行注释
            tokens = []
            for token in statement.flatten():
                if token.ttype not in (sqlparse.tokens.Comment.Single,
                                        sqlparse.tokens.Comment.Multiline):
                    tokens.append(token.value)
            cleaned.append(''.join(tokens))

        # 3. 统一大小写（关键字大写）
        normalized = ' '.join(cleaned)
        normalized = sqlparse.format(
            normalized,
            keyword_case='upper',
            strip_comments=True,
        )

        return normalized

    @staticmethod
    def _url_decode(s: str) -> str:
        import urllib.parse
        try:
            return urllib.parse.unquote(s)
        except Exception:
            return s

3.2 LLM 语义判断

import json

class LLMSQLiDetector:
    """使用 LLM 进行 SQL 注入的语义判断"""

    def __init__(self, llm_client):
        self.llm = llm_client

    def detect(self, sql: str, context: dict = None) -> dict:
        """语义层面的注入检测"""
        prompt = f"""分析以下 SQL 语句是否存在注入攻击意图。

SQL 语句: {sql}
业务上下文: {context or '未知'}

判断标准：
1. 恒真条件：WHERE 子句中是否存在永真表达式（如 1=1, 'a'='a', TRUE）
2. 越权访问：是否尝试访问非授权数据（如 UNION 获取其他表）
3. 破坏性操作：是否包含 DROP, TRUNCATE 等危险操作
4. 绕过手法：是否使用编码、注释、等价表达式绕过过滤
5. 堆叠注入：是否包含分号分隔的多条语句

注意区分：
- 合法的 UNION 查询（报表、数据同步）vs 注入型 UNION
- 合法的 OR 条件（多条件筛选）vs 恒真 OR
- 合法的子查询 vs 嵌套注入

输出 JSON:
{{"is_injection": true/false, "confidence": 0.0-1.0, "reason": "...", "attack_type": "..."}}"""

        response = self.llm.chat(prompt)
        try:
            result = json.loads(response)
        except json.JSONDecodeError:
            result = {"is_injection": False, "confidence": 0.0, "reason": "解析失败"}

        return result

3.3 多层防御管线

class SQLiDefensePipeline:
    """规则引擎 + AST + LLM 的多层防御管线"""

    def __init__(self, llm_client):
        self.rule_detector = RuleBasedSQLiDetector()
        self.normalizer = SQLNormalizer()
        self.llm_detector = LLMSQLiDetector(llm_client)

    def analyze(self, raw_sql: str, context: dict = None) -> dict:
        """多层分析：快速规则过滤 + 语义深度判断"""

        # 层1：规则引擎快速过滤（<1ms）
        rule_result = self.rule_detector.detect(raw_sql)
        if rule_result['detected'] and rule_result['confidence'] > 0.9:
            return {
                'action': 'block',
                'reason': f"规则引擎拦截: {rule_result['rule']}",
                'confidence': rule_result['confidence'],
                'layer': 'rule',
            }

        # 层2：规范化 + AST 解析（<5ms）
        normalized = self.normalizer.normalize(raw_sql)
        try:
            parsed = sqlparse.parse(normalized)
            if not parsed:
                return {
                    'action': 'block',
                    'reason': 'SQL 解析失败，可能为编码绕过',
                    'confidence': 0.8,
                    'layer': 'ast',
                }
        except Exception:
            return {
                'action': 'block',
                'reason': 'SQL 语法异常',
                'confidence': 0.75,
                'layer': 'ast',
            }

        # 层3：LLM 语义判断（50-200ms）
        llm_result = self.llm_detector.detect(normalized, context)

        if llm_result['is_injection'] and llm_result['confidence'] > 0.7:
            return {
                'action': 'block',
                'reason': llm_result['reason'],
                'confidence': llm_result['confidence'],
                'layer': 'llm',
                'attack_type': llm_result.get('attack_type'),
            }

        # 层3 低置信度告警
        if llm_result['is_injection'] and llm_result['confidence'] > 0.4:
            return {
                'action': 'alert',
                'reason': llm_result['reason'],
                'confidence': llm_result['confidence'],
                'layer': 'llm',
            }

        return {
            'action': 'allow',
            'confidence': max(rule_result['confidence'], llm_result.get('confidence', 0)),
            'layer': 'all',
        }

四、边界分析与架构权衡

4.1 LLM 的误报与漏报

LLM 可能将合法的复杂 SQL 误判为注入（如包含 UNION 的报表查询），也可能被精心构造的"语义伪装"绕过（如使用业务术语包装的注入语句）。缓解策略：提供业务上下文（表结构、应用场景）帮助 LLM 理解查询意图；对 LLM 判定结果设置置信度阈值，低置信度时转人工审核。

4.2 延迟对业务的影响

规则引擎检测耗时 <1ms，LLM 语义判断耗时 50-200ms。在请求延迟敏感的 API 网关中，200ms 的额外延迟不可接受。优化方案：规则引擎作为同步拦截层（<1ms），LLM 作为异步审计层（事后分析），两者解耦运行。

4.3 提示注入攻击

攻击者可能构造"针对 LLM 的提示注入"——在 SQL 中嵌入指令，试图让 LLM 输出"非注入"的判断。防御措施：将 SQL 严格作为数据输入，使用系统提示约束 LLM 的输出格式，禁止 LLM 执行 SQL 中的任何指令。

4.4 合规审计的日志要求

安全审计要求所有拦截和告警事件可追溯。多层管线的每层判断都需要记录：原始 SQL、规范化结果、规则匹配详情、LLM 判断理由和置信度。日志存储需满足合规保留期限（通常 6-12 个月）。

五、总结

大模型辅助的 SQL 注入检测，在传统规则引擎的基础上增加了语义理解层。规则引擎快速过滤已知模式（<1ms），AST 解析检测编码绕过，LLM 判断变异注入的语义意图。三层管线按延迟递增排列，高置信度拦截在低延迟层完成，低置信度告警交由 LLM 深度分析。工程实践中需注意 LLM 的误报和漏报、延迟对业务的影响、提示注入攻击，以及合规审计的日志要求。LLM 检测最适合作为异步审计层，与同步规则引擎解耦运行，在保证业务延迟的前提下提升零日变异的检测率。