ZapcoMan's world
python

从零打造漏洞扫描器：端口扫描 + Web 检测 + 报告生成

2026-07-06 #渗透测试#信息收集#python#端口扫描

安全从业者的工具箱里，漏洞扫描器是出场率最高的家伙。市面上的成品要么贵（Nessus），要么重（OpenVAS），要么只做一件事（Nmap）。这篇带你从零搭一个能打的全功能扫描器——端口扫描、目录爆破、SQL 注入/XSS/命令注入检测、子域名枚举、HTML 可视化报告，一个不少。

最终成品

一个 Python CLI 工具，输入一个域名或 IP，自动跑完四条扫描管线，生成带风险可视化图表的 HTML 报告。

1
2
3
4
5
6
7
8
# 单目标全量扫描
python main.py --target example.com

# 批量扫描 + 自定义输出格式
python main.py --targets targets.txt --format html,json,excel

# 只做端口 + Web 漏洞扫描
python main.py --target example.com --scan port,vulnerability

完整代码：CyberThreatHunter（基于 J09715/Vulnerability-Scanning-tool 二次开发）

一、架构设计：流水线 + 并发调度

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
main.py (CLI入口)
    │
    ▼
VulnerabilityScannerMain (核心引擎)
    │
    ├─► PortScanner        ──► TCP/UDP端口扫描 + 服务识别
    ├─► DirectoryBuster    ──► Web目录爆破 + 敏感文件检查
    ├─► VulnerabilityScanner ──► SQL注入 + XSS + 命令注入 + 安全头
    ├─► SubdomainEnumerator ──► DNS子域名枚举 + 安全记录检查
    │
    ▼
OutputReporter (报告引擎)
    ├─► HTML (交互式风险看板)
    ├─► JSON (API集成)
    ├─► CSV (数据分析)
    └─► Excel (两个工作表：详情 + 摘要)

核心设计决策：

  1. 目标级并发 + 管线内顺序执行：多个目标用 ThreadPoolExecutor 并行跑，但每个目标内部四个模块按顺序执行（端口 → 目录 → 漏洞 → 子域名）。因为漏洞扫描需要先知道开放端口，目录爆破需要先确认 Web 服务可达。
  2. 统一的发现项数据模型：所有子模块产出的都是一个 List[Dict]，每个 dict 有固定的字段结构（type/name/severity/cvss_score/evidence/remediation）。统一的输出格式让报告生成器不用关心数据来源。

二、统一的数据模型：ScanResult

1
2
3
4
5
6
7
8
9
10
11
12
13
from dataclasses import dataclass, field
from typing import List, Dict, Any

@dataclass
class ScanResult:
    target: str
    scan_type: str
    start_time: str
    end_time: str = ""
    findings: List[Dict[str, Any]] = field(default_factory=list)
    status: str = "pending"    # pending → running → completed → failed
    errors: List[str] = field(default_factory=list)
    scan_config: Dict[str, Any] = field(default_factory=dict)

每个 finding 的字段契约：

1
2
3
4
5
6
7
8
9
10
11
12
13
{
    "type": "sql_injection",        # 漏洞类型标识
    "name": "SQL注入 - 目标参数",    # 人类可读名称
    "description": "...",           # 详细描述
    "severity": "high",             # critical/high/medium/low/info
    "cvss_score": 8.5,             # CVSS v3 评分
    "evidence": "原始请求/响应证据",  # 具体证据
    "remediation": "修复建议",       # 可操作的修复步骤
    "details": {                    # 额外上下文
        "target": "...",
        "timestamp": "..."
    }
}

为什么选 dataclass 而不是 Pydantic：这个项目的场景是内部数据传递，不需要运行时验证和序列化。dataclass 零依赖、性能更好、代码更简洁。

三、核心引擎：VulnerabilityScannerMain

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class VulnerabilityScannerMain:
    def __init__(self, config_path=None):
        self.config = self._load_config(config_path)
        self._setup_logging()
        self._init_scanners()

    def _init_scanners(self):
        self.scanners = {
            'port': PortScanner(self.config),
            'directory': DirectoryBuster(self.config),
            'vulnerability': VulnerabilityScanner(self.config),
            'subdomain': SubdomainEnumerator(self.config),
        }

    def scan(self, target, scan_types=None):
        """单目标扫描：四个模块顺序执行"""
        scan_types = scan_types or list(self.scanners.keys())
        result = ScanResult(target=target, scan_type=','.join(scan_types),
                            start_time=datetime.now().isoformat())

        try:
            for st in scan_types:
                findings = self.scanners[st].scan(target)
                result.findings.extend(findings)
            result.status = "completed"
        except Exception as e:
            result.status = "failed"
            result.errors.append(str(e))

        result.end_time = datetime.now().isoformat()
        return result

    def scan_multiple(self, targets, scan_types=None, max_workers=50):
        """多目标并发扫描"""
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {
                executor.submit(self.scan, t, scan_types): t
                for t in targets
            }
            results = []
            for future in as_completed(futures, timeout=300):
                try:
                    results.append(future.result())
                except Exception as e:
                    ...
            return results

三个关键设计：

  1. 顺序管线：端口 → 目录 → 漏洞 → 子域名。这保证了依赖关系——如果 80 端口没开放，目录爆破和漏洞扫描自然跳过。
  2. 异常隔离：单个目标扫描失败不影响其他目标，错误被收集到 ScanResult.errors 中。
  3. 配置驱动：所有参数（超时、线程数、字典路径、端口范围）都在 config.yaml 中，不需要改代码。

四、模块一：端口扫描

端口扫描是信息收集的第一步。这个模块实现了双引擎策略：

1
2
3
try nmap (SYN 半开扫描) → 成功 → 返回
                        → 失败 → fallback socket (CONNECT 全连接)
                              → 可选 UDP 扫描

4.1 Nmap 引擎（优先）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import nmap

def _nmap_scan_tcp(self, target, ports):
    nm = nmap.PortScanner()
    nm.scan(target, ports, arguments='-sS -T4 --min-rate 1000')
    # -sS: TCP SYN 半开扫描（不完成三次握手）
    # -T4: 激进的时间模板
    # --min-rate 1000: 每秒最少发 1000 个包

    results = []
    for port in nm[target]['tcp']:
        if nm[target]['tcp'][port]['state'] == 'open':
            service = nm[target]['tcp'][port].get('name', 'unknown')
            results.append({'port': port, 'service': service, 'protocol': 'tcp'})
    return results

4.2 Socket 回退引擎

1
2
3
4
5
6
7
8
import socket

def _socket_scan_tcp(self, target, port, timeout=2):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(timeout)
    result = sock.connect_ex((target, port))
    sock.close()
    return result == 0  # 0 表示连接成功

4.3 Banner 抓取 + 服务识别

1
2
开放端口 → 发送 GET / HTTP/1.0 → 正则匹配响应头
         → 端口映射表兜底 (22→SSH, 3306→MySQL, 6379→Redis...)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def _grab_banner(self, target, port, timeout=3):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(timeout)
    sock.connect((target, port))

    if port == 80 or port == 8080:
        sock.send(b"GET / HTTP/1.0\r\n\r\n")
        response = sock.recv(1024).decode('utf-8', errors='ignore')

        if 'Apache' in response: return 'Apache'
        if 'nginx' in response: return 'Nginx'
        if 'IIS' in response: return 'IIS'
        # ...

    sock.close()
    return 'unknown'

4.4 端口风险评估

每个端口预设了风险等级，不是所有开放端口都需要紧张：

1
2
3
4
5
6
7
8
9
port_risks = {
    23: 'high',      # Telnet - 明文传输
    445: 'high',     # SMB - 永恒之蓝
    3389: 'high',    # RDP - 暴力破解重灾区
    21: 'medium',    # FTP - 可能匿名登录
    22: 'medium',    # SSH - 暴力破解入口
    80: 'low',       # HTTP - 正常业务
    443: 'low',      # HTTPS - 正常业务
}

五、模块二：Web 目录爆破

5.1 字典 + 扩展名变换

1
2
3
4
5
6
7
8
9
10
11
# 基础路径: admin, backup, config, .git, .env, phpinfo.php...
base_paths = load_wordlist('config/wordlists/directories.txt')  # 200条

# 扩展名变换
extensions = ['.php', '.html', '.asp', '.aspx', '.jsp',
              '.txt', '.xml', '.json', '.bak', '.old', '.backup']

# admin → admin.php, admin.html, admin.asp, admin.aspx...
for path in base_paths:
    for ext in extensions:
        test_url = f"{target}/{path}{ext}"

5.2 并发请求 + 指纹伪装

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from concurrent.futures import ThreadPoolExecutor
import random

user_agents = [  # config.yaml 中 70+ 条
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) ...',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) ...',
    'Mozilla/5.0 (X11; Linux x86_64) ...',
    # ...
]

def _check_url(self, session, url):
    headers = {'User-Agent': random.choice(user_agents)}
    resp = session.get(url, headers=headers, allow_redirects=False, timeout=10)

    if resp.status_code in [200, 301, 302, 403]:
        severity = self._assess_risk(url, resp.status_code)
        return {'url': url, 'status': resp.status_code, 'severity': severity}
    return None

with ThreadPoolExecutor(max_workers=50) as executor:
    futures = [executor.submit(self._check_url, session, url) for url in all_urls]

5.3 敏感路径自动分级

1
2
3
4
5
6
7
8
sensitive_paths = {
    '.git': 'critical',    # 源码泄露
    '.env': 'critical',    # 环境变量泄露
    'config': 'critical',  # 配置文件
    'admin': 'high',       # 管理后台
    'backup': 'high',      # 备份文件
    'phpinfo.php': 'high', # PHP 信息泄露
}

5.4 HTTP 方法检查

除了目录存在性，还检查目标支持哪些 HTTP 方法——PUT 和 DELETE 如果开放而无认证，危害巨大：

1
2
3
4
5
methods = ['GET', 'POST', 'PUT', 'DELETE', 'OPTIONS', 'HEAD', 'PATCH', 'TRACE']
for method in methods:
    resp = session.request(method, target)
    if resp.status_code != 405:  # 非 Method Not Allowed
        findings.append(f"HTTP方法 {method} 允许")

六、模块三：Web 漏洞扫描（核心）

这是整个扫描器里最有技术含量的部分。实现了三种注入检测。

6.1 页面爬取

先爬取目标页面，提取所有链接和表单参数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import re
from bs4 import BeautifulSoup

def _extract_links(self, html):
    soup = BeautifulSoup(html, 'html.parser')
    links = []

    # 提取 <a href="...">
    for a in soup.find_all('a', href=True):
        links.append({'url': a['href'], 'method': 'GET', 'params': []})

    # 提取 <form action="..." method="...">
    for form in soup.find_all('form'):
        action = form.get('action', '')
        method = form.get('method', 'GET').upper()
        # 提取所有 <input name="...">
        params = [inp.get('name') for inp in form.find_all('input') if inp.get('name')]
        links.append({'url': action, 'method': method, 'params': params})

    return links

6.2 SQL 注入检测：基于错误回显

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
sql_payloads = [
    "'",                               # 基础闭合
    "1' OR '1'='1",                    # 永真注入
    "1' AND sleep(5)--",              # 时间盲注
    # ...共 13 种 payload
]

error_patterns = {
    'MySQL': [
        r"SQL syntax.*MySQL",
        r"mysql_fetch",
        r"You have an error in your SQL syntax",
    ],
    'PostgreSQL': [
        r"PostgreSQL.*ERROR",
        r"pg_query\(",
    ],
    'SQLite': [
        r"SQLite/JDBCDriver",
        r"sqlite3\.OperationalError",
    ],
    'Oracle': [
        r"ORA-[0-9]{5}",
        r"Oracle error",
    ],
    'MSSQL': [
        r"Microsoft SQL Server",
        r"\[SQL Server\]",
    ],
}

检测逻辑：对每个 GET/POST 参数注入 payload，检查响应中是否匹配上述 20 个错误特征正则。命中任何一个即判定存在 SQL 注入。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
def _test_sql_injection_param(self, url, param, method='GET'):
    for payload in sql_payloads:
        test_params = {param: payload}
        if method == 'GET':
            resp = requests.get(url, params=test_params)
        else:
            resp = requests.post(url, data=test_params)

        for db, patterns in error_patterns.items():
            for pattern in patterns:
                if re.search(pattern, resp.text, re.IGNORECASE):
                    return {
                        'type': 'sql_injection',
                        'database': db,
                        'payload': payload,
                        'param': param,
                        'severity': 'critical',
                        'cvss_score': 9.0,
                    }
    return None

6.3 XSS 检测：反射验证法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
xss_payloads = [
    "<script>alert('XSS')</script>",           # 经典
    "<img src=x onerror=alert(1)>",            # 事件注入
    "\" onmouseover=\"alert(1)",               # 属性分离
    "<svg onload=alert(1)>",                   # SVG 注入
    "javascript:alert(1)",                     # 协议注入
]

def _test_xss(self, url, param):
    for payload in xss_payloads:
        resp = requests.get(url, params={param: payload})
        # 核心检测：响应中完整包含 payload
        if payload in resp.text:
            return {
                'type': 'xss',
                'payload': payload,
                'severity': 'medium',
                'cvss_score': 6.1,
            }
    return None

为什么用”完整包含”而不是更复杂的 DOM 分析：反射型 XSS 的核心特征是 payload 原样出现在响应中。这个简单方法误报率低、速度快、不需要浏览器渲染。

6.4 命令注入检测

1
2
3
4
5
6
7
8
9
10
11
12
13
cmd_payloads = [
    ";ls", "|ls", "|| ls", "&& ls",          # 分隔符注入
    "`ls`", "$(ls)",                          # 命令替换
    ";cat /etc/passwd", "`cat /etc/passwd`",  # 文件读取
]

indicators = [
    r"root:.*:0:0:",              # /etc/passwd 内容特征
    r"bin/.*sh",                  # Shell 路径
    r"Permission denied",         # 执行成功但无权限
    r"command not found",         # 已尝试执行命令
    r"uid=\d+\(.*\) gid=\d+",     # id 命令输出
]

七、模块四：DNS 子域名枚举

7.1 字典枚举（第一层）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import dns.resolver

resolver = dns.resolver.Resolver()
resolver.nameservers = ['8.8.8.8', '1.1.1.1']  # 可配置

def enumerate_subdomains(self, domain, wordlist):
    results = []
    for prefix in wordlist:
        subdomain = f"{prefix}.{domain}"
        try:
            answers = resolver.resolve(subdomain, 'A')
            for answer in answers:
                results.append({
                    'subdomain': subdomain,
                    'ip': str(answer),
                    'type': 'A',
                })
        except dns.resolver.NXDOMAIN:
            continue
    return results

7.2 通配符 DNS 检测

有些域名配置了通配符 DNS（*.example.com），会导致所有不存在的子域名都解析成功。用随机子域名测试：

1
2
3
4
5
6
7
8
9
import time

def _detect_wildcard(self, domain):
    test_subdomain = f"random{int(time.time())}.{domain}"
    try:
        resolver.resolve(test_subdomain, 'A')
        return True  # 通配符 DNS 已启用
    except dns.resolver.NXDOMAIN:
        return False

7.3 安全记录检查

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# SPF 记录（防邮件伪造）
try:
    answers = resolver.resolve(domain, 'TXT')
    for a in answers:
        if 'v=spf1' in str(a):
            print(f"[+] SPF 记录: {a}")
except:
    print("[-] 未配置 SPF 记录")

# DMARC 记录（邮件认证策略）
try:
    answers = resolver.resolve(f"_dmarc.{domain}", 'TXT')
    for a in answers:
        if 'v=DMARC1' in str(a):
            print(f"[+] DMARC 记录: {a}")
except:
    print("[-] 未配置 DMARC 记录")

八、报告引擎：让扫描结果可读

扫描结果再多，老板/客户看不懂也白搭。HTML 报告是默认输出，包含：

8.1 统计卡片

1
2
3
4
┌──────────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
│  总发现数    │ │ 严重: 3  │ │ 高危: 5  │ │ 中危: 12 │
│     25       │ │  ████    │ │  ██████  │ │  ███████ │
└──────────────┘ └──────────┘ └──────────┘ └──────────┘

8.2 可折叠详情面板

每个发现项默认折叠，点击展开显示 CVSS 评分、原始证据、修复建议和 JSON 数据：

1
2
3
4
5
6
7
8
9
10
11
12
13
<div class="finding" onclick="toggleDetails(this)">
    <span class="severity-badge critical">严重</span>
    <span>SQL注入 - id参数</span>
    <span>CVSS: 9.0</span>
</div>
<div class="details" style="display:none">
    <h4>描述</h4>
    <p>在 /login.php?id= 参数发现 SQL 注入漏洞</p>
    <h4>证据</h4>
    <pre>MySQL error: You have an error in your SQL syntax...</pre>
    <h4>修复建议</h4>
    <p>使用参数化查询，对所有用户输入进行过滤和转义</p>
</div>

8.3 风险分布可视化

1
2
3
4
5
6
7
8
9
10
11
.risk-bar {
    display: flex;
    height: 24px;
    border-radius: 12px;
    overflow: hidden;
}
.risk-critical { background: #dc3545; width: 12%; }   /* 红色  */
.risk-high     { background: #fd7e14; width: 20%; }   /* 橙色  */
.risk-medium   { background: #ffc107; width: 48%; }   /* 黄色  */
.risk-low      { background: #28a745; width: 16%; }   /* 绿色  */
.risk-info     { background: #17a2b8; width: 4%; }    /* 蓝色  */

8.4 Excel 双工作表

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import pandas as pd

# 工作表 1：漏洞详情
df_details = pd.DataFrame(findings)
df_details.columns = ['ID', '漏洞类型', '漏洞名称', '风险等级',
                       'CVSS评分', '描述', '证据', '修复建议',
                       '发现时间', '目标']
# 工作表 2：扫描摘要
df_summary = pd.DataFrame({
    '统计项': ['总发现数', '严重', '高危', '中危', '低危'],
    '数量': [25, 3, 5, 12, 5],
})
with pd.ExcelWriter('report.xlsx', engine='openpyxl') as writer:
    df_details.to_excel(writer, sheet_name='漏洞详情', index=False)
    df_summary.to_excel(writer, sheet_name='扫描摘要', index=False)

九、配置系统设计

200+ 行的 config.yaml 集中管理所有参数，改了配置不需要动代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
scanning:
  timeout: 10           # 单次请求超时
  threads: 50           # 默认并发数
  retries: 2            # 失败重试
  delay: 0.1            # 请求间延迟

network:
  default_ports: "22,23,25,53,80,110,135,139,143,443,445,993,995,
                   1433,1521,3306,3389,5432,5900,6379,8080,8443"
  tcp_scan_timeout: 2   # TCP 连接超时
  tcp_scan_threads: 100 # TCP 扫描并发
  udp_scan_threads: 50  # UDP 扫描并发（比 TCP 慢）

web:
  extensions: [php, html, asp, aspx, jsp, txt, xml, json, bak, old, backup]
  check_methods: [GET, POST, PUT, DELETE, OPTIONS, HEAD, PATCH, TRACE]

dns:
  servers: [8.8.8.8, 8.8.4.4, 1.1.1.1, 9.9.9.9]
  record_types: [A, AAAA, CNAME, MX, NS, SOA, TXT, SRV]

output:
  default_format: html
  report_dir: reports

十、这个项目中值得复用的设计模式

10.1 策略模式——双引擎端口扫描

nmap 和 socket 两个引擎实现了同一接口。调用方不关心用的是哪个引擎：

1
2
3
4
5
def scan(self, target):
    try:
        return self._nmap_scan(target)     # 主引擎
    except NmapNotAvailable:
        return self._socket_scan(target)   # 备用引擎

10.2 管线模式——模块串联

四个模块串联成一条流水线，数据单向流动（端口结果 → 目录扫描 → 漏洞测试 → 子域名枚举），每个模块的输出是下一个模块的上下文。

10.3 工厂模式——扫描器注册

1
2
3
4
5
6
self.scanners = {
    'port': PortScanner(config),
    'directory': DirectoryBuster(config),
    'vulnerability': VulnerabilityScanner(config),
    'subdomain': SubdomainEnumerator(config),
}

要加新模块？在 _init_scanners 里加一行，在 scan_types 列表里加个名字就行。

十一、总结

这个扫描器从零写下来大概 2000+ 行 Python，核心逻辑其实不复杂，但架构设计决定了可维护性和可扩展性：

  • 统一数据模型 → 报告引擎零耦合
  • 配置驱动 → 改参数不碰代码
  • 双引擎策略 → nmap 不可用时自动降级
  • 异常隔离 → 单目标失败不影响批量

法律声明：此工具仅限授权安全测试使用。未经授权扫描他人系统可能违反《网络安全法》。

评论
分享

相关文章

查看更多
自动化资产信息收集：CNVD 漏洞报送辅助工具 2026-07-06
WAF 指纹识别原理与实现：从 sqlmap 源码学起 2026-07-06
图像预处理 + 二维码识别全攻略：9 种方法让模糊二维码无处遁形 2026-07-06
Vue3 + Flask 全栈实战：从零构建在线二维码解析器 2026-07-06