lou
c4822207ff
- ctypes 后端修复: 所有 lib 调用显式 argtypes/restype(未声明时 64 位指针被 c_int 截断→段错误); nfq_set_verdict2 的 buffer 参数必须 cast(c_void_p 直接传 buffer→内核坏 payload→全连接断) - 计数: rewritten/passed/failed + 退出统计打印 - 铁律注释: 改包后原包 ACCEPT / 异常一律放行
302 lines
11 KiB
Python
302 lines
11 KiB
Python
"""NFQUEUE worker(部署在软路由/本机, 双后端自动选择)。
|
|
|
|
职责: 把出向 SYN 送进 rewrite_syn 改包, 改完原包 ACCEPT 放行。
|
|
铁律(9-06 + 9-09 三次断网复盘, 勿改):
|
|
- 改包后原包 ACCEPT, 绝不走 zapret/nfqws 的 desync 重发
|
|
- 任何异常一律放行(宁可不伪装, 不能断网); 回调抛异常 -> ACCEPT
|
|
- 内核规则必须带 --queue-bypass(worker 不在时包直接放行)
|
|
- 只挂 SYN 进队列(内核侧 iptables: -p tcp --syn), 用户态不额外过滤
|
|
|
|
双后端:
|
|
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 中断)
|
|
|
|
纯函数核心 rewrite_syn 不依赖本模块, 单测照常跑。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import ctypes
|
|
import ctypes.util
|
|
import os
|
|
import select
|
|
import signal
|
|
import struct
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
|
|
from .rewrite import Fingerprint, fingerprint_from_config, load_config, rewrite_syn
|
|
|
|
NF_ACCEPT = 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 _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 _OfficialBackend:
|
|
"""官方 netfilterqueue 绑定(事件循环在 run 内驱动)。"""
|
|
|
|
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:
|
|
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 _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:
|
|
# 取 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
|
|
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
|
|
return 0
|
|
except Exception:
|
|
# 铁律: 异常一律放行
|
|
self.failed += 1
|
|
return NF_ACCEPT
|
|
|
|
def run(self) -> None:
|
|
lib = self._lib
|
|
h = lib.nfq_open()
|
|
if not h:
|
|
raise RuntimeError("nfq_open failed")
|
|
self._handle = ctypes.c_void_p(h)
|
|
try:
|
|
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:
|
|
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:
|
|
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: 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
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main(sys.argv[1:]))
|