Compare commits

4 Commits
Author SHA1 Message Date
Mxmilu666 c879e10ac1 feat(tls): add support for RDP negotiation before TLS handshake 2026-09-30 16:54:53 +08:00
Mxmilu666andClaude Opus 5.5 ae7135ab64 feat(tls): reassemble hello messages split across TLS records
A handshake message may span any number of records, and splitting the
ClientHello into tiny records hides the SNI from analyzers that expect it
in the first one. The tls analyzer read the first record header and then
cut the handshake body straight out of the stream, so fragmented hellos
either stalled until Limit() or failed to parse, and every SNI rule was
bypassed.

The record layer is now peeled off and the hello rebuilt from the record
bodies before parsing, for both ClientHello and ServerHello. The number of
records a hello spanned is exposed as tls.req_records / tls.resp_records,
kept outside req/resp so a non-nil req still means a fully parsed hello.
Fragmentation is published as soon as a second record shows up, since
1-byte records can stretch a hello past Limit() and it never completes.

docs: add TLS record fragmentation doc, correct the claim in the TCP
window evasion doc that tls is unaffected by splitting.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-24 17:31:57 +08:00
Mxmilu666 8f8cc1d4d9 feat(tcp, udp): implement stream state release after verdict and adjust flush interval logic 2026-09-12 17:20:24 +08:00
Mxmilu666andClaude Fable 5.1 cc6ad72706 feat(udp): add EasyTier analyzer for tcp:// and udp:// transports
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
2026-09-12 17:05:39 +08:00
18 changed files with 2584 additions and 216 deletions
+1 -1
View File
@@ -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] 機械学習に基づくトラフィック分類
+1 -1
View File
@@ -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
+7 -1
View File
@@ -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,7 +37,10 @@ Telegram 群组: https://t.me/OpGFW
## 本仓库内的文档
- [Minecraft 分析器](docs/minecraft.zh.md)
- [EasyTier 分析器](docs/easytier.zh.md)
- [TCP 窗口操纵规避与检测](docs/tcp-window-evasion.zh.md)
- [TLS record 分片规避与检测](docs/tls-record-fragmentation.zh.md)
- [拦截 TLS、放行 RDP](docs/rdp-tls.zh.md)
## 使用场景
+183 -169
View File
@@ -22,38 +22,72 @@ func (a *TLSAnalyzer) NewTCP(info analyzer.TCPInfo, logger analyzer.Logger) anal
return newTLSStream(logger)
}
const (
// tlsRecordHeaderSize is content type (1 byte) + legacy protocol version
// (2 bytes) + length (2 bytes).
tlsRecordHeaderSize = 5
// tlsMaxPlaintextLen is the largest body a plaintext record may carry.
tlsMaxPlaintextLen = 1 << 14
// tlsHandshakeHeaderSize is message type (1 byte) + length (3 bytes).
tlsHandshakeHeaderSize = 4
)
type tlsStream struct {
logger analyzer.Logger
reqBuf *utils.ByteBuffer
reqMap analyzer.PropMap
reqUpdated bool
reqLSM *utils.LinearStateMachine
reqDone bool
req tlsHello
resp tlsHello
}
respBuf *utils.ByteBuffer
respMap analyzer.PropMap
respUpdated bool
respLSM *utils.LinearStateMachine
respDone bool
// tlsHello reads one hello message off one direction of the stream.
//
// A handshake message may be split across any number of records (RFC 8446
// 5.1), and splitting the ClientHello into tiny records is a common way to
// hide the SNI from middleboxes that expect it in the first record. So the
// record layer is peeled off first and the message is rebuilt from the record
// bodies before it is parsed.
type tlsHello struct {
rec utils.ByteBuffer // record layer bytes not consumed yet
msg utils.ByteBuffer // handshake layer bytes collected so far
lsm *utils.LinearStateMachine
clientHelloLen int
serverHelloLen int
msgType byte
x224Code byte // X.224 connection PDU that may precede the hello on RDP
minLen int // smallest valid message body
parse func(*utils.ByteBuffer) analyzer.PropMap // parses the message body
recLeft int // body bytes of the current record not consumed yet
msgLen int // message body length, -1 until the handshake header is read
records int // records that have carried the message so far
reported int // value of records last published
m analyzer.PropMap
updated bool
done bool
rdp analyzer.PropMap // RDP negotiation, nil if there was none
rdpUpdated bool
}
func newTLSStream(logger analyzer.Logger) *tlsStream {
s := &tlsStream{logger: logger, reqBuf: &utils.ByteBuffer{}, respBuf: &utils.ByteBuffer{}}
s.reqLSM = utils.NewLinearStateMachine(
s.tlsClientHelloPreprocess,
s.parseClientHelloData,
)
s.respLSM = utils.NewLinearStateMachine(
s.tlsServerHelloPreprocess,
s.parseServerHelloData,
)
s := &tlsStream{logger: logger}
// minimal ClientHello body: protocol version (2 bytes) + random (32 bytes) +
// + session ID (1 byte) + cipher suites (4 bytes) +
// + compression methods (2 bytes) + no extensions
s.req.init(internal.TypeClientHello, x224CodeConnectionRequest, 41, internal.ParseTLSClientHelloMsgData)
// minimal ServerHello body: server version (2 bytes) + random (32 bytes) +
// + session ID (>=1 byte) + cipher suite (2 bytes) +
// + compression method (1 byte) + no extensions
s.resp.init(internal.TypeServerHello, x224CodeConnectionConfirm, 38, internal.ParseTLSServerHelloMsgData)
return s
}
func (h *tlsHello) init(msgType, x224Code byte, minLen int, parse func(*utils.ByteBuffer) analyzer.PropMap) {
h.msgType, h.x224Code, h.minLen, h.parse = msgType, x224Code, minLen, parse
h.msgLen = -1
h.lsm = utils.NewLinearStateMachine(h.readPrelude, h.readMessage, h.parseMessage)
}
func (s *tlsStream) Feed(rev, start, end bool, skip int, data []byte) (u *analyzer.PropUpdate, done bool) {
if skip != 0 {
return nil, true
@@ -61,166 +95,146 @@ func (s *tlsStream) Feed(rev, start, end bool, skip int, data []byte) (u *analyz
if len(data) == 0 {
return nil, false
}
var update *analyzer.PropUpdate
var cancelled bool
h, key := &s.req, "req"
if rev {
s.respBuf.Append(data)
s.respUpdated = false
cancelled, s.respDone = s.respLSM.Run()
if s.respUpdated {
update = &analyzer.PropUpdate{
Type: analyzer.PropUpdateMerge,
M: analyzer.PropMap{"resp": s.respMap},
}
s.respUpdated = false
h, key = &s.resp, "resp"
}
if h.done {
return nil, false
}
h.rec.Append(data)
h.updated = false
var cancelled bool
cancelled, h.done = h.lsm.Run()
// A fragmented hello is published as soon as it shows up, not only once
// the message is complete: a peer can make it too large to ever complete
// within Limit(), and a rule should not need the SNI to act on it.
if m := h.update(key, h.updated || (h.records > 1 && h.reported <= 1)); m != nil {
u = &analyzer.PropUpdate{Type: analyzer.PropUpdateMerge, M: m}
}
return u, cancelled || (s.req.done && s.resp.done)
}
// update returns the properties to publish for this direction, or nil if
// there is nothing to publish. The parsed message goes under key, the record
// count under key + "_records". The record count is kept out of the message
// map so that a non-nil message map still means a fully parsed hello. An RDP
// negotiation goes under "rdp_" + key as soon as it is parsed, whatever
// publish says.
func (h *tlsHello) update(key string, publish bool) analyzer.PropMap {
m := analyzer.PropMap{}
if h.rdpUpdated {
m["rdp_"+key] = h.rdp
h.rdpUpdated = false
}
if publish && h.records > 0 {
m[key+"_records"] = h.records
if h.updated {
m[key] = h.m
h.updated = false
}
} else {
s.reqBuf.Append(data)
s.reqUpdated = false
cancelled, s.reqDone = s.reqLSM.Run()
if s.reqUpdated {
update = &analyzer.PropUpdate{
Type: analyzer.PropUpdateMerge,
M: analyzer.PropMap{"req": s.reqMap},
h.reported = h.records
}
if len(m) == 0 {
return nil
}
return m
}
// readMessage consumes records until a whole hello message is collected in
// h.msg.
func (h *tlsHello) readMessage() utils.LSMAction {
for {
if h.msgLen < 0 && h.msg.Len() >= tlsHandshakeHeaderSize {
header := h.msg.Buf
if header[0] != h.msgType {
return utils.LSMActionCancel
}
s.reqUpdated = false
h.msgLen = int(header[1])<<16 | int(header[2])<<8 | int(header[3])
if h.msgLen < h.minLen {
return utils.LSMActionCancel
}
}
if h.msgLen >= 0 && h.msg.Len() >= tlsHandshakeHeaderSize+h.msgLen {
return utils.LSMActionNext
}
if h.recLeft == 0 {
header, ok := h.rec.Get(tlsRecordHeaderSize, true)
if !ok {
return utils.LSMActionPause
}
if header[0] != internal.RecordTypeHandshake {
// Nothing else may come between the records of a handshake
// message.
return utils.LSMActionCancel
}
h.recLeft = int(header[3])<<8 | int(header[4])
if h.recLeft > tlsMaxPlaintextLen {
return utils.LSMActionCancel
}
// Empty records are forbidden for handshake messages, but still
// counted rather than rejected: a peer that sends them anyway is
// fragmenting, and rejecting them would stop the analysis.
h.records++
continue
}
n := min(h.recLeft, h.rec.Len())
if n == 0 {
return utils.LSMActionPause
}
body, _ := h.rec.Get(n, true)
h.recLeft -= n
if h.msg.Len() == 0 {
// Usually the whole message sits in this one chunk, so reference
// it instead of copying. The capacity is capped so that a later
// Append reallocates rather than writing into h.rec.
h.msg.Buf = body[:n:n]
} else {
h.msg.Append(body)
}
}
return update, cancelled || (s.reqDone && s.respDone)
}
// tlsClientHelloPreprocess validates ClientHello message.
// parseMessage converts the collected hello message into h.m.
//
// During validation, message header and first handshake header may be removed
// from `s.reqBuf`.
func (s *tlsStream) tlsClientHelloPreprocess() utils.LSMAction {
// headers size: content type (1 byte) + legacy protocol version (2 bytes) +
// + content length (2 bytes) + message type (1 byte) +
// + handshake length (3 bytes)
const headersSize = 9
// minimal data size: protocol version (2 bytes) + random (32 bytes) +
// + session ID (1 byte) + cipher suites (4 bytes) +
// + compression methods (2 bytes) + no extensions
const minDataSize = 41
header, ok := s.reqBuf.Get(headersSize, true)
if !ok {
// not a full header yet
return utils.LSMActionPause
}
if header[0] != internal.RecordTypeHandshake || header[5] != internal.TypeClientHello {
return utils.LSMActionCancel
}
s.clientHelloLen = int(header[6])<<16 | int(header[7])<<8 | int(header[8])
if s.clientHelloLen < minDataSize {
return utils.LSMActionCancel
}
// TODO: something is missing. See:
// const messageHeaderSize = 4
// fullMessageLen := int(header[3])<<8 | int(header[4])
// msgNo := fullMessageLen / int(messageHeaderSize+s.serverHelloLen)
// if msgNo != 1 {
// // what here?
// }
// if messageNo != int(messageNo) {
// // what here?
// }
return utils.LSMActionNext
}
// tlsServerHelloPreprocess validates ServerHello message.
//
// During validation, message header and first handshake header may be removed
// from `s.reqBuf`.
func (s *tlsStream) tlsServerHelloPreprocess() utils.LSMAction {
// header size: content type (1 byte) + legacy protocol version (2 byte) +
// + content length (2 byte) + message type (1 byte) +
// + handshake length (3 byte)
const headersSize = 9
// minimal data size: server version (2 byte) + random (32 byte) +
// + session ID (>=1 byte) + cipher suite (2 byte) +
// + compression method (1 byte) + no extensions
const minDataSize = 38
header, ok := s.respBuf.Get(headersSize, true)
if !ok {
// not a full header yet
return utils.LSMActionPause
}
if header[0] != internal.RecordTypeHandshake || header[5] != internal.TypeServerHello {
return utils.LSMActionCancel
}
s.serverHelloLen = int(header[6])<<16 | int(header[7])<<8 | int(header[8])
if s.serverHelloLen < minDataSize {
return utils.LSMActionCancel
}
// TODO: something is missing. See example:
// const messageHeaderSize = 4
// fullMessageLen := int(header[3])<<8 | int(header[4])
// msgNo := fullMessageLen / int(messageHeaderSize+s.serverHelloLen)
// if msgNo != 1 {
// // what here?
// }
// if messageNo != int(messageNo) {
// // what here?
// }
return utils.LSMActionNext
}
// parseClientHelloData converts valid ClientHello message data (without
// headers) into `analyzer.PropMap`.
//
// Parsing error may leave `s.reqBuf` in an unusable state.
func (s *tlsStream) parseClientHelloData() utils.LSMAction {
chBuf, ok := s.reqBuf.GetSubBuffer(s.clientHelloLen, true)
if !ok {
// Not a full client hello yet
return utils.LSMActionPause
}
m := internal.ParseTLSClientHelloMsgData(chBuf)
// Anything after the message, such as the rest of the server's flight, is
// ignored.
func (h *tlsHello) parseMessage() utils.LSMAction {
body := &utils.ByteBuffer{Buf: h.msg.Buf[tlsHandshakeHeaderSize : tlsHandshakeHeaderSize+h.msgLen]}
m := h.parse(body)
h.rec.Reset()
h.msg.Reset()
if m == nil {
return utils.LSMActionCancel
} else {
s.reqUpdated = true
s.reqMap = m
return utils.LSMActionNext
}
}
// parseServerHelloData converts valid ServerHello message data (without
// headers) into `analyzer.PropMap`.
//
// Parsing error may leave `s.respBuf` in an unusable state.
func (s *tlsStream) parseServerHelloData() utils.LSMAction {
shBuf, ok := s.respBuf.GetSubBuffer(s.serverHelloLen, true)
if !ok {
// Not a full server hello yet
return utils.LSMActionPause
}
m := internal.ParseTLSServerHelloMsgData(shBuf)
if m == nil {
return utils.LSMActionCancel
} else {
s.respUpdated = true
s.respMap = m
return utils.LSMActionNext
}
h.m = m
h.updated = true
return utils.LSMActionNext
}
func (s *tlsStream) Close(limited bool) *analyzer.PropUpdate {
s.reqBuf.Reset()
s.respBuf.Reset()
s.reqMap = nil
s.respMap = nil
return nil
// Report how far a fragmented hello got if it never completed, e.g.
// because it was stretched past Limit().
m := s.req.update("req", s.req.records > 1 && s.req.records > s.req.reported)
if rm := s.resp.update("resp", s.resp.records > 1 && s.resp.records > s.resp.reported); rm != nil {
if m == nil {
m = rm
} else {
for k, v := range rm {
m[k] = v
}
}
}
s.req.close()
s.resp.close()
if m == nil {
return nil
}
return &analyzer.PropUpdate{Type: analyzer.PropUpdateMerge, M: m}
}
func (h *tlsHello) close() {
h.rec.Reset()
h.msg.Reset()
h.m = nil
}
+123
View File
@@ -0,0 +1,123 @@
package tcp
import (
"bytes"
"encoding/binary"
"github.com/apernet/OpenGFW/analyzer"
"github.com/apernet/OpenGFW/analyzer/internal"
"github.com/apernet/OpenGFW/analyzer/utils"
)
// RDP does not start TLS right away: the client first sends an X.224
// Connection Request and the server answers with a Connection Confirm, each in
// a TPKT packet, to agree on a security protocol ([MS-RDPBCGR] 1.3.1.1). Only
// then does the TLS handshake begin, on the same connection.
//
// The tls analyzer used to give up on the first byte of such a stream, which
// left RDP invisible to TLS rules, and with it anything that puts a fake RDP
// negotiation in front of its TLS. So the negotiation is parsed as an optional
// prelude of the hello, and published as rdp_req / rdp_resp.
const (
tpktVersion = 0x03
tpktHeaderSize = 4 // version (1 byte) + reserved (1 byte) + length (2 bytes)
// x224FixedSize is length indicator (1 byte) + code (1 byte) +
// DST-REF (2 bytes) + SRC-REF (2 bytes) + class option (1 byte).
x224FixedSize = 7
x224CodeConnectionRequest = 0xe0
x224CodeConnectionConfirm = 0xd0
rdpNegReq = 0x01
rdpNegRsp = 0x02
rdpNegFailure = 0x03
rdpNegSize = 8
)
var rdpCookiePrefix = []byte("Cookie: ")
// readPrelude consumes the X.224 connection PDU that precedes the hello on an
// RDP connection, if there is one.
func (h *tlsHello) readPrelude() utils.LSMAction {
b, ok := h.rec.GetByte(false)
if !ok {
return utils.LSMActionPause
}
if b == internal.RecordTypeHandshake {
return utils.LSMActionNext
}
if b != tpktVersion {
return utils.LSMActionCancel
}
header, ok := h.rec.Get(tpktHeaderSize, false)
if !ok {
return utils.LSMActionPause
}
n := int(binary.BigEndian.Uint16(header[2:4]))
if header[1] != 0 || n < tpktHeaderSize+x224FixedSize {
return utils.LSMActionCancel
}
pdu, ok := h.rec.Get(n, true)
if !ok {
return utils.LSMActionPause
}
m := parseX224Connection(pdu[tpktHeaderSize:], h.x224Code)
if m == nil {
return utils.LSMActionCancel
}
h.rdp = m
h.rdpUpdated = true
return utils.LSMActionNext
}
// parseX224Connection parses an X.224 Connection Request or Confirm TPDU
// carrying the RDP negotiation, or returns nil if tpdu is not one.
func parseX224Connection(tpdu []byte, code byte) analyzer.PropMap {
// The length indicator does not count itself. The low nibble of the code
// is the credit, and the high nibble of the class option the class, which
// is always 0 for RDP.
if int(tpdu[0]) != len(tpdu)-1 || tpdu[1]&0xf0 != code || tpdu[6]&0xf0 != 0 {
return nil
}
rest := tpdu[x224FixedSize:]
m := analyzer.PropMap{}
if code == x224CodeConnectionRequest {
// Both the cookie ("Cookie: mstshash=<user>") and the load balancer
// routing token ("Cookie: msts=<token>") are terminated by CRLF.
if bytes.HasPrefix(rest, rdpCookiePrefix) {
i := bytes.Index(rest, []byte("\r\n"))
if i < 0 {
return nil
}
m["cookie"] = string(rest[len(rdpCookiePrefix):i])
rest = rest[i+2:]
}
// A client that sends no RDP_NEG_REQ can only do standard RDP
// security (PROTOCOL_RDP).
m["protocols"] = 0
} else {
m["protocol"] = 0
}
if len(rest) == 0 {
return m
}
// Anything after the negotiation structure, such as the correlation info
// a client may append, is ignored.
if len(rest) < rdpNegSize || binary.LittleEndian.Uint16(rest[2:4]) != rdpNegSize {
return nil
}
v := int(binary.LittleEndian.Uint32(rest[4:8]))
switch {
case code == x224CodeConnectionRequest && rest[0] == rdpNegReq:
m["protocols"] = v
case code == x224CodeConnectionConfirm && rest[0] == rdpNegRsp:
m["protocol"] = v
case code == x224CodeConnectionConfirm && rest[0] == rdpNegFailure:
delete(m, "protocol")
m["failure"] = v
default:
return nil
}
return m
}
+131
View File
@@ -0,0 +1,131 @@
package tcp
import (
"reflect"
"testing"
"github.com/apernet/OpenGFW/analyzer"
)
// rdpConnectionRequest is what mstsc sends first: a cookie with the user name
// and an RDP_NEG_REQ asking for PROTOCOL_SSL | PROTOCOL_HYBRID |
// PROTOCOL_HYBRID_EX.
var rdpConnectionRequest = []byte{
0x03, 0x00, 0x00, 0x2c, // TPKT, length 44
0x27, 0xe0, 0x00, 0x00, 0x00, 0x00, 0x00, // X.224 CR
'C', 'o', 'o', 'k', 'i', 'e', ':', ' ', 'm', 's', 't', 's', 'h', 'a', 's', 'h', '=', 'e', 'l', 't', 'o', 'n', 's', '\r', '\n',
0x01, 0x00, 0x08, 0x00, 0x0b, 0x00, 0x00, 0x00, // RDP_NEG_REQ
}
func rdpConnectionConfirm(negType byte, v byte) []byte {
return []byte{
0x03, 0x00, 0x00, 0x13, // TPKT, length 19
0x0e, 0xd0, 0x00, 0x00, 0x12, 0x34, 0x00, // X.224 CC
negType, 0x00, 0x08, 0x00, v, 0x00, 0x00, 0x00,
}
}
var rdpTLSClientHelloProps = analyzer.PropMap{"cookie": "mstshash=eltons", "protocols": 11}
func TestTlsRDPClientHello(t *testing.T) {
data := append(append([]byte{}, rdpConnectionRequest...), tlsExampleClientHello...)
for name, segs := range map[string][][]byte{
"one segment": {data},
"byte by byte": splitEvery(data, 1),
"pdu, hello": {rdpConnectionRequest, tlsExampleClientHello},
} {
props, _ := feedTLS(newTLSStream(nil), false, segs...)
if got := props.Get("rdp_req"); !reflect.DeepEqual(got, rdpTLSClientHelloProps) {
t.Errorf("%s: rdp_req = %v, want %v", name, got, rdpTLSClientHelloProps)
}
if got := props.Get("req"); !reflect.DeepEqual(got, tlsExampleClientHelloProps) {
t.Errorf("%s: req = %v, want %v", name, got, tlsExampleClientHelloProps)
}
}
}
// The negotiation is published before any TLS shows up, so that rules can
// already see the stream is RDP.
func TestTlsRDPNegotiationReportedEarly(t *testing.T) {
s := newTLSStream(nil)
props, done := feedTLS(s, false, rdpConnectionRequest)
if done {
t.Fatal("stream done after connection request")
}
if got := props.Get("rdp_req"); !reflect.DeepEqual(got, rdpTLSClientHelloProps) {
t.Errorf("rdp_req = %v, want %v", got, rdpTLSClientHelloProps)
}
props, done = feedTLS(s, true, rdpConnectionConfirm(0x02, 0x02))
if done {
t.Fatal("stream done after connection confirm")
}
if got, want := props.Get("rdp_resp"), (analyzer.PropMap{"protocol": 2}); !reflect.DeepEqual(got, want) {
t.Errorf("rdp_resp = %v, want %v", got, want)
}
}
func TestTlsRDPConnectionConfirm(t *testing.T) {
for name, tc := range map[string]struct {
pdu []byte
want analyzer.PropMap
}{
"ssl": {rdpConnectionConfirm(0x02, 0x01), analyzer.PropMap{"protocol": 1}},
"hybrid_ex": {rdpConnectionConfirm(0x02, 0x08), analyzer.PropMap{"protocol": 8}},
"failure": {rdpConnectionConfirm(0x03, 0x05), analyzer.PropMap{"failure": 5}},
"no neg rsp": {[]byte{0x03, 0x00, 0x00, 0x0b, 0x06, 0xd0, 0x00, 0x00, 0x12, 0x34, 0x00}, analyzer.PropMap{"protocol": 0}},
} {
props, _ := feedTLS(newTLSStream(nil), true, tc.pdu)
if got := props.Get("rdp_resp"); !reflect.DeepEqual(got, tc.want) {
t.Errorf("%s: rdp_resp = %v, want %v", name, got, tc.want)
}
}
}
// Standard RDP security has no TLS: the client goes on with MCS, which is not
// a TLS record, so the analyzer stops with only the negotiation published.
func TestTlsRDPWithoutTLS(t *testing.T) {
mcs := []byte{0x03, 0x00, 0x01, 0xa0, 0x02, 0xf0, 0x80, 0x7f, 0x65}
props, done := feedTLS(newTLSStream(nil), false, rdpConnectionRequest, mcs)
if !done {
t.Error("stream not done")
}
if props.Get("rdp_req") == nil || props.Get("req") != nil {
t.Errorf("props = %v, want rdp_req only", props)
}
}
// truncatedNegReq is rdpConnectionRequest with the last two bytes of the
// RDP_NEG_REQ cut off and the lengths adjusted to match.
func truncatedNegReq() []byte {
pdu := append([]byte{}, rdpConnectionRequest[:len(rdpConnectionRequest)-2]...)
pdu[3] -= 2
pdu[4] -= 2
return pdu
}
func TestTlsRDPRejectsMalformedPrelude(t *testing.T) {
mutate := func(i int, b byte) []byte {
pdu := append([]byte{}, rdpConnectionRequest...)
pdu[i] = b
return pdu
}
for name, pdu := range map[string][]byte{
"reserved set": mutate(1, 0x01),
"wrong code": mutate(5, 0xd0),
"length mismatch": mutate(4, 0x26),
"class 2": mutate(10, 0x20),
"cookie w/o crlf": mutate(len(rdpConnectionRequest)-10, 'x'),
"bad neg type": mutate(len(rdpConnectionRequest)-8, 0x02),
"tpkt too short": {0x03, 0x00, 0x00, 0x05, 0x00},
"not tpkt nor tls": {0x05, 0x01, 0x00},
"truncated neg req": truncatedNegReq(),
} {
props, done := feedTLS(newTLSStream(nil), false, pdu)
if !done {
t.Errorf("%s: stream not done", name)
}
if len(props) != 0 {
t.Errorf("%s: props = %v, want none", name, props)
}
}
}
+150 -31
View File
@@ -7,39 +7,42 @@ import (
"github.com/apernet/OpenGFW/analyzer"
)
func TestTlsStreamParsing_ClientHello(t *testing.T) {
// example packet taken from <https://tls12.xargs.org/#client-hello/annotated>
clientHello := []byte{
0x16, 0x03, 0x01, 0x00, 0xa5, 0x01, 0x00, 0x00, 0xa1, 0x03, 0x03, 0x00,
0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c,
0x0d, 0x0e, 0x0f, 0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18,
0x19, 0x1a, 0x1b, 0x1c, 0x1d, 0x1e, 0x1f, 0x00, 0x00, 0x20, 0xcc, 0xa8,
0xcc, 0xa9, 0xc0, 0x2f, 0xc0, 0x30, 0xc0, 0x2b, 0xc0, 0x2c, 0xc0, 0x13,
0xc0, 0x09, 0xc0, 0x14, 0xc0, 0x0a, 0x00, 0x9c, 0x00, 0x9d, 0x00, 0x2f,
0x00, 0x35, 0xc0, 0x12, 0x00, 0x0a, 0x01, 0x00, 0x00, 0x58, 0x00, 0x00,
0x00, 0x18, 0x00, 0x16, 0x00, 0x00, 0x13, 0x65, 0x78, 0x61, 0x6d, 0x70,
0x6c, 0x65, 0x2e, 0x75, 0x6c, 0x66, 0x68, 0x65, 0x69, 0x6d, 0x2e, 0x6e,
0x65, 0x74, 0x00, 0x05, 0x00, 0x05, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00,
0x0a, 0x00, 0x0a, 0x00, 0x08, 0x00, 0x1d, 0x00, 0x17, 0x00, 0x18, 0x00,
0x19, 0x00, 0x0b, 0x00, 0x02, 0x01, 0x00, 0x00, 0x0d, 0x00, 0x12, 0x00,
0x10, 0x04, 0x01, 0x04, 0x03, 0x05, 0x01, 0x05, 0x03, 0x06, 0x01, 0x06,
0x03, 0x02, 0x01, 0x02, 0x03, 0xff, 0x01, 0x00, 0x01, 0x00, 0x00, 0x12,
0x00, 0x00,
}
want := analyzer.PropMap{
"ciphers": []uint16{52392, 52393, 49199, 49200, 49195, 49196, 49171, 49161, 49172, 49162, 156, 157, 47, 53, 49170, 10},
"compression": []uint8{0},
"random": []uint8{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31},
"session": []uint8{},
"sni": "example.ulfheim.net",
"version": uint16(771),
}
// tlsExampleClientHello is taken from <https://tls12.xargs.org/#client-hello/annotated>.
var tlsExampleClientHello = []byte{
0x16, 0x03, 0x01, 0x00, 0xa5, 0x01, 0x00, 0x00, 0xa1, 0x03, 0x03, 0x00,
0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c,
0x0d, 0x0e, 0x0f, 0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18,
0x19, 0x1a, 0x1b, 0x1c, 0x1d, 0x1e, 0x1f, 0x00, 0x00, 0x20, 0xcc, 0xa8,
0xcc, 0xa9, 0xc0, 0x2f, 0xc0, 0x30, 0xc0, 0x2b, 0xc0, 0x2c, 0xc0, 0x13,
0xc0, 0x09, 0xc0, 0x14, 0xc0, 0x0a, 0x00, 0x9c, 0x00, 0x9d, 0x00, 0x2f,
0x00, 0x35, 0xc0, 0x12, 0x00, 0x0a, 0x01, 0x00, 0x00, 0x58, 0x00, 0x00,
0x00, 0x18, 0x00, 0x16, 0x00, 0x00, 0x13, 0x65, 0x78, 0x61, 0x6d, 0x70,
0x6c, 0x65, 0x2e, 0x75, 0x6c, 0x66, 0x68, 0x65, 0x69, 0x6d, 0x2e, 0x6e,
0x65, 0x74, 0x00, 0x05, 0x00, 0x05, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00,
0x0a, 0x00, 0x0a, 0x00, 0x08, 0x00, 0x1d, 0x00, 0x17, 0x00, 0x18, 0x00,
0x19, 0x00, 0x0b, 0x00, 0x02, 0x01, 0x00, 0x00, 0x0d, 0x00, 0x12, 0x00,
0x10, 0x04, 0x01, 0x04, 0x03, 0x05, 0x01, 0x05, 0x03, 0x06, 0x01, 0x06,
0x03, 0x02, 0x01, 0x02, 0x03, 0xff, 0x01, 0x00, 0x01, 0x00, 0x00, 0x12,
0x00, 0x00,
}
var tlsExampleClientHelloProps = analyzer.PropMap{
"ciphers": []uint16{52392, 52393, 49199, 49200, 49195, 49196, 49171, 49161, 49172, 49162, 156, 157, 47, 53, 49170, 10},
"compression": []uint8{0},
"random": []uint8{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31},
"session": []uint8{},
"sni": "example.ulfheim.net",
"version": uint16(771),
}
func TestTlsStreamParsing_ClientHello(t *testing.T) {
s := newTLSStream(nil)
u, _ := s.Feed(false, false, false, 0, clientHello)
got := u.M.Get("req")
if !reflect.DeepEqual(got, want) {
t.Errorf("%d B parsed = %v, want %v", len(clientHello), got, want)
u, _ := s.Feed(false, false, false, 0, tlsExampleClientHello)
if got := u.M.Get("req"); !reflect.DeepEqual(got, tlsExampleClientHelloProps) {
t.Errorf("parsed = %v, want %v", got, tlsExampleClientHelloProps)
}
if got := u.M.Get("req_records"); got != 1 {
t.Errorf("req_records = %v, want 1", got)
}
}
@@ -67,3 +70,119 @@ func TestTlsStreamParsing_ServerHello(t *testing.T) {
t.Errorf("%d B parsed = %v, want %v", len(serverHello), got, want)
}
}
// tlsFragment re-encodes the handshake message in a single-record hello as
// records carrying at most size bytes of it each.
func tlsFragment(hello []byte, size int) []byte {
var out []byte
for msg := hello[5:]; len(msg) > 0; {
n := min(size, len(msg))
out = append(out, hello[0], hello[1], hello[2], byte(n>>8), byte(n))
out = append(out, msg[:n]...)
msg = msg[n:]
}
return out
}
// feedTLS feeds each segment in turn and returns the properties as the engine
// would have merged them, along with whether the stream said it was done.
func feedTLS(s *tlsStream, rev bool, segs ...[]byte) (analyzer.PropMap, bool) {
props := analyzer.PropMap{}
done := false
for _, seg := range segs {
var u *analyzer.PropUpdate
u, done = s.Feed(rev, false, false, 0, seg)
if u != nil {
for k, v := range u.M {
props[k] = v
}
}
}
return props, done
}
// Splitting the ClientHello across records must not hide it: the parse result
// has to be the same wherever the record boundaries fall.
func TestTlsClientHelloRecordFragmentation(t *testing.T) {
msgLen := len(tlsExampleClientHello) - 5
for _, size := range []int{1, 2, 3, 4, 5, 9, 50, msgLen - 1} {
data := tlsFragment(tlsExampleClientHello, size)
wantRecords := (msgLen + size - 1) / size
for name, segs := range map[string][][]byte{
"one segment": {data},
"byte by byte": splitEvery(data, 1),
"7 byte segs": splitEvery(data, 7),
} {
props, _ := feedTLS(newTLSStream(nil), false, segs...)
if got := props.Get("req"); !reflect.DeepEqual(got, tlsExampleClientHelloProps) {
t.Errorf("size %d, %s: parsed = %v, want %v", size, name, got, tlsExampleClientHelloProps)
}
if got := props.Get("req_records"); got != wantRecords {
t.Errorf("size %d, %s: req_records = %v, want %d", size, name, got, wantRecords)
}
}
}
}
// Fragmentation is published before the hello is complete, so that a rule can
// act on it even if the hello never completes.
func TestTlsClientHelloFragmentationReportedEarly(t *testing.T) {
data := tlsFragment(tlsExampleClientHello, 1)
s := newTLSStream(nil)
props, done := feedTLS(s, false, data[:12])
if done {
t.Fatal("stream done after two records")
}
if props.Get("req") != nil {
t.Errorf("req published early: %v", props.Get("req"))
}
if got := props.Get("req_records"); got != 2 {
t.Errorf("req_records = %v, want 2", got)
}
// Further records are not published one by one...
if u, _ := s.Feed(false, false, false, 0, data[12:60]); u != nil {
t.Errorf("unexpected update %v", u.M)
}
// ...but the count reached is, if the stream gets cut off.
u := s.Close(true)
if u == nil {
t.Fatal("no update on close")
}
if got := u.M.Get("req_records"); got != 10 {
t.Errorf("req_records on close = %v, want 10", got)
}
}
func TestTlsClientHelloRejectsInterleavedRecords(t *testing.T) {
data := tlsFragment(tlsExampleClientHello, 50)
// Turn the second record into an alert.
data[5+50] = 0x15
props, done := feedTLS(newTLSStream(nil), false, data)
if !done {
t.Error("stream not done")
}
if props.Get("req") != nil {
t.Errorf("req = %v, want nil", props.Get("req"))
}
}
func TestTlsStreamIgnoresUnfragmentedPartialHelloOnClose(t *testing.T) {
s := newTLSStream(nil)
if props, _ := feedTLS(s, false, tlsExampleClientHello[:100]); len(props) != 0 {
t.Errorf("props = %v, want none", props)
}
if u := s.Close(true); u != nil {
t.Errorf("close update = %v, want nil", u.M)
}
}
func splitEvery(data []byte, n int) [][]byte {
var segs [][]byte
for len(data) > 0 {
k := min(n, len(data))
segs = append(segs, data[:k])
data = data[k:]
}
return segs
}
+766
View File
@@ -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
}
+562
View File
@@ -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")
}
+1
View File
@@ -96,6 +96,7 @@ var analyzers = []analyzer.Analyzer{
&tcp.TLSAnalyzer{},
&tcp.TrojanAnalyzer{},
&udp.DNSAnalyzer{},
&udp.EasyTierAnalyzer{},
&udp.OpenVPNAnalyzer{},
&udp.QUICAnalyzer{},
&udp.WireGuardAnalyzer{},
+241
View File
@@ -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
```
+93
View File
@@ -0,0 +1,93 @@
# 拦截 TLS、放行 RDP
## 问题
RDP 不是一连上就开始 TLS:客户端先在 TPKT 里发一个 X.224 Connection Request,服务端回一个 Connection Confirm,双方借此协商安全协议([MS-RDPBCGR] 1.3.1.1),协商完才在同一条连接上开始 TLS 握手。
`tls` 分析器以前要求流的第一个字节就是 TLS record(`0x16`),碰到 RDP 开头的 `03 00` 直接放弃。结果是:
- RDP 里的 TLS 对 `tls` 规则完全不可见,没法按 RDP 单独写规则;
- 反过来,任何流量只要在 TLS 前面垫一段假的 X.224 协商,就能绕过"拦截所有 TLS"的规则。
## 做法
`tls` 分析器现在把 RDP 协商当作 hello 之前可选的一段前导来解析:
- 第一个字节是 `0x16`:和以前一样,直接当 TLS 解析。
- 第一个字节是 `0x03`:按 TPKT + X.224 解析。客户端方向要求是 Connection Request,服务端方向要求是 Connection Confirm;长度、class、协商结构任何一处不对都视为非法,分析器取消。
- 前导解析完后照常解析 ClientHello / ServerHello,record 分片重组等逻辑完全一样。
协商一解析出来就上报,不用等 TLS:
| 属性 | 类型 | 说明 |
| --- | --- | --- |
| `tls.rdp_req.cookie` | string | Connection Request 里的 cookie,去掉 `Cookie: ` 前缀和 CRLF,如 `mstshash=alice`;没有 cookie 时不存在 |
| `tls.rdp_req.protocols` | int | RDP_NEG_REQ 里客户端请求的协议位图;没有 RDP_NEG_REQ 时为 0 |
| `tls.rdp_resp.protocol` | int | RDP_NEG_RSP 里服务端选定的协议;没有 RDP_NEG_RSP 时为 0 |
| `tls.rdp_resp.failure` | int | 服务端回的是 RDP_NEG_FAILURE 时的失败码(此时没有 `protocol`) |
协议取值:
| 值 | 名称 | 是否走 TLS |
| --- | --- | --- |
| 0 | PROTOCOL_RDP(标准 RDP 安全,RC4) | 否 |
| 1 | PROTOCOL_SSL | 是 |
| 2 | PROTOCOL_HYBRID(CredSSP / NLA) | 是 |
| 4 | PROTOCOL_RDSTLS | 是 |
| 8 | PROTOCOL_HYBRID_EX | 是 |
| 16 | PROTOCOL_RDSAAD | 是 |
## 规则
规则按顺序匹配、第一条命中的生效,所以放行 RDP 的规则要写在拦截 TLS 的前面:
```yaml
- name: allow-rdp-tls
action: allow
log: true
expr: >
tls != nil && tls.req != nil &&
tls.rdp_req != nil && tls.rdp_resp != nil &&
tls.rdp_resp.protocol in [1, 2, 4, 8, 16]
- name: block-tls
action: block
log: true
expr: tls != nil && tls.req != nil
```
放行条件同时要求:
- **`tls.req != nil`**:ClientHello 已经完整解析。只看到协商就放行的话,协商之后跟什么都行。
- **`tls.rdp_resp != nil` 且 `protocol` 是走 TLS 的协议**:真实的 RDP 只有在服务端选了 TLS 类协议之后才会发 ClientHello。服务端选了 PROTOCOL_RDP 或者回了协商失败,后面却出现 TLS,说明是伪造的,交给 `block-tls` 拦掉。
正常 RDP 的时序是 CR → CC → ClientHello,ClientHello 到的时候 `rdp_resp` 早已上报,不存在规则先看到 `req` 再看到 `rdp_resp` 的竞争。
标准 RDP 安全(`protocol == 0`)的连接里根本没有 TLS,两条规则都不会命中,按默认放行。
### 收紧放行条件
上面的规则只验证"协商格式正确、协商结果和后续 TLS 一致"。客户端和服务端串通的话,完全可以按规范发一套协商再跑任意 TLS,所以按需叠加:
```yaml
# 只放行标准端口
expr: ... && port.dst == 3389
# 只放行指定服务器
expr: ... && ip.dst in ["10.0.0.10", "10.0.0.11"]
# 要求客户端带 mstshash cookie(mstsc 通常会带,但不保证,先观察日志再启用)
expr: ... && tls.rdp_req.cookie != nil && tls.rdp_req.cookie startsWith "mstshash="
```
## 行为变化
**以前按 `tls != nil && tls.req != nil` 写的规则现在会命中 RDP。** 以前 RDP 里的 TLS 看不到,所以不受影响;现在看得到了。不想拦 RDP 的话,必须像上面那样在前面加一条放行规则。
## 局限
**只认 RDP 这一种前导。** 其他协议里的 STARTTLS(SMTP、IMAP、XMPP 等)和任何非 TLS、非 TPKT 开头的前缀仍然会让 `tls` 分析器直接放弃。要做到"拦截所有 TLS",还需要配合 `fet` 等不依赖协议开头的检测。
**RD Gateway 不在此列。** 经由 RD 网关的 RDP 跑在 443 端口的 HTTPS 上,开头就是普通 TLS,没有 X.224 协商,只能按 SNI / IP 放行。
**RDP 的 UDP 传输不在此列。** RDP-UDP 用的是 DTLS,走 UDP,与 `tls` 分析器无关。
+1 -1
View File
@@ -28,7 +28,7 @@ table inet filter {
- **快路径**:首个分段本身就够长时(正常流量的首包都是几百字节),直接在原始切片上分类,不拷贝、不分配,开销与改动前完全一致。只有小于 16 字节的首段才会走缓冲。
- **段数上限**:最多等 `fetMaxSegments`(4)个分段。否则一个只发 1 字节然后装死的对端,就能让流一直留在活跃状态、拿不到 connmark 快速通道,直到 TCP 超时(默认 10 分钟)为止——那会变成一个可被主动利用的资源消耗面。
`http`、`tls`、`minecraft` 等分析器本来就用 `utils.ByteBuffer` 做增量解析,不受这个手法影响。
`http`、`tls`、`minecraft` 等分析器本来就用 `utils.ByteBuffer` 做增量解析,不受这个手法影响。注意这只针对 TCP 分段;TLS 自己记录层上的分片是另一回事,见 [TLS record 分片规避与检测](tls-record-fragmentation.zh.md)。
## 第二层防御:把窗口改写本身当指纹
+70
View File
@@ -0,0 +1,70 @@
# TLS record 分片规避与检测
## 规避手法
TLS 允许一个握手消息跨任意多个 record 传输(RFC 8446 §5.1)。客户端把 ClientHello 拆成很多个小 record,服务端照常把它们拼回去完成握手;但只看第一个 record 的 DPI 就拿不到完整的 ClientHello,SNI 也就看不到了。Xray 的 `fragment`(`packets: "tlshello"`)以及各类"TLS 分片"工具用的都是这个手法。
它和 [TCP 窗口操纵](tcp-window-evasion.zh.md) 是两层不同的东西:TCP 分段切的是字节流,OpenGFW 的 TCP 重组会把它抹平;record 分片切的是 TLS 自己的记录层,重组后的字节流里每一片前面都还带着 5 字节的 record 头。
`tls` 分析器以前默认 ClientHello 只在一个 record 里,读完第一个 record 头就直接按握手长度切数据,碰到分片会:
- 第一个 record 少于 4 字节握手数据时,把下一个 record 的头误当成握手长度(例如 `16 03 01` 被读成 0x160301),一直等到 8192 字节的上限;
- 其他情况下,后续 record 的头混进消息体导致解析失败,分析器取消。
两种情况都不产出任何属性,所有基于 `tls.req.sni` 的规则都会被绕过。
## 第一层防御:在 record 层重组
现在分析器先剥掉 record 层,把各个 record 的内容拼回完整的握手消息再解析,所以无论切在哪里、切成多少片,`tls.req` 的解析结果都和不分片时完全一样。ServerHello 方向也做了同样的处理。
- 握手消息中间夹了其他类型的 record(alert、change_cipher_spec 等)会被视为非法,分析器取消。
- record 长度超过 2^14 视为非法。
- 空 record 按规范不允许出现在握手中,但这里不拒绝,只是计数——拒绝的话反而给了对方一个让分析中止的开关。
- 常见情况下整个 ClientHello 在一个 record 里,此时直接引用原始缓冲区,不做额外拷贝。
## 第二层防御:把分片本身当指纹
| 属性 | 类型 | 说明 |
| --- | --- | --- |
| `tls.req_records` | int | ClientHello 跨了几个 record |
| `tls.resp_records` | int | ServerHello 跨了几个 record |
主流浏览器和 TLS 库都把 ClientHello 放在一个 record 里,即使是带后量子密钥交换、将近 2KB 的 ClientHello 也远小于 16KB 的 record 上限。所以 `req_records > 1` 本身就是很强的规避信号,误报率很低。
这个计数放在 `req` 外面,是为了保证 `tls.req != nil` 仍然表示"ClientHello 已完整解析",现有规则的语义不变。
### 提前上报
分片一出现(第二个 record 开始)就会上报一次 `req_records`,不用等消息收完。原因是分析器有 8192 字节的 `Limit()`:切成每片 1 字节时,每个字节要额外带 5 字节的 record 头,一个 1.8KB 的 ClientHello 会膨胀到 11KB 左右,永远收不完。所以规则不能依赖 SNI 才动手。
之后的 record 不会逐个上报(否则每片都要跑一次规则匹配);消息收完时,或者分析器因为超限被关闭时,再上报最终的计数。所以规则里看到的值可能是 2,也可能是最终值——写阈值时按 `> 1` 来写。
### 规则示例
直接拦截分片的 ClientHello:
```yaml
- name: tls-record-fragmentation
action: block
log: true
expr: tls != nil && tls.req_records != nil && tls.req_records > 1
```
> [!IMPORTANT]
> 要判断 `tls.req_records != nil`,不能只判 `tls != nil`:只看到服务端方向时,`tls` 里可能只有 `resp` 相关的字段。
按 SNI 拦截的规则不需要任何改动,分片后照样能匹配:
```yaml
- name: block-example
action: block
expr: tls != nil && tls.req != nil && tls.req.sni endsWith "example.com"
```
## 局限
**只统计 hello 消息本身。** ClientHello 收完就停止计数,后面的 record 不影响 `req_records`。
**超长的分片拿不到 SNI。** 分片把 ClientHello 撑过 8192 字节时,`tls.req` 不会出现,只能靠 `req_records` 拦截。需要同时拿到 SNI 的话可以调大 `Limit()`,代价是每条 TLS 连接的缓冲上限相应变大。
**QUIC 不在此列。** QUIC 的 ClientHello 在 CRYPTO 帧里,走的是 `quic` 分析器,与这里无关。
+199
View File
@@ -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)
}
}
}
+30 -5
View File
@@ -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
}
+14 -5
View File
@@ -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
}
+11 -2
View File
@@ -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 {