NOCAU

网络攻防与白帽渗透基础

白帽视角的渗透测试方法论：授权前提、信息收集、端口扫描、漏洞验证与修复闭环。

场景：安全测试 技术栈：Python / nmap / OWASP 来源：OWASP / NIST / nmap / Python 官方文档
合规声明

本页面内容仅用于安全学习与合法授权测试（如 CTF、自建靶场、获得授权的渗透测试项目）。严禁用于未授权的系统入侵、数据窃取或任何违法活动。使用者须自行承担一切法律责任。

项目简介

渗透测试（Penetration Testing）是 “在授权范围内模拟攻击以发现漏洞” 的工程方法。流程：信息收集 → 资产测绘 → 漏洞扫描 → 利用验证 → 报告与修复。所有步骤都必须有书面授权，否则即为违法。

代码示例：端口扫描（合规环境内）

py
import socket

def scan_port(host: str, port: int, timeout: float = 0.8) -> bool:
    """仅用于你拥有授权的目标。对互联网随机扫描是违法的。"""
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.settimeout(timeout)
            return s.connect_ex((host, port)) == 0
    except OSError:
        return False

if __name__ == "__main__":
    target = "127.0.0.1"   # 仅演示，必须是你的授权目标
    for port in [21, 22, 80, 443, 3306, 8080]:
        if scan_port(target, port):
            print(f"{port} open")

代码示例：信息收集与防御闭环

py
import requests

def recon(url: str):
    r = requests.get(url, timeout=8)
    print("Server:", r.headers.get("Server"))
    print("Security:", r.headers.get("Content-Security-Policy", "未配置"))
    robots = requests.get(url.rstrip("/") + "/robots.txt", timeout=8)
    if robots.status_code == 200:
        print(robots.text[:500])

FIXES = [
    "启用 CSP、HSTS、X-Frame-Options 响应头",
    "关闭非必要端口与默认口令",
    "及时升级依赖，消除已知 CVE",
    "对敏感接口做鉴权、限流与审计日志",
]
# recon("https://your-authorized-target.com")

操作步骤

  1. 准备授权书与测试范围文档，明确目标、时间窗口与允许的手段。
  2. 用 nmap 做资产测绘：nmap -sV -p- 授权目标，再按开放端口缩小攻击面。
  3. 漏洞验证必须在隔离靶场（DVWA / Metasploitable / Vulfocus）完成。
  4. 输出报告：漏洞描述、复现步骤、风险等级、修复建议，推动闭环。

完整可运行工程

可直接运行Python 标准库，零依赖仅限授权目标

运行环境：Python 3.8+（Windows / macOS / Linux 均可）。把下面 4 个文件保存到同一目录后直接按 README 的命令运行；两个 --selftest 完全离线，只在本地起临时服务，可放心先跑一遍确认环境正常。扫描任何非本机目标前，必须先取得书面授权。。
依赖清单：仅 Python 标准库（socket / concurrent.futures / urllib / ssl / http.server / unittest），无第三方依赖，无需 pip install。

security-pentest/ ├── port_scan.py 授权范围内的 TCP 端口扫描（默认只允许本机） ├── http_audit.py HTTP 响应头安全配置体检（CSP / HSTS / 版本泄露等） ├── selftest.py 7 个 unittest 用例，验证两项工具的判定逻辑 └── README.md 运行说明、依赖清单、合规边界

port_scan.py

并发 socket 探测 + 授权前置校验（非本机目标未声明授权直接退出），自带 --selftest。

py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""授权范围内的 TCP 端口连通性检查（仅标准库，无第三方依赖）。

合规边界（务必先读）：
  1. 默认只允许扫描本机 127.0.0.1 / localhost / ::1；
  2. 扫描其他目标，必须由你本人持有目标所有者的书面授权，并显式追加
     --i-have-written-authorization，否则脚本直接退出；
  3. 严禁对互联网随机目标扫描，未授权扫描在多数司法辖区属违法行为。

用法：
  python port_scan.py 127.0.0.1
  python port_scan.py 127.0.0.1 --ports 22,80,443,8000 --timeout 0.5
  python port_scan.py your-authorized-target.example --i-have-written-authorization
  python port_scan.py --selftest
"""
from __future__ import annotations

import argparse
import socket
import sys
from concurrent.futures import ThreadPoolExecutor

COMMON_PORTS = [21, 22, 23, 25, 53, 80, 110, 143, 443, 445, 587, 993, 995,
                1433, 1521, 3306, 3389, 5432, 6379, 8000, 8080, 8443, 9000, 27017]
LOCAL_HOSTS = {"127.0.0.1", "localhost", "::1", "0.0.0.0"}


def is_local(host: str) -> bool:
    """只有本机地址可以不声明授权。"""
    return host.strip().lower() in LOCAL_HOSTS


def check_authorization(host: str, confirmed: bool) -> None:
    """非本机目标必须显式确认持有书面授权，否则终止（防止误扫）。"""
    if is_local(host):
        return
    if not confirmed:
        sys.exit(
            "已阻止：%s 不是本机地址。\n"
            "仅当你已获得目标所有者的书面授权时，才可追加 --i-have-written-authorization 参数。"
            % host
        )


def scan_port(host: str, port: int, timeout: float = 0.8) -> bool:
    """TCP 三次握手能否建立：True=open，False=closed / filtered / timeout。"""
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.settimeout(timeout)
            return s.connect_ex((host, port)) == 0
    except OSError:
        return False


def scan(host: str, ports, timeout: float = 0.8, workers: int = 64):
    """并发扫描，返回按端口升序的 [(port, is_open), ...]。"""
    with ThreadPoolExecutor(max_workers=workers) as ex:
        results = list(ex.map(lambda p: (p, scan_port(host, p, timeout)), ports))
    return sorted(results)


def parse_ports(text: str):
    """解析 '22,80,443' 或 '1-1024' 或两者混写。"""
    ports = []
    for part in text.split(","):
        part = part.strip()
        if not part:
            continue
        if "-" in part:
            a, b = part.split("-", 1)
            ports += list(range(int(a), int(b) + 1))
        else:
            ports.append(int(part))
    ports = sorted({p for p in ports if 0 < p < 65536})
    if not ports:
        sys.exit("端口列表为空或非法")
    return ports


def selftest() -> int:
    """最小自测：本地开临时监听端口验证 open/closed 判定。"""
    import threading

    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as srv:
        srv.bind(("127.0.0.1", 0))
        srv.listen(1)
        open_port = srv.getsockname()[1]
        threading.Thread(target=srv.accept, daemon=True).start()
        assert scan_port("127.0.0.1", open_port, 1.0) is True, "已监听端口应判定为 open"

    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as tmp:
        tmp.bind(("127.0.0.1", 0))
        closed_port = tmp.getsockname()[1]          # 绑定后立即释放，基本无人监听
    assert scan_port("127.0.0.1", closed_port, 0.5) is False, "未监听端口应判定为 closed"

    print("port_scan selftest OK：open / closed 判定正确")
    return 0


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description="授权范围内的 TCP 端口连通性检查")
    ap.add_argument("host", nargs="?", default="127.0.0.1", help="目标主机（默认本机 127.0.0.1）")
    ap.add_argument("--ports", default=",".join(str(p) for p in COMMON_PORTS), help="端口列表，如 22,80,443 或 1-1024")
    ap.add_argument("--timeout", type=float, default=0.8, help="单端口超时秒数")
    ap.add_argument("--workers", type=int, default=64, help="并发线程数")
    ap.add_argument("--selftest", action="store_true", help="只跑最小自测，不扫描任何目标")
    ap.add_argument("--i-have-written-authorization", dest="authorized", action="store_true",
                    help="声明你已获得目标所有者的书面授权（扫描非本机目标时必需）")
    args = ap.parse_args(argv)

    if args.selftest:
        return selftest()

    check_authorization(args.host, args.authorized)
    ports = parse_ports(args.ports)
    print("目标 %s | 端口数 %d | timeout=%.2fs" % (args.host, len(ports), args.timeout))
    opened = []
    for port, is_open in scan(args.host, ports, args.timeout, args.workers):
        if is_open:
            opened.append(port)
            print("  [open]   %d" % port)
    print("开放端口 %d 个：%s" % (len(opened), ", ".join(map(str, opened)) or "无"))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

http_audit.py

响应头基线体检：CSP / HSTS / XFO / nosniff / Referrer-Policy / Permissions-Policy / 版本泄露，支持 JSON 输出。

py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""HTTP 响应头安全配置体检（仅标准库）。

检查你自己的站点 / 已授权目标的响应头是否满足基线：
  Content-Security-Policy、Strict-Transport-Security、X-Frame-Options、
  X-Content-Type-Options、Referrer-Policy、Permissions-Policy，
  以及 Server、X-Powered-By 是否泄露版本号。

用法：
  python http_audit.py --selftest                     # 本地起服务自测（不需要联网）
  python http_audit.py https://your-own-site.example  # 体检自己的站点
  python http_audit.py https://your-site --json       # 输出 JSON，便于写进报告
"""
from __future__ import annotations

import argparse
import json
import ssl
import sys
import urllib.request

UA = "Mozilla/5.0 (compatible; nocau-http-audit/1.0)"


def audit_headers(headers, scheme: str = "https"):
    """headers 为普通 dict（大小写不敏感地取用）；返回逐项检查结果列表。"""
    h = {str(k).lower(): str(v) for k, v in headers.items()}
    csp = h.get("content-security-policy", "")
    csp_low = csp.lower()
    out = []

    out.append({"name": "Content-Security-Policy", "ok": bool(csp), "detail": csp or "未配置",
                "fix": "以 default-src 'self' 起步，逐步去掉 unsafe-inline / unsafe-eval"})

    hsts = h.get("strict-transport-security", "")
    if scheme != "https":
        out.append({"name": "Strict-Transport-Security", "ok": True, "detail": "非 HTTPS，跳过",
                    "fix": "站点上 HTTPS 后再配置 max-age=31536000; includeSubDomains"})
    else:
        out.append({"name": "Strict-Transport-Security", "ok": bool(hsts), "detail": hsts or "未配置",
                    "fix": "配置 max-age=31536000; includeSubDomains"})

    xfo = h.get("x-frame-options", "")
    out.append({"name": "X-Frame-Options", "ok": bool(xfo) or "frame-ancestors" in csp_low,
                "detail": xfo or ("由 CSP frame-ancestors 覆盖" if "frame-ancestors" in csp_low else "未配置"),
                "fix": "设置 DENY / SAMEORIGIN，或用 CSP frame-ancestors 限制嵌入"})

    xcto = h.get("x-content-type-options", "")
    out.append({"name": "X-Content-Type-Options", "ok": xcto.strip().lower() == "nosniff",
                "detail": xcto or "未配置", "fix": "设置为 nosniff，禁止 MIME 嗅探"})

    rp = h.get("referrer-policy", "")
    out.append({"name": "Referrer-Policy", "ok": bool(rp), "detail": rp or "未配置",
                "fix": "建议 strict-origin-when-cross-origin"})

    pp = h.get("permissions-policy", "")
    out.append({"name": "Permissions-Policy", "ok": bool(pp), "detail": pp or "未配置",
                "fix": "按需关闭 camera / microphone / geolocation 等能力"})

    server = h.get("server", "")
    leak = bool(server) and any(ch.isdigit() for ch in server)
    out.append({"name": "Server 版本泄露", "ok": not leak, "detail": server or "未返回",
                "fix": "隐藏或裁剪版本号（如 nginx 的 server_tokens off）"})

    xpb = h.get("x-powered-by", "")
    out.append({"name": "X-Powered-By", "ok": not xpb, "detail": xpb or "未返回",
                "fix": "移除该响应头，避免暴露技术栈与版本"})
    return out


def fetch_headers(url: str, timeout: float = 15.0):
    """返回 (status, headers_dict, final_url)。"""
    req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "*/*"})
    ctx = ssl.create_default_context()
    with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r:
        return r.status, dict(r.headers.items()), r.geturl()


def http_server(secure: bool):
    """起一个本地 HTTP 服务：secure=True 返回完整安全头，False 返回带版本号的裸服务。"""
    import threading
    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

    class Handler(BaseHTTPRequestHandler):
        server_version = "demo-srv" if secure else "TestServer/1.2.3"
        sys_version = ""

        def do_GET(self):
            self.send_response(200)
            self.send_header("Content-Type", "text/plain; charset=utf-8")
            if secure:
                self.send_header("Content-Security-Policy", "default-src 'self'")
                self.send_header("X-Frame-Options", "DENY")
                self.send_header("X-Content-Type-Options", "nosniff")
                self.send_header("Referrer-Policy", "strict-origin-when-cross-origin")
                self.send_header("Permissions-Policy", "camera=(), microphone=()")
            else:
                self.send_header("X-Powered-By", "Demo/0.1")
            self.end_headers()
            self.wfile.write(b"ok")

        def log_message(self, *args):
            pass

    srv = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    return srv, srv.server_address[1]


def selftest() -> int:
    """最小自测：不安全站点应被判不合格，安全站点全部通过。"""
    bad_srv, bad_port = http_server(secure=False)
    good_srv, good_port = http_server(secure=True)
    need = ("Content-Security-Policy", "X-Frame-Options", "X-Content-Type-Options",
            "Referrer-Policy", "Permissions-Policy")
    try:
        _, h_bad, _ = fetch_headers("http://127.0.0.1:%d/" % bad_port)
        bad = {x["name"]: x for x in audit_headers(h_bad, "http")}
        for name in need:
            assert not bad[name]["ok"], "%s 在裸服务上应判定为不合格" % name
        assert not bad["Server 版本泄露"]["ok"], "带版本号的 Server 头应被标记"

        _, h_good, _ = fetch_headers("http://127.0.0.1:%d/" % good_port)
        good = {x["name"]: x for x in audit_headers(h_good, "http")}
        for name in need:
            assert good[name]["ok"], "%s 在安全服务上应通过" % name
        assert good["Server 版本泄露"]["ok"], "无版本号 Server 头应通过"
        assert good["X-Powered-By"]["ok"], "无 X-Powered-By 应通过"
    finally:
        bad_srv.shutdown()
        good_srv.shutdown()
    print("http_audit selftest OK：裸服务被判不合格，安全配置全部通过")
    return 0


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description="HTTP 响应头安全配置体检")
    ap.add_argument("url", nargs="?", help="要体检的地址（请只填自己拥有或已获授权的站点）")
    ap.add_argument("--selftest", action="store_true", help="本地起服务自测，不需要联网")
    ap.add_argument("--json", action="store_true", help="以 JSON 输出，便于写入报告")
    ap.add_argument("--timeout", type=float, default=15.0, help="请求超时秒数")
    args = ap.parse_args(argv)

    if args.selftest:
        return selftest()
    if not args.url:
        ap.error("请提供要体检的 URL，或使用 --selftest")

    scheme = args.url.split("://", 1)[0].lower()
    status, headers, final = fetch_headers(args.url, args.timeout)
    rows = audit_headers(headers, scheme)

    if args.json:
        print(json.dumps({"url": args.url, "final_url": final, "status": status, "checks": rows},
                         ensure_ascii=False, indent=2))
        return 0

    print("体检地址：%s（最终 %s，HTTP %d）" % (args.url, final, status))
    passed = 0
    for r in rows:
        flag = "PASS" if r["ok"] else "FIX "
        passed += 1 if r["ok"] else 0
        print("  [%s] %-26s %s" % (flag, r["name"], r["detail"][:70]))
        if not r["ok"]:
            print("        建议：%s" % r["fix"])
    print("通过 %d / %d 项" % (passed, len(rows)))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

selftest.py

unittest：授权拦截、端口 open/closed 判定、裸服务与安全服务的审计结果对照。

py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""security-pentest 最小自测（仅标准库 unittest）：串联验证两个工具的判定逻辑。

运行：
  python selftest.py
  python -m unittest selftest -v
"""
from __future__ import annotations

import socket
import threading
import unittest

import http_audit
import port_scan


class PortScanTest(unittest.TestCase):
    def test_local_target_allowed(self):
        port_scan.check_authorization("127.0.0.1", False)   # 不抛异常即通过

    def test_non_local_target_blocked(self):
        with self.assertRaises(SystemExit):
            port_scan.check_authorization("example.com", False)

    def test_non_local_target_allowed_with_authorization(self):
        port_scan.check_authorization("example.com", True)  # 声明授权后放行

    def test_open_and_closed_detection(self):
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as srv:
            srv.bind(("127.0.0.1", 0))
            srv.listen(1)
            open_port = srv.getsockname()[1]
            threading.Thread(target=srv.accept, daemon=True).start()
            self.assertTrue(port_scan.scan_port("127.0.0.1", open_port, 1.0))
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as tmp:
            tmp.bind(("127.0.0.1", 0))
            closed_port = tmp.getsockname()[1]
        self.assertFalse(port_scan.scan_port("127.0.0.1", closed_port, 0.5))

    def test_parse_ports(self):
        self.assertEqual(port_scan.parse_ports("80, 443, 8000-8002"),
                         [80, 443, 8000, 8001, 8002])


class HttpAuditTest(unittest.TestCase):
    def setUp(self):
        self.bad_srv, self.bad_port = http_audit.http_server(secure=False)
        self.good_srv, self.good_port = http_audit.http_server(secure=True)

    def tearDown(self):
        self.bad_srv.shutdown()
        self.good_srv.shutdown()

    def _audit(self, port):
        _, headers, _ = http_audit.fetch_headers("http://127.0.0.1:%d/" % port)
        return {x["name"]: x for x in http_audit.audit_headers(headers, "http")}

    def test_insecure_server_flagged(self):
        rows = self._audit(self.bad_port)
        self.assertFalse(rows["Content-Security-Policy"]["ok"])
        self.assertFalse(rows["X-Content-Type-Options"]["ok"])
        self.assertFalse(rows["Server 版本泄露"]["ok"])
        self.assertFalse(rows["X-Powered-By"]["ok"])

    def test_secure_server_passes(self):
        rows = self._audit(self.good_port)
        for name in ("Content-Security-Policy", "X-Frame-Options", "X-Content-Type-Options",
                     "Referrer-Policy", "Permissions-Policy", "Server 版本泄露", "X-Powered-By"):
            self.assertTrue(rows[name]["ok"], name)


def main() -> int:
    suite = unittest.defaultTestLoader.loadTestsFromModule(__import__(__name__))
    result = unittest.TextTestRunner(verbosity=2).run(suite)
    print("security-pentest selftest：%d 个用例，失败 %d" % (result.testsRun, len(result.failures) + len(result.errors)))
    return 0 if result.wasSuccessful() else 1


if __name__ == "__main__":
    raise SystemExit(main())

README.md

运行环境、依赖清单、可直接运行说明、需先改的配置项与最小自测命令。

md
# 授权渗透测试工具包（security-pentest）

两个零依赖命令行工具：TCP 端口连通性检查、HTTP 响应头安全体检，外加一个 unittest 自测脚本。
**仅可用于你自己拥有、或已获得所有者书面授权的目标。**

## 目录结构

```
security-pentest/
├── port_scan.py    授权范围内的 TCP 端口扫描（默认只允许本机）
├── http_audit.py   HTTP 响应头安全配置体检（CSP / HSTS / 版本泄露等）
├── selftest.py     7 个 unittest 用例，验证两项工具的判定逻辑
└── README.md       运行说明、依赖清单、合规边界
```

## 运行环境与依赖清单

- Python 3.8+，**仅标准库**：socket / concurrent.futures / urllib / ssl / http.server / json / argparse / unittest
- 无第三方依赖，不需要 `pip install`；无需安装 nmap（脚本内置并发 socket 探测）

## 运行方式（可直接运行）

```bash
python port_scan.py --selftest                        # 不扫描任何外部目标，仅自检
python port_scan.py 127.0.0.1                         # 扫本机常见端口
python port_scan.py 127.0.0.1 --ports 22,80,443,8000 --timeout 0.5
python http_audit.py --selftest                       # 本地起服务自测，不需要联网
python http_audit.py https://你自己的站点 --json       # 体检响应头，输出 JSON
python selftest.py                                    # 全部用例应 OK
```

## 需先改的配置项

- **扫描非本机目标**：必须显式追加 `--i-have-written-authorization`，否则脚本会直接拦截退出；
  这是刻意的合规保护，不建议绕过。
- **端口清单**：改 `port_scan.py` 顶部 `COMMON_PORTS`；单次扫描建议不超过 1024 个端口，并发数用 `--workers` 调整。
- **体检基线**：`http_audit.py` 中 `audit_headers()` 的判定与 `fix` 建议可按团队规范调整（如 HSTS 时长）。

## 最小自测

```bash
python port_scan.py --selftest    # 期望：port_scan selftest OK
python http_audit.py --selftest   # 期望：http_audit selftest OK
python selftest.py                # 期望：Ran 7 tests ... OK
```

## 合规声明

本工具包用于安全学习、CTF、自建靶场与获得授权的渗透测试项目。
严禁用于未授权的系统探测、入侵或数据获取；使用者须自行承担一切法律责任。
运行标注：可直接运行（Python 3.8+ 标准库，零第三方依赖；两个 --selftest 都不联网、不扫描外部目标）。需先改的配置项：扫描非本机目标必须携带 --i-have-written-authorization（默认拦截）；端口清单在 port_scan.py 顶部 COMMON_PORTS；体检基线在 http_audit.py 的 audit_headers()。最小自测：python port_scan.py --selftest / python http_audit.py --selftest / python selftest.py（期望 Ran 7 tests ... OK）。
复制结果为 # ===== 文件名 ===== 分隔的多文件文本，按文件名分别保存即可；也可逐块用右上角“复制代码”单独取用。

来源参考

实现所依据的官方规范与文档（均已实测可直接访问）：

OWASP · Web 安全测试指南（WSTG，渗透测试流程权威参考） NIST SP 800-115 · 信息安全测试与评估技术指南 nmap 官方手册（资产测绘与端口扫描参数） Python 官方文档 · socket（本工程探测逻辑所用模块） OWASP · HTTP 安全响应头速查表
← 返回代码开发合集