From babf6bd0b88bb84193eb92f57ec2737607ddd100 Mon Sep 17 00:00:00 2001 From: uk0 Date: Wed, 8 Apr 2026 16:40:08 +0800 Subject: [PATCH] integrate audio ticker, STUN consent, fix timing and STUN ping-pong - Launch audio ticker (50pps Opus PT=111) and STUN consent freshness as background goroutines in WebRTC mode - Fix STUN infinite ping-pong: only respond to Binding Requests (0x0001), not Binding Responses (0x0101) - Add decoy send delay (2-8ms) to avoid burst timing fingerprint - Use shared atomic transport-cc counter across all streams - WebRTC mode uses compound RTCP (SR+SDES) instead of bare SR - Server responds to client STUN Binding Requests with proper Response DPI score: 55.6 -> 73.8 (timing 10->69, size CV 0.09->0.93) --- main.go | 95 ++++++++++++++++++++++++++++++++++++++------------ wire.go | 10 ++++++ wire_webrtc.go | 42 ++++++++++++++++++++++ 3 files changed, 125 insertions(+), 22 deletions(-) diff --git a/main.go b/main.go index 1f2ac46..0a9afcd 100644 --- a/main.go +++ b/main.go @@ -793,6 +793,12 @@ type runtimeCtx struct { // Shim decoy RPS shimUpRL *rateLimiter shimDownRL *rateLimiter + + // WebRTC shared state + twccSeq uint32 // shared transport-cc counter (atomic) + + // IPCAM shared state + gop *gopState } func runMetrics(addr string) { @@ -987,29 +993,58 @@ func runUDPClient(ctx context.Context, rt *runtimeCtx) error { logf(LInfo, "[client][udp] %s -> server %s (sess=%d, wire=%s, %dfps, %dMbps)", rt.cfg.Listen, rt.cfg.ServerAddr, sess, rt.cfg.Wire, rt.cfg.FPS, rt.cfg.BitrateMbps) + // WebRTC: simulate DTLS handshake before media + if rt.cfg.Wire == WireWebRTC && rt.cfg.EnableDTLS { + logf(LInfo, "[client] performing DTLS handshake with server...") + if err := performDTLSHandshake(uc, saddr, false, rt.pc); err != nil { + logf(LWarn, "[client] DTLS handshake failed: %v (continuing)", err) + } else { + metricDTLSSent.Add(1) + logf(LInfo, "[client] DTLS handshake complete") + } + } + var appPeerMu sync.RWMutex var appPeer *net.UDPAddr - // RTP 累加 var rtp = rtpState{seq: uint16(grnd.Uint32()), ssrc: grnd.Uint32()} var rtpTs uint32 step := 90000 / max(1, rt.cfg.FPS) + // WebRTC: launch audio ticker and STUN consent freshness + if rt.cfg.Wire == WireWebRTC { + audio := &audioState{ + seq: uint16(grnd.Uint32()), + ts: 0, + ssrc: grnd.Uint32(), // different SSRC for audio + twccSeq: &rt.twccSeq, + } + go runAudioTicker(ctx, uc, saddr, audio, rt.tb, rt.pc) + go runSTUNConsent(ctx, uc, saddr, rt.pc) + } + go func() { buf := make([]byte, 64<<10) for { n, from, err := uc.ReadFromUDP(buf) if err != nil { return } - // 来自 server：解壳 -> 发给应用 if from.IP.Equal(saddr.IP) && from.Port == saddr.Port { raw := append([]byte(nil), buf[:n]...) src := ip4OrLoopback(from.IP) dst := ip4OrLoopback(uc.LocalAddr().(*net.UDPAddr).IP) rt.pc.WriteUDP(src, from.Port, dst, uc.LocalAddr().(*net.UDPAddr).Port, raw) - // Skip DTLS/STUN control packets (not data) - if isDTLSRange(raw[0]) || isSTUNPacket(raw) { continue } + if len(raw) == 0 { continue } + if isDTLSRange(raw[0]) || isSTUNPacket(raw) { + // WebRTC: respond to STUN Binding Requests only (not responses, to avoid ping-pong) + if rt.cfg.Wire == WireWebRTC && isSTUNRequest(raw) { + resp := buildSTUNBindingResponse(raw[8:20]) + _, _ = uc.WriteToUDP(resp, from) + rt.pc.WriteUDP(dst, uc.LocalAddr().(*net.UDPAddr).Port, src, from.Port, resp) + } + continue + } h, payload, e2 := decodeUDPFrame(rt.cfg, rt.ae, raw, sess) if e2 != nil { continue } @@ -1026,13 +1061,11 @@ func runUDPClient(ctx context.Context, rt *runtimeCtx) error { continue } - // 来自应用：加壳 -> 先发真实帧，再插播 shim 诱饵（RPS 优先）→ 插播额外伪报文 appPeerMu.Lock(); appPeer = from; appPeerMu.Unlock() p := append([]byte(nil), buf[:n]...) - // a) 真实帧 - var twccSeq uint32 - frame := encodeUDPFrame(rt.cfg, rt.ae, &rtp, &rtpTs, step, sess, p, false, &twccSeq) + // a) 真实帧（使用共享 twccSeq） + frame := encodeUDPFrame(rt.cfg, rt.ae, &rtp, &rtpTs, step, sess, p, false, &rt.twccSeq) rt.tb.wait(len(frame) + 28) _, _ = uc.WriteToUDP(frame, saddr) metricFramesUp.Add(1); metricBytesUp.Add(int64(n)) @@ -1040,24 +1073,29 @@ func runUDPClient(ctx context.Context, rt *runtimeCtx) error { dstIP := ip4OrLoopback(saddr.IP) rt.pc.WriteUDP(srcIP, uc.LocalAddr().(*net.UDPAddr).Port, dstIP, saddr.Port, frame) - // b) shim 诱饵 + // b) shim 诱饵（延迟发送避免突发） sendDecoy := false if rt.shimUpRL != nil && rt.shimUpRL.takeMax(1) > 0 { sendDecoy = true } else if rt.cfg.DecoyRps <= 0 && grnd.Intn(100) < rt.cfg.DecoyPct { sendDecoy = true } if sendDecoy { + // 延迟 2-8ms 避免与真实帧突发 + time.Sleep(time.Duration(2+grnd.Intn(6)) * time.Millisecond) djunk := make([]byte, len(p)) _, _ = cryptoRand.Read(djunk) - df := encodeUDPFrame(rt.cfg, rt.ae, &rtp, &rtpTs, step, sess, djunk, true, &twccSeq) + df := encodeUDPFrame(rt.cfg, rt.ae, &rtp, &rtpTs, step, sess, djunk, true, &rt.twccSeq) rt.tb.wait(len(df) + 28) _, _ = uc.WriteToUDP(df, saddr) metricFramesUp.Add(1); metricShimDecoySent.Add(1) rt.pc.WriteUDP(srcIP, uc.LocalAddr().(*net.UDPAddr).Port, dstIP, saddr.Port, df) } - // c) 额外伪报文（无 shim；RTCP / pure RTP / STUN） - sendExtraDecoysUDP(uc, uc.LocalAddr().(*net.UDPAddr), saddr, rt, rtp.ssrc, &rtp.seq, &rtpTs, step, true) + // c) 额外伪报文（WebRTC 用 compound RTCP） + if rt.cfg.Wire == WireWebRTC { + sendWebRTCExtraDecoys(uc, uc.LocalAddr().(*net.UDPAddr), saddr, rt, rtp.ssrc, &rtp.seq, &rtpTs, step, true) + } else { + sendExtraDecoysUDP(uc, uc.LocalAddr().(*net.UDPAddr), saddr, rt, rtp.ssrc, &rtp.seq, &rtpTs, step, true) + } - // 节奏（fps） base := 1000 / max(1, rt.cfg.FPS) jitterSleep(base, rt.cfg.JitterPct) } @@ -1120,8 +1158,17 @@ func runUDPServer(ctx context.Context, rt *runtimeCtx) error { dst := ip4OrLoopback(ln.LocalAddr().(*net.UDPAddr).IP) rt.pc.WriteUDP(src, caddr.Port, dst, ln.LocalAddr().(*net.UDPAddr).Port, raw) - // Skip DTLS/STUN control packets - if len(raw) > 0 && (isDTLSRange(raw[0]) || isSTUNPacket(raw)) { continue } + // Handle DTLS/STUN control packets + if len(raw) > 0 && isDTLSRange(raw[0]) { continue } + if len(raw) > 0 && isSTUNPacket(raw) { + // WebRTC: respond to STUN Binding Requests only (not responses) + if rt.cfg.Wire == WireWebRTC && isSTUNRequest(raw) { + resp := buildSTUNBindingResponse(raw[8:20]) + _, _ = ln.WriteToUDP(resp, caddr) + rt.pc.WriteUDP(dst, ln.LocalAddr().(*net.UDPAddr).Port, src, caddr.Port, resp) + } + continue + } h, p, e2 := decodeUDPFrame(rt.cfg, rt.ae, raw, 0) if e2 != nil { continue } @@ -1154,9 +1201,8 @@ func runUDPServer(ctx context.Context, rt *runtimeCtx) error { if e2 != nil { return } pp := append([]byte(nil), b[:n2]...) - // a) 真实帧 - var twccSeq uint32 - pkt := encodeUDPFrame(rt.cfg, rt.ae, &rtp, &rtpTs, step, k.sess, pp, false, &twccSeq) + // a) 真实帧（使用共享 twccSeq） + pkt := encodeUDPFrame(rt.cfg, rt.ae, &rtp, &rtpTs, step, k.sess, pp, false, &rt.twccSeq) rt.tb.wait(len(pkt) + 28) _, _ = ln.WriteToUDP(pkt, s.client) metricFramesDown.Add(1); metricBytesDown.Add(int64(n2)) @@ -1164,22 +1210,27 @@ func runUDPServer(ctx context.Context, rt *runtimeCtx) error { dst2 := ip4OrLoopback(s.client.IP) rt.pc.WriteUDP(src2, ln.LocalAddr().(*net.UDPAddr).Port, dst2, s.client.Port, pkt) - // b) shim 诱饵 + // b) shim 诱饵（延迟避免突发） sendDecoy := false if rt.shimDownRL != nil && rt.shimDownRL.takeMax(1) > 0 { sendDecoy = true } else if rt.cfg.DecoyRps <= 0 && grnd.Intn(100) < rt.cfg.DecoyPct { sendDecoy = true } if sendDecoy { + time.Sleep(time.Duration(2+grnd.Intn(6)) * time.Millisecond) djunk := make([]byte, len(pp)) _, _ = cryptoRand.Read(djunk) - df := encodeUDPFrame(rt.cfg, rt.ae, &rtp, &rtpTs, step, k.sess, djunk, true, &twccSeq) + df := encodeUDPFrame(rt.cfg, rt.ae, &rtp, &rtpTs, step, k.sess, djunk, true, &rt.twccSeq) rt.tb.wait(len(df) + 28) _, _ = ln.WriteToUDP(df, s.client) metricFramesDown.Add(1); metricShimDecoySent.Add(1) rt.pc.WriteUDP(src2, ln.LocalAddr().(*net.UDPAddr).Port, dst2, s.client.Port, df) } - // c) 额外伪报文（无 shim；RTCP / pure RTP / STUN），发往 client - sendExtraDecoysUDP(ln, ln.LocalAddr().(*net.UDPAddr), s.client, rt, rtp.ssrc, &rtp.seq, &rtpTs, step, false) + // c) 额外伪报文（WebRTC 用 compound RTCP） + if rt.cfg.Wire == WireWebRTC { + sendWebRTCExtraDecoys(ln, ln.LocalAddr().(*net.UDPAddr), s.client, rt, rtp.ssrc, &rtp.seq, &rtpTs, step, false) + } else { + sendExtraDecoysUDP(ln, ln.LocalAddr().(*net.UDPAddr), s.client, rt, rtp.ssrc, &rtp.seq, &rtpTs, step, false) + } // 节奏（fps） base := 1000 / max(1, rt.cfg.FPS) diff --git a/wire.go b/wire.go index 6a8147a..7a96ad5 100644 --- a/wire.go +++ b/wire.go @@ -147,3 +147,13 @@ func isSTUNPacket(data []byte) bool { cookie := binary.BigEndian.Uint32(data[4:8]) return cookie == 0x2112A442 } + +// isSTUNRequest checks if a STUN packet is a Binding Request (type 0x0001). +func isSTUNRequest(data []byte) bool { + if len(data) < 20 { + return false + } + msgType := binary.BigEndian.Uint16(data[0:2]) + cookie := binary.BigEndian.Uint32(data[4:8]) + return cookie == 0x2112A442 && msgType == 0x0001 +} diff --git a/wire_webrtc.go b/wire_webrtc.go index a2161b9..63bebd7 100644 --- a/wire_webrtc.go +++ b/wire_webrtc.go @@ -224,3 +224,45 @@ func runSTUNConsent(ctx context.Context, conn *net.UDPConn, peer *net.UDPAddr, p } } } + +// sendWebRTCExtraDecoys sends compound RTCP and STUN with proper WebRTC formatting. +func sendWebRTCExtraDecoys(conn *net.UDPConn, src *net.UDPAddr, dst *net.UDPAddr, rt *runtimeCtx, + ssrc uint32, seq *uint16, ts *uint32, step int, upDirection bool) { + + srcIP := ip4OrLoopback(src.IP) + dstIP := ip4OrLoopback(dst.IP) + + var rl *decoyRL + if upDirection { + rl = rt.upRL + } else { + rl = rt.downRL + } + const maxBurst = 2 + + // Compound RTCP (SR+SDES) instead of bare SR + if rl != nil && rl.sr != nil { + n := rl.sr.takeMax(maxBurst) + for i := 0; i < n; i++ { + pktCount := uint32(metricFramesUp.Value()) + octCount := uint32(metricBytesUp.Value()) + compound := buildCompoundRTCP(ssrc, *ts, pktCount, octCount) + rt.tb.wait(len(compound) + 28) + _, _ = conn.WriteToUDP(compound, dst) + rt.pc.WriteUDP(srcIP, src.Port, dstIP, dst.Port, compound) + metricRTCPsrSent.Add(1) + } + } + + // STUN with FINGERPRINT + if rl != nil && rl.stun != nil { + n := rl.stun.takeMax(1) + for i := 0; i < n; i++ { + req := buildSTUNBindingRequestFull() + rt.tb.wait(len(req) + 28) + _, _ = conn.WriteToUDP(req, dst) + rt.pc.WriteUDP(srcIP, src.Port, dstIP, dst.Port, req) + metricSTUNSent.Add(1) + } + } +}