worker: 双后端(官方netfilterqueue优先, ctypes兜底OpenWrt) + ctypes 全签名声明/verdict cast 修复
- ctypes 后端修复: 所有 lib 调用显式 argtypes/restype(未声明时 64 位指针被 c_int 截断→段错误); nfq_set_verdict2 的 buffer 参数必须 cast(c_void_p 直接传 buffer→内核坏 payload→全连接断) - 计数: rewritten/passed/failed + 退出统计打印 - 铁律注释: 改包后原包 ACCEPT / 异常一律放行
This commit is contained in:
lou
+257
-89
@@ -1,131 +1,299 @@
|
||||
"""NFQUEUE worker(软路由 OpenWrt / 本机 Linux)。
|
||||
"""NFQUEUE worker(部署在软路由/本机, 双后端自动选择)。
|
||||
|
||||
职责: 把 FORWARD(转发流量)与 OUTPUT(本机自身流量)的出向 SYN
|
||||
送进 rewrite_syn, 改完原包 ACCEPT 放行。
|
||||
职责: 把出向 SYN 送进 rewrite_syn 改包, 改完原包 ACCEPT 放行。
|
||||
铁律(9-06 + 9-09 三次断网复盘, 勿改):
|
||||
- 改包后原包 ACCEPT, 绝不走 zapret/nfqws 的 desync 重发
|
||||
- 任何异常一律放行(宁可不伪装, 不能断网); 回调抛异常 -> ACCEPT
|
||||
- 内核规则必须带 --queue-bypass(worker 不在时包直接放行)
|
||||
- 只挂 SYN 进队列(内核侧 iptables: -p tcp --syn), 用户态不额外过滤
|
||||
|
||||
实现用官方 python3-netfilterqueue 绑定(ctypes 手写版因 ABI 细节
|
||||
在真实环境 verdict 发坏包, 2026-09-09 首跑瘫痪复盘后弃用):
|
||||
- OpenWrt: opkg install python3-netfilterqueue
|
||||
- Ubuntu: apt install python3-netfilterqueue
|
||||
双后端:
|
||||
1. netfilterqueue(官方绑定, 本机等有 pip 环境)
|
||||
2. ctypes 直调 libnetfilter_queue(OpenWrt: 官方包 sstrip 掉 .so 的
|
||||
section headers, ld 无法链接 -> pip 编译必败; 但 dlopen 只需
|
||||
program headers, ctypes 可直调。9-09 本机 ctypes ABI 六坑全修:
|
||||
回调 4 参 / COPY_PACKET=2 / packet_id 取自 nfqnl_msg_packet_hdr
|
||||
首字段(宏 nfq_nfah_get_packet_id 是 static inline 不导出) /
|
||||
nfq_get_payload 是 char** 出参 / select 轮询可被 SIGTERM 中断)
|
||||
|
||||
铁律(9-06 zapret 断网 + 9-09 首跑瘫痪复盘):
|
||||
- 改包后原包 ACCEPT, 绝不走 desync 重发
|
||||
- 异常一律 ACCEPT(宁可不伪装, 不能断网)
|
||||
- verdict 由 netfilterqueue 内部处理, 不手搓 ctypes
|
||||
纯函数核心 rewrite_syn 不依赖本模块, 单测照常跑。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import ctypes
|
||||
import ctypes.util
|
||||
import os
|
||||
import select
|
||||
import signal
|
||||
import struct
|
||||
import sys
|
||||
from typing import Any
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable
|
||||
|
||||
from .rewrite import Fingerprint, fingerprint_from_config, load_config, rewrite_syn
|
||||
|
||||
NF_ACCEPT = 1
|
||||
DEBUG_NFQ = os.environ.get("DEBUG_NFQ") == "1"
|
||||
NFQNL_COPY_PACKET = 2
|
||||
|
||||
# (payload) -> bytes | None; None=原样放行, bytes=替换放行
|
||||
PacketHandler = Callable[[bytes], bytes | None]
|
||||
|
||||
try: # 后端 1: 官方绑定
|
||||
import netfilterqueue # type: ignore[no-redef]
|
||||
|
||||
_HAS_OFFICIAL = True
|
||||
except ImportError:
|
||||
_HAS_OFFICIAL = False
|
||||
|
||||
try: # 后端 2: ctypes 直调
|
||||
_lib_name = ctypes.util.find_library("netfilter_queue")
|
||||
if _lib_name is None:
|
||||
raise OSError("libnetfilter_queue not found")
|
||||
_nfq: ctypes.CDLL | None = ctypes.CDLL(_lib_name)
|
||||
except OSError as exc: # pragma: no cover
|
||||
_nfq = None
|
||||
_NFQ_IMPORT_ERR = str(exc)
|
||||
else:
|
||||
_NFQ_IMPORT_ERR = ""
|
||||
|
||||
|
||||
def _dbg(msg: str) -> None:
|
||||
if DEBUG_NFQ:
|
||||
print(f"[dbg] {msg}", file=sys.stderr, flush=True)
|
||||
def _pick_backend() -> str:
|
||||
if _HAS_OFFICIAL:
|
||||
return "official"
|
||||
if _nfq is not None:
|
||||
return "ctypes"
|
||||
raise RuntimeError(
|
||||
"no NFQUEUE backend: netfilterqueue not installed and libnetfilter_queue "
|
||||
"unavailable: " + _NFQ_IMPORT_ERR
|
||||
)
|
||||
|
||||
|
||||
class Worker:
|
||||
"""NFQUEUE 伪装 worker。用法: run() 阻塞直到 SIGTERM/SIGINT。"""
|
||||
class _OfficialBackend:
|
||||
"""官方 netfilterqueue 绑定(事件循环在 run 内驱动)。"""
|
||||
|
||||
def __init__(self, queue_num: int, fp: Fingerprint) -> None:
|
||||
if queue_num < 0 or queue_num > 65535:
|
||||
raise ValueError(f"bad queue num: {queue_num}")
|
||||
self.queue_num = queue_num
|
||||
self.fp = fp
|
||||
def __init__(self, queue_num: int, on_packet: PacketHandler) -> None:
|
||||
import netfilterqueue # noqa: PLC0415
|
||||
|
||||
self._q = netfilterqueue.NetfilterQueue()
|
||||
self._q.bind(queue_num, self._cb)
|
||||
self._on_packet = on_packet
|
||||
|
||||
def _cb(self, pkt: Any) -> None:
|
||||
try:
|
||||
from netfilterqueue import NetfilterQueue
|
||||
except ImportError as exc: # 未安装绑定的部署提示路径
|
||||
raise RuntimeError(
|
||||
"python3-netfilterqueue 未安装: "
|
||||
"Ubuntu 用 apt install python3-netfilterqueue, "
|
||||
"OpenWrt 用 opkg install python3-netfilterqueue"
|
||||
) from exc
|
||||
self._nfq = NetfilterQueue()
|
||||
new_pkt = self._on_packet(bytes(pkt.get_payload()))
|
||||
if new_pkt is None:
|
||||
pkt.accept()
|
||||
else:
|
||||
pkt.set_payload(new_pkt)
|
||||
pkt.accept()
|
||||
except Exception:
|
||||
# 铁律: 异常一律放行
|
||||
pkt.accept()
|
||||
|
||||
def run(self) -> None:
|
||||
try:
|
||||
self._q.run()
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
|
||||
def close(self) -> None:
|
||||
try:
|
||||
self._q.unbind()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
class _CtypesBackend:
|
||||
"""ctypes 直调 libnetfilter_queue(OpenWrt sstrip 环境)。
|
||||
|
||||
9-09 本机 ctypes 六坑修复清单(勿回退):
|
||||
1. 回调 4 参 (qh, nfmsg, nfad, data), 3 参会让 nfad 拿到 nfgenmsg
|
||||
2. nfq_set_mode mode=2 (NFQNL_COPY_PACKET), 0=COPY_NONE 不拷包
|
||||
3. verdict id 必须取 nfqnl_msg_packet_hdr.packet_id(首字段大端),
|
||||
static inline 宏 nfq_nfah_get_packet_id 不在 .so 导出
|
||||
4. nfq_get_payload 是 char** 出参: 传 byref(c_void_p) 再 string_at
|
||||
5. 主循环 select 轮询(阻塞 os.read 让 SIGTERM 无法中断,
|
||||
queue 残留占用 -> 新实例 nfq_create_queue failed)
|
||||
6. nfq_handle_packet 收 bytes 需 create_string_buffer 拷贝
|
||||
(c_char_p 遇内嵌 null 截断)
|
||||
"""
|
||||
|
||||
def __init__(self, queue_num: int, on_packet: PacketHandler) -> None:
|
||||
if _nfq is None:
|
||||
raise RuntimeError("libnetfilter_queue unavailable")
|
||||
self._lib = _nfq
|
||||
self._queue_num = queue_num
|
||||
self._on_packet = on_packet
|
||||
self._handle: ctypes.c_void_p = ctypes.c_void_p()
|
||||
self._queue: ctypes.c_void_p = ctypes.c_void_p()
|
||||
self._cb_ref: Any = None # 防 GC
|
||||
self.rewritten = 0
|
||||
self.passed = 0
|
||||
self.failed = 0
|
||||
self._declare_signatures()
|
||||
|
||||
def _handle(self, pkt: Any) -> None:
|
||||
"""netfilterqueue 回调。任何异常都必须放行。"""
|
||||
def _declare_signatures(self) -> None:
|
||||
"""统一声明 C 函数签名。
|
||||
|
||||
铁律: 不设 argtypes 的调用 ctypes 按 32 位 c_int 转换实参,
|
||||
64 位指针被截断 -> 段错误(9-09 软路由实测 segfault 根因)。
|
||||
必须在回调外一次性声明, 回调内只调用。
|
||||
"""
|
||||
lib = self._lib
|
||||
cb_type = ctypes.CFUNCTYPE(
|
||||
ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p,
|
||||
ctypes.c_void_p, ctypes.c_void_p,
|
||||
)
|
||||
lib.nfq_open.restype = ctypes.c_void_p
|
||||
lib.nfq_open.argtypes = []
|
||||
lib.nfq_unbind_pf.argtypes = [ctypes.c_void_p, ctypes.c_int]
|
||||
lib.nfq_bind_pf.argtypes = [ctypes.c_void_p, ctypes.c_int]
|
||||
lib.nfq_create_queue.restype = ctypes.c_void_p
|
||||
lib.nfq_create_queue.argtypes = [
|
||||
ctypes.c_void_p, ctypes.c_uint16, cb_type, ctypes.c_void_p,
|
||||
]
|
||||
lib.nfq_set_mode.argtypes = [ctypes.c_void_p, ctypes.c_uint8, ctypes.c_uint32]
|
||||
lib.nfq_set_mode.restype = ctypes.c_int
|
||||
lib.nfq_fd.argtypes = [ctypes.c_void_p]
|
||||
lib.nfq_fd.restype = ctypes.c_int
|
||||
lib.nfq_get_msg_packet_hdr.argtypes = [ctypes.c_void_p]
|
||||
lib.nfq_get_msg_packet_hdr.restype = ctypes.c_void_p
|
||||
lib.nfq_get_payload.argtypes = [
|
||||
ctypes.c_void_p, ctypes.POINTER(ctypes.c_void_p),
|
||||
]
|
||||
lib.nfq_get_payload.restype = ctypes.c_int
|
||||
lib.nfq_set_verdict2.argtypes = [
|
||||
ctypes.c_void_p, ctypes.c_uint32, ctypes.c_uint32,
|
||||
ctypes.c_uint32, ctypes.c_void_p,
|
||||
]
|
||||
lib.nfq_set_verdict2.restype = ctypes.c_int
|
||||
lib.nfq_handle_packet.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_int]
|
||||
lib.nfq_handle_packet.restype = ctypes.c_int
|
||||
lib.nfq_destroy_queue.argtypes = [ctypes.c_void_p]
|
||||
lib.nfq_close.argtypes = [ctypes.c_void_p]
|
||||
|
||||
def _cb(self, _qh: Any, _nfmsg: Any, nfad: Any, _data: Any) -> int:
|
||||
lib = self._lib
|
||||
try:
|
||||
raw = bytes(pkt.get_payload())
|
||||
new_pkt = rewrite_syn(raw, self.fp)
|
||||
_dbg(
|
||||
f"pkt len={len(raw)} rewrite="
|
||||
f"{'None' if new_pkt is None else len(new_pkt)}B"
|
||||
)
|
||||
# 取 packet_id: nfq_get_msg_packet_hdr -> 首字段 __be32 packet_id
|
||||
hdr = lib.nfq_get_msg_packet_hdr(nfad)
|
||||
if not hdr:
|
||||
return NF_ACCEPT
|
||||
packet_id = struct.unpack("!I", ctypes.string_at(hdr, 4))[0]
|
||||
# 取 payload: char** 出参
|
||||
buf_ptr = ctypes.c_void_p()
|
||||
n = lib.nfq_get_payload(nfad, ctypes.byref(buf_ptr))
|
||||
if n <= 0 or not buf_ptr.value:
|
||||
return NF_ACCEPT
|
||||
raw = ctypes.string_at(buf_ptr.value, n)
|
||||
new_pkt = self._on_packet(raw)
|
||||
if new_pkt is None:
|
||||
self.passed += 1
|
||||
pkt.accept()
|
||||
return
|
||||
pkt.set_payload(new_pkt)
|
||||
pkt.accept()
|
||||
return NF_ACCEPT
|
||||
out = ctypes.create_string_buffer(new_pkt)
|
||||
# 铁律: c_void_p 参数必须显式 cast, 直接传 buffer 对象
|
||||
# ctypes 转不成指针 -> 内核收到坏 payload -> 全连接断
|
||||
rc = lib.nfq_set_verdict2(
|
||||
self._queue, packet_id, NF_ACCEPT, len(new_pkt),
|
||||
ctypes.cast(out, ctypes.c_void_p),
|
||||
)
|
||||
if os.environ.get("DEBUG_NFQ"):
|
||||
print(
|
||||
f"[dbg] verdict id={packet_id} rc={rc} "
|
||||
f"len={len(new_pkt)} errno={ctypes.get_errno()}",
|
||||
flush=True,
|
||||
)
|
||||
self.rewritten += 1
|
||||
except Exception as exc:
|
||||
return 0
|
||||
except Exception:
|
||||
# 铁律: 异常一律放行
|
||||
self.failed += 1
|
||||
_dbg(f"EXC {exc!r}")
|
||||
try:
|
||||
pkt.accept()
|
||||
except Exception:
|
||||
pass
|
||||
return NF_ACCEPT
|
||||
|
||||
def run(self) -> None:
|
||||
"""绑定队列并进入处理循环, 直到 SIGTERM/SIGINT。"""
|
||||
stop = False
|
||||
|
||||
def _on_signal(signum: int, _frame: Any) -> None:
|
||||
nonlocal stop
|
||||
stop = True
|
||||
_dbg(f"signal {signum}, 准备退出")
|
||||
|
||||
signal.signal(signal.SIGTERM, _on_signal)
|
||||
signal.signal(signal.SIGINT, _on_signal)
|
||||
self._nfq.bind(self.queue_num, self._handle)
|
||||
print(
|
||||
f"[masquerader] queue={self.queue_num} "
|
||||
f"fingerprint=win{self.fp.window}/ws{self.fp.wscale} "
|
||||
f"strip_ts={self.fp.strip_timestamps}",
|
||||
file=sys.stderr,
|
||||
flush=True,
|
||||
)
|
||||
lib = self._lib
|
||||
h = lib.nfq_open()
|
||||
if not h:
|
||||
raise RuntimeError("nfq_open failed")
|
||||
self._handle = ctypes.c_void_p(h)
|
||||
try:
|
||||
# netfilterqueue 的 run() 阻塞; 用 run(True)? 不支持超时,
|
||||
# 改在主线程跑, 信号 handler 置位后从回调侧退出不可行,
|
||||
# 因此 SIGTERM 后直接抛 KeyboardInterrupt 由外层收尾。
|
||||
lib.nfq_unbind_pf(self._handle, 2) # AF_INET, 忽略失败
|
||||
lib.nfq_bind_pf(self._handle, 2)
|
||||
self._cb_ref = ctypes.CFUNCTYPE(
|
||||
ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p,
|
||||
ctypes.c_void_p, ctypes.c_void_p,
|
||||
)(self._cb)
|
||||
q = lib.nfq_create_queue(
|
||||
self._handle, self._queue_num, self._cb_ref, None
|
||||
)
|
||||
if not q:
|
||||
raise RuntimeError("nfq_create_queue failed")
|
||||
self._queue = ctypes.c_void_p(q)
|
||||
lib.nfq_set_mode(self._queue, NFQNL_COPY_PACKET, 0xFFFF)
|
||||
fd = lib.nfq_fd(self._handle)
|
||||
print(
|
||||
f"[masquerader] backend=ctypes queue={self._queue_num} fd={fd}",
|
||||
flush=True,
|
||||
)
|
||||
stop = False
|
||||
|
||||
def _on_signal(_sig: int, _fr: Any) -> None:
|
||||
nonlocal stop
|
||||
stop = True
|
||||
|
||||
signal.signal(signal.SIGTERM, _on_signal)
|
||||
signal.signal(signal.SIGINT, _on_signal)
|
||||
while not stop:
|
||||
self._nfq.run(1) # 非阻塞轮询, 100ms 粒度(仅当支持)
|
||||
except Exception:
|
||||
# run() 在无包时也可能抛 select 超时, 忽略并检查 stop
|
||||
pass
|
||||
r, _, _ = select.select([fd], [], [], 0.5)
|
||||
if not r:
|
||||
continue
|
||||
data = os.read(fd, 65536)
|
||||
cbuf = ctypes.create_string_buffer(data)
|
||||
lib.nfq_handle_packet(self._handle, cbuf, len(data))
|
||||
finally:
|
||||
try:
|
||||
self._nfq.unbind()
|
||||
except Exception:
|
||||
pass
|
||||
print(
|
||||
f"[masquerader] 退出: rewritten={self.rewritten} "
|
||||
f"passed={self.passed} failed={self.failed}",
|
||||
file=sys.stderr,
|
||||
flush=True,
|
||||
)
|
||||
if self._queue:
|
||||
lib.nfq_destroy_queue(self._queue)
|
||||
if self._handle:
|
||||
lib.nfq_close(self._handle)
|
||||
print(
|
||||
f"[masquerader] exit rewritten={self.rewritten} "
|
||||
f"passed={self.passed} failed={self.failed}",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
def close(self) -> None:
|
||||
pass # run() 的 finally 已清理
|
||||
|
||||
|
||||
def run_worker(queue_num: int, fp: Fingerprint) -> None:
|
||||
backend = _pick_backend()
|
||||
|
||||
def on_packet(raw: bytes) -> bytes | None:
|
||||
return rewrite_syn(raw, fp)
|
||||
|
||||
if backend == "official":
|
||||
print(f"[masquerader] backend=official queue={queue_num}", flush=True)
|
||||
w = _OfficialBackend(queue_num, on_packet)
|
||||
else:
|
||||
w = _CtypesBackend(queue_num, on_packet)
|
||||
print(
|
||||
f"[masquerader] fingerprint=win{fp.window}/ws{fp.wscale} "
|
||||
f"strip_ts={fp.strip_timestamps}",
|
||||
flush=True,
|
||||
)
|
||||
w.run()
|
||||
|
||||
|
||||
def main(argv: list[str]) -> int:
|
||||
if not argv:
|
||||
print("usage: python -m masquerader.worker <config.json>", file=sys.stderr)
|
||||
return 2
|
||||
cfg = load_config(argv[0])
|
||||
fp = fingerprint_from_config(cfg)
|
||||
queue_num = int(json.dumps(cfg.get("nfqueue", {}).get("queue_num", 100)))
|
||||
Worker(queue_num, fp).run()
|
||||
cfg: dict[str, Any] = load_config(Path(argv[0]))
|
||||
nfq_cfg: dict[str, Any] = cfg.get("nfqueue", {})
|
||||
run_worker(
|
||||
int(nfq_cfg.get("queue_num", 100)),
|
||||
fingerprint_from_config(cfg),
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user