Ruoyi-Scan 插件开发教程#

from abc import ABC, abstractmethod

class PluginBase(ABC):
    # 类属性：插件元信息（子类覆盖）
    name = ""
    cve = ""
    severity = "low"   # high / medium / low
    category = ""
    # ... 其他元属性

    @abstractmethod
    def verify(self, target: str, session: "SessionManager") -> ScanResult:
        """执行检测，返回 ScanResult（三态判定）"""
        raise NotImplementedError

STATUS_CONFIRMED = "CONFIRMED"   # 确认漏洞存在（有明确证据）
STATUS_SAFE       = "SAFE"       # 确认漏洞不存在（有明确反证）
STATUS_UNKNOWN    = "UNKNOWN"    # 无法判定（网络异常、超时、响应异常等）

load  →  route  →  verify  →  result
 │        │        │          │
 │        │        │          └─ ScanResult（CONFIRMED / SAFE / UNKNOWN）
 │        │        └─ 调用 plugin.verify(target, session)
 │        └─ 按指纹 cms 选择插件包（plugins/<cms>/__init__.py 的 plugin_list）
 └─ core/loader.py: load_plugins / load_external_plugins / load_entry_point_plugins

python main.py --plugin-init my_plugin --category ruoyi

[*]生成插件模板
    名称: my_plugin
    类别: ruoyi
[+]插件已生成: plugins/ruoyi/my_plugin.py
[*]下一步:
    1. 编辑 plugins/ruoyi/my_plugin.py 完善检测逻辑
    2. 运行 python main.py --plugin-check plugins/ruoyi/my_plugin.py 验证
    3. 运行 python main.py -u http://target/ 扫描

# plugins/ruoyi/my_plugin.py
# 最小可运行插件示例：检测 /demo/test 端点是否暴露
from common.models import STATUS_CONFIRMED, STATUS_SAFE, STATUS_UNKNOWN, ScanResult
from core.http import join_url
from lib.colors import ok, no
from plugins.base import PluginBase


class MyPluginPlugin(PluginBase):
    """最小可运行插件示例"""
    name = "测试端点暴露"
    cve = "N/A"
    severity = "low"
    category = "vuln"
    description = "检测 /demo/test 端点是否可被未授权访问"
    fix = "关闭 /demo/test 端点或添加鉴权"
    # CVSS v3.1 向量 + 合规映射（可省略，默认 0.0 / 空）
    cvss_vector = "AV:N/AC:L/PR:N/UI:N/S:U/C:L/I:N/A:N"
    compliance = "等保2.0:8.1.4;OWASP:A05:2021"
    # WAF 绕过支持
    vuln_type = "info"          # info / sqli / xss / rce / file_read / auth / other
    supports_waf_bypass = False

    def verify(self, target, session):
        url = join_url(target, "/demo/test")
        try:
            resp = session.get(url, timeout=10)
        except Exception as e:
            # 铁律：网络异常 → UNKNOWN，绝不判 SAFE
            print(no(f"{self.name}（网络异常）"))
            return ScanResult(
                kind="vuln", name=self.name, status=STATUS_UNKNOWN,
                url=url, evidence=str(e),
            )

        text = resp.text or ""
        # 明确证据 → CONFIRMED
        if "demo-test-marker" in text:
            print(ok(f"存在{self.name}"))
            return ScanResult(
                kind="vuln", name=self.name, severity=self.severity,
                status=STATUS_CONFIRMED, url=url,
                evidence="响应含 demo-test-marker 特征",
                fix=self.fix,
            )
        # 明确反证 → SAFE
        print(no(f"不存在{self.name}"))
        return ScanResult(
            kind="vuln", name=self.name, status=STATUS_SAFE,
            url=url, evidence="响应不含特征",
        )

from plugins.ruoyi.my_plugin import MyPluginPlugin

plugin_list = [
    # ... 其他插件
    MyPluginPlugin,  # 新增插件
]

# 静态 + 导入双重验证
python main.py --plugin-check plugins/ruoyi/my_plugin.py

# 列出所有插件元数据
python main.py --plugin-list

# 扫描
python main.py -u http://target/ -p

plugins/
└── ruoyi/                       # 插件包（按 CMS 分类）
    ├── __init__.py              # 声明 plugin_list = [PluginA, PluginB, ...]
    ├── base.py → (实际位于 plugins/base.py)
    ├── file_read.py             # FileReadPlugin
    ├── sql_inject_role.py       # SqlInjectRolePlugin
    ├── ...
    └── my_plugin.py             # MyPluginPlugin（新插件）

@abstractmethod
def verify(self, target: str, session: "SessionManager") -> ScanResult:
    """执行检测，返回 ScanResult（三态判定）

    网络异常等不可判定情形必须返回 status=UNKNOWN，不得判为 SAFE。
    """
    raise NotImplementedError

def get(self, url: str, headers: Optional[Dict[str, str]] = None, **kwargs) -> requests.Response
def post(self, url: str, headers: Optional[Dict[str, str]] = None,
         data: Optional[Dict[str, str]] = None, **kwargs) -> requests.Response
def request(self, method: str, url: str, headers: Optional[Dict[str, str]] = None, **kwargs) -> requests.Response
def close(self) -> None

def _build_result(
    self, status: str, url: str = "", evidence: str = "",
    extra: Optional[Dict[str, Any]] = None,
) -> ScanResult:
    """辅助方法：构造 ScanResult 并自动填充 kind/name/severity/fix

    插件在 verify() 中可直接用此方法构建结果，自动继承插件类属性。
    D12：自动填充 cve/cvss_score/cvss_vector/compliance。
    D18/D24：自动填充 fix_detail/reproduce。
    """
    return ScanResult(
        kind="vuln" if status == STATUS_CONFIRMED else "info",
        name=self.name,
        severity=self.severity,
        status=status,
        url=url,
        evidence=evidence,
        extra=extra or {},
        fix=self.fix or "",
        fix_detail=self.fix_detail or "",
        reproduce=self.reproduce or "",
        cve=self.cve or "",
        cvss_score=cvss_score(self.cvss_vector) if self.cvss_vector else 0.0,
        cvss_vector=self.cvss_vector or "",
        compliance=parse_compliance(self.compliance) if self.compliance else {},
    )

def verify(self, target, session):
    url = join_url(target, "/some/path")
    try:
        resp = session.get(url)
    except Exception as e:
        return self._build_result(STATUS_UNKNOWN, url=url, evidence=str(e))

    if "vuln-marker" in resp.text:
        return self._build_result(
            STATUS_CONFIRMED, url=url,
            evidence="响应含 vuln-marker 特征",
            extra={"vuln_type": "sqli", "payload_class": "marker_probe"},
        )
    return self._build_result(STATUS_SAFE, url=url)

def verify_with_bypass(
    self, target: str, bypass_session: "SessionManager", bypass_ctx: Any
) -> ScanResult:
    """WAF 绕过验证（D7）：子类覆盖以实现绕过逻辑

    默认实现：复用 verify()，但使用 BypassSession（已应用传输层变换）。
    子类可覆盖此方法，利用 bypass_ctx.original_payload 和策略变形函数
    构造绕过 payload。
    """
    return self.verify(target, bypass_session)

def meta(self) -> Dict[str, str]:
    """返回插件元信息字典"""
    return {
        "name": self.name,
        "cve": self.cve,
        "severity": self.severity,
        "category": self.category,
        "description": self.description,
        "fix": self.fix,
        "fix_detail": self.fix_detail,
        "reproduce": self.reproduce,
        "cvss_vector": self.cvss_vector,
        "cvss_score": cvss_score(self.cvss_vector),
        "compliance": parse_compliance(self.compliance),
        "affected_versions": self.affected_versions,
    }

# 好：双关键词联合判定，降低误报
if match_all(resp.text, ["root", ":/"]):
    return self._build_result(
        STATUS_CONFIRMED, url=url,
        evidence="响应含 root 与 :/ 特征（/etc/passwd）",
    )

# 好：响应不含任何漏洞特征，且非异常状态码
if resp.status_code == 200 and "vuln-marker" not in resp.text:
    return self._build_result(STATUS_SAFE, url=url)

# 铁律：网络异常 → UNKNOWN
try:
    resp = session.get(url, timeout=10)
except Exception as e:
    return self._build_result(
        STATUS_UNKNOWN, url=url,
        evidence=f"网络异常: {e}",
    )

# 5xx 也归 UNKNOWN（可能服务过载或 WAF 拦截）
if resp.status_code >= 500:
    return self._build_result(
        STATUS_UNKNOWN, url=url,
        evidence=f"服务端错误: HTTP {resp.status_code}",
    )

from lib.matcher import match_all

# 差：单关键字容易误报（业务页面里恰好出现 "root"）
if "root" in resp.text:
    return self._build_result(STATUS_CONFIRMED, ...)

# 好：多关键字联合（/etc/passwd 同时含 "root" 和 ":/"）
if match_all(resp.text, ["root", ":/"]):
    return self._build_result(STATUS_CONFIRMED, ...)

from lib.matcher import match_spring_actuator_env

if match_spring_actuator_env(resp.text):
    return self._build_result(
        STATUS_CONFIRMED, url=url,
        evidence="响应含 propertySources / activeProfiles 特征",
    )

plugins/
└── my_cms/                    # 新增 CMS 框架时只需建包
    ├── __init__.py             # 声明 plugin_list
    ├── vuln_a.py               # VulnAPlugin
    └── vuln_b.py               # VulnBPlugin

from plugins.my_cms.vuln_a import VulnAPlugin
from plugins.my_cms.vuln_b import VulnBPlugin

plugin_list = [
    # 按危险度从高到低排序（high 在前）
    VulnAPlugin,  # 高危漏洞
    VulnBPlugin,  # 中危漏洞
]

# core/loader.py 第 140-161 行
def discover_plugin_packages() -> List[str]:
    """自动发现所有插件包（消除硬编码，P1: 插件自动发现）"""
    _EXCLUDED = {"chain"}
    plugins_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "plugins")
    packages = []
    for name in sorted(os.listdir(plugins_dir)):
        if name in _EXCLUDED:
            continue
        pkg_path = os.path.join(plugins_dir, name)
        if os.path.isdir(pkg_path) and os.path.isfile(os.path.join(pkg_path, "__init__.py")):
            packages.append(f"plugins.{name}")
    return packages

/path/to/my_plugins/
├── __init__.py             # 含 plugin_list = [MyPlugin, ...]
├── plugin_a.py             # PluginA
└── plugin_b.py             # PluginB

from my_plugins.plugin_a import PluginA
from my_plugins.plugin_b import PluginB

plugin_list = [
    PluginA,
    PluginB,
]

# /path/to/my_standalone_plugin.py
from plugins.base import PluginBase
from common.models import STATUS_CONFIRMED, STATUS_SAFE, STATUS_UNKNOWN, ScanResult
from core.http import join_url
from lib.colors import ok, no


class StandalonePlugin(PluginBase):
    name = "独立插件示例"
    cve = "N/A"
    severity = "medium"
    category = "vuln"
    description = "通过 --plugin-path 加载的单文件插件"
    fix = "修复建议"

    def verify(self, target, session):
        url = join_url(target, "/some/path")
        try:
            resp = session.get(url)
        except Exception as e:
            return ScanResult(
                kind="vuln", name=self.name, status=STATUS_UNKNOWN,
                url=url, evidence=str(e),
            )
        if "marker" in resp.text:
            return ScanResult(
                kind="vuln", name=self.name, severity=self.severity,
                status=STATUS_CONFIRMED, url=url, fix=self.fix,
            )
        return ScanResult(
            kind="vuln", name=self.name, status=STATUS_SAFE, url=url,
        )

# 加载目录 + 单文件
python main.py -u http://target/ -p \
    --plugin-path /path/to/my_plugins \
    --plugin-path /path/to/my_standalone_plugin.py

ruoyi_scan_extra/
├── pyproject.toml
├── README.md
└── ruoyi_scan_extra/
    ├── __init__.py           # 声明 plugin_list
    └── extra_vuln.py        # ExtraVulnPlugin

[build-system]
requires = ["setuptools>=68.0", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "ruoyi-scan-extra"
version = "0.1.0"
description = "Ruoyi-Scan 第三方插件扩展包"
requires-python = ">=3.8"
dependencies = [
    "ruoyi-scan>=1.1.0",  # 依赖主程序
    "requests>=2.28",
]

# 关键：注册到 ruoyi_scan.plugins entry-point 组
[project.entry-points."ruoyi_scan.plugins"]
extra-vuln = "ruoyi_scan_extra:plugin_list"

from ruoyi_scan_extra.extra_vuln import ExtraVulnPlugin

# entry-point 指向 plugin_list 变量
plugin_list = [
    ExtraVulnPlugin,
]

from common.models import STATUS_CONFIRMED, STATUS_SAFE, STATUS_UNKNOWN, ScanResult
from core.http import join_url
from lib.colors import ok, no
from plugins.base import PluginBase


class ExtraVulnPlugin(PluginBase):
    name = "第三方扩展插件示例"
    cve = "CVE-2026-XXXXX"
    severity = "high"
    category = "vuln"
    description = "通过 pip install 自动发现的第三方插件"
    fix = "升级至最新版本"
    cvss_vector = "AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H"
    compliance = "等保2.0:8.1.3;OWASP:A03:2021"

    def verify(self, target, session):
        url = join_url(target, "/some/vulnerable/path")
        try:
            resp = session.get(url)
        except Exception as e:
            return ScanResult(
                kind="vuln", name=self.name, status=STATUS_UNKNOWN,
                url=url, evidence=str(e),
            )
        if "extra-vuln-marker" in resp.text:
            return ScanResult(
                kind="vuln", name=self.name, severity=self.severity,
                status=STATUS_CONFIRMED, url=url, fix=self.fix,
            )
        return ScanResult(
            kind="vuln", name=self.name, status=STATUS_SAFE, url=url,
        )

# 发布到 PyPI（或本地安装）
cd ruoyi_scan_extra
pip install .

# 安装后 ruoyi-scan 自动发现，无需 --plugin-path
python main.py -u http://target/ -p
python main.py --plugin-list   # 应能看到 ExtraVulnPlugin

# 查找所有 ruoyi_scan.plugins 组的 entry-points
eps = entry_points(group="ruoyi_scan.plugins")
for ep in eps:
    plugin_list = ep.load()  # 加载 entry-point 指向的对象
    if isinstance(plugin_list, list):
        result.extend(plugin_list)
    else:
        # 加载的是模块，尝试取 plugin_list 属性
        plugin_list = getattr(plugin_list, "plugin_list", [])
        if isinstance(plugin_list, list):
            result.extend(plugin_list)

class MyPlugin(PluginBase):
    # ...
    vuln_type = "sqli"                  # 必填，供策略匹配
    supports_waf_bypass = True           # 启用绕过
    bypass_max_attempts = 3             # 最大尝试次数（默认 3）

from lib.waf_bypass import BypassContext, BypassSession
from common.models import STATUS_CONFIRMED, STATUS_SAFE, STATUS_UNKNOWN, ScanResult


class SqlInjectPlugin(PluginBase):
    # ... 其他属性
    vuln_type = "sqli"
    supports_waf_bypass = True

    # 原始 payload（供绕过时变形）
    _ORIGINAL_PAYLOAD = "and extractvalue(1,concat(0x7e,(select database()),0x7e))"

    def verify(self, target, session):
        # 正常检测逻辑（不带绕过）
        ...

    def verify_with_bypass(self, target, bypass_session, bypass_ctx):
        """WAF 绕过验证

        Args:
            target: 目标 URL
            bypass_session: BypassSession 实例（已应用传输层变换）
            bypass_ctx: BypassContext（含 waf_type/vuln_type/strategy 等）
        """
        # 1. 用当前策略变形 payload
        strategy = bypass_ctx.strategy
        if strategy is not None:
            tampered_payload = strategy.tamper_payload(self._ORIGINAL_PAYLOAD, bypass_ctx)
        else:
            tampered_payload = self._ORIGINAL_PAYLOAD

        # 2. 用变形后的 payload 发请求（bypass_session 已应用传输层变换）
        url = join_url(target, "/vulnerable/endpoint")
        data = {"param": tampered_payload}

        try:
            resp = bypass_session.post(url, data=data)
        except Exception as e:
            # 铁律：绕过异常 → UNKNOWN（绝不判 SAFE）
            return ScanResult(
                kind="vuln", name=self.name, status=STATUS_UNKNOWN,
                url=url, evidence=f"绕过异常: {e}",
            )

        # 3. 判定（与 verify() 一致的三态）
        if "XPATH syntax error" in resp.text or "database()" in resp.text:
            return ScanResult(
                kind="vuln", name=self.name, severity=self.severity,
                status=STATUS_CONFIRMED, url=url,
                evidence=f"绕过成功（策略={bypass_ctx.strategy.strategy_id}）",
                fix=self.fix,
            )

        # 4. 绕过未成功，返回 SAFE 或 UNKNOWN（不能 CONFIRMED）
        #    引擎会自动尝试下一个策略
        return ScanResult(
            kind="vuln", name=self.name, status=STATUS_SAFE,
            url=url, evidence="绕过后仍未命中特征",
        )

{
  "waf_bypass": {
    "strategy_used": "BP-GEN-1",
    "strategy_name": "大小写混淆",
    "layer": "L1",
    "attempt": 1,
    "waf_type": "cloudflare"
  }
}

cvss_vector = "AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H"

compliance = "等保2.0:8.1.4;OWASP:A01:2021"

{
    "等保2.0": "8.1.4",
    "OWASP":   "A01:2021",
}

from plugins.base import cvss_score

# 完整向量
cvss_score("AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H")
# → 9.8（Critical）

# 文件读取（仅机密性影响）
cvss_score("AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N")
# → 7.5（High）

# SQL 报错注入（需要低权限）
cvss_score("AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:N/A:N")
# → 6.5（Medium）

python main.py --plugin-check plugins/ruoyi/my_plugin.py

[*]验证插件: plugins/ruoyi/my_plugin.py
────────────────────────────────────────────────
静态检查:
  ✓ 通过
导入检查:
  ✓ 通过
────────────────────────────────────────────────
[+]插件验证通过

# tests/test_my_plugin.py
import os
import sys

import requests_mock

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from common.models import STATUS_CONFIRMED, STATUS_SAFE
from core.session import SessionManager
from plugins.ruoyi.my_plugin import MyPluginPlugin


def test_my_plugin_confirmed():
    """命中漏洞特征 → CONFIRMED"""
    plugin = MyPluginPlugin()
    session = SessionManager(timeout=5)

    with requests_mock.Mocker() as m:
        m.get(
            "http://target/demo/test",
            text="this is a demo-test-marker page",
            status_code=200,
        )
        result = plugin.verify("http://target/", session)

    assert result.status == STATUS_CONFIRMED
    assert "demo-test-marker" in result.evidence
    session.close()


def test_my_plugin_safe():
    """正常页面 → SAFE"""
    plugin = MyPluginPlugin()
    session = SessionManager(timeout=5)

    with requests_mock.Mocker() as m:
        m.get(
            "http://target/demo/test",
            text="normal page without any marker",
            status_code=200,
        )
        result = plugin.verify("http://target/", session)

    assert result.status == STATUS_SAFE
    session.close()


def test_my_plugin_network_error():
    """网络异常 → UNKNOWN（铁律）"""
    from common.models import STATUS_UNKNOWN

    plugin = MyPluginPlugin()
    session = SessionManager(timeout=5)

    with requests_mock.Mocker() as m:
        m.get(
            "http://target/demo/test",
            exc=requests_mock.exceptions.ConnectTimeout,
        )
        result = plugin.verify("http://target/", session)

    assert result.status == STATUS_UNKNOWN
    session.close()

pip install requests-mock
pytest tests/test_my_plugin.py -v

# tests/test_my_plugin_e2e.py
import os
import sys

import pytest

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from common.models import STATUS_CONFIRMED
from core.session import SessionManager
from plugins.ruoyi.my_plugin import MyPluginPlugin


@pytest.mark.skipif(
    not os.environ.get("RUN_E2E_TESTS"),
    reason="需启动签名靶场并设置 RUN_E2E_TESTS=1",
)
def test_my_plugin_e2e():
    """端到端：对真实签名靶场执行扫描"""
    plugin = MyPluginPlugin()
    session = SessionManager(timeout=10)

    # 假设靶场已启动在 http://127.0.0.1:5000/
    result = plugin.verify("http://127.0.0.1:5000/", session)

    assert result.status == STATUS_CONFIRMED
    session.close()

# 1. 启动签名靶场
cd lab && python app.py &

# 2. 运行端到端测试
RUN_E2E_TESTS=1 pytest tests/test_my_plugin_e2e.py -v

def test_all_bypass_plugins_support_waf_bypass():
    """所有支持绕过的插件 supports_waf_bypass=True"""
    for plugin_cls, _ in _BYPASS_PLUGINS:
        assert plugin_cls.supports_waf_bypass is True, \
            f'{plugin_cls.__name__}.supports_waf_bypass 应为 True'

# plugins/ruoyi/sql_inject_demo.py
# SQL 报错注入（demo）：完整插件示例，含 WAF 绕过、CVSS、合规、修复详情
from common.models import STATUS_CONFIRMED, STATUS_SAFE, STATUS_UNKNOWN, ScanResult
from core.http import host_of, join_url
from lib.colors import no, ok
from lib.matcher import match_sql_error
from plugins.base import PluginBase


class SqlInjectDemoPlugin(PluginBase):
    """SQL 报错注入检测插件（完整示例）

    演示内容：
    - 所有类属性（name/cve/severity/category/description/fix/fix_detail/reproduce）
    - CVSS v3.1 向量与合规映射
    - WAF 绕过支持（vuln_type/supports_waf_bypass/bypass_max_attempts）
    - verify() 三态判定实现
    - verify_with_bypass() payload 变形实现
    """

    # ===== 基础元信息 =====
    name = "SQL报错注入（demo）"
    cve = "CVE-2026-XXXXX"
    severity = "high"
    category = "vuln"
    description = (
        "/demo/sqlInject 端点的 username 参数拼接 SQL，"
        "可通过 extractvalue 报错注入泄露 database()、version() 等敏感信息"
    )
    fix = "对 username 参数做白名单校验，使用参数化查询，禁止拼接 SQL"

    # ===== 修复详情（多行字符串） =====
    fix_detail = (
        "【升级方案】升级至 demo-app 2.0.0+（该版本已修复 SQL 注入）\n"
        "【代码修复】修改 DemoMapper.xml，对 username 参数做白名单校验：\n"
        "  - 修改前：SELECT * FROM user WHERE username = '${username}'\n"
        "  - 修改后：SELECT * FROM user WHERE username = #{username}（参数化）\n"
        "【配置加固】启用 MyBatis 参数化：\n"
        "  mybatis.configuration.safe-result-handler-enabled: true\n"
        "【WAF 规则】拦截包含 extractvalue/updatexml/concat 的 username 参数\n"
        "【合规】OWASP A03:2021 注入；等保 2.0 8.1.3 输入校验"
    )

    # ===== 复现命令（可直接复制执行） =====
    reproduce = (
        'curl -X POST "http://target/demo/sqlInject" \\\n'
        '  -H "Content-Type: application/x-www-form-urlencoded" \\\n'
        '  -H "Accept: application/json" \\\n'
        "  -d 'username=admin%27+and+extractvalue(1,concat(0x7e,(select+database()),0x7e))--' \\\n"
        '  --cookie ""\n'
        "\n"
        '# 预期响应：HTTP 500 + 响应体含 "XPATH syntax error" 或 "database()" 报错特征'
    )

    # ===== 影响版本 =====
    affected_versions = ">=1.0,<2.0"

    # ===== CVSS v3.1 + 合规映射 =====
    cvss_vector = "AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N"
    compliance = "等保2.0:8.1.3;OWASP:A03:2021"
    # CVSS Base Score 自动计算为 7.5（High）

    # ===== WAF 绕过支持 =====
    vuln_type = "sqli"
    supports_waf_bypass = True
    bypass_max_attempts = 3

    # 原始 payload（供 verify_with_bypass 变形）
    _ORIGINAL_PAYLOAD = (
        "admin' and extractvalue(1,concat(0x7e,(select database()),0x7e))--"
    )

    def verify(self, target, session):
        """执行 SQL 报错注入检测

        Args:
            target: 目标 URL（已归一化，以 / 结尾）
            session: SessionManager 实例

        Returns:
            ScanResult（CONFIRMED/SAFE/UNKNOWN 三态判定）
        """
        host = host_of(target)
        headers = {
            "Host": host,
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:98.0) Gecko/20100101 Firefox/98.0",
            "Accept": "application/json, text/javascript, */*; q=0.01",
            "Content-Type": "application/x-www-form-urlencoded",
            "X-Requested-With": "XMLHttpRequest",
            "Origin": f"http://{host}",
            "Referer": f"http://{host}/demo",
            "Connection": "close",
        }
        data = {"username": self._ORIGINAL_PAYLOAD}

        url = join_url(target, "/demo/sqlInject")
        try:
            resp = session.post(url, headers=headers, data=data, timeout=10)
        except Exception as e:
            # 铁律：网络异常 → UNKNOWN
            print(no(f"{self.name}（网络异常）"))
            return ScanResult(
                kind="vuln", name=self.name, status=STATUS_UNKNOWN,
                url=url, evidence=str(e),
            )

        text = resp.text or ""

        # 判定：SQL 报错特征（用 lib/matcher.py 统一降误报工具）
        if match_sql_error(text) or "database()" in text:
            print(ok(f"存在{self.name}"))
            return ScanResult(
                kind="vuln", name=self.name, severity=self.severity,
                status=STATUS_CONFIRMED, url=url,
                evidence="响应含 SQL 报错特征（XPATH syntax error / database()）",
                fix=self.fix, fix_detail=self.fix_detail,
                reproduce=self.reproduce, cve=self.cve,
                cvss_vector=self.cvss_vector,
                extra={
                    "vuln_type": "sqli",
                    "payload_class": "extractvalue_database",
                    "plugin_name": "sql_inject_demo",
                },
            )

        # 明确反证：正常业务响应
        print(no(f"不存在{self.name}"))
        return ScanResult(
            kind="vuln", name=self.name, status=STATUS_SAFE,
            url=url, evidence="响应不含 SQL 报错特征",
        )

    def verify_with_bypass(self, target, bypass_session, bypass_ctx):
        """WAF 绕过验证：用当前策略变形 payload 后重试

        Args:
            target: 目标 URL
            bypass_session: BypassSession 实例（已应用传输层变换）
            bypass_ctx: BypassContext（含 strategy/waf_type/attempt 等）

        Returns:
            ScanResult（三态判定，与 verify() 一致）
        """
        # 1. 用当前策略变形 payload
        strategy = bypass_ctx.strategy
        if strategy is not None:
            tampered_payload = strategy.tamper_payload(
                self._ORIGINAL_PAYLOAD, bypass_ctx
            )
        else:
            tampered_payload = self._ORIGINAL_PAYLOAD

        # 2. 构造请求（bypass_session 已应用 headers/chunked/origin IP 变换）
        url = join_url(target, "/demo/sqlInject")
        headers = {
            "Content-Type": "application/x-www-form-urlencoded",
            "Accept": "application/json",
        }
        data = {"username": tampered_payload}

        try:
            resp = bypass_session.post(url, headers=headers, data=data, timeout=10)
        except Exception as e:
            # 铁律：绕过异常 → UNKNOWN（绝不判 SAFE）
            return ScanResult(
                kind="vuln", name=self.name, status=STATUS_UNKNOWN,
                url=url, evidence=f"绕过异常（策略={strategy.strategy_id if strategy else 'N/A'}）: {e}",
            )

        text = resp.text or ""

        # 3. 判定（与 verify() 一致）
        if match_sql_error(text) or "database()" in text:
            return ScanResult(
                kind="vuln", name=self.name, severity=self.severity,
                status=STATUS_CONFIRMED, url=url,
                evidence=(
                    f"绕过成功（策略={strategy.strategy_id if strategy else 'N/A'}, "
                    f"layer={strategy.layer if strategy else 'N/A'}, "
                    f"attempt={bypass_ctx.attempt}）"
                ),
                fix=self.fix, fix_detail=self.fix_detail,
                reproduce=self.reproduce, cve=self.cve,
                cvss_vector=self.cvss_vector,
                extra={
                    "vuln_type": "sqli",
                    "payload_class": "extractvalue_database_bypassed",
                    "plugin_name": "sql_inject_demo",
                },
            )

        # 4. 绕过未命中，返回 SAFE（引擎会自动尝试下一策略）
        return ScanResult(
            kind="vuln", name=self.name, status=STATUS_SAFE,
            url=url,
            evidence=f"绕过后仍未命中特征（策略={strategy.strategy_id if strategy else 'N/A'}）",
        )

# 1. 注册到 plugin_list
# 编辑 plugins/ruoyi/__init__.py，添加：
#   from plugins.ruoyi.sql_inject_demo import SqlInjectDemoPlugin
#   plugin_list = [..., SqlInjectDemoPlugin]

# 2. 验证插件完整性
python main.py --plugin-check plugins/ruoyi/sql_inject_demo.py

# 3. 列出所有插件（确认已加载）
python main.py --plugin-list

# 4. 扫描
python main.py -u http://target/ -p

# 5. 启用 WAF 绕过扫描
python main.py -u http://target/ -p --bypass-waf auto