feat(tcp, udp): implement stream state release after verdict and adjust flush interval logic

Port of NoBrand-Cloud/OpenGFW-RS commit 1177ff7 to Go.

EasyTier's ciphers cover only the payload: the 16-byte PeerManagerHeader
stays in the clear, so peer ids, packet type and two redundant length
fields are readable even on an encrypted mesh. The plaintext handshake
also carries the network name, which lets a rule block one mesh and
leave the rest alone (easytier.network_name). A Syn answered by a Sack
echoing the same connection id and magic, or a handshake carrying the
0xd1e1a5e1 magic, settles a flow on its own; otherwise two consistent
frames are required. Hole-punch probes are reported but never set yes.

The wg://, quic:// and wss:// transports are left to the wireguard, quic
and tls analyzers. fet does not cover the tcp:// transport because it
classifies the plaintext handshake segment and is then done.

Rules must be written as `easytier != nil && ...`: the rule engine only
exposes analyzers that have published properties.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RjQ1CTBn8j65cUZ2JKsY7L

README.ja.md

@@ -18,7 +18,7 @@ Telegram グループ: https://t.me/OpGFW
 ## 特徴
 
 - フル IP/TCP 再アセンブル、各種プロトコルアナライザー
-  - HTTP、TLS、QUIC、DNS、SSH、SOCKS4/5、WireGuard、OpenVPN、その他多数
+  - HTTP、TLS、QUIC、DNS、SSH、SOCKS4/5、WireGuard、OpenVPN、Minecraft、EasyTier、その他多数
   - Shadowsocks、VMess の「完全に暗号化されたトラフィック」の検出など (https://gfw.report/publications/usenixsecurity23/en/)
   - Trojan プロキシプロトコルの検出
   - [WIP] 機械学習に基づくトラフィック分類

README.md

@@ -24,7 +24,7 @@ Telegram group: https://t.me/OpGFW
 ## Features
 
 - Full IP/TCP reassembly, various protocol analyzers
-  - HTTP, TLS, QUIC, DNS, SSH, SOCKS4/5, WireGuard, OpenVPN, Minecraft, and many more to come
+  - HTTP, TLS, QUIC, DNS, SSH, SOCKS4/5, WireGuard, OpenVPN, Minecraft, EasyTier, and many more to come
   - "Fully encrypted traffic" detection for Shadowsocks, VMess,
     etc. (https://gfw.report/publications/usenixsecurity23/en/)
   - Trojan (proxy protocol) detection

README.zh.md

@@ -15,10 +15,13 @@ Telegram 群组： https://t.me/OpGFW
 > [!CAUTION]
 > 本项目仍处于早期开发阶段。测试时自行承担风险。我们正在寻求贡献者一起完善本项目。
 
+> [!CAUTION]
+> 此版本为 LoliaFRP 二次开发版本，加入了更多的识别规则，如果您正在寻找原版请移步 original 分支
+
 ## 功能
 
 - 完整的 IP/TCP 重组，各种协议解析器
-  - HTTP, TLS, QUIC, DNS, SSH, SOCKS4/5, WireGuard, OpenVPN, Minecraft, 更多协议正在开发中
+  - HTTP, TLS, QUIC, DNS, SSH, SOCKS4/5, WireGuard, OpenVPN, Minecraft, EasyTier, 更多协议正在开发中
   - Shadowsocks, VMess 等 "全加密流量" 检测 (https://gfw.report/publications/usenixsecurity23/zh/)
   - Trojan 协议检测
   - [开发中] 基于机器学习的流量分类
@@ -34,6 +37,7 @@ Telegram 群组： https://t.me/OpGFW
 ## 本仓库内的文档
 
 - [Minecraft 分析器](docs/minecraft.zh.md)
+- [EasyTier 分析器](docs/easytier.zh.md)
 - [TCP 窗口操纵规避与检测](docs/tcp-window-evasion.zh.md)
 
 ## 使用场景

analyzer/udp/easytier.go

@@ -0,0 +1,766 @@
+package udp
+
+import (
+	"encoding/binary"
+	"maps"
+	"unicode"
+	"unicode/utf8"
+
+	"github.com/apernet/OpenGFW/analyzer"
+	"github.com/apernet/OpenGFW/analyzer/utils"
+)
+
+// EasyTierAnalyzer 识别 EasyTier（https://github.com/EasyTier/EasyTier）
+type EasyTierAnalyzer struct{}
+
+var (
+	_ analyzer.UDPAnalyzer = (*EasyTierAnalyzer)(nil)
+	_ analyzer.TCPAnalyzer = (*EasyTierAnalyzer)(nil)
+	_ analyzer.UDPStream   = (*easyTierUDPStream)(nil)
+	_ analyzer.TCPStream   = (*easyTierTCPStream)(nil)
+)
+
+const (
+	// easyTierUDPTunnelHeaderSize UDPTunnelHeader：conn_id u32、msg_type u8、padding u8、len u16，全部小端
+	easyTierUDPTunnelHeaderSize = 8
+	// easyTierTCPTunnelHeaderSize TCPTunnelHeader：一个小端 u32 帧体长度
+	easyTierTCPTunnelHeaderSize = 4
+	// easyTierPeerManagerHeaderSize PeerManagerHeader：from u32、to u32、type u8、flags u8、fwd u8、reserved u8、len u32
+	easyTierPeerManagerHeaderSize = 16
+	// easyTierAEADTailSize AEAD 加密追加的 16 字节 tag 加 12 字节 nonce，头里声明的长度不包含它
+	easyTierAEADTailSize = 28
+
+	// easyTierMaxFrameBody 帧体长度上限。上游自己的限制更低（TCP 隧道 2000，字节流隧道 4096，
+	// UDP 会话 8192），这里取最大的那个，精确的约束交给长度不变量
+	easyTierMaxFrameBody = 8192
+	// easyTierMaxForwardCounter 路由在 >7 时丢弃后再自增，所以线上最大能见到 8
+	easyTierMaxForwardCounter = 8
+
+	// easyTierHandshakeMagic HandshakeRequest.magic，见上游 peers/conn/peer_conn.rs
+	easyTierHandshakeMagic uint64 = 0xd1e1a5e1
+	// easyTierSecretDigestLen NetworkSecretDigest = [u8; 32]
+	easyTierSecretDigestLen = 32
+	// easyTierNoiseEphemeralLen Noise XX 首条消息以 32 字节临时公钥开头
+	easyTierNoiseEphemeralLen = 32
+
+	// easyTierFramesForYes 没有握手可指的情况下，置 yes 需要攒够的自洽帧数
+	easyTierFramesForYes = 2
+	// easyTierUDPInvalidCountThreshold UDP 流上连续无法解析的数据报容忍数，与 wireguard 分析器一致
+	easyTierUDPInvalidCountThreshold = 4
+	// easyTierPacketLimit 超过这个包数仍未定性就结束该流，与 openvpn 分析器一致
+	easyTierPacketLimit = 256
+)
+
+// easyTierHolePunchBodyLens 上游打洞包的两种体长：连通性路径 16 字节，UDP 会话层 32 字节。
+// 体本身是随机数据，长度是唯一能看的东西。
+var easyTierHolePunchBodyLens = [...]int{16, 32}
+
+// UDP 隧道消息类型（UdpPacketType）
+const (
+	easyTierMsgSyn         = 1
+	easyTierMsgSack        = 2
+	easyTierMsgData        = 3
+	easyTierMsgFin         = 4
+	easyTierMsgHolePunch   = 5
+	easyTierMsgV4HolePunch = 6
+	easyTierMsgV6HolePunch = 7
+)
+
+// 需要特殊处理的 PeerManagerHeader packet_type（PacketType）
+const (
+	easyTierPktHandshake = 2
+	easyTierPktNoiseMsg1 = 13
+	easyTierPktNoiseMsg3 = 15
+)
+
+// PeerManagerHeaderFlags
+const (
+	easyTierFlagEncrypted    = 0b0000_0001
+	easyTierFlagLatencyFirst = 0b0000_0010
+	easyTierFlagExitNode     = 0b0000_0100
+	easyTierFlagNoProxy      = 0b0000_1000
+	easyTierFlagCompressed   = 0b0001_0000
+	// easyTierFlagLiveness 两个存活探测标志，是仅有的让 reserved 字节有意义的标志
+	easyTierFlagLiveness = 0b1010_0000
+)
+
+func (a *EasyTierAnalyzer) Name() string {
+	return "easytier"
+}
+
+func (a *EasyTierAnalyzer) Limit() int {
+	// 定长帧协议，按字节截断会让帧读取失步；用 easyTierPacketLimit 约束开销
+	return 0
+}
+
+func (a *EasyTierAnalyzer) NewUDP(info analyzer.UDPInfo, logger analyzer.Logger) analyzer.UDPStream {
+	return newEasyTierUDPStream(logger)
+}
+
+func (a *EasyTierAnalyzer) NewTCP(info analyzer.TCPInfo, logger analyzer.Logger) analyzer.TCPStream {
+	return newEasyTierTCPStream(logger)
+}
+
+// easyTierEvidence 一帧能证明多少
+type easyTierEvidence int
+
+const (
+	// easyTierEvidenceWeak 格式正确但单独不足为据：体是随机数据的短定长数据报。字段照发，不推进 yes
+	easyTierEvidenceWeak easyTierEvidence = iota
+	// easyTierEvidenceFrame 带 PeerManagerHeader 且通过全部校验的帧
+	easyTierEvidenceFrame
+	// easyTierEvidenceConclusive 匹配上的握手，别的东西不会碰巧产生它
+	easyTierEvidenceConclusive
+)
+
+// easyTierCommon 对外发布的属性，加上 yes 背后的计数。
+//
+// 描述最近一帧的字段每次都会覆盖；从握手里读出来的字段是粘性的，
+// 因为网络名只发一次，而按 mesh 名写的规则必须在握手滚过去之后还能用。
+type easyTierCommon struct {
+	props         analyzer.PropMap
+	frames        int
+	conclusive    bool
+	packetCount   int
+	rxPacketCount int
+	txPacketCount int
+}
+
+func newEasyTierCommon(transport string) easyTierCommon {
+	return easyTierCommon{
+		props: analyzer.PropMap{
+			"yes":        false,
+			"transport":  transport,
+			"tx_pkt_cnt": 0,
+			"rx_pkt_cnt": 0,
+		},
+	}
+}
+
+func (c *easyTierCommon) countPacket(rev bool) {
+	c.packetCount++
+	if rev {
+		c.rxPacketCount++
+		c.props["rx_pkt_cnt"] = c.rxPacketCount
+	} else {
+		c.txPacketCount++
+		c.props["tx_pkt_cnt"] = c.txPacketCount
+	}
+}
+
+func (c *easyTierCommon) record(evidence easyTierEvidence) {
+	switch evidence {
+	case easyTierEvidenceConclusive:
+		c.conclusive = true
+	case easyTierEvidenceFrame:
+		c.frames++
+	}
+	c.props["yes"] = c.conclusive || c.frames >= easyTierFramesForYes
+}
+
+func (c *easyTierCommon) set(key string, value interface{}) {
+	c.props[key] = value
+}
+
+// setSticky 只写一次、之后不再覆盖，用于从握手里读出的字段
+func (c *easyTierCommon) setSticky(key string, value interface{}) {
+	if _, ok := c.props[key]; !ok {
+		c.props[key] = value
+	}
+}
+
+// update 发布一份属性快照。引擎会持有这份 map，所以必须复制，不能把内部的交出去
+func (c *easyTierCommon) update() *analyzer.PropUpdate {
+	return &analyzer.PropUpdate{
+		Type: analyzer.PropUpdateReplace,
+		M:    maps.Clone(c.props),
+	}
+}
+
+func (c *easyTierCommon) exhausted() bool {
+	return c.packetCount > easyTierPacketLimit
+}
+
+// applyHeader 记录 PeerManagerHeader 的字段，并返回这一帧能证明多少
+func (c *easyTierCommon) applyHeader(h easyTierPeerManagerHeader, body []byte) easyTierEvidence {
+	c.set("packet_type", h.packetType)
+	c.set("packet_type_name", easyTierPacketTypeName(h.packetType))
+	c.set("from_peer_id", h.fromPeerID)
+	c.set("to_peer_id", h.toPeerID)
+	c.set("forward_counter", h.forwardCounter)
+	c.set("payload_len", h.length)
+	c.set("encrypted", h.flags&easyTierFlagEncrypted != 0)
+	c.set("compressed", h.flags&easyTierFlagCompressed != 0)
+	c.set("latency_first", h.flags&easyTierFlagLatencyFirst != 0)
+	c.set("exit_node", h.flags&easyTierFlagExitNode != 0)
+	c.set("no_proxy", h.flags&easyTierFlagNoProxy != 0)
+
+	payload := body[easyTierPeerManagerHeaderSize:]
+	switch h.packetType {
+	case easyTierPktHandshake:
+		return c.applyPlainHandshake(payload)
+	case easyTierPktNoiseMsg1:
+		c.setSticky("secure", true)
+		c.applyNoiseMsg1(payload)
+		// 只有临时公钥和一段 protobuf，随机流也能凑出来，不算确证
+		return easyTierEvidenceFrame
+	case easyTierPktNoiseMsg3:
+		c.setSticky("secure", true)
+		return easyTierEvidenceFrame
+	default:
+		return easyTierEvidenceFrame
+	}
+}
+
+// applyPlainHandshake 明文 HandshakeRequest，magic 对上就定性
+func (c *easyTierCommon) applyPlainHandshake(payload []byte) easyTierEvidence {
+	hs, ok := easyTierParseHandshakeRequest(payload)
+	if !ok {
+		return easyTierEvidenceFrame
+	}
+	c.setSticky("magic_ok", true)
+	c.setSticky("secure", false)
+	if hs.hasVersion {
+		c.setSticky("version", hs.version)
+	}
+	if hs.networkName != "" {
+		c.setSticky("network_name", hs.networkName)
+	}
+	return easyTierEvidenceConclusive
+}
+
+// applyNoiseMsg1 Noise XX 的首条消息此时还没有密钥，protobuf 载荷是明文。
+// 机会性地读：解析不干净就什么都不说，不猜。
+func (c *easyTierCommon) applyNoiseMsg1(payload []byte) {
+	if len(payload) <= easyTierNoiseEphemeralLen {
+		return
+	}
+	msg, ok := easyTierParseNoiseMsg1(payload[easyTierNoiseEphemeralLen:])
+	if !ok {
+		return
+	}
+	if msg.hasVersion {
+		c.setSticky("version", msg.version)
+	}
+	if msg.networkName != "" {
+		c.setSticky("network_name", msg.networkName)
+	}
+}
+
+type easyTierPeerManagerHeader struct {
+	fromPeerID     uint32
+	toPeerID       uint32
+	packetType     uint8
+	flags          uint8
+	forwardCounter uint8
+	length         uint32
+}
+
+// easyTierReadPeerManagerHeader 读取并校验帧体开头的 PeerManagerHeader。
+//
+// 声明的载荷长度要和实际收到的帧体对照。AEAD 加密会追加 28 字节尾部而不改声明长度，
+// 压缩会缩短载荷也不改声明长度，所以只有两个标志都没置位时关系才是精确相等。
+func easyTierReadPeerManagerHeader(body []byte) (easyTierPeerManagerHeader, bool) {
+	var h easyTierPeerManagerHeader
+	if len(body) < easyTierPeerManagerHeaderSize {
+		return h, false
+	}
+	packetType := body[8]
+	if !easyTierIsKnownPacketType(packetType) {
+		return h, false
+	}
+	flags := body[9]
+	forwardCounter := body[10]
+	if forwardCounter > easyTierMaxForwardCounter {
+		return h, false
+	}
+	// reserved 只在存活探测时装 token，其余时候恒为 0
+	if body[11] != 0 && flags&easyTierFlagLiveness == 0 {
+		return h, false
+	}
+
+	length := binary.LittleEndian.Uint32(body[12:16])
+	if length > easyTierMaxFrameBody {
+		return h, false
+	}
+	wire := uint32(len(body) - easyTierPeerManagerHeaderSize)
+	var ok bool
+	switch {
+	case flags&easyTierFlagCompressed != 0:
+		// 压缩器只在能省下自己那 1 字节尾时才压
+		ok = wire >= 1 && wire <= length+easyTierAEADTailSize
+	case flags&easyTierFlagEncrypted != 0:
+		// XOR 不改长度；AEAD 追加尾部
+		ok = wire == length || wire == length+easyTierAEADTailSize
+	default:
+		ok = wire == length
+	}
+	if !ok {
+		return h, false
+	}
+
+	return easyTierPeerManagerHeader{
+		fromPeerID:     binary.LittleEndian.Uint32(body[0:4]),
+		toPeerID:       binary.LittleEndian.Uint32(body[4:8]),
+		packetType:     packetType,
+		flags:          flags,
+		forwardCounter: forwardCounter,
+		length:         length,
+	}, true
+}
+
+func easyTierIsKnownPacketType(packetType uint8) bool {
+	return packetType >= 1 && packetType <= 21
+}
+
+func easyTierPacketTypeName(packetType uint8) string {
+	switch packetType {
+	case 1:
+		return "data"
+	case 2:
+		return "handshake"
+	case 3:
+		return "route_packet"
+	case 4:
+		return "ping"
+	case 5:
+		return "pong"
+	case 6:
+		return "ta_rpc"
+	case 7:
+		return "route"
+	case 8:
+		return "rpc_req"
+	case 9:
+		return "rpc_resp"
+	case 10:
+		return "foreign_network_packet"
+	case 11:
+		return "kcp_src"
+	case 12:
+		return "kcp_dst"
+	case 13:
+		return "noise_handshake_msg1"
+	case 14:
+		return "noise_handshake_msg2"
+	case 15:
+		return "noise_handshake_msg3"
+	case 16:
+		return "quic_src"
+	case 17:
+		return "quic_dst"
+	case 18:
+		return "data_with_kcp_src_modified"
+	case 19:
+		return "data_with_quic_src_modified"
+	case 20:
+		return "relay_handshake"
+	case 21:
+		return "relay_handshake_ack"
+	default:
+		return "unknown"
+	}
+}
+
+func easyTierMsgTypeName(msgType uint8) string {
+	switch msgType {
+	case easyTierMsgSyn:
+		return "syn"
+	case easyTierMsgSack:
+		return "sack"
+	case easyTierMsgData:
+		return "data"
+	case easyTierMsgFin:
+		return "fin"
+	case easyTierMsgHolePunch:
+		return "hole_punch"
+	case easyTierMsgV4HolePunch:
+		return "v4_hole_punch"
+	case easyTierMsgV6HolePunch:
+		return "v6_hole_punch"
+	default:
+		return "unknown"
+	}
+}
+
+// easyTierPendingSyn 一个等着 Sack 的 Syn
+type easyTierPendingSyn struct {
+	connID uint32
+	magic  uint64
+	rev    bool
+}
+
+type easyTierUDPStream struct {
+	logger       analyzer.Logger
+	common       easyTierCommon
+	invalidCount int
+	pendingSyn   *easyTierPendingSyn
+}
+
+func newEasyTierUDPStream(logger analyzer.Logger) *easyTierUDPStream {
+	return &easyTierUDPStream{
+		logger: logger,
+		common: newEasyTierCommon("udp"),
+	}
+}
+
+func (s *easyTierUDPStream) Feed(rev bool, data []byte) (u *analyzer.PropUpdate, done bool) {
+	// EasyTier 把 STUN 复用在同一个 socket 上，并且在自己的帧解析之前先分流 STUN；
+	// 照做：既不算证据，也不算错误
+	if easyTierIsSTUNDatagram(data) {
+		s.common.packetCount++
+		return nil, s.common.exhausted()
+	}
+
+	evidence, ok := s.parseDatagram(rev, data)
+	if !ok {
+		s.invalidCount++
+		return nil, s.invalidCount >= easyTierUDPInvalidCountThreshold
+	}
+
+	s.invalidCount = 0
+	s.common.countPacket(rev)
+	s.common.record(evidence)
+	return s.common.update(), s.common.exhausted()
+}
+
+func (s *easyTierUDPStream) Close(limited bool) *analyzer.PropUpdate {
+	return nil
+}
+
+func (s *easyTierUDPStream) parseDatagram(rev bool, data []byte) (easyTierEvidence, bool) {
+	if len(data) < easyTierUDPTunnelHeaderSize {
+		return 0, false
+	}
+	connID := binary.LittleEndian.Uint32(data[0:4])
+	msgType := data[4]
+	// padding 只为对齐长度字段而存在
+	if data[5] != 0 {
+		return 0, false
+	}
+	declared := int(binary.LittleEndian.Uint16(data[6:8]))
+	body := data[easyTierUDPTunnelHeaderSize:]
+	// 上游要求这两者相等，否则整个数据报直接丢弃
+	if declared != len(body) {
+		return 0, false
+	}
+
+	var evidence easyTierEvidence
+	switch msgType {
+	case easyTierMsgSyn:
+		if len(body) != 8 {
+			return 0, false
+		}
+		s.pendingSyn = &easyTierPendingSyn{connID: connID, magic: binary.LittleEndian.Uint64(body), rev: rev}
+		evidence = easyTierEvidenceWeak
+	case easyTierMsgSack:
+		// 反向回来、同一 conn_id、同一随机 64 位 magic 的 Sack，别的东西做不到
+		if len(body) != 8 {
+			return 0, false
+		}
+		magic := binary.LittleEndian.Uint64(body)
+		syn := s.pendingSyn
+		if syn != nil && syn.rev != rev && syn.connID == connID && syn.magic == magic {
+			evidence = easyTierEvidenceConclusive
+		} else {
+			evidence = easyTierEvidenceWeak
+		}
+	case easyTierMsgData:
+		h, ok := easyTierReadPeerManagerHeader(body)
+		if !ok {
+			return 0, false
+		}
+		evidence = s.common.applyHeader(h, body)
+	case easyTierMsgFin:
+		if len(body) != 0 {
+			return 0, false
+		}
+		evidence = easyTierEvidenceWeak
+	case easyTierMsgHolePunch:
+		// 打洞探测是定长随机体，往可能根本没跑 EasyTier 的地址上喷，
+		// 不足以下结论，但 msg_type_name 照样说明它是什么
+		if !easyTierIsHolePunchBodyLen(len(body)) {
+			return 0, false
+		}
+		evidence = easyTierEvidenceWeak
+	case easyTierMsgV4HolePunch:
+		// 两种定向变体在 conn_id 里回显目标端口。上游只从回环地址接受它们，主机间线上少见
+		if len(body) != 6 || connID != uint32(binary.LittleEndian.Uint16(body[4:6])) {
+			return 0, false
+		}
+		evidence = easyTierEvidenceWeak
+	case easyTierMsgV6HolePunch:
+		if len(body) != 38 || connID != uint32(binary.LittleEndian.Uint16(body[16:18])) {
+			return 0, false
+		}
+		evidence = easyTierEvidenceWeak
+	default:
+		return 0, false
+	}
+
+	s.common.set("conn_id", connID)
+	s.common.set("msg_type", msgType)
+	s.common.set("msg_type_name", easyTierMsgTypeName(msgType))
+	return evidence, true
+}
+
+func easyTierIsHolePunchBodyLen(n int) bool {
+	for _, l := range easyTierHolePunchBodyLens {
+		if n == l {
+			return true
+		}
+	}
+	return false
+}
+
+// easyTierIsSTUNDatagram 和 EasyTier 自己先做的判断一样：
+// 它的长度字段位置上是 STUN magic cookie，且前两位为零
+func easyTierIsSTUNDatagram(data []byte) bool {
+	return len(data) >= easyTierUDPTunnelHeaderSize &&
+		data[4] == 0x21 && data[5] == 0x12 && data[6] == 0xa4 && data[7] == 0x42 &&
+		data[0]&0xc0 == 0
+}
+
+type easyTierTCPStream struct {
+	logger  analyzer.Logger
+	common  easyTierCommon
+	reqBuf  utils.ByteBuffer
+	respBuf utils.ByteBuffer
+}
+
+func newEasyTierTCPStream(logger analyzer.Logger) *easyTierTCPStream {
+	return &easyTierTCPStream{
+		logger: logger,
+		common: newEasyTierCommon("tcp"),
+	}
+}
+
+// easyTierFrameStatus 从一个方向的缓冲里读一帧的结果
+type easyTierFrameStatus int
+
+const (
+	// easyTierFrameRead 头已校验通过，帧体可以发布
+	easyTierFrameRead easyTierFrameStatus = iota
+	// easyTierFrameNeedMore 声明的帧体还没到齐
+	easyTierFrameNeedMore
+	// easyTierFrameNo 不可能是 EasyTier 的帧
+	easyTierFrameNo
+)
+
+func (s *easyTierTCPStream) Feed(rev, start, end bool, skip int, data []byte) (u *analyzer.PropUpdate, done bool) {
+	// 重组有空洞时帧边界无从推断
+	if skip != 0 {
+		return nil, true
+	}
+	if len(data) == 0 {
+		return nil, false
+	}
+
+	buf := &s.reqBuf
+	if rev {
+		buf = &s.respBuf
+	}
+	buf.Append(data)
+
+	updated := false
+	for {
+		h, body, status := easyTierReadFrame(buf)
+		switch status {
+		case easyTierFrameRead:
+			evidence := s.common.applyHeader(h, body)
+			s.common.countPacket(rev)
+			s.common.record(evidence)
+			updated = true
+			if s.common.exhausted() {
+				return s.common.update(), true
+			}
+		case easyTierFrameNeedMore:
+			if updated {
+				return s.common.update(), false
+			}
+			return nil, false
+		case easyTierFrameNo:
+			if updated {
+				return s.common.update(), true
+			}
+			return nil, true
+		}
+	}
+}
+
+func (s *easyTierTCPStream) Close(limited bool) *analyzer.PropUpdate {
+	s.reqBuf.Reset()
+	s.respBuf.Reset()
+	return nil
+}
+
+// easyTierReadFrame 帧结构是 [u32 LE 帧体长度][帧体]，帧体总是以 PeerManagerHeader 开头，
+// 上游会把更短的帧体按 "body too short" 拒绝。
+func easyTierReadFrame(buf *utils.ByteBuffer) (easyTierPeerManagerHeader, []byte, easyTierFrameStatus) {
+	var h easyTierPeerManagerHeader
+	bodyLen, ok := buf.GetUint32(true, false)
+	if !ok {
+		return h, nil, easyTierFrameNeedMore
+	}
+	if bodyLen < easyTierPeerManagerHeaderSize || bodyLen > easyTierMaxFrameBody {
+		return h, nil, easyTierFrameNo
+	}
+	total := easyTierTCPTunnelHeaderSize + int(bodyLen)
+	frame, ok := buf.Get(total, false)
+	if !ok {
+		return h, nil, easyTierFrameNeedMore
+	}
+	body := frame[easyTierTCPTunnelHeaderSize:]
+	h, ok = easyTierReadPeerManagerHeader(body)
+	if !ok {
+		return h, nil, easyTierFrameNo
+	}
+	// 只在帧确认合法后才消费；被拒绝的流反正要结束，留着字节让缓冲状态更好推理
+	buf.Skip(total)
+	return h, body, easyTierFrameRead
+}
+
+// easyTierHandshakeInfo 从握手里读出来的东西。networkName 为空表示没有可用的名字
+type easyTierHandshakeInfo struct {
+	version     uint64
+	hasVersion  bool
+	networkName string
+}
+
+// easyTierParseHandshakeRequest 解析 HandshakeRequest { magic = 1, my_peer_id = 2,
+// version = 3, features = 4, network_name = 5, network_secret_digest = 6 }。
+//
+// 只有 magic 存在且正确才返回 true，调用方可以把 true 当作证据而不是提示。
+func easyTierParseHandshakeRequest(payload []byte) (easyTierHandshakeInfo, bool) {
+	var info easyTierHandshakeInfo
+	var magic uint64
+	hasMagic := false
+	digestLen := -1
+
+	ok := easyTierForEachProtoField(payload, func(f easyTierProtoField) {
+		switch {
+		case f.isVarint && f.number == 1:
+			magic, hasMagic = f.varint, true
+		case f.isVarint && f.number == 3:
+			info.version, info.hasVersion = f.varint, true
+		case f.isBytes && f.number == 5:
+			info.networkName, _ = easyTierPrintableUTF8(f.bytes)
+		case f.isBytes && f.number == 6:
+			digestLen = len(f.bytes)
+		}
+	})
+	if !ok || !hasMagic || magic != easyTierHandshakeMagic {
+		return easyTierHandshakeInfo{}, false
+	}
+	// 上游会拒绝 digest 宽度不对的握手
+	if digestLen >= 0 && digestLen != easyTierSecretDigestLen {
+		return easyTierHandshakeInfo{}, false
+	}
+	return info, true
+}
+
+// easyTierParseNoiseMsg1 解析 PeerConnNoiseMsg1Pb { version = 1, a_network_name = 2, ... }。
+//
+// 这里没有 magic 可查，所以消息必须完整解析并声明一个合理的版本，否则返回 false。
+func easyTierParseNoiseMsg1(payload []byte) (easyTierHandshakeInfo, bool) {
+	var info easyTierHandshakeInfo
+
+	ok := easyTierForEachProtoField(payload, func(f easyTierProtoField) {
+		switch {
+		case f.isVarint && f.number == 1:
+			info.version, info.hasVersion = f.varint, true
+		case f.isBytes && f.number == 2:
+			info.networkName, _ = easyTierPrintableUTF8(f.bytes)
+		}
+	})
+	// 协议版本从有这个字段起就一直是 1，很大的值说明这些字节不是这条消息
+	if !ok || !info.hasVersion || info.version > 16 {
+		return easyTierHandshakeInfo{}, false
+	}
+	return info, true
+}
+
+type easyTierProtoField struct {
+	number   uint32
+	isVarint bool
+	varint   uint64
+	isBytes  bool
+	bytes    []byte
+}
+
+// easyTierForEachProtoField 遍历 protobuf 字段，跳过调用方不认识的字段，
+// 遇到编码站不住脚（截断、字段号为 0、这两条消息从不使用的 wire type）时返回 false。
+func easyTierForEachProtoField(data []byte, fn func(easyTierProtoField)) bool {
+	pos := 0
+	for pos < len(data) {
+		tag, ok := easyTierReadVarint(data, &pos)
+		if !ok {
+			return false
+		}
+		number := uint32(tag >> 3)
+		if number == 0 {
+			return false
+		}
+		switch tag & 0x7 {
+		case 0:
+			value, ok := easyTierReadVarint(data, &pos)
+			if !ok {
+				return false
+			}
+			fn(easyTierProtoField{number: number, isVarint: true, varint: value})
+		case 2:
+			length, ok := easyTierReadVarint(data, &pos)
+			if !ok || length > uint64(len(data)-pos) {
+				return false
+			}
+			end := pos + int(length)
+			fn(easyTierProtoField{number: number, isBytes: true, bytes: data[pos:end]})
+			pos = end
+		case 5:
+			if len(data)-pos < 4 {
+				return false
+			}
+			pos += 4
+		case 1:
+			if len(data)-pos < 8 {
+				return false
+			}
+			pos += 8
+		default:
+			// group 类型，两条消息都不用
+			return false
+		}
+	}
+	return true
+}
+
+func easyTierReadVarint(data []byte, pos *int) (uint64, bool) {
+	var value uint64
+	for i := 0; i < 10; i++ {
+		if *pos >= len(data) {
+			return 0, false
+		}
+		b := data[*pos]
+		*pos++
+		value |= uint64(b&0x7f) << (7 * i)
+		if b&0x80 == 0 {
+			return value, true
+		}
+	}
+	return 0, false
+}
+
+// easyTierPrintableUTF8 网络名是运营者自己起的文本。拒绝控制字符，
+// 免得一段碰巧的字节最后被引用进日志或审计记录里。
+func easyTierPrintableUTF8(b []byte) (string, bool) {
+	if len(b) == 0 || !utf8.Valid(b) {
+		return "", false
+	}
+	s := string(b)
+	for _, r := range s {
+		if unicode.IsControl(r) {
+			return "", false
+		}
+	}
+	return s, true
+}

analyzer/udp/easytier_test.go

@@ -0,0 +1,562 @@
+package udp
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"testing"
+
+	"github.com/apernet/OpenGFW/analyzer"
+	"github.com/apernet/OpenGFW/analyzer/tcp"
+)
+
+func etVarint(v uint64) []byte {
+	var out []byte
+	for {
+		b := byte(v & 0x7f)
+		v >>= 7
+		if v != 0 {
+			b |= 0x80
+		}
+		out = append(out, b)
+		if v == 0 {
+			return out
+		}
+	}
+}
+
+func etProtoVarint(field uint32, v uint64) []byte {
+	return append(etVarint(uint64(field)<<3), etVarint(v)...)
+}
+
+func etProtoBytes(field uint32, v []byte) []byte {
+	out := etVarint(uint64(field)<<3 | 2)
+	out = append(out, etVarint(uint64(len(v)))...)
+	return append(out, v...)
+}
+
+// etHandshakeRequest HandshakeRequest，按 send_handshake 的构造方式
+func etHandshakeRequest(peerID uint32, networkName string) []byte {
+	var out []byte
+	out = append(out, etProtoVarint(1, easyTierHandshakeMagic)...)
+	out = append(out, etProtoVarint(2, uint64(peerID))...)
+	out = append(out, etProtoVarint(3, 1)...)
+	out = append(out, etProtoBytes(4, []byte("liveness-echo-v1"))...)
+	out = append(out, etProtoBytes(5, []byte(networkName))...)
+	out = append(out, etProtoBytes(6, make([]byte, easyTierSecretDigestLen))...)
+	return out
+}
+
+// etNoiseMsg1 Noise XX 先写的临时公钥，后面跟 PeerConnNoiseMsg1Pb
+func etNoiseMsg1(networkName string) []byte {
+	out := bytes.Repeat([]byte{0xab}, easyTierNoiseEphemeralLen)
+	out = append(out, etProtoVarint(1, 1)...)
+	return append(out, etProtoBytes(2, []byte(networkName))...)
+}
+
+// etBody 帧体：16 字节 PeerManagerHeader 加载荷
+func etBody(packetType uint8, from, to uint32, payload []byte) []byte {
+	return etBodyRaw(packetType, from, to, 0, 1, 0, uint32(len(payload)), payload)
+}
+
+func etBodyRaw(packetType uint8, from, to uint32, flags, forwardCounter, reserved uint8, declaredLen uint32, payload []byte) []byte {
+	out := make([]byte, 0, easyTierPeerManagerHeaderSize+len(payload))
+	out = binary.LittleEndian.AppendUint32(out, from)
+	out = binary.LittleEndian.AppendUint32(out, to)
+	out = append(out, packetType, flags, forwardCounter, reserved)
+	out = binary.LittleEndian.AppendUint32(out, declaredLen)
+	return append(out, payload...)
+}
+
+// etDatagram 8 字节 UDPTunnelHeader 加帧体
+func etDatagram(connID uint32, msgType uint8, body []byte) []byte {
+	out := make([]byte, 0, easyTierUDPTunnelHeaderSize+len(body))
+	out = binary.LittleEndian.AppendUint32(out, connID)
+	out = append(out, msgType, 0)
+	out = binary.LittleEndian.AppendUint16(out, uint16(len(body)))
+	return append(out, body...)
+}
+
+func etSyn(connID uint32, magic uint64) []byte {
+	return etDatagram(connID, easyTierMsgSyn, binary.LittleEndian.AppendUint64(nil, magic))
+}
+
+func etSack(connID uint32, magic uint64) []byte {
+	return etDatagram(connID, easyTierMsgSack, binary.LittleEndian.AppendUint64(nil, magic))
+}
+
+// etSegment 4 字节 TCPTunnelHeader 加帧体
+func etSegment(body []byte) []byte {
+	out := binary.LittleEndian.AppendUint32(nil, uint32(len(body)))
+	return append(out, body...)
+}
+
+func etUDP() analyzer.UDPStream {
+	return (&EasyTierAnalyzer{}).NewUDP(analyzer.UDPInfo{}, nil)
+}
+
+func etTCP() analyzer.TCPStream {
+	return (&EasyTierAnalyzer{}).NewTCP(analyzer.TCPInfo{}, nil)
+}
+
+func etFeedUDP(s analyzer.UDPStream, rev bool, data []byte) (analyzer.PropMap, bool) {
+	u, done := s.Feed(rev, data)
+	if u == nil {
+		return analyzer.PropMap{}, done
+	}
+	return u.M, done
+}
+
+func etFeedTCP(s analyzer.TCPStream, rev bool, data []byte) (analyzer.PropMap, bool) {
+	u, done := s.Feed(rev, false, false, 0, data)
+	if u == nil {
+		return analyzer.PropMap{}, done
+	}
+	return u.M, done
+}
+
+func etText(m analyzer.PropMap, key string) string {
+	v, ok := m[key]
+	if !ok {
+		return "<missing>"
+	}
+	return fmt.Sprint(v)
+}
+
+func etFlag(m analyzer.PropMap, key string) bool {
+	v, _ := m[key].(bool)
+	return v
+}
+
+func etExpect(t *testing.T, m analyzer.PropMap, key, want string) {
+	t.Helper()
+	if got := etText(m, key); got != want {
+		t.Fatalf("%s = %q, want %q (props: %v)", key, got, want, m)
+	}
+}
+
+func TestEasyTier_HandshakeMagicEncodesToDocumentedBytes(t *testing.T) {
+	want := []byte{0x08, 0xe1, 0xcb, 0x86, 0x8f, 0x0d}
+	if got := etProtoVarint(1, easyTierHandshakeMagic); !bytes.Equal(got, want) {
+		t.Fatalf("magic encodes to % x, want % x", got, want)
+	}
+}
+
+func TestEasyTier_UDPHandshakeNamesTheMeshOnFirstDatagram(t *testing.T) {
+	payload := etHandshakeRequest(0x1a2b3c4d, "office-mesh")
+	m, done := etFeedUDP(etUDP(), false, etDatagram(7, easyTierMsgData, etBody(easyTierPktHandshake, 0x1a2b3c4d, 0, payload)))
+
+	if done {
+		t.Fatal("stream done after a single handshake")
+	}
+	if !etFlag(m, "yes") {
+		t.Fatal("the magic alone should settle it")
+	}
+	if !etFlag(m, "magic_ok") {
+		t.Fatal("magic_ok not set")
+	}
+	etExpect(t, m, "network_name", "office-mesh")
+	etExpect(t, m, "version", "1")
+	etExpect(t, m, "transport", "udp")
+	etExpect(t, m, "msg_type_name", "data")
+	etExpect(t, m, "packet_type_name", "handshake")
+	etExpect(t, m, "from_peer_id", "439041101")
+	etExpect(t, m, "to_peer_id", "0")
+	etExpect(t, m, "secure", "false")
+}
+
+func TestEasyTier_TCPHandshakeNamesTheMeshOnFirstFrame(t *testing.T) {
+	payload := etHandshakeRequest(9, "office-mesh")
+	m, done := etFeedTCP(etTCP(), false, etSegment(etBody(easyTierPktHandshake, 9, 0, payload)))
+
+	if done {
+		t.Fatal("stream done after a single handshake")
+	}
+	if !etFlag(m, "yes") {
+		t.Fatal("yes not set")
+	}
+	etExpect(t, m, "network_name", "office-mesh")
+	etExpect(t, m, "transport", "tcp")
+	etExpect(t, m, "tx_pkt_cnt", "1")
+}
+
+func TestEasyTier_HandshakeWithWrongMagicIsNotConclusive(t *testing.T) {
+	payload := append(etProtoVarint(1, 0xd1e1a5e0), etProtoBytes(5, []byte("office-mesh"))...)
+	m, _ := etFeedUDP(etUDP(), false, etDatagram(7, easyTierMsgData, etBody(easyTierPktHandshake, 1, 0, payload)))
+
+	// 帧结构仍然自洽，字段照发；但一帧不够，而且对网络不做任何声明
+	if etFlag(m, "yes") {
+		t.Fatal("yes set on a wrong magic")
+	}
+	etExpect(t, m, "network_name", "<missing>")
+	etExpect(t, m, "magic_ok", "<missing>")
+}
+
+func TestEasyTier_HandshakeWithWrongDigestWidthIsRejected(t *testing.T) {
+	// 上游这种情况会拒连，报匹配等于报一个根本连不起来的 mesh 名
+	payload := etProtoVarint(1, easyTierHandshakeMagic)
+	payload = append(payload, etProtoBytes(5, []byte("office-mesh"))...)
+	payload = append(payload, etProtoBytes(6, make([]byte, 16))...)
+	m, _ := etFeedUDP(etUDP(), false, etDatagram(7, easyTierMsgData, etBody(easyTierPktHandshake, 1, 0, payload)))
+
+	if etFlag(m, "yes") {
+		t.Fatal("yes set on a wrong digest width")
+	}
+	etExpect(t, m, "network_name", "<missing>")
+}
+
+func TestEasyTier_NoiseFirstMessageStillGivesUpTheNetworkName(t *testing.T) {
+	// Noise XX 写 msg1 时还没有密钥，临时公钥后面的载荷是明文
+	s := etUDP()
+	m, _ := etFeedUDP(s, false, etDatagram(7, easyTierMsgData, etBody(easyTierPktNoiseMsg1, 5, 0, etNoiseMsg1("secure-mesh"))))
+
+	if !etFlag(m, "secure") {
+		t.Fatal("secure not set")
+	}
+	etExpect(t, m, "network_name", "secure-mesh")
+	if etFlag(m, "yes") {
+		t.Fatal("one frame with no magic to check should not settle it")
+	}
+
+	// 反方向再来一帧就到阈值了
+	m, _ = etFeedUDP(s, true, etDatagram(7, easyTierMsgData, etBody(14, 6, 5, bytes.Repeat([]byte{0xcd}, 48))))
+	if !etFlag(m, "yes") {
+		t.Fatal("two consistent frames should settle it")
+	}
+	etExpect(t, m, "packet_type_name", "noise_handshake_msg2")
+}
+
+func TestEasyTier_OpaqueNoiseMessageSaysNothingAboutTheNetwork(t *testing.T) {
+	m, _ := etFeedUDP(etUDP(), false, etDatagram(7, easyTierMsgData, etBody(easyTierPktNoiseMsg1, 5, 0, bytes.Repeat([]byte{0x5a}, 96))))
+
+	if !etFlag(m, "secure") {
+		t.Fatal("secure not set")
+	}
+	etExpect(t, m, "network_name", "<missing>")
+}
+
+func TestEasyTier_SynAnsweredByItsSackSettlesTheFlow(t *testing.T) {
+	s := etUDP()
+	m, _ := etFeedUDP(s, false, etSyn(0x12345678, 0x0102030405060708))
+	if etFlag(m, "yes") {
+		t.Fatal("a lone syn is 16 bytes of nearly nothing")
+	}
+	etExpect(t, m, "msg_type_name", "syn")
+
+	m, _ = etFeedUDP(s, true, etSack(0x12345678, 0x0102030405060708))
+	if !etFlag(m, "yes") {
+		t.Fatal("a matching sack should settle it")
+	}
+	etExpect(t, m, "msg_type_name", "sack")
+	etExpect(t, m, "conn_id", "305419896")
+	etExpect(t, m, "tx_pkt_cnt", "1")
+	etExpect(t, m, "rx_pkt_cnt", "1")
+}
+
+func TestEasyTier_SackEchoingTheWrongMagicSettlesNothing(t *testing.T) {
+	s := etUDP()
+	etFeedUDP(s, false, etSyn(0x12345678, 1))
+	m, _ := etFeedUDP(s, true, etSack(0x12345678, 2))
+	if etFlag(m, "yes") {
+		t.Fatal("yes set on a mismatched magic")
+	}
+}
+
+func TestEasyTier_SackDownTheSameDirectionSettlesNothing(t *testing.T) {
+	s := etUDP()
+	etFeedUDP(s, false, etSyn(9, 42))
+	m, _ := etFeedUDP(s, false, etSack(9, 42))
+	if etFlag(m, "yes") {
+		t.Fatal("yes set on a same-direction sack")
+	}
+}
+
+func TestEasyTier_HolePunchProbesAreReportedButDoNotSettleTheFlow(t *testing.T) {
+	// 往可能根本没跑 EasyTier 的地址上喷的，规则得点名才拦
+	s := etUDP()
+	var m analyzer.PropMap
+	for i := 0; i < 3; i++ {
+		m, _ = etFeedUDP(s, false, etDatagram(1, easyTierMsgHolePunch, bytes.Repeat([]byte{0x7f}, 16)))
+	}
+	if etFlag(m, "yes") {
+		t.Fatal("hole punches settled the flow")
+	}
+	etExpect(t, m, "msg_type_name", "hole_punch")
+}
+
+func TestEasyTier_BothHolePunchBodyLengthsAreRecognised(t *testing.T) {
+	// 连通性路径 16 字节，UDP 会话层 32 字节；只有长度能把它们和噪声分开，两个都得列
+	for _, n := range []int{16, 32} {
+		m, _ := etFeedUDP(etUDP(), false, etDatagram(1, easyTierMsgHolePunch, bytes.Repeat([]byte{0x7f}, n)))
+		if got := etText(m, "msg_type_name"); got != "hole_punch" {
+			t.Fatalf("body length %d: msg_type_name = %q", n, got)
+		}
+	}
+	m, _ := etFeedUDP(etUDP(), false, etDatagram(1, easyTierMsgHolePunch, bytes.Repeat([]byte{0x7f}, 24)))
+	etExpect(t, m, "msg_type_name", "<missing>")
+}
+
+func TestEasyTier_DirectedHolePunchMustEchoTheDestinationPortInConnID(t *testing.T) {
+	punch := []byte{203, 0, 113, 9}
+	punch = binary.LittleEndian.AppendUint16(punch, 11010)
+
+	m, _ := etFeedUDP(etUDP(), false, etDatagram(11010, easyTierMsgV4HolePunch, punch))
+	etExpect(t, m, "msg_type_name", "v4_hole_punch")
+
+	m, done := etFeedUDP(etUDP(), false, etDatagram(11011, easyTierMsgV4HolePunch, punch))
+	etExpect(t, m, "msg_type_name", "<missing>")
+	if done {
+		t.Fatal("one bad datagram is not enough to give up")
+	}
+}
+
+func TestEasyTier_UDPLengthFieldMustMatchTheDatagram(t *testing.T) {
+	wrong := etSyn(1, 2)
+	wrong[6] = 9 // 声明 9，体是 8
+	m, _ := etFeedUDP(etUDP(), false, wrong)
+	etExpect(t, m, "msg_type_name", "<missing>")
+}
+
+func TestEasyTier_UDPPaddingByteMustBeZero(t *testing.T) {
+	wrong := etSyn(1, 2)
+	wrong[5] = 1
+	m, _ := etFeedUDP(etUDP(), false, wrong)
+	etExpect(t, m, "msg_type_name", "<missing>")
+}
+
+func TestEasyTier_FourUnparseableDatagramsEndAUDPFlow(t *testing.T) {
+	s := etUDP()
+	for i := 0; i < 3; i++ {
+		if _, done := etFeedUDP(s, false, []byte("not easytier at all")); done {
+			t.Fatalf("gave up after %d datagrams", i+1)
+		}
+	}
+	if _, done := etFeedUDP(s, false, []byte("not easytier at all")); !done {
+		t.Fatal("did not give up after four unparseable datagrams")
+	}
+}
+
+func TestEasyTier_TwoLengthFieldsOfATCPFrameMustAgree(t *testing.T) {
+	// 16 字节头后面声明载荷 4，实际到了 8
+	frame := etBodyRaw(1, 1, 2, 0, 1, 0, 4, bytes.Repeat([]byte{0xaa}, 8))
+	m, done := etFeedTCP(etTCP(), false, etSegment(frame))
+	if !done {
+		t.Fatal("stream not done on a length mismatch")
+	}
+	etExpect(t, m, "packet_type_name", "<missing>")
+}
+
+func TestEasyTier_AEADTailIsAllowedToOutrunTheDeclaredLength(t *testing.T) {
+	// 加密器追加 16 字节 tag 和 12 字节 nonce，不改头里声明的长度
+	payload := bytes.Repeat([]byte{0x11}, 40+easyTierAEADTailSize)
+	frame := etBodyRaw(1, 1, 2, easyTierFlagEncrypted, 1, 0, 40, payload)
+	m, _ := etFeedTCP(etTCP(), false, etSegment(frame))
+
+	if !etFlag(m, "encrypted") {
+		t.Fatal("encrypted not set")
+	}
+	etExpect(t, m, "payload_len", "40")
+}
+
+func TestEasyTier_CompressedPayloadMayComeInUnderTheDeclaredLength(t *testing.T) {
+	frame := etBodyRaw(1, 1, 2, easyTierFlagCompressed, 1, 0, 900, bytes.Repeat([]byte{0x22}, 61))
+	m, _ := etFeedTCP(etTCP(), false, etSegment(frame))
+
+	if !etFlag(m, "compressed") {
+		t.Fatal("compressed not set")
+	}
+	etExpect(t, m, "payload_len", "900")
+}
+
+func TestEasyTier_UnencryptedPayloadMustMatchTheDeclaredLengthExactly(t *testing.T) {
+	frame := etBodyRaw(1, 1, 2, 0, 1, 0, 40, bytes.Repeat([]byte{0x11}, 40+easyTierAEADTailSize))
+	if _, done := etFeedTCP(etTCP(), false, etSegment(frame)); !done {
+		t.Fatal("stream not done on an unencrypted length mismatch")
+	}
+}
+
+func TestEasyTier_ReservedByteIsZeroUnlessALivenessFlagIsSet(t *testing.T) {
+	withFlag := etBodyRaw(1, 1, 2, 0b0010_0000, 1, 0x5c, 4, make([]byte, 4))
+	m, _ := etFeedTCP(etTCP(), false, etSegment(withFlag))
+	etExpect(t, m, "packet_type_name", "data")
+
+	withoutFlag := etBodyRaw(1, 1, 2, 0, 1, 0x5c, 4, make([]byte, 4))
+	m, done := etFeedTCP(etTCP(), false, etSegment(withoutFlag))
+	if !done {
+		t.Fatal("stream not done on a non-zero reserved byte")
+	}
+	etExpect(t, m, "packet_type_name", "<missing>")
+}
+
+func TestEasyTier_ForwardCounterPastTheRoutersOwnCapIsRejected(t *testing.T) {
+	frame := etBodyRaw(1, 1, 2, 0, 9, 0, 4, make([]byte, 4))
+	m, done := etFeedTCP(etTCP(), false, etSegment(frame))
+	if !done {
+		t.Fatal("stream not done on forward counter 9")
+	}
+	etExpect(t, m, "packet_type_name", "<missing>")
+}
+
+func TestEasyTier_UnnamedPacketTypeIsRejected(t *testing.T) {
+	frame := etBodyRaw(22, 1, 2, 0, 1, 0, 4, make([]byte, 4))
+	if _, done := etFeedTCP(etTCP(), false, etSegment(frame)); !done {
+		t.Fatal("stream not done on packet type 22")
+	}
+}
+
+func TestEasyTier_FrameSplitAcrossSegmentsIsReassembled(t *testing.T) {
+	wire := etSegment(etBody(easyTierPktHandshake, 9, 0, etHandshakeRequest(9, "office-mesh")))
+
+	s := etTCP()
+	m, done := etFeedTCP(s, false, wire[:3])
+	if done {
+		t.Fatal("gave up on a length field that had not fully arrived")
+	}
+	etExpect(t, m, "packet_type_name", "<missing>")
+
+	m, done = etFeedTCP(s, false, wire[3:])
+	if done {
+		t.Fatal("stream done after a valid handshake")
+	}
+	if !etFlag(m, "yes") {
+		t.Fatal("yes not set after reassembly")
+	}
+	etExpect(t, m, "network_name", "office-mesh")
+}
+
+func TestEasyTier_ByteAtATimeDeliveryStillReadsTheFrame(t *testing.T) {
+	wire := etSegment(etBody(easyTierPktHandshake, 9, 0, etHandshakeRequest(9, "mesh")))
+	s := etTCP()
+	last := analyzer.PropMap{}
+	for _, b := range wire {
+		m, done := etFeedTCP(s, false, []byte{b})
+		if done {
+			t.Fatal("stream done mid-frame")
+		}
+		if len(m) != 0 {
+			last = m
+		}
+	}
+	if !etFlag(last, "yes") {
+		t.Fatal("yes not set after byte-at-a-time delivery")
+	}
+	etExpect(t, last, "network_name", "mesh")
+}
+
+func TestEasyTier_SeveralFramesPipelinedIntoOneSegmentAreAllRead(t *testing.T) {
+	wire := etSegment(etBody(1, 9, 8, bytes.Repeat([]byte{0x01}, 32)))
+	wire = append(wire, etSegment(etBody(4, 9, 8, nil))...)
+	wire = append(wire, etSegment(etBody(1, 9, 8, bytes.Repeat([]byte{0x03}, 64)))...)
+
+	m, done := etFeedTCP(etTCP(), false, wire)
+	if done {
+		t.Fatal("stream done after valid frames")
+	}
+	if !etFlag(m, "yes") {
+		t.Fatal("yes not set after three consistent frames")
+	}
+	etExpect(t, m, "tx_pkt_cnt", "3")
+	// map 描述的是最新一帧
+	etExpect(t, m, "payload_len", "64")
+}
+
+func TestEasyTier_NetworkNameSurvivesLaterFrames(t *testing.T) {
+	// 只发一次。按 mesh 名写的规则得在握手滚过去之后还能用
+	s := etTCP()
+	etFeedTCP(s, false, etSegment(etBody(easyTierPktHandshake, 9, 0, etHandshakeRequest(9, "office-mesh"))))
+	m, _ := etFeedTCP(s, false, etSegment(etBody(1, 9, 8, bytes.Repeat([]byte{0xee}, 128))))
+
+	etExpect(t, m, "network_name", "office-mesh")
+	etExpect(t, m, "packet_type_name", "data")
+}
+
+func TestEasyTier_GapInTheTCPStreamEndsIt(t *testing.T) {
+	u, done := etTCP().Feed(false, false, false, 7, []byte("whatever"))
+	if u != nil {
+		t.Fatal("update published on a gap")
+	}
+	if !done {
+		t.Fatal("stream not done on a gap")
+	}
+}
+
+func TestEasyTier_OrdinaryTrafficIsNotMistakenForATunnel(t *testing.T) {
+	m, done := etFeedTCP(etTCP(), false, []byte("GET / HTTP/1.1\r\nHost: example.com\r\n\r\n"))
+	if !done || etFlag(m, "yes") {
+		t.Fatalf("HTTP: done=%v yes=%v", done, etFlag(m, "yes"))
+	}
+
+	// TLS ClientHello 记录：16 03 01 <len> ... 读出来是 21 MB 的帧体长度，直接拒绝
+	m, done = etFeedTCP(etTCP(), false, []byte{0x16, 0x03, 0x01, 0x01, 0x4a, 0x01, 0x00, 0x01, 0x46})
+	if !done || etFlag(m, "yes") {
+		t.Fatalf("TLS: done=%v yes=%v", done, etFlag(m, "yes"))
+	}
+}
+
+func TestEasyTier_FETDoesNotCatchTheTCPHandshake(t *testing.T) {
+	// fet 只看客户端第一个分段，判完就 done。EasyTier 连接的第一个分段永远是明文握手，
+	// 整条连接上唯一低熵的东西，所以 fet 读作"非全加密"，引擎放行，后面真正加密的帧再也不看。
+	// 而且能不能中取决于握手里碰巧有多少随机内容（ex1 落在 3.4 豁免线附近，随网络名长短摇摆），
+	// 一个随运营者起的名字长短而变的检测器不是检测器。
+	digest := make([]byte, 32)
+	for i := range digest {
+		digest[i] = byte(uint32(i) * 2654435761)
+	}
+	payload := etProtoVarint(1, easyTierHandshakeMagic)
+	payload = append(payload, etProtoVarint(2, 0x1a2b3c4d)...)
+	payload = append(payload, etProtoVarint(3, 1)...)
+	payload = append(payload, etProtoBytes(4, []byte("liveness-echo-v1"))...)
+	payload = append(payload, etProtoBytes(5, []byte("mesh"))...)
+	payload = append(payload, etProtoBytes(6, digest)...)
+	wire := etSegment(etBody(easyTierPktHandshake, 0x1a2b3c4d, 0, payload))
+
+	fet := (&tcp.FETAnalyzer{}).NewTCP(analyzer.TCPInfo{}, nil)
+	u, done := fet.Feed(false, true, false, 0, wire)
+	if !done {
+		t.Fatal("fet should decide on the first segment and not look again")
+	}
+	if u != nil && etFlag(u.M, "yes") {
+		t.Fatal("fet claimed this handshake, so this analyzer would be redundant")
+	}
+
+	// 同样的字节，给本分析器
+	m, _ := etFeedTCP(etTCP(), false, wire)
+	if !etFlag(m, "yes") {
+		t.Fatal("easytier did not claim its own handshake")
+	}
+	etExpect(t, m, "network_name", "mesh")
+}
+
+func TestEasyTier_WireGuardAnalyzerDoesNotClaimUDPTransport(t *testing.T) {
+	// wireguard 覆盖的是 EasyTier 的 wg:// 监听，六分之一。udp:// 上没有任何东西读起来像 WireGuard 报文
+	datagrams := [][]byte{
+		etSyn(0x12345678, 0x0102030405060708),
+		etDatagram(7, easyTierMsgData, etBody(easyTierPktHandshake, 0x1a2b3c4d, 0, etHandshakeRequest(9, "mesh"))),
+		etDatagram(1, easyTierMsgHolePunch, bytes.Repeat([]byte{0x7f}, 16)),
+	}
+	for i, data := range datagrams {
+		wg := (&WireGuardAnalyzer{}).NewUDP(analyzer.UDPInfo{}, nil)
+		if u, _ := wg.Feed(false, data); u != nil {
+			t.Fatalf("wireguard claimed easytier datagram %d: %v", i, u.M)
+		}
+	}
+}
+
+func TestEasyTier_STUNOnTheSameSocketIsNeitherEvidenceNorHeldAgainstTheFlow(t *testing.T) {
+	// EasyTier 把 STUN 复用在自己的 UDP socket 上，并在自己的帧解析之前分流；四个 STUN 不能结束流
+	stun := append([]byte{0x00, 0x01, 0x00, 0x00, 0x21, 0x12, 0xa4, 0x42}, bytes.Repeat([]byte{0x11}, 12)...)
+
+	s := etUDP()
+	for i := 0; i < 8; i++ {
+		m, done := etFeedUDP(s, false, stun)
+		if done {
+			t.Fatalf("STUN datagram %d ended the flow", i+1)
+		}
+		if len(m) != 0 {
+			t.Fatalf("STUN datagram %d published props: %v", i+1, m)
+		}
+	}
+
+	m, _ := etFeedUDP(s, false, etSyn(1, 2))
+	etExpect(t, m, "msg_type_name", "syn")
+}

cmd/root.go

@@ -96,6 +96,7 @@ var analyzers = []analyzer.Analyzer{
 	&tcp.TLSAnalyzer{},
 	&tcp.TrojanAnalyzer{},
 	&udp.DNSAnalyzer{},
+	&udp.EasyTierAnalyzer{},
 	&udp.OpenVPNAnalyzer{},
 	&udp.QUICAnalyzer{},
 	&udp.WireGuardAnalyzer{},

docs/easytier.zh.md

@@ -0,0 +1,241 @@
+# EasyTier 分析器
+
+识别 [EasyTier](https://github.com/EasyTier/EasyTier) 去中心化 Mesh VPN 的对等连接。
+
+## 传输方式与可见性
+
+EasyTier 的对等连接可以走六种传输，节点默认全部监听：
+
+| scheme | 默认端口 | 线上是什么样 | 谁负责 |
+| --- | --- | --- | --- |
+| `tcp://` | 11010 | EasyTier 自己的定长帧，明文头 | **本分析器** |
+| `udp://` | 11010 | EasyTier 自己的数据报头，明文头 | **本分析器** |
+| `ws://` | 11011 | HTTP Upgrade，之后是 WebSocket 二进制帧 | `http`（握手可见），帧内容不解 |
+| `wss://` | 11012 | TLS（自签证书，CN=`localhost`） | `tls`（SNI/证书层面） |
+| `wg://` | 11013 | 标准 WireGuard 报文 | `wireguard` |
+| `quic://` | 11010 | QUIC，未设置 ALPN | `quic` |
+
+只有前两种把 EasyTier 的帧结构直接放在网络层上，也是唯二能拿到网络名、对端 ID 这些信息的。下面说的都是这两种。
+
+## 为什么 `wireguard` 和 `fet` 不够
+
+这两个现有分析器看起来应该能覆盖，实测都不行。
+
+### `wireguard`：只挡得住六分之一
+
+它匹配的是真正的 WireGuard 报文，也就是 EasyTier 的 `wg://` 传输（默认 11013）。问题在于**节点默认把六种传输全部监听起来**，连接方按拿到的 URI 列表逐个尝试，哪个通用哪个。只挡 `wg://`，mesh 换 `udp://` 照样建起来，而 `udp://` 恰恰是 NAT 穿透场景下的首选。
+
+另外 WireGuard 报文里没有网络名，拦了也不知道拦的是哪个 mesh，而且会一并拦掉合规的 WireGuard VPN。粒度太粗。
+
+`TestEasyTier_WireGuardAnalyzerDoesNotClaimUDPTransport` 这个用例把 `udp://` 上的 Syn、握手、打洞包依次喂给 `wireguard` 分析器，确认它一个都不认。
+
+### `fet`：判在了整条连接熵最低的那一段上
+
+`fet` 检测「全加密流量」，只看客户端**第一个**分段，判完就 `done`。而 EasyTier 连接的第一个分段永远是明文握手帧，整条连接上唯一不加密的东西。
+
+那一帧长什么样：`to_peer_id` 四个零、`flags`/`reserved` 是零、两个长度字段的高位字节是零、`liveness-echo-v1` 和网络名是 ASCII。实测 `ex1`（平均 popcount）落在 3.4 这条豁免线附近，而且**结果取决于网络名有多长**：
+
+| 网络名 | `fet.yes` 命中率（200 组随机 digest） | `ex1` 区间 |
+| --- | --- | --- |
+| `a` | 0/200 | 3.045..3.258 |
+| `mesh` | 0/200 | 3.076..3.283 |
+| `office-mesh` | 3/200 | 3.242..3.434 |
+| `my-home-network` | 52/200 | 3.320..3.505 |
+
+握手不带真实 digest 时（跨网络/共享节点场景，字段填 32 个零）更低，四种名字全是 0/200，`ex1` 只有 1.6..2.2。
+
+也就是说 `fet` 在这里是**大概率漏、小概率中**，而且能不能中取决于对方给自己的网络起了个多长的名字。而后续真正加密的数据帧 `ex1` 有 3.92，`fet` 本来能认，但它早就判完退出了。
+
+更麻烦的是漏判的后果：`fet` 报 `yes=false` 且没有别的规则拦截时，引擎会把这条流放进快速通道（connmark bypass），之后的包**根本不再进用户态**。一次漏判就是永久放行，和 `docs/tcp-window-evasion.zh.md` 里描述的是同一个后果。
+
+`TestEasyTier_FETDoesNotCatchTheTCPHandshake` 这个用例把这一点钉住了。
+
+## 报文结构
+
+三个头，全部小端，全部 `#[repr(C, packed)]`，没有任何对齐填充。
+
+**`UDPTunnelHeader`（8 字节，仅 UDP）**
+
+| 偏移 | 字段 | 说明 |
+| --- | --- | --- |
+| 0..4 | `conn_id` u32 | 会话 ID，发起方随机选取 |
+| 4 | `msg_type` u8 | 1=Syn 2=Sack 3=Data 4=Fin\* 5=HolePunch 6=V4HolePunch 7=V6HolePunch（\* 枚举里有，当前实现从不发送） |
+| 5 | `padding` u8 | 恒为 0，只为对齐长度字段而存在 |
+| 6..8 | `len` u16 | 后面的字节数，**必须**等于 `数据报长度 - 8` |
+
+**`TCPTunnelHeader`（4 字节，仅 TCP）**
+
+一个 u32 长度前缀，值等于后面帧体的字节数。上游要求 `16 ≤ len ≤ max_packet_size`（TCP 隧道 2000，字节流隧道 4096）。
+
+**`PeerManagerHeader`（16 字节，两种传输都有）**
+
+| 偏移 | 字段 | 说明 |
+| --- | --- | --- |
+| 0..4 | `from_peer_id` u32 | 源节点 ID |
+| 4..8 | `to_peer_id` u32 | 目的节点 ID，首包为 0 |
+| 8 | `packet_type` u8 | 见下表，取值 1..=21 |
+| 9 | `flags` u8 | bit0 加密 bit1 延迟优先 bit2 出口节点 bit3 不代理 bit4 压缩 bit5/7 存活探测 bit6 不送 TUN |
+| 10 | `forward_counter` u8 | 中继跳数，路由在 >7 时丢弃后再自增，所以线上最大 8 |
+| 11 | `reserved` u8 | 存活标志位（bit5/bit7）置位时是探测 token，否则恒为 0 |
+| 12..16 | `len` u32 | 载荷长度 |
+
+`packet_type` 取值：1 Data、2 HandShake、3 RoutePacket*、4 Ping、5 Pong、6 TaRpc*、7 Route*、8 RpcReq、9 RpcResp、10 ForeignNetworkPacket、11 KcpSrc、12 KcpDst、13/14/15 NoiseHandshakeMsg1/2/3、16 QuicSrc、17 QuicDst、18/19 DataWith{Kcp,Quic}SrcModified、20 RelayHandshake、21 RelayHandshakeAck（\* = 已废弃但仍占位）。
+
+### 关键点：头不加密
+
+EasyTier 的三种加密（AES-GCM / ChaCha20 / XOR）**只覆盖载荷**，`PeerManagerHeader` 自始至终是明文，加密只是把 `flags` 的 bit0 置位（`tunnel/encrypt/aes_gcm.rs` 等）。压缩同理，置 bit4。
+
+所以哪怕整个 mesh 开了加密，对端 ID、报文类型、跳数、声明的载荷长度仍然一直挂在明面上。这是本分析器的立足点。
+
+## 判定信号
+
+分三档，从弱到强。
+
+### 一、帧结构自洽（`easyTierEvidenceFrame`）
+
+每一帧里有两个互相冗余的长度字段，必须和实际收到的字节数对得上：
+
+```
+UDP:  len(6..8)  == 数据报长度 - 8
+TCP:  len(0..4)  == 帧体长度
+两者: PeerManagerHeader.len 与实际载荷长度的关系
+```
+
+最后一条要分情况，因为**加密和压缩都不更新 `PeerManagerHeader.len`**：
+
+| flags | 实际载荷长度 |
+| --- | --- |
+| 都没置位 | `== len`，精确相等 |
+| 仅加密 | `== len`（XOR）或 `== len + 28`（AEAD 的 16 字节 tag + 12 字节 nonce） |
+| 压缩 | `≤ len + 28`（压缩器只在能省下自己那 1 字节尾时才压） |
+
+再加上 `padding == 0`、`packet_type ∈ 1..=21`、`forward_counter ≤ 8`、`reserved` 的取值约束，单帧大约值三十几个比特。分析器要求**攒够 2 帧**才把 `yes` 置真。
+
+### 二、UDP 会话握手（`easyTierEvidenceConclusive`）
+
+发起方发一个 16 字节的 `Syn`（`len == 8`，体是随机 u64 magic），对端回一个 16 字节的 `Sack`，**同一个 `conn_id`、同一个 magic**，方向相反。96 比特的巧合概率，一次就够。
+
+单独一个 `Syn` 或 magic 对不上的 `Sack` 只算「见过」，不推进 `yes`。
+
+### 三、对等握手（`easyTierEvidenceConclusive`）
+
+默认（非 secure）模式下，连接建立后第一帧就是 `packet_type = 2` 的 protobuf `HandshakeRequest`，第一个字段是常量 `magic = 0xd1e1a5e1`。protobuf 编码后是固定的 6 字节：
+
+```
+08 e1 cb 86 8f 0d
+```
+
+在 TCP 上，它落在整条连接载荷的固定偏移 20..26：
+
+```
+0..4    TCPTunnelHeader.len
+4..20   PeerManagerHeader（type=2, to_peer_id=0）
+20..26  08 e1 cb 86 8f 0d
+```
+
+同一条消息还带着 `network_name`（字段 5，明文）和 32 字节的 `network_secret_digest`（字段 6）。**digest 宽度不对的握手上游会直接拒连**，所以分析器也把它当作否定条件，避免报一个根本连不起来的 mesh 名。
+
+### 关于 secure mode
+
+`--enable-secure-mode` 会把首帧换成 Noise XX 的 msg1（`packet_type = 13`）。Noise XX 的第一条消息此时还没有密钥，载荷是**明文**，所以 32 字节临时公钥后面的 protobuf 仍然能读出 `a_network_name`。
+
+分析器对这段是「能读就读、读不出就不说」：protobuf 必须完整解析、version 合理、网络名是不含控制字符的 UTF-8，否则只置 `secure`，不填 `network_name`。secure mode 下没有 magic 可查，所以要靠第一档攒够 2 帧才判 `yes`。
+
+## 属性
+
+| 属性 | 类型 | 说明 |
+| --- | --- | --- |
+| `yes` | bool | 综合判定。一定存在 |
+| `transport` | string | `udp` / `tcp`。一定存在 |
+| `tx_pkt_cnt` / `rx_pkt_cnt` | int | 两个方向已解析的帧数。一定存在 |
+| `msg_type` | uint8 | UDP 隧道消息类型，仅 UDP |
+| `msg_type_name` | string | `syn` / `sack` / `data` / `fin` / `hole_punch` / `v4_hole_punch` / `v6_hole_punch` |
+| `conn_id` | uint32 | UDP 会话 ID，仅 UDP |
+| `packet_type` | uint8 | `PeerManagerHeader.packet_type` |
+| `packet_type_name` | string | 上表的可读形式，如 `handshake`、`ping`、`noise_handshake_msg1` |
+| `from_peer_id` / `to_peer_id` | uint32 | 对端 ID |
+| `forward_counter` | uint8 | 中继跳数 |
+| `payload_len` | uint32 | 头里声明的载荷长度 |
+| `encrypted` / `compressed` / `latency_first` / `exit_node` / `no_proxy` | bool | `flags` 各位 |
+| `magic_ok` | bool | 明文握手 magic 匹配。**粘性** |
+| `secure` | bool | true = 见到 Noise 握手，false = 见到明文握手。**粘性** |
+| `version` | uint64 | 握手里的协议版本。**粘性** |
+| `network_name` | string | mesh 名。**粘性** |
+
+标「粘性」的字段一旦写入就不再被后续帧覆盖：网络名只在握手里出现一次，一条按 mesh 名写的规则必须在握手滚过去之后还能用。其余字段描述的是**最近一帧**。整数字段的具体宽度不影响规则，expr 比较时会自动对齐。
+
+`packet_type` 及其后的字段只在该帧带 `PeerManagerHeader` 时存在（UDP 的 `Data`、全部 TCP 帧）。`Syn`/`Sack`/打洞包没有。
+
+> [!NOTE]
+> 规则里要写 `easytier != nil && ...`。规则引擎只把已经发布过属性的分析器放进表达式环境（见 `ruleset/expr.go` 的 `streamInfoToExprEnv`），流上一帧都没解析出来时 `easytier` 是 `nil`，直接取 `easytier.yes` 会报错。map 里不存在的键读出来是 `nil`，比较结果为假，不报错：所以 `easytier.exit_node == true` 在 `Syn`/`Sack`/打洞包上只是**不匹配**，`exit_node` 要等一个带 `PeerManagerHeader` 的帧到达才有值；同理 `easytier.network_name == "..."` 在握手到达之前一直不匹配。
+
+## 规则示例
+
+拦截所有 EasyTier 对等连接：
+
+```yaml
+- name: block-easytier
+  action: block
+  log: true
+  expr: easytier != nil && easytier.yes
+```
+
+只拦某一个 mesh，其余放行（网络名明文可见，这是最精确的一档）：
+
+```yaml
+- name: block-one-easytier-mesh
+  action: block
+  log: true
+  expr: easytier != nil && easytier.network_name == "office-mesh"
+```
+
+只拦出口节点，即把本网流量往公网转发的那些，通常是策略上真正在意的：
+
+```yaml
+- name: block-easytier-exit-node
+  action: block
+  log: true
+  expr: easytier != nil && easytier.yes && easytier.exit_node == true
+```
+
+打洞包不参与 `yes`，要单独点名（见下节的取舍说明）：
+
+```yaml
+- name: log-easytier-hole-punch
+  log: true
+  expr: easytier != nil && easytier.msg_type_name == "hole_punch"
+```
+
+只观察不拦截，先看清网络里有什么：
+
+```yaml
+- name: log-easytier
+  log: true
+  expr: easytier != nil && easytier.yes
+```
+
+规则按顺序匹配，第一条带 action 的命中即生效。想同时用「只拦一个 mesh」和「拦全部」，具体的要写在前面，否则被兜底的那条挡住永远到不了。
+
+配套的 WireGuard 传输要另外拦，走 `wireguard` 分析器。
+
+## 检不到什么
+
+- **`wg://` / `quic://` / `wss://` 传输**。线上分别就是 WireGuard、QUIC、TLS，交给对应分析器。`wss://` 的服务端用的是 `rcgen` 现签的自签证书，CN 为 `localhost`，这本身是个可用的特征，但它属于 `tls` 的范畴。
+- **`ws://` 帧内容**。HTTP Upgrade 握手 `http` 能看到，之后的 WebSocket 二进制帧里才是 `PeerManagerHeader`（无长度前缀，`ZCPacketType::DummyTunnel`），本分析器不做 WebSocket 拆帧。
+- **中途接入的流**。TCP 侧遇到 `skip != 0`（重组有空洞）会直接放弃，因为帧边界无从推断。
+- **只发打洞包的流**。打洞探测是往可能根本没跑 EasyTier 的地址上喷的随机数据，除了长度没有任何结构（连通性路径 16 字节，UDP 会话层 32 字节，两个值都认），把它算作确证会在正常网络里产生误报，所以它只填 `msg_type_name`、不推进 `yes`。要抓就按上面的例子点名。
+
+## 误报边界
+
+- UDP 侧连续 4 个无法解析的数据报就放弃该流（和 `wireguard` 分析器一致）；TCP 侧一旦帧结构对不上立刻结束。
+- EasyTier 把 STUN 复用在同一个 UDP socket 上，并且在自己的帧解析**之前**先分流 STUN。分析器照做：识别出 STUN（`data[4..8] == 21 12 A4 42` 且 `data[0] & 0xC0 == 0`）的数据报既不算证据也不算错误，只计入总包数，免得一串 STUN 把流提前判死。
+- 任一传输超过 256 个包仍未定性就结束该流（和 `openvpn` 分析器一致），避免长流一直占着分析器状态。
+- `Limit()` 返回 0。这是个定长帧协议，按字节数截断会让帧读取失步；上面的包数上限替代它约束开销。
+
+## 复现
+
+字节级的边界条件都在 `analyzer/udp/easytier_test.go` 里，包括 magic 的编码、两个长度字段的关系、AEAD 尾、压缩、`reserved` 的取值规则、帧切在长度前缀中间的重组、逐字节投递、STUN 复用，以及上面提到的 `fet` 和 `wireguard` 对照用例：
+
+```bash
+go test ./analyzer/udp -run EasyTier -v
+```

engine/release_test.go

@@ -0,0 +1,199 @@
+package engine
+
+import (
+	"net"
+	"testing"
+	"time"
+
+	"github.com/apernet/OpenGFW/analyzer"
+	"github.com/apernet/OpenGFW/analyzer/tcp"
+	"github.com/apernet/OpenGFW/analyzer/udp"
+	"github.com/apernet/OpenGFW/io"
+	"github.com/apernet/OpenGFW/ruleset"
+	"github.com/apernet/OpenGFW/ruleset/builtins/geo"
+
+	"github.com/bwmarrin/snowflake"
+	"github.com/google/gopacket"
+	"github.com/google/gopacket/layers"
+	"github.com/google/gopacket/reassembly"
+)
+
+type nullEngineLogger struct{}
+
+func (nullEngineLogger) WorkerStart(int)                                          {}
+func (nullEngineLogger) WorkerStop(int)                                           {}
+func (nullEngineLogger) TCPStreamNew(int, ruleset.StreamInfo)                     {}
+func (nullEngineLogger) TCPStreamPropUpdate(ruleset.StreamInfo, bool)             {}
+func (nullEngineLogger) TCPStreamAction(ruleset.StreamInfo, ruleset.Action, bool) {}
+func (nullEngineLogger) TCPFlush(int, int, int)                                   {}
+func (nullEngineLogger) UDPStreamNew(int, ruleset.StreamInfo)                     {}
+func (nullEngineLogger) UDPStreamPropUpdate(ruleset.StreamInfo, bool)             {}
+func (nullEngineLogger) UDPStreamAction(ruleset.StreamInfo, ruleset.Action, bool) {}
+func (nullEngineLogger) ModifyError(ruleset.StreamInfo, error)                    {}
+func (nullEngineLogger) AnalyzerDebugf(int64, string, string, ...interface{})     {}
+func (nullEngineLogger) AnalyzerInfof(int64, string, string, ...interface{})      {}
+func (nullEngineLogger) AnalyzerErrorf(int64, string, string, ...interface{})     {}
+
+type nullRulesetLogger struct{}
+
+func (nullRulesetLogger) Log(ruleset.StreamInfo, string)               {}
+func (nullRulesetLogger) MatchError(ruleset.StreamInfo, string, error) {}
+
+func testRuleset(t *testing.T, rules ...ruleset.ExprRule) ruleset.Ruleset {
+	t.Helper()
+	ans := []analyzer.Analyzer{
+		&tcp.FETAnalyzer{}, &tcp.HTTPAnalyzer{}, &tcp.TLSAnalyzer{},
+		&udp.QUICAnalyzer{}, &udp.WireGuardAnalyzer{},
+	}
+	rs, err := ruleset.CompileExprRules(rules, ans, nil, &ruleset.BuiltinConfig{Logger: nullRulesetLogger{}, GeoMatcher: geo.NewGeoMatcher("", "")})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return rs
+}
+
+// capturingFactory 记下工厂创建的每条 tcpStream，让测试能检查判定之后它的状态
+type capturingFactory struct {
+	inner   *tcpStreamFactory
+	streams []*tcpStream
+}
+
+func (f *capturingFactory) New(ipFlow, tcpFlow gopacket.Flow, tcp *layers.TCP, ac reassembly.AssemblerContext) reassembly.Stream {
+	s := f.inner.New(ipFlow, tcpFlow, tcp, ac).(*tcpStream)
+	f.streams = append(f.streams, s)
+	return s
+}
+
+func newTestAssembler(t *testing.T, rs ruleset.Ruleset) (*reassembly.Assembler, *capturingFactory) {
+	t.Helper()
+	node, err := snowflake.NewNode(1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	cf := &capturingFactory{inner: &tcpStreamFactory{WorkerID: 1, Logger: nullEngineLogger{}, Node: node, Ruleset: rs}}
+	asm := reassembly.NewAssembler(reassembly.NewStreamPool(cf))
+	asm.MaxBufferedPagesTotal = defaultTCPMaxBufferedPagesTotal
+	asm.MaxBufferedPagesPerConnection = defaultTCPMaxBufferedPagesPerConnection
+	return asm, cf
+}
+
+var testIPFlow = gopacket.NewFlow(layers.EndpointIPv4, net.IPv4(10, 0, 0, 1).To4(), net.IPv4(203, 0, 113, 1).To4())
+
+func tcpSegment(seq uint32, syn bool, payload []byte) *layers.TCP {
+	seg := &layers.TCP{SrcPort: 40000, DstPort: 80, Seq: seq, SYN: syn, ACK: !syn, PSH: len(payload) > 0, Window: 65535}
+	seg.Payload = payload
+	return seg
+}
+
+func feedTCP(asm *reassembly.Assembler, seg *layers.TCP, ts time.Time) tcpVerdict {
+	ctx := &tcpContext{PacketMetadata: &gopacket.PacketMetadata{CaptureInfo: gopacket.CaptureInfo{Timestamp: ts}}, Verdict: tcpVerdictAccept}
+	asm.AssembleWithContext(testIPFlow, seg, ctx)
+	return ctx.Verdict
+}
+
+var httpRequest = []byte("GET / HTTP/1.1\r\nHost: example.com\r\n\r\n")
+
+func TestTCPStreamReleasesStateAfterVerdict(t *testing.T) {
+	rs := testRuleset(t, ruleset.ExprRule{Name: "block-http", Action: "block", Expr: "http?.req != nil"})
+	asm, cf := newTestAssembler(t, rs)
+	ts := time.Now()
+
+	feedTCP(asm, tcpSegment(1000, true, nil), ts)
+	if v := feedTCP(asm, tcpSegment(1001, false, httpRequest), ts); v != tcpVerdictDropStream {
+		t.Fatalf("verdict = %v, want drop stream", v)
+	}
+	if len(cf.streams) != 1 {
+		t.Fatalf("streams created = %d, want 1", len(cf.streams))
+	}
+	s := cf.streams[0]
+	if !s.released || s.activeEntries != nil || s.info.Props != nil || s.ruleset != nil {
+		t.Fatalf("stream still holds state after verdict: released=%v entries=%d props=%v ruleset=%v", s.released, len(s.activeEntries), s.info.Props, s.ruleset != nil)
+	}
+
+	// 后续报文沿用判定，且不能碰已经释放的状态
+	if v := feedTCP(asm, tcpSegment(1001+uint32(len(httpRequest)), false, []byte("more")), ts); v != tcpVerdictDropStream {
+		t.Fatalf("verdict on a later segment = %v, want drop stream", v)
+	}
+}
+
+func TestTCPStreamReleasedWhenAllAnalyzersFinish(t *testing.T) {
+	// 规则只引用 tls；一个 HTTP 请求让 tls 分析器直接放弃，没有裁决，引擎应当发 AcceptStream 并释放
+	rs := testRuleset(t, ruleset.ExprRule{Name: "block-tls", Action: "block", Expr: "tls?.req != nil"})
+	asm, cf := newTestAssembler(t, rs)
+	ts := time.Now()
+
+	feedTCP(asm, tcpSegment(1000, true, nil), ts)
+	if v := feedTCP(asm, tcpSegment(1001, false, httpRequest), ts); v != tcpVerdictAcceptStream {
+		t.Fatalf("verdict = %v, want accept stream", v)
+	}
+	if s := cf.streams[0]; !s.released || s.info.Props != nil {
+		t.Fatalf("stream not released after all analyzers finished: released=%v props=%v", s.released, s.info.Props)
+	}
+}
+
+// 判定之前缓冲的乱序页会在超时 flush 时被补送到 ReassembledSG，那时流已经释放，不能崩
+func TestTCPStreamSurvivesFlushDeliveringBufferedPageAfterRelease(t *testing.T) {
+	rs := testRuleset(t, ruleset.ExprRule{Name: "block-http", Action: "block", Expr: "http?.req != nil"})
+	asm, cf := newTestAssembler(t, rs)
+	ts := time.Now()
+
+	feedTCP(asm, tcpSegment(1000, true, nil), ts)
+	// 乱序段：序号跳过一大截，被缓冲起来等空洞补齐
+	feedTCP(asm, tcpSegment(9000, false, []byte("out of order")), ts)
+	if v := feedTCP(asm, tcpSegment(1001, false, httpRequest), ts); v != tcpVerdictDropStream {
+		t.Fatalf("verdict = %v, want drop stream", v)
+	}
+	s := cf.streams[0]
+	if !s.released {
+		t.Fatal("stream not released after verdict")
+	}
+
+	// 超时 flush 会带着空洞把那一页送进来
+	flushed, closed := asm.FlushCloseOlderThan(ts.Add(time.Hour))
+	if flushed == 0 || closed == 0 {
+		t.Fatalf("flush did nothing: flushed=%d closed=%d", flushed, closed)
+	}
+	if s.info.Props != nil || s.activeEntries != nil {
+		t.Fatal("flush resurrected released state")
+	}
+}
+
+func TestUDPStreamReleasesStateAfterVerdict(t *testing.T) {
+	rs := testRuleset(t, ruleset.ExprRule{Name: "allow-443", Action: "allow", Expr: "port.dst == 443"})
+	w, err := newWorker(workerConfig{ID: 1, Logger: nullEngineLogger{}, Ruleset: rs})
+	if err != nil {
+		t.Fatal(err)
+	}
+	pkt := &layers.UDP{SrcPort: 40000, DstPort: 443}
+	pkt.Payload = []byte("hello")
+
+	if v, _ := w.handleUDP(7, testIPFlow, pkt); v != io.VerdictAcceptStream {
+		t.Fatalf("verdict = %v, want accept stream", v)
+	}
+	value, ok := w.udpStreamManager.streams.Get(7)
+	if !ok {
+		t.Fatal("stream missing from the LRU")
+	}
+	if s := value.Stream; s.activeEntries != nil || s.info.Props != nil || s.ruleset != nil {
+		t.Fatalf("udp stream still holds state after verdict: entries=%d props=%v", len(s.activeEntries), s.info.Props)
+	}
+	if v, _ := w.handleUDP(7, testIPFlow, pkt); v != io.VerdictAcceptStream {
+		t.Fatalf("verdict on a later packet = %v, want accept stream", v)
+	}
+}
+
+func TestTCPFlushIntervalFollowsTimeout(t *testing.T) {
+	cases := []struct {
+		timeout, want time.Duration
+	}{
+		{30 * time.Second, 15 * time.Second},
+		{10 * time.Minute, time.Minute},
+		{time.Second, time.Second},
+		{0, time.Second},
+	}
+	for _, c := range cases {
+		if got := tcpFlushIntervalFor(c.timeout); got != c.want {
+			t.Errorf("timeout %v: interval = %v, want %v", c.timeout, got, c.want)
+		}
+	}
+}

engine/tcp.go

@@ -124,8 +124,8 @@ type tcpStream struct {
 	logger        Logger
 	ruleset       ruleset.Ruleset
 	activeEntries []*tcpStreamEntry
-	doneEntries   []*tcpStreamEntry
 	lastVerdict   tcpVerdict
+	released      bool // 最终判定已发出，分析器和属性表已经释放
 	firstSegSeen  bool // true once the client's first data segment was measured
 	synAckWindow  int  // window of the observed SYN-ACK, -1 if none seen yet
 	synAckWscale  int  // its window scale factor, -1 if the option was absent
@@ -139,6 +139,11 @@ type tcpStreamEntry struct {
 }
 
 func (s *tcpStream) Accept(tcp *layers.TCP, ci gopacket.CaptureInfo, dir reassembly.TCPFlowDirection, nextSeq reassembly.Sequence, start *bool, ac reassembly.AssemblerContext) bool {
+	if s.released {
+		// 最终判定已发出，后续报文一律沿用；Props 已经释放，不能再碰
+		ac.(*tcpContext).Verdict = s.lastVerdict
+		return false
+	}
 	if tcp.SYN && tcp.ACK && s.synAckWindow < 0 {
 		s.synAckWindow, s.synAckWscale = int(tcp.Window), tcpWindowScale(tcp)
 		if s.synAckWindow <= tcpSynAckWindowMax {
@@ -210,6 +215,10 @@ func (s *tcpStream) recordFirstSegLen(n int) bool {
 }
 
 func (s *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.AssemblerContext) {
+	if s.released {
+		// 判定已发出。还会走到这里只可能是超时 flush 补送的乱序页，没有东西可处理
+		return
+	}
 	dir, start, end, skip := sg.Info()
 	rev := dir == reassembly.TCPDirServerToClient
 	avail, _ := sg.Lengths()
@@ -229,7 +238,6 @@ func (s *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 		updated = updated || up1 || up2
 		if done {
 			s.activeEntries = append(s.activeEntries[:i], s.activeEntries[i+1:]...)
-			s.doneEntries = append(s.doneEntries, entry)
 		}
 	}
 	ctx := ac.(*tcpContext)
@@ -254,15 +262,20 @@ func (s *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 		ctx.Verdict = tcpVerdictAcceptStream
 		s.logger.TCPStreamAction(s.info, ruleset.ActionAllow, true)
 	}
+	if len(s.activeEntries) == 0 {
+		// 不会再有分析器读数据，也不会再匹配规则：从这一刻起 Accept 只看 lastVerdict
+		s.release()
+	}
 }
 
 func (s *tcpStream) ReassemblyComplete(ac reassembly.AssemblerContext) bool {
 	s.closeActiveEntries()
+	s.release()
 	return true
 }
 
 func (s *tcpStream) closeActiveEntries() {
-	// Signal close to all active entries & move them to doneEntries
+	// Signal close to all active entries
 	updated := false
 	for _, entry := range s.activeEntries {
 		update := entry.Stream.Close(false)
@@ -272,10 +285,22 @@ func (s *tcpStream) closeActiveEntries() {
 	if updated {
 		s.logger.TCPStreamPropUpdate(s.info, true)
 	}
-	s.doneEntries = append(s.doneEntries, s.activeEntries...)
 	s.activeEntries = nil
 }
 
+// release 在最终判定发出后丢掉不会再用到的状态。
+//
+// 判定之后 Accept 只看 lastVerdict，分析器和属性表都不会再被读。但这条流本身还要在
+// 连接池里挂到超时被 flush，而 gopacket 回收连接槽位时并不清掉 stream 指针，
+// 要等下一条连接复用那个槽位才会覆盖。不主动放手的话，每条判定完的流会带着
+// 全部分析器对象一直活到那时候；节点上的 heap profile 里这部分占到了八成。
+func (s *tcpStream) release() {
+	s.released = true
+	s.activeEntries = nil
+	s.info.Props = nil
+	s.ruleset = nil
+}
+
 func (s *tcpStream) feedEntry(entry *tcpStreamEntry, rev, start, end bool, skip int, data []byte) (update *analyzer.PropUpdate, closeUpdate *analyzer.PropUpdate, done bool) {
 	if !entry.HasLimit {
 		update, done = entry.Stream.Feed(rev, start, end, skip, data)
@@ -287,7 +312,7 @@ func (s *tcpStream) feedEntry(entry *tcpStreamEntry, rev, start, end bool, skip
 		update, done = entry.Stream.Feed(rev, start, end, skip, qData)
 		entry.Quota -= len(qData)
 		if entry.Quota <= 0 {
-			// Quota exhausted, signal close & move to doneEntries
+			// Quota exhausted, signal close & drop the entry
 			closeUpdate = entry.Stream.Close(true)
 			done = true
 		}

engine/udp.go

@@ -160,7 +160,6 @@ type udpStream struct {
 	logger        Logger
 	ruleset       ruleset.Ruleset
 	activeEntries []*udpStreamEntry
-	doneEntries   []*udpStreamEntry
 	lastVerdict   udpVerdict
 }
 
@@ -194,7 +193,6 @@ func (s *udpStream) Feed(udp *layers.UDP, rev bool, uc *udpContext) {
 		updated = updated || up1 || up2
 		if done {
 			s.activeEntries = append(s.activeEntries[:i], s.activeEntries[i+1:]...)
-			s.doneEntries = append(s.doneEntries, entry)
 		}
 	}
 	if updated || s.virgin {
@@ -236,14 +234,19 @@ func (s *udpStream) Feed(udp *layers.UDP, rev bool, uc *udpContext) {
 		uc.Verdict = udpVerdictAcceptStream
 		s.logger.UDPStreamAction(s.info, ruleset.ActionAllow, true)
 	}
+	if len(s.activeEntries) == 0 {
+		// 同 tcpStream.release：这条流会在 LRU 里一直待到被挤出去，别让它带着分析器一起等
+		s.release()
+	}
 }
 
 func (s *udpStream) Close() {
 	s.closeActiveEntries()
+	s.release()
 }
 
 func (s *udpStream) closeActiveEntries() {
-	// Signal close to all active entries & move them to doneEntries
+	// Signal close to all active entries
 	updated := false
 	for _, entry := range s.activeEntries {
 		update := entry.Stream.Close(false)
@@ -253,16 +256,22 @@ func (s *udpStream) closeActiveEntries() {
 	if updated {
 		s.logger.UDPStreamPropUpdate(s.info, true)
 	}
-	s.doneEntries = append(s.doneEntries, s.activeEntries...)
 	s.activeEntries = nil
 }
 
+// release 在最终判定发出后丢掉不会再用到的状态，Accept 此后只看 lastVerdict
+func (s *udpStream) release() {
+	s.activeEntries = nil
+	s.info.Props = nil
+	s.ruleset = nil
+}
+
 func (s *udpStream) feedEntry(entry *udpStreamEntry, rev bool, data []byte) (update *analyzer.PropUpdate, closeUpdate *analyzer.PropUpdate, done bool) {
 	update, done = entry.Stream.Feed(rev, data)
 	if entry.HasLimit {
 		entry.Quota -= len(data)
 		if entry.Quota <= 0 {
-			// Quota exhausted, signal close & move to doneEntries
+			// Quota exhausted, signal close & drop the entry
 			closeUpdate = entry.Stream.Close(true)
 			done = true
 		}

engine/worker.go

@@ -20,9 +20,18 @@ const (
 	defaultTCPTimeout                       = 10 * time.Minute
 	defaultUDPMaxStreams                    = 4096
 
-	tcpFlushInterval = 1 * time.Minute
+	// 超时 flush 的间隔范围。判定后的流走 connmark 旁路，FIN 到不了用户态，
+	// 只能靠定时 flush 把它们从连接池里清掉。间隔取超时的一半：
+	// 之前固定 1 分钟，30 秒的超时实际会让流在池里活到 90 秒。
+	tcpMaxFlushInterval = 1 * time.Minute
+	tcpMinFlushInterval = 1 * time.Second
 )
 
+// tcpFlushIntervalFor 返回给定 TCP 超时下的 flush 间隔
+func tcpFlushIntervalFor(timeout time.Duration) time.Duration {
+	return min(tcpMaxFlushInterval, max(timeout/2, tcpMinFlushInterval))
+}
+
 type workerPacket struct {
 	StreamID   uint32
 	Packet     gopacket.Packet
@@ -122,7 +131,7 @@ func (w *worker) Run(ctx context.Context) {
 	w.logger.WorkerStart(w.id)
 	defer w.logger.WorkerStop(w.id)
 
-	tcpFlushTicker := time.NewTicker(tcpFlushInterval)
+	tcpFlushTicker := time.NewTicker(tcpFlushIntervalFor(w.tcpTimeout))
 	defer tcpFlushTicker.Stop()
 
 	for {