vpn-loopback-firewall/masquerader/rewrite.py

"""SYN 包重写核心(纯函数, 不依赖 netfilter)。

把出站 TCP SYN 伪装成 Windows 10 布局:
- window 64240 + wscale 8(实测安全: 接收窗口高估不触发对端问题)
- 剥掉 TS option(协商级移除: SYN 不带 TS 则全连接无 TS, 无 PAWS 回显问题,
  多设备时钟分离信号直接失效)
- MSS 钳 1460(Windows 以太网默认)
- TTL 128(可选手动设, 常规由 nftables/iptables 层做)

核心约束(断网事故复盘):
- 只处理出向 SYN(带 SYN 标志且无 ACK)
- 不认识的包一律返回 None(worker 原样放行)
"""
from __future__ import annotations

import json
import struct
from dataclasses import dataclass
from pathlib import Path
from typing import Any

TCP_OPT_EOL = 0
TCP_OPT_NOP = 1
TCP_OPT_MSS = 2
TCP_OPT_WS = 3
TCP_OPT_SACK = 4
TCP_OPT_TS = 8

TCP_HDR_LEN = 20
IP_PROTO_TCP = 6

SYN_FLAG = 0x02
ACK_FLAG = 0x10


@dataclass(frozen=True)
class Fingerprint:
    """Windows 10 TCP 指纹模板(值全部来自 config/masquerader.json)。"""

    ttl: int = 128
    window: int = 64240
    wscale: int = 8
    mss: int = 1460
    set_mss: bool = True
    strip_timestamps: bool = True


def fingerprint_from_config(cfg: dict[str, Any]) -> Fingerprint:
    fp: dict[str, Any] = cfg.get("fingerprint", {})
    return Fingerprint(
        ttl=int(fp.get("ttl", 128)),
        window=int(fp.get("window", 64240)),
        wscale=int(fp.get("wscale", 8)),
        mss=int(fp.get("mss", 1460)),
        set_mss=bool(fp.get("set_mss", True)),
        strip_timestamps=bool(fp.get("strip_timestamps", True)),
    )


def load_config(path: str | Path) -> dict[str, Any]:
    p = Path(path)
    if not p.is_file():
        raise FileNotFoundError(f"config not found: {p}")
    return json.loads(p.read_text(encoding="utf-8"))


def ip_checksum(data: bytes) -> int:
    """RFC 1071 网际校验和(补码和取反)。"""
    if len(data) & 1:
        data += b"\x00"
    total = sum(struct.unpack(f"!{len(data) // 2}H", data))
    total = (total >> 16) + (total & 0xFFFF)
    total += total >> 16
    return (~total) & 0xFFFF


def tcp_checksum(src: bytes, dst: bytes, tcp_seg: bytes) -> int:
    """TCP 校验和(含 IPv4 伪头)。"""
    pseudo = struct.pack("!4s4sBBH", src, dst, 0, IP_PROTO_TCP, len(tcp_seg))
    if len(tcp_seg) & 1:
        tcp_seg += b"\x00"
    words = pseudo + tcp_seg
    if len(words) & 1:
        words += b"\x00"
    total = sum(struct.unpack(f"!{len(words) // 2}H", words))
    total = (total >> 16) + (total & 0xFFFF)
    total += total >> 16
    return (~total) & 0xFFFF


def _parse_options(data: bytes) -> list[tuple[int, bytes]]:
    """解析 TCP options 区, 返回 [(kind, value_bytes)], value 不含 kind/len。"""
    opts: list[tuple[int, bytes]] = []
    i = 0
    while i < len(data):
        kind = data[i]
        if kind == TCP_OPT_EOL:
            opts.append((kind, b""))
            break
        if kind == TCP_OPT_NOP:
            opts.append((kind, b""))
            i += 1
            continue
        if i + 1 >= len(data):
            break
        length = data[i + 1]
        if length < 2 or i + length > len(data):
            break
        opts.append((kind, data[i + 2 : i + length]))
        i += length
    return opts


def _build_options(
    parsed: list[tuple[int, bytes]], fp: Fingerprint
) -> bytes:
    """按 Windows 模板重建 options: [MSS][SACK][TS?][NOP][WS]。

    strip_timestamps=True 时剥 TS(多设备时钟分离失效);
    False 时保留原 TS 值(单设备场景, 保留 TS 更像 Windows 10)。
    顺序模板化而不是保留原序, 保证所有设备出站 SYN 布局一致。
    """
    has_sack = any(kind == TCP_OPT_SACK for kind, _ in parsed)
    ts_val: bytes | None = None
    mss_val = fp.mss
    has_mss = False
    for kind, val in parsed:
        if kind == TCP_OPT_TS and len(val) == 8:
            ts_val = val
        elif kind == TCP_OPT_MSS and len(val) == 2:
            orig = struct.unpack("!H", val)[0]
            mss_val = fp.mss if fp.set_mss else orig
            has_mss = True
    out = bytearray()
    if has_mss:
        out += bytes((TCP_OPT_MSS, 4)) + struct.pack("!H", mss_val)
    if has_sack:
        out += bytes((TCP_OPT_SACK, 2))
    if ts_val is not None and not fp.strip_timestamps:
        out += bytes((TCP_OPT_TS, 10)) + ts_val
    out += bytes((TCP_OPT_NOP,))
    out += bytes((TCP_OPT_WS, 3, fp.wscale))
    return bytes(out)


def _is_syn_only(tcp: bytes) -> bool:
    flags = tcp[13]
    return bool(flags & SYN_FLAG) and not bool(flags & ACK_FLAG)


def rewrite_syn(packet: bytes, fp: Fingerprint) -> bytes | None:
    """重写一个出向 IPv4 TCP SYN 包, 返回新包; 非 SYN/非 IPv4/畸形返回 None。

    包结构: 仅 TCP 头长度可能变化(剥 TS 10 字节), IP total length 与
    两个校验和同步重算。IP 头其余字段(TTL 由外层 nftables 管)不动。
    """
    if len(packet) < TCP_HDR_LEN + TCP_HDR_LEN:
        return None
    version = packet[0] >> 4
    if version != 4:
        return None
    ihl = (packet[0] & 0x0F) * 4
    if ihl < 20 or len(packet) < ihl + TCP_HDR_LEN:
        return None
    proto = packet[9]
    if proto != IP_PROTO_TCP:
        return None
    tcp_start = ihl
    tcp = packet[tcp_start:]
    if len(tcp) < TCP_HDR_LEN:
        return None
    if not _is_syn_only(tcp):
        return None
    data_offset = (tcp[12] >> 4) * 4
    if data_offset < TCP_HDR_LEN or data_offset > len(tcp):
        return None
    src_ip = packet[12:16]
    dst_ip = packet[16:20]

    opts_raw = tcp[TCP_HDR_LEN:data_offset]
    parsed = _parse_options(opts_raw)
    if fp.strip_timestamps:
        parsed = [(k, v) for k, v in parsed if k != TCP_OPT_TS]
    new_opts = _build_options(parsed, fp)
    if len(new_opts) % 4 != 0:
        new_opts += b"\x00" * (4 - len(new_opts) % 4)
    new_tcp_len = TCP_HDR_LEN + len(new_opts)

    new_tcp = bytearray(tcp[:TCP_HDR_LEN])
    new_tcp[12] = ((new_tcp_len // 4) << 4) | (new_tcp[12] & 0x0F)
    new_tcp[14:16] = struct.pack("!H", fp.window)
    new_tcp[16:18] = b"\x00\x00"
    payload = tcp[data_offset:]
    seg = bytes(new_tcp[:TCP_HDR_LEN]) + new_opts + payload
    cs = tcp_checksum(src_ip, dst_ip, seg)
    seg = seg[:16] + struct.pack("!H", cs) + seg[18:]

    new_ip = bytearray(packet[:ihl])
    new_ip[2:4] = struct.pack("!H", ihl + len(seg))
    new_ip[10:12] = b"\x00\x00"
    ip_cs = ip_checksum(bytes(new_ip))
    new_ip[10:12] = struct.pack("!H", ip_cs)
    return bytes(new_ip) + seg