"""渗透测试框架 — 共享核心模块（Day 161）。

本模块是整个框架的骨架，只有四件事，缺一件都不叫框架：

1. **范围模型 Scope**：把授权范围变成数据结构。越界在"创建 socket 之前"
   就抛 AuthorizationError —— 这是硬门禁，不可被参数关闭。
2. **审计日志 AuditLog**：谁、何时、对什么、做了什么。JSONL 追加写。
3. **只读探测 tcp_probe**：TCP connect + 读服务主动发来的问候语（banner）。
   **绝不发送任何自定义数据**；不爆破、不提交表单、不构造求解请求。
4. **编排器 Framework**：令牌桶限速 + 线程池并发 + 阶段结果状态机
   （ok / skipped / failed），并保证"空输入不得算出 0 个问题"。

安全设计（刻意为之，不是遗漏）：

* 只接受 IP / CIDR 形式的目标，**拒绝域名** → 避免隐式 DNS 外发。
* 不提供 raw socket / SYN 扫描（需要 root，且越界风险更高）。
* "利用"阶段为空实现（见 04-pentest-framework.py），本模块不含任何载荷。
* 默认范围 = 回环地址 + 8000-8099 实验端口段。

运行自检：
    python3 pentest_core.py --self-test     # 输出 SELF-TEST OK
"""

from __future__ import annotations

import ipaddress
import json
import socket
import threading
import time
from dataclasses import asdict, dataclass, field
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Callable, Iterable, Sequence

# ── 常量 ───────────────────────────────────────────────────────────────

CST = timezone(timedelta(hours=8))
DEFAULT_TIMEOUT = 0.5      # 秒；实验环境够用，报告里必须回显
DEFAULT_WORKERS = 8        # 同时在途连接数（背压）
DEFAULT_RATE = 50.0        # 每秒新增连接数（礼貌）

# 端口 → 服务名（教学用对照表，识别依据是 IANA 分配 + banner）
SERVICE_MAP = {
    20: "ftp-data", 21: "ftp", 22: "ssh", 23: "telnet", 25: "smtp",
    53: "domain", 80: "http", 110: "pop3", 143: "imap", 443: "https",
    445: "microsoft-ds", 3306: "mysql", 3389: "rdp", 5432: "postgresql",
    6379: "redis", 8000: "http-alt", 8080: "http-proxy", 8443: "https-alt",
}

# 本课实验服务（全部绑定在回环地址，仅供自测）
LAB_SERVICES: dict[int, str] = {
    8000: "SSH-2.0-OpenSSH_8.9p1 Ubuntu-3ubuntu0.4\r\n",       # 伪装 SSH
    8080: "HTTP/1.1 200 OK\r\nServer: lab-nginx/1.18.0\r\n\r\n",  # 伪装 HTTP
    8888: "220 lab-smtp.example ESMTP LabSMTP 1.0\r\n",        # 范围外端口
}

SEVERITY_ORDER = ["info", "low", "medium", "high", "critical"]


class AuthorizationError(RuntimeError):
    """越界访问被拒绝。**不可能被重试或参数绕过** —— 策略拒绝，不是网络错误。"""


class ScopeError(ValueError):
    """范围文件本身不合法（缺字段、端口越界、时间格式错等）。"""


# ── 基础工具 ───────────────────────────────────────────────────────────

def now_iso() -> str:
    return datetime.now(CST).isoformat(timespec="seconds")


def parse_port_spec(spec: str) -> list[int]:
    """解析端口表达式： "80,443,8000-8010" → 有序去重列表。

    >>> parse_port_spec("80,443")
    [80, 443]
    >>> parse_port_spec("8000-8002")
    [8000, 8001, 8002]
    """
    ports: set[int] = set()
    for chunk in str(spec).split(","):
        chunk = chunk.strip()
        if not chunk:
            continue
        if "-" in chunk:
            lo_s, _, hi_s = chunk.partition("-")
            lo, hi = int(lo_s), int(hi_s)
            if lo > hi:
                raise ScopeError(f"端口区间反向: {chunk}")
            ports.update(range(lo, hi + 1))
        else:
            ports.add(int(chunk))
    for p in ports:
        if not 1 <= p <= 65535:
            raise ScopeError(f"端口超出 1..65535: {p}")
    return sorted(ports)


def parse_host_spec(spec: str) -> list[str]:
    """把 IP 或 CIDR 展开成主机列表。

    **只接受 IP 字面量**：域名会被拒绝，因为解析域名本身就会产生 DNS 外发，
    而 DNS 查询可能落到不相关的基础设施上（也是一种越界）。
    """
    spec = str(spec).strip()
    if not spec:
        return []
    try:
        net = ipaddress.ip_network(spec, strict=False)
    except ValueError as exc:
        raise ScopeError(f"只接受 IP / CIDR 目标，拒绝域名或非法输入: {spec!r} ({exc})")
    if net.num_addresses == 1:
        return [str(net.network_address)]
    # /31 /32 等特殊段：ip.hosts() 在 /31、/32 上只会返回部分地址
    return [str(ip) for ip in net.hosts()] or [str(net.network_address)]


# ── 范围模型 ───────────────────────────────────────────────────────────

@dataclass(frozen=True)
class Target:
    """一个"主机:端口"探测单元。frozen 是为了能放进 set 去重。"""
    host: str
    port: int

    @property
    def key(self) -> str:
        return f"{self.host}:{self.port}"

    def __str__(self) -> str:  # pragma: no cover - 仅用于打印
        return self.key


@dataclass
class Scope:
    """授权范围。黑名单（excluded）优先于白名单（networks）。"""

    networks: list[str] = field(default_factory=list)
    ports: list[tuple[int, int]] = field(default_factory=list)
    excluded: list[str] = field(default_factory=list)
    allow_raw_sockets: bool = False       # 本课恒为 False
    allow_write_actions: bool = False     # 本课恒为 False
    ticket: str = ""
    expires: str = ""                     # ISO8601；空 = 不限
    note: str = ""

    # --- 构造 ---
    @classmethod
    def from_dict(cls, data: dict) -> "Scope":
        try:
            networks = list(data.get("networks", []))
            raw_ports = data.get("ports", [])
        except AttributeError as exc:  # pragma: no cover
            raise ScopeError(f"scope 必须是 JSON 对象: {exc}")
        if not networks:
            raise ScopeError("scope.networks 不能为空（没有范围的授权等于没有授权）")
        ports: list[tuple[int, int]] = []
        for item in raw_ports:
            if isinstance(item, str):
                for p in parse_port_spec(item):
                    ports.append((p, p))
            elif isinstance(item, (list, tuple)) and len(item) == 2:
                lo, hi = int(item[0]), int(item[1])
                if lo > hi:
                    raise ScopeError(f"端口区间反向: {item}")
                ports.append((lo, hi))
            else:
                raise ScopeError(f"无法解析端口条目: {item!r}")
        if not ports:
            raise ScopeError("scope.ports 不能为空（无端口范围的授权无法执行）")
        # 硬性拒绝：本课框架不提供 raw socket / 写操作能力
        if data.get("allow_raw_sockets") or data.get("allow_write_actions"):
            raise ScopeError("本课框架不支持 raw socket / 写操作，请勿在范围里声明")
        return cls(
            networks=networks,
            ports=ports,
            excluded=list(data.get("excluded", [])),
            ticket=str(data.get("ticket", "")),
            expires=str(data.get("expires", "")),
            note=str(data.get("note", "")),
        )

    @classmethod
    def load(cls, path: str | Path) -> "Scope":
        raw = json.loads(Path(path).read_text(encoding="utf-8"))
        return cls.from_dict(raw)

    def to_dict(self) -> dict:
        d = asdict(self)
        d["ports"] = [[lo, hi] for lo, hi in self.ports]
        return d

    # --- 判定 ---
    def _excluded(self, host: str) -> bool:
        ip = ipaddress.ip_address(host)
        for entry in self.excluded:
            if ip in ipaddress.ip_network(entry, strict=False):
                return True
        return False

    def allows_host(self, host: str) -> bool:
        ip = ipaddress.ip_address(host)
        if self._excluded(host):
            return False
        return any(ip in ipaddress.ip_network(n, strict=False) for n in self.networks)

    def allows_port(self, port: int) -> bool:
        return any(lo <= port <= hi for lo, hi in self.ports)

    def expired(self) -> bool:
        if not self.expires:
            return False
        try:
            deadline = datetime.fromisoformat(self.expires)
        except ValueError:
            raise ScopeError(f"scope.expires 不是 ISO8601 时间: {self.expires!r}")
        if deadline.tzinfo is None:
            deadline = deadline.replace(tzinfo=CST)
        return datetime.now(deadline.tzinfo) > deadline

    def check(self, target: Target) -> None:
        """不通过就抛 AuthorizationError。**唯一**的越界出口。"""
        if self.expired():
            raise AuthorizationError(f"授权已于 {self.expires} 到期，拒绝执行: {target.key}")
        if self._excluded(target.host):
            raise AuthorizationError(f"目标命中排除名单（黑名单优先）: {target.host}")
        if not self.allows_host(target.host):
            raise AuthorizationError(f"目标不在授权网段内: {target.host}")
        if not self.allows_port(target.port):
            raise AuthorizationError(f"端口不在授权端口区间内: {target.port}")


def default_lab_scope() -> Scope:
    """默认实验范围：回环地址 + 8000-8099。**供本机实验使用**。"""
    return Scope(
        networks=["127.0.0.1/32"],
        ports=[(8000, 8099)],
        excluded=[],
        ticket="LAB-SELF-001",
        note="self-owned loopback lab only",
    )


# ── 审计日志 ───────────────────────────────────────────────────────────

@dataclass
class AuditLog:
    """过程证据。报告是结论，审计日志是**可复核的事实**。"""

    entries: list[dict] = field(default_factory=list)
    path: Path | None = None
    _lock: threading.Lock = field(default_factory=threading.Lock, repr=False)

    def record(self, event: str, subject: str = "-", result: str = "", **detail) -> dict:
        entry = {
            "ts": now_iso(),
            "event": event,
            "subject": subject,
            "result": result,
            "detail": detail or {},
        }
        with self._lock:
            self.entries.append(entry)
            if self.path is not None:
                with open(self.path, "a", encoding="utf-8") as fh:
                    fh.write(json.dumps(entry, ensure_ascii=False) + "\n")
        return entry

    def to_jsonl(self) -> str:
        return "\n".join(json.dumps(e, ensure_ascii=False) for e in self.entries)

    def counts_by_event(self) -> dict[str, int]:
        out: dict[str, int] = {}
        for e in self.entries:
            out[e["event"]] = out.get(e["event"], 0) + 1
        return out


# ── 探测（只读） ───────────────────────────────────────────────────────

@dataclass
class ScanResult:
    target: Target
    state: str                  # open / closed / filtered / error
    banner: str = ""
    service: str = ""
    elapsed_ms: float = 0.0
    error: str = ""
    retried: int = 0

    def to_dict(self) -> dict:
        d = asdict(self)
        d["target"] = self.target.key
        return d


def guess_service(port: int, banner: str = "") -> str:
    """先看 banner（更可靠），再看端口号（只是约定）。"""
    low = banner.lower()
    if "ssh-" in low:
        return "ssh"
    if low.startswith("http/") or "server:" in low:
        return "http"
    if "smtp" in low:
        return "smtp"
    if "mysql" in low:
        return "mysql"
    return SERVICE_MAP.get(port, "unknown")


def _read_banner(sock: socket.socket, max_bytes: int = 256, timeout: float = 0.3) -> str:
    """只读服务**主动发来**的问候语。不发送任何字节。"""
    try:
        sock.settimeout(timeout)
        data = sock.recv(max_bytes)
        return data.decode("utf-8", "replace").strip()
    except (socket.timeout, OSError):
        return ""


def tcp_probe(
    target: Target,
    timeout: float = DEFAULT_TIMEOUT,
    retries: int = 1,
    grab_banner: bool = True,
) -> ScanResult:
    """TCP connect 探测（只读）。三态语义见 README 3.1。

    重试规则（README 4.5）：只重试 timeout / EHOSTUNREACH 这类抖动错误；
    ECONNREFUSED 是有效结论（closed），不重试。
    """
    started = time.perf_counter()
    attempt = 0
    last_err = ""
    while attempt <= retries:
        attempt += 1
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            # ★ 必须显式设置超时：否则 connect 走 OS 默认超时（可达数十秒），
            #   一个不可达目标就能让整轮扫描卡住 —— 这是最隐蔽的一个坑。
            sock.settimeout(timeout)
            code = sock.connect_ex((target.host, target.port))
            elapsed = (time.perf_counter() - started) * 1000
            if code == 0:
                banner = _read_banner(sock, timeout=min(timeout, 0.5)) if grab_banner else ""
                return ScanResult(
                    target=target, state="open", banner=banner,
                    service=guess_service(target.port, banner),
                    elapsed_ms=round(elapsed, 2), retried=attempt - 1,
                )
            if code == 111:  # ECONNREFUSED
                return ScanResult(
                    target=target, state="closed",
                    elapsed_ms=round(elapsed, 2), retried=attempt - 1,
                )
            last_err = f"errno={code}"
        except socket.gaierror as exc:      # 目标不是 IP 时才会出现
            return ScanResult(target=target, state="error", error=f"gaierror: {exc}",
                              elapsed_ms=round((time.perf_counter() - started) * 1000, 2))
        except OSError as exc:
            last_err = f"{type(exc).__name__}: {exc}"
        finally:
            sock.close()
        # 只有抖动类错误才重试，且不重试 last attempt
    elapsed = (time.perf_counter() - started) * 1000
    state = "filtered" if "timed out" in last_err.lower() or "errno=11" in last_err else "error"
    return ScanResult(
        target=target, state=state, error=last_err,
        elapsed_ms=round(elapsed, 2), retried=attempt - 1,
    )


# ── 令牌桶 ─────────────────────────────────────────────────────────────

class TokenBucket:
    """限速器：每秒最多放行 rate 个动作，容量上限 capacity。

    与 workers 的分工见 README 4.2：
      workers → 同时在途连接数（背压）
      rate    → 每秒新增连接数（礼貌）
    """

    def __init__(self, rate: float = DEFAULT_RATE, capacity: int = DEFAULT_WORKERS):
        if rate <= 0:
            raise ValueError("rate 必须 > 0")
        self.rate = float(rate)
        self.capacity = float(max(1, capacity))
        self._tokens = self.capacity
        self._updated = time.monotonic()
        self._lock = threading.Lock()
        self.waited_s = 0.0

    def acquire(self) -> float:
        """取一个令牌；不足则睡眠等待。返回本次等待秒数。"""
        waited = 0.0
        while True:
            with self._lock:
                now = time.monotonic()
                self._tokens = min(self.capacity, self._tokens + (now - self._updated) * self.rate)
                self._updated = now
                if self._tokens >= 1.0:
                    self._tokens -= 1.0
                    self.waited_s += waited
                    return waited
                need = (1.0 - self._tokens) / self.rate
            time.sleep(min(need, 0.05))
            waited += min(need, 0.05)


# ── Finding 与评分（沿用 Day 159/160 的模型） ──────────────────────────

@dataclass
class Finding:
    key: str
    title: str
    severity: str        # info/low/medium/high/critical
    confidence: str      # info/low/medium/high
    target: str = ""
    evidence: str = ""
    detail: dict = field(default_factory=dict)

    @property
    def priority(self) -> str:
        s = SEVERITY_ORDER.index(self.severity) if self.severity in SEVERITY_ORDER else 0
        c = SEVERITY_ORDER.index(self.confidence) if self.confidence in SEVERITY_ORDER else 0
        score = (s + 1) * (c + 1)
        return "P0" if score >= 20 else "P1" if score >= 12 else "P2" if score >= 6 else "P3"

    def to_dict(self) -> dict:
        d = asdict(self)
        d["priority"] = self.priority
        return d


# ── 插件注册表 ─────────────────────────────────────────────────────────

@dataclass
class PluginContext:
    """只读上下文：插件只能读，结果必须 return Finding（防止隐式耦合）。"""
    framework: "Framework"
    asset: ScanResult
    assets: list[ScanResult]


class Registry:
    """有序插件注册表。priority 小的先跑；同 priority 按注册顺序（稳定）。"""

    def __init__(self) -> None:
        self._items: list[tuple[str, int, Callable]] = []

    def plugin(self, name: str, priority: int = 100):
        def deco(fn: Callable[[PluginContext], list[Finding] | None]):
            self.register(name, fn, priority)
            return fn
        return deco

    def register(self, name: str, fn: Callable, priority: int = 100) -> None:
        if any(n == name for n, _, _ in self._items):
            raise ValueError(f"插件重名: {name}")
        self._items.append((name, priority, fn))

    def ordered(self) -> list[tuple[str, int, Callable]]:
        return sorted(self._items, key=lambda item: item[1])

    def names(self) -> list[str]:
        return [n for n, _, _ in self.ordered()]


# ── 阶段 ───────────────────────────────────────────────────────────────

@dataclass
class PhaseResult:
    name: str
    status: str          # ok / skipped / failed
    reason: str = ""
    findings: list[Finding] = field(default_factory=list)

    @property
    def usable(self) -> bool:
        return self.status == "ok"


@dataclass
class Phase:
    name: str
    fn: Callable[["Framework", list[ScanResult]], PhaseResult]
    optional: bool = False
    critical: bool = False


# ── 框架 ───────────────────────────────────────────────────────────────

class Framework:
    """编排器：范围门禁 + 去重缓存 + 限速并发 + 阶段状态机 + 审计留痕。"""

    def __init__(
        self,
        scope: Scope,
        timeout: float = DEFAULT_TIMEOUT,
        workers: int = DEFAULT_WORKERS,
        rate: float = DEFAULT_RATE,
        audit_path: str | Path | None = None,
    ) -> None:
        self.scope = scope
        self.timeout = timeout
        self.workers = workers
        self.bucket = TokenBucket(rate=rate, capacity=workers)
        self.audit = AuditLog(path=Path(audit_path) if audit_path else None)
        self.registry = Registry()
        self.cache: dict[str, ScanResult] = {}

    # 便捷注册（@fw.plugin("name")）
    def plugin(self, name: str, priority: int = 100):
        return self.registry.plugin(name, priority)

    # --- 目标构造（一定先校验，再返回） ---
    def build_targets(self, host_specs: Sequence[str], port_specs: Sequence[str]) -> list[Target]:
        hosts: list[str] = []
        for spec in host_specs:
            hosts.extend(parse_host_spec(spec))
        ports: list[int] = []
        for spec in port_specs:
            ports.extend(parse_port_spec(spec))
        targets = [Target(h, p) for h in hosts for p in ports]
        seen: set[Target] = set()
        unique = [t for t in targets if not (t in seen or seen.add(t))]
        self.audit.record("build_targets", "-", "ok",
                          hosts=len(hosts), ports=len(ports), targets=len(unique))
        return unique

    def authorize(self, targets: Iterable[Target]) -> list[Target]:
        """整批校验：**任何一个越界都会终止整轮执行**（fail-fast）。

        为什么是 fail-fast 而不是"跳过越界的"？
        因为越界意味着"我对范围的理解与授权不一致"——
        这种情况下正确的动作是停下来问人，而不是"悄悄少扫几台"。
        """
        approved: list[Target] = []
        for t in targets:
            try:
                self.scope.check(t)
            except AuthorizationError as exc:
                self.audit.record("scope_check", t.key, "denied", reason=str(exc))
                self.audit.record("run_abort", "-", "exit=2", reason=str(exc))
                raise
            self.audit.record("scope_check", t.key, "in_scope", ticket=self.scope.ticket)
            approved.append(t)
        return approved

    # --- 探测（唯一会产生网络行为的入口） ---
    def probe_one(self, target: Target) -> ScanResult:
        if target.key in self.cache:
            result = self.cache[target.key]
            self.audit.record("probe", target.key, "cache_hit", state=result.state)
            return result
        self.bucket.acquire()
        result = tcp_probe(target, timeout=self.timeout)
        self.cache[target.key] = result
        self.audit.record("probe", target.key, result.state,
                          elapsed_ms=result.elapsed_ms, retried=result.retried,
                          errno_or_error=result.error)
        return result

    def probe_many(self, targets: Sequence[Target]) -> list[ScanResult]:
        """线程池并发 + 令牌桶限速。**结果顺序与输入一致**（编排需要确定性）。"""
        if not targets:
            return []
        from concurrent.futures import ThreadPoolExecutor

        self.audit.record("phase_begin", "collect", "", targets=len(targets),
                          workers=self.workers, rate=self.bucket.rate)
        started = time.perf_counter()
        with ThreadPoolExecutor(max_workers=self.workers) as pool:
            results = list(pool.map(self.probe_one, targets))
        elapsed = round(time.perf_counter() - started, 3)
        stats = self.state_counts(results)
        self.audit.record("phase_end", "collect", "ok", assets=stats["open"],
                          **{f"n_{k}": v for k, v in stats.items()},
                          elapsed_s=elapsed, throttle_wait_s=round(self.bucket.waited_s, 3))
        return results

    @staticmethod
    def state_counts(results: Sequence[ScanResult]) -> dict[str, int]:
        out = {"open": 0, "closed": 0, "filtered": 0, "error": 0}
        for r in results:
            out[r.state] = out.get(r.state, 0) + 1
        return out

    # --- 插件执行 ---
    def run_plugins(self, assets: Sequence[ScanResult]) -> list[Finding]:
        """按 priority 依次执行插件。

        单插件崩溃**只影响它自己**：记 `plugin_end / failed` 后继续。
        为什么不直接抛出去？因为"插件崩溃"和"插件没发现问题"在报告里
        必须能被区分，而终止整轮会让后面所有插件的结论一起消失。
        """
        findings: list[Finding] = []
        for name, _priority, fn in self.registry.ordered():
            for asset in assets:
                ctx = PluginContext(framework=self, asset=asset, assets=list(assets))
                try:
                    produced = fn(ctx) or []
                except AuthorizationError:
                    raise                      # 越界永远不外吞
                except Exception as exc:       # noqa: BLE001 - 兜底必须宽
                    self.audit.record("plugin_end", asset.target.key, "failed",
                                      plugin=name, error=f"{type(exc).__name__}: {exc}")
                    continue
                self.audit.record("plugin_end", asset.target.key,
                                  f"ok findings={len(produced)}", plugin=name)
                for f in produced:
                    self.audit.record("finding", f.target or asset.target.key,
                                      f"{f.severity}/{f.confidence}",
                                      plugin=name, key=f.key, priority=f.priority)
                findings.extend(produced)
        return findings

    # --- 阶段编排 ---
    def run_phases(
        self,
        phases: Sequence[Phase],
        assets: list[ScanResult],
    ) -> tuple[list[PhaseResult], list[ScanResult], list[Finding]]:
        results: list[PhaseResult] = []
        findings: list[Finding] = []
        for phase in phases:
            # 依赖检查：上游没有可用资产时，**必须 skipped**（绝不能算出 0 个问题）
            if phase.name != "collect" and not assets:
                res = PhaseResult(phase.name, "skipped", "上游阶段无可用资产（全部 filtered/error）")
                self.audit.record("phase_skip", phase.name, "skipped", reason=res.reason)
                results.append(res)
                continue
            self.audit.record("phase_begin", phase.name, "")
            try:
                res = phase.fn(self, assets)
            except AuthorizationError:
                raise
            except Exception as exc:  # noqa: BLE001 - 框架层兜底
                res = PhaseResult(phase.name, "failed", f"{type(exc).__name__}: {exc}")
            res.name = phase.name
            if res.status == "failed" and phase.critical:
                self.audit.record("run_abort", phase.name, "exit=4", reason=res.reason)
                results.append(res)
                raise RuntimeError(f"关键阶段失败: {phase.name}: {res.reason}")
            self.audit.record(
                "phase_end", phase.name, res.status,
                findings=len(res.findings), reason=res.reason,
            )
            findings.extend(res.findings)
            results.append(res)
        return results, assets, findings


# ── 报告 ───────────────────────────────────────────────────────────────

def findings_table(findings: Sequence[Finding]) -> str:
    if not findings:
        return "（无）"
    lines = ["| 优先级 | 严重度 | 置信度 | 插件 | 目标 | 说明 |",
             "| --- | --- | --- | --- | --- | --- |"]
    rank = {"P0": 0, "P1": 1, "P2": 2, "P3": 3}
    for f in sorted(findings, key=lambda x: (rank[x.priority], x.key)):
        lines.append(
            f"| {f.priority} | {f.severity} | {f.confidence} | {f.key} | "
            f"{f.target or '-'} | {f.title} |"
        )
    return "\n".join(lines)


def build_report(
    scope: Scope,
    results: Sequence[ScanResult],
    phases: Sequence[PhaseResult],
    findings: Sequence[Finding],
    params: dict,
    audit: AuditLog,
) -> dict:
    stats = Framework.state_counts(results)
    by_prio: dict[str, int] = {}
    for f in findings:
        by_prio[f.priority] = by_prio.get(f.priority, 0) + 1
    open_assets = [
        {"target": r.target.key, "service": r.service, "banner": r.banner}
        for r in results if r.state == "open"
    ]
    return {
        "generated_at": now_iso(),
        "scope": scope.to_dict(),
        "params": params,
        "coverage": {
            "targets": len(results),
            **stats,
            "conclusive": stats["open"] + stats["closed"],
            "inconclusive_filtered": stats["filtered"],
        },
        "assets": open_assets,
        "findings": [f.to_dict() for f in findings],
        "phases": [asdict(p) | {"findings": [f.key for f in p.findings]} for p in phases],
        "audit_events": audit.counts_by_event(),
        "exit_code": exit_code(findings, stats),
    }


def exit_code(findings: Sequence[Finding], stats: dict[str, int]) -> int:
    """退出码语义（优先级 2 > 4 > 3 > 0）：

    2  用法/授权错误（越界、参数非法）——不可执行
    4  结果不完整（存在 filtered/error，本轮无结论）
    3  有达到阈值的发现（P0/P1）
    0  干净完成
    """
    if stats.get("filtered", 0) or stats.get("error", 0):
        return 4
    if any(f.priority in ("P0", "P1") for f in findings):
        return 3
    return 0


def render_markdown(report: dict) -> str:
    cov = report["coverage"]
    lines = [
        "# 渗透测试扫描报告（自动化阶段 1–2）",
        "",
        f"- 生成时间：{report['generated_at']}",
        f"- 授权单号：{report['scope'].get('ticket') or '（未填写）'}",
        f"- 范围：{report['scope']['networks']} · 端口 {report['scope']['ports']}",
        f"- 参数：{report['params']}",
        "",
        "## 覆盖与结论可信度",
        "",
        f"- 目标数 {cov['targets']}：open {cov['open']} / closed {cov['closed']} / "
        f"filtered {cov['filtered']} / error {cov['error']}",
        f"- **可判定 {cov['conclusive']} 条；不可判定 {cov['inconclusive_filtered']} 条**"
        "（filtered = 被过滤或链路问题，不能当作关闭）",
        "",
        "## 开放端口与服务",
        "",
    ]
    if report["assets"]:
        lines += ["| 目标 | 服务 | banner（截断） |", "| --- | --- | --- |"]
        for a in report["assets"]:
            lines.append(f"| {a['target']} | {a['service']} | {a['banner'][:60]} |")
    else:
        lines.append("（无开放端口）")
    lines += ["", "## 发现", "", findings_table([Finding(**{k: v for k, v in f.items() if k != 'priority'})
                                                for f in report["findings"]]), ""]
    lines += ["## 阶段执行情况", "", "| 阶段 | 状态 | 说明 | 发现数 |", "| --- | --- | --- | --- |"]
    for p in report["phases"]:
        lines.append(f"| {p['name']} | {p['status']} | {p['reason'] or '-'} | {len(p['findings'])} |")
    lines += ["", "## 审计事件统计", "", f"`{report['audit_events']}`", ""]
    return "\n".join(lines)


# ── 实验服务（自测用，仅供回环） ───────────────────────────────────────

class LabServer:
    """最小实验服务：每个端口一个线程，连接后**只发一句问候语**就关闭。

    用于让框架在完全离线的环境里被测出 open / banner / service。
    """

    def __init__(self, services: dict[int, str] | None = None, host: str = "127.0.0.1"):
        self.services = dict(services or LAB_SERVICES)
        self.host = host
        self._socks: list[socket.socket] = []
        self._threads: list[threading.Thread] = []
        self._stop = threading.Event()

    def start(self) -> list[int]:
        started: list[int] = []
        for port, greeting in self.services.items():
            try:
                srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                srv.bind((self.host, port))
                srv.listen(16)
            except OSError:
                continue        # 端口被占：跳过（实验环境很常见，不算失败）
            self._socks.append(srv)
            th = threading.Thread(target=self._serve, args=(srv, greeting), daemon=True)
            th.start()
            self._threads.append(th)
            started.append(port)
        return started

    def _serve(self, srv: socket.socket, greeting: str) -> None:
        srv.settimeout(0.2)
        while not self._stop.is_set():
            try:
                conn, _ = srv.accept()
            except socket.timeout:
                continue
            except OSError:
                return
            try:
                conn.sendall(greeting.encode("utf-8"))
            except OSError:
                pass
            finally:
                conn.close()

    def stop(self) -> None:
        self._stop.set()
        for srv in self._socks:
            try:
                srv.close()
            except OSError:
                pass
        for th in self._threads:
            th.join(timeout=1.0)

    def __enter__(self) -> "LabServer":
        self.start()
        return self

    def __exit__(self, *exc) -> None:
        self.stop()


# ── 自检 ───────────────────────────────────────────────────────────────

def _self_test() -> None:
    # 端口表达式
    assert parse_port_spec("80,443,8000-8002") == [80, 443, 8000, 8001, 8002]
    try:
        parse_host_spec("example.com")
        raise AssertionError("域名必须被拒绝")
    except ScopeError:
        pass
    assert parse_host_spec("127.0.0.0/30") == ["127.0.0.1", "127.0.0.2"]

    # 范围判定：黑名单优先
    sc = default_lab_scope()
    sc.excluded = ["127.0.0.1/32"]
    try:
        sc.check(Target("127.0.0.1", 8000))
        raise AssertionError("命中排除名单必须被拒绝")
    except AuthorizationError:
        pass
    sc.excluded = []
    sc.check(Target("127.0.0.1", 8000))
    for bad in (Target("10.0.0.5", 8000), Target("127.0.0.1", 22)):
        try:
            sc.check(bad)
            raise AssertionError(f"越界目标必须被拒绝: {bad}")
        except AuthorizationError:
            pass

    # 探测三态 + 编排 + 审计
    with LabServer({8000: LAB_SERVICES[8000]}) as lab:
        fw = Framework(default_lab_scope(), timeout=0.3, workers=4, rate=100)
        targets = fw.build_targets(["127.0.0.1"], ["8000-8002"])
        assert len(targets) == 3
        fw.authorize(targets)
        results = fw.probe_many(targets)
        states = {r.target.key: r.state for r in results}
        assert states["127.0.0.1:8000"] == "open", states
        assert states["127.0.0.1:8001"] == "closed", states
        opened = [r for r in results if r.state == "open"]
        assert opened and opened[0].service == "ssh", opened[0]

        # 越界必须让整轮终止（fail-fast）
        denied = fw.build_targets(["10.0.0.5"], ["8000"])
        try:
            fw.authorize(denied)
            raise AssertionError("越界必须抛 AuthorizationError")
        except AuthorizationError:
            pass

        # 缓存命中：同一目标二次探测不再产生新 probe 记录
        before = len(fw.audit.entries)
        fw.probe_one(Target("127.0.0.1", 8000))
        added = fw.audit.entries[before:]
        assert len(added) == 1 and added[0]["result"] == "cache_hit", added

        # 空资产 → 阶段必须 skipped（禁止在空输入上算出 0 个问题）
        fw2 = Framework(default_lab_scope(), timeout=0.3)
        empty_phase = Phase("detect", lambda fw, assets: PhaseResult("detect", "ok"))
        res, _, _ = fw2.run_phases([empty_phase], [])
        assert res[0].status == "skipped", res

    print("SELF-TEST OK")


if __name__ == "__main__":
    import sys

    if "--self-test" in sys.argv:
        _self_test()
    else:
        print(__doc__)