lou
efa634b574
真凶(两次"阿里云黑洞"实为系统卡死): server 每连接新建 SSLContext (每次重读证书, 内存+盘 IO 双爆点) + ThreadPoolExecutor 无界队列 (4MB 帧在内存堆积) -> 1.6G 云服 OOM -> swap 抖动 -> 系统盘 IOPS/BPS 打满 -> 整机假死(ping 丢+应用全断, 形似黑洞)。 - server: SSLContext 启动建一次复用; TLS 握手移入连接线程(不再 阻塞 accept 主循环); 在飞请求/并发连接双上限(64); 响应上限 8MB; 内存自愈(RSS 达阈值主动退出, 交由 systemd 重启释放) - proxy_dns: 通道不可用 -> 直连 fallback_dns 降级(全屋 DNS 不断), 再失败回 SERVFAIL(客户端立即切备用 DNS, 不再干等超时); 重连改 后台限频, 不再阻塞查询路径 - client: DNS 请求超时 15s -> 3.5s(让降级快速触发) - watchdog: 判据由端口/进程探测改为直连云端完成 TLS+AUTH(端口在听、 本地 DNS 探针都会给假健康信号 - 降级路径照样能答); 通道不健康即 撤 L2 劫持规则, DNS 回落 smartdns 直连; HTTP 规则排除内网段 - 配置外置: fallback_dns + 服务端防护参数 - 新增 scripts/test_dns_fallback.py(降级链单测, 2/2 PASS) QEMU OpenWrt lab 全链路验证: 降级 / 判据 / 故障演练(云挂->自动撤 规则、云回->自动重挂) 全过。
217 lines
7.0 KiB
Python
217 lines
7.0 KiB
Python
"""L2 通道客户端(软路由侧)。
|
|
|
|
职责: 维护一条到云服务器的 TLS 长连接, 多路复用 HTTP/DNS 请求帧。
|
|
- req_id 多路复用: HTTP 大响应不阻塞 DNS 小查询
|
|
- AUTH 鉴权(首帧 token), 防公网滥用
|
|
- 断线自动重连(最多 retries 次)
|
|
- 全标准库(socket/ssl/threading/queue), OpenWrt python3 可直接跑
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import queue
|
|
import socket
|
|
import ssl
|
|
import threading
|
|
import time
|
|
from typing import Any
|
|
|
|
from .proto import (
|
|
FrameReader,
|
|
TYPE_AUTH,
|
|
TYPE_DNS_REQ,
|
|
TYPE_ERROR,
|
|
TYPE_HTTP_REQ,
|
|
TYPE_PING,
|
|
TYPE_PONG,
|
|
pack_frame,
|
|
)
|
|
|
|
CONNECT_TIMEOUT = 10
|
|
REQUEST_TIMEOUT = 90
|
|
DNS_REQUEST_TIMEOUT = 3.5 # 配合 proxy_dns 降级: 云通道无响应时快速失败转直连
|
|
_ERR = object() # waiter 哨兵: 连接中断
|
|
|
|
|
|
class ChannelError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class ChannelClient:
|
|
"""到云服务器的通道客户端。线程安全: 任意线程可并发 request。"""
|
|
|
|
def __init__(
|
|
self,
|
|
host: str,
|
|
port: int,
|
|
token: str,
|
|
ca_path: str = "",
|
|
heartbeat: float = 30.0,
|
|
retries: int = 3,
|
|
) -> None:
|
|
if not host or not token:
|
|
raise ValueError("host/token required")
|
|
self.host = host
|
|
self.port = port
|
|
self.token = token
|
|
self.ca_path = ca_path
|
|
self.heartbeat = heartbeat
|
|
self.retries = retries
|
|
self._sock: socket.socket | None = None
|
|
self._reader_stop = threading.Event()
|
|
self._reader_thread: threading.Thread | None = None
|
|
self._write_lock = threading.RLock()
|
|
self._waiters_lock = threading.Lock()
|
|
self._waiters: dict[int, queue.SimpleQueue[Any]] = {}
|
|
self._next_id = 1
|
|
self._id_lock = threading.Lock()
|
|
self._alive = False
|
|
self._heartbeat_thread: threading.Thread | None = None
|
|
|
|
def _alloc_id(self) -> int:
|
|
with self._id_lock:
|
|
req_id = self._next_id
|
|
self._next_id = (self._next_id + 1) & 0xFFFFFFFF
|
|
return req_id
|
|
|
|
def _make_ssl_context(self) -> ssl.SSLContext:
|
|
ctx = ssl.create_default_context()
|
|
if self.ca_path:
|
|
ctx.load_verify_locations(cafile=self.ca_path)
|
|
ctx.check_hostname = True
|
|
else:
|
|
# 自签证书部署: 关闭对端校验, 安全靠 token 鉴权(生产应换 CA 证书)
|
|
ctx.check_hostname = False
|
|
ctx.verify_mode = ssl.CERT_NONE
|
|
return ctx
|
|
|
|
def _open_socket(self) -> None:
|
|
ctx = self._make_ssl_context()
|
|
raw = socket.create_connection(
|
|
(self.host, self.port), timeout=CONNECT_TIMEOUT
|
|
)
|
|
try:
|
|
sock = ctx.wrap_socket(raw, server_hostname=self.host)
|
|
except Exception:
|
|
raw.close()
|
|
raise
|
|
sock.settimeout(None)
|
|
self._sock = sock
|
|
self._reader_stop.clear()
|
|
self._reader_thread = threading.Thread(
|
|
target=self._reader_loop, name="channel-reader", daemon=True
|
|
)
|
|
self._reader_thread.start()
|
|
self._alive = True
|
|
|
|
def _reader_loop(self) -> None:
|
|
reader = FrameReader()
|
|
assert self._sock is not None
|
|
while not self._reader_stop.is_set():
|
|
try:
|
|
data = self._sock.recv(65536)
|
|
except OSError:
|
|
break
|
|
if not data:
|
|
break
|
|
reader.feed(data)
|
|
while True:
|
|
frame = reader.poll()
|
|
if frame is None:
|
|
break
|
|
self._dispatch(*frame)
|
|
self._fail_all()
|
|
|
|
def _dispatch(self, frame_type: int, req_id: int, payload: bytes) -> None:
|
|
if frame_type == TYPE_PING:
|
|
self._send_raw(pack_frame(TYPE_PONG, req_id))
|
|
return
|
|
with self._waiters_lock:
|
|
q = self._waiters.pop(req_id, None)
|
|
if q is not None:
|
|
if frame_type == TYPE_ERROR:
|
|
q.put(_ERR)
|
|
else:
|
|
q.put(payload)
|
|
|
|
def _send_raw(self, data: bytes) -> None:
|
|
with self._write_lock:
|
|
if self._sock is None:
|
|
raise ChannelError("not connected")
|
|
self._sock.sendall(data)
|
|
|
|
def _fail_all(self) -> None:
|
|
self._alive = False
|
|
with self._waiters_lock:
|
|
waiters = list(self._waiters.values())
|
|
self._waiters.clear()
|
|
for q in waiters:
|
|
q.put(_ERR)
|
|
|
|
def connect(self) -> None:
|
|
last_err: Exception | None = None
|
|
for attempt in range(self.retries):
|
|
try:
|
|
self._open_socket()
|
|
resp = self.request(TYPE_AUTH, self.token.encode("utf-8"), timeout=10)
|
|
if resp is None:
|
|
raise ChannelError("auth failed: connection dropped")
|
|
return
|
|
except Exception as exc: # noqa: BLE001 - 重连必须兜住所有异常
|
|
last_err = exc
|
|
self.close()
|
|
if attempt < self.retries - 1:
|
|
time.sleep(1.5 * (attempt + 1))
|
|
raise ChannelError(f"connect failed after {self.retries} tries: {last_err}")
|
|
|
|
def ensure_connected(self) -> None:
|
|
"""调用方在 request 抛 ChannelError 后重连用(内部锁防并发)。"""
|
|
with self._write_lock:
|
|
if self._sock is None or not self._alive:
|
|
self.connect()
|
|
|
|
def request(
|
|
self, frame_type: int, payload: bytes = b"", timeout: float = REQUEST_TIMEOUT
|
|
) -> bytes | None:
|
|
"""发送请求帧并等待响应; 超时/断线返回 None(ERROR 帧也返回 None)。"""
|
|
q: queue.SimpleQueue[Any] = queue.SimpleQueue()
|
|
req_id = self._alloc_id()
|
|
with self._waiters_lock:
|
|
self._waiters[req_id] = q
|
|
try:
|
|
self._send_raw(pack_frame(frame_type, req_id, payload))
|
|
except Exception:
|
|
with self._waiters_lock:
|
|
self._waiters.pop(req_id, None)
|
|
if not self._alive:
|
|
raise ChannelError("connection lost") from None
|
|
raise
|
|
try:
|
|
result = q.get(timeout=timeout)
|
|
except queue.Empty:
|
|
with self._waiters_lock:
|
|
self._waiters.pop(req_id, None)
|
|
return None
|
|
if result is _ERR:
|
|
raise ChannelError("connection lost")
|
|
return result # type: ignore[return-value]
|
|
|
|
def send_http_request(self, raw_request: bytes) -> bytes | None:
|
|
return self.request(TYPE_HTTP_REQ, raw_request)
|
|
|
|
def send_dns_query(self, raw_query: bytes) -> bytes | None:
|
|
# 短超时: DNS 是用户感知最敏感的路径, 超时即降级直连
|
|
return self.request(TYPE_DNS_REQ, raw_query, timeout=DNS_REQUEST_TIMEOUT)
|
|
|
|
def close(self) -> None:
|
|
self._reader_stop.set()
|
|
self._alive = False
|
|
with self._write_lock:
|
|
sock = self._sock
|
|
self._sock = None
|
|
if sock is not None:
|
|
try:
|
|
sock.close()
|
|
except OSError:
|
|
pass
|
|
self._fail_all()
|