diff --git a/internal/network/helpers.go b/internal/network/helpers.go index bad6c45..54d7833 100644 --- a/internal/network/helpers.go +++ b/internal/network/helpers.go @@ -59,6 +59,15 @@ func KeepAliveAll(p *PeerConn) { case <-time.After(keepaliveInterval): } for _, addr := range p.establishedPeers() { + // Any authenticated packet proves the peer is alive. Under a heavy + // transfer the pong replies can be lost to congestion for long + // stretches; without this, a peer whose data is streaming in + // perfectly would be torn down mid-transfer. + if p.RecentActivity(addr, keepaliveInterval) { + p.mu.Lock() + p.missedPings[addr.String()] = 0 + p.mu.Unlock() + } p.mu.Lock() missed := p.missedPings[addr.String()] p.mu.Unlock() diff --git a/internal/network/peer.go b/internal/network/peer.go index c60115a..b64cabc 100644 --- a/internal/network/peer.go +++ b/internal/network/peer.go @@ -75,6 +75,19 @@ type peerSession struct { rxAEAD cipher.AEAD txCtr uint64 // next send counter (the AEAD nonce), guarded by mu + // lastRx is when the last packet that passed AEAD authentication and the + // replay window arrived from this peer. Guarded by mu. Any authenticated + // packet proves liveness, so the keepalive must not declare a peer dead + // while this is fresh — pongs alone are unreliable under congestion. + lastRx time.Time + + // rekeyNotice is a prebuilt CtrlRekey packet sealed under the previous + // session keys. driveHandshake retransmits it until the new handshake + // establishes, so a peer that still holds the old session tears it down + // and re-handshakes instead of answering fresh msg1s with its stale + // cached msg2 (which deadlocks both sides until its keepalive times out). + rekeyNotice []byte + // Anti-replay window for the receive direction, guarded by mu like the rest // of the session state and consulted only after AEAD authentication. replay ReplayWindow @@ -116,6 +129,7 @@ func (s *peerSession) installTransportKeysLocked(cs0, cs1 *noise.CipherState) er } s.txAEAD, s.rxAEAD = txAEAD, rxAEAD s.txCtr = 0 + s.rekeyNotice = nil return nil } @@ -144,6 +158,9 @@ func (s *peerSession) decrypt(dst []byte, pkt []byte) ([]byte, error) { } s.mu.Lock() fresh := s.replay.Check(counter) + if fresh { + s.lastRx = time.Now() + } s.mu.Unlock() if !fresh { return nil, fmt.Errorf("replayed or out-of-window packet from %s", s.addr) @@ -162,8 +179,9 @@ type rxPacket struct { } type PeerConn struct { - bind wgconn.Bind - batch int + bind wgconn.Bind + batch int // pipeline batch: rx queue, consumer batches, send chunking + recvBatch int // datagrams per receive-function call (the bind's own batch) static noise.DHKey psk []byte @@ -205,6 +223,7 @@ func NewPeerConn(t *Transport, privateKey, publicKey, psk, prologue []byte) *Pee p := &PeerConn{ bind: t.bind, batch: t.batch, + recvBatch: t.recvBatch, static: noise.DHKey{Private: privateKey, Public: publicKey}, psk: psk, prologue: prologue, @@ -268,9 +287,9 @@ func buildPacket(pktType byte, body []byte) []byte { // decrypts them in parallel. func (p *PeerConn) recvLoop(fn wgconn.ReceiveFunc) { defer p.recvWG.Done() - bufs := make([][]byte, p.batch) - sizes := make([]int, p.batch) - eps := make([]wgconn.Endpoint, p.batch) + bufs := make([][]byte, p.recvBatch) + sizes := make([]int, p.recvBatch) + eps := make([]wgconn.Endpoint, p.recvBatch) for i := range bufs { bufs[i] = make([]byte, maxUDPPacket) } @@ -345,6 +364,16 @@ func (p *PeerConn) handlePacket(pkt []byte, ep wgconn.Endpoint) { // Authenticated "I'm leaving" from the peer: drop its session and // surface a Dead event — other peers are unaffected. p.RemovePeer(s.addr) + case CtrlRekey: + // The peer lost this session (keepalive timeout on its side) and is + // re-handshaking. Drop our copy too — answering its fresh msg1 with + // the stale cached msg2 would deadlock both sides until our own + // keepalive timed out (~30s of blackout mid-transfer). No Dead event: + // mappings stay valid and the tunnel heals in one handshake round + // trip; watchRearm surfaces Dead only if the re-handshake never lands. + if gen, ok := p.rearmSession(s, false); ok { + go p.watchRearm(s, gen) + } } case PacketData, PacketTun: s := p.sessionForEndpoint(ep) @@ -486,6 +515,7 @@ func (p *PeerConn) driveHandshake(s *peerSession, gen uint64, deadline time.Time established := s.established initiator := s.initiator initMsg := s.initMsg + notice := s.rekeyNotice ep := s.ep s.mu.Unlock() if superseded || established { @@ -494,6 +524,13 @@ func (p *PeerConn) driveHandshake(s *peerSession, gen uint64, deadline time.Time if !deadline.IsZero() && time.Now().After(deadline) { return } + if notice != nil { + // Rekey notice first, so a peer still holding the old session tears + // it down before our msg1 arrives. Retransmitting the same sealed + // packet is safe: once one copy is delivered, later copies fail the + // peer's replay window (or, after its rekey, the AEAD) and are dropped. + p.bind.Send([][]byte{notice}, ep) + } if initiator && initMsg != nil { p.bind.Send([][]byte{initMsg}, ep) } else { @@ -931,21 +968,36 @@ func (p *PeerConn) TimeoutPeer(addr *net.UDPAddr) { // as the pinned key for the next one. The retry is bounded by retryWindow; after // that the session is parked until an inbound msg1 or AddKnownPeer revives it. func (p *PeerConn) retryHandshake(addr *net.UDPAddr) { - s := p.sessionByAddr(addr) - if s == nil { - return + if s := p.sessionByAddr(addr); s != nil { + p.rearmSession(s, true) } +} + +// rearmSession resets an established session back to the handshake phase and +// starts a bounded driver, returning the driver generation. With notify set, a +// CtrlRekey notice sealed under the outgoing keys is prepared before they are +// wiped; the driver retransmits it so the peer — which may still hold the old +// session and would otherwise answer our fresh msg1 only with its stale cached +// msg2 — tears its copy down and re-handshakes immediately. +func (p *PeerConn) rearmSession(s *peerSession, notify bool) (uint64, bool) { s.mu.Lock() defer s.mu.Unlock() if !s.established || s.driving { - return + return 0, false } remote := s.remoteStatic if remote == nil { remote = s.expected } if remote == nil { - return + return 0, false + } + var notice []byte + if notify && s.txAEAD != nil { + pkt := sealPacket(s.txAEAD, PacketControl, s.txCtr, []byte{CtrlRekey}) + s.txCtr++ + notice = append([]byte(nil), pkt...) + putPacketBuf(pkt) } s.established = false s.txAEAD, s.rxAEAD = nil, nil @@ -954,9 +1006,42 @@ func (p *PeerConn) retryHandshake(addr *net.UDPAddr) { s.replay = ReplayWindow{} gen, err := p.armHandshakeLocked(s, remote) if err != nil { - return + return 0, false } + s.rekeyNotice = notice go p.driveHandshake(s, gen, time.Now().Add(retryWindow)) + return gen, true +} + +// watchRearm fires Dead if a rekey-triggered re-handshake never completes +// within the retry window. A CtrlRekey rearm deliberately emits no Dead event +// (mappings stay valid for a fast heal), so a session that parks anyway must +// still be surfaced to consumers or they would route into it forever. +func (p *PeerConn) watchRearm(s *peerSession, gen uint64) { + select { + case <-p.stop: + case <-time.After(retryWindow + time.Second): + s.mu.Lock() + parked := !s.established && s.driverGen == gen + s.mu.Unlock() + if parked { + p.fireDead(s.addr) + } + } +} + +// RecentActivity reports whether a packet that passed authentication and the +// replay window arrived from addr within d. The keepalive uses it so a peer +// whose data is flowing is never declared dead just because its pongs are +// being lost to congestion. +func (p *PeerConn) RecentActivity(addr *net.UDPAddr, d time.Duration) bool { + s := p.sessionByAddr(addr) + if s == nil { + return false + } + s.mu.Lock() + defer s.mu.Unlock() + return !s.lastRx.IsZero() && time.Since(s.lastRx) < d } // establishedPeers returns the addresses of all currently established peers. diff --git a/internal/network/protocol.go b/internal/network/protocol.go index 87c808d..526fe2a 100644 --- a/internal/network/protocol.go +++ b/internal/network/protocol.go @@ -52,8 +52,15 @@ const ( // Inner control opcodes, carried as the plaintext of a PacketControl transport // packet (authenticated and anti-replay-protected like any other payload). +// +// CtrlRekey is sent (sealed under the keys being discarded) by a peer that is +// resetting an established session back to the handshake phase, so the other +// side drops its copy and re-handshakes immediately instead of answering the +// fresh msg1 with a stale cached msg2. Peers that don't know the opcode simply +// ignore it and heal via their own keepalive timeout, as before. const ( - CtrlPing byte = 0x01 - CtrlPong byte = 0x02 - CtrlDead byte = 0x03 + CtrlPing byte = 0x01 + CtrlPong byte = 0x02 + CtrlDead byte = 0x03 + CtrlRekey byte = 0x04 ) diff --git a/internal/network/rio_diag_test.go b/internal/network/rio_diag_test.go new file mode 100644 index 0000000..cf5691a --- /dev/null +++ b/internal/network/rio_diag_test.go @@ -0,0 +1,163 @@ +package network + +// Diagnostic benchmarks over the REAL platform bind (WinRingBind on Windows), +// not the WrapUDPConn test adapter. Temporary: used to chase the 0–2 MB/s +// oscillation seen in real transfers. + +import ( + "net" + "testing" + "time" + + "github.com/neozmmv/blindspot/internal/crypto" +) + +type rioSide struct { + pc *PeerConn + addr *net.UDPAddr + kp *crypto.KeyPair +} + +func newRIOSide(t *testing.B, psk []byte) *rioSide { + t.Helper() + tr, err := OpenTransport() + if err != nil { + t.Fatalf("OpenTransport: %v", err) + } + kp, err := crypto.GenerateKeyPair() + if err != nil { + t.Fatalf("GenerateKeyPair: %v", err) + } + pc := NewPeerConn(tr, kp.PrivateKey, kp.PublicKey, psk, Prologue("bench-session")) + addr := &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: tr.Port()} + return &rioSide{pc: pc, addr: addr, kp: kp} +} + +func rioPair(b *testing.B) (*rioSide, *rioSide) { + b.Helper() + psk := crypto.DerivePSK("bench-pass-123", "bench-session") + sa := newRIOSide(b, psk) + sb := newRIOSide(b, psk) + if err := sa.pc.AddKnownPeer(sb.addr, sb.kp.PublicKey); err != nil { + b.Fatalf("AddKnownPeer: %v", err) + } + if err := sb.pc.AddKnownPeer(sa.addr, sa.kp.PublicKey); err != nil { + b.Fatalf("AddKnownPeer: %v", err) + } + for _, s := range []*rioSide{sa, sb} { + select { + case <-s.pc.Connected: + case <-time.After(5 * time.Second): + b.Fatal("handshake did not complete") + } + } + b.Cleanup(func() { + sa.pc.Close() + sb.pc.Close() + }) + return sa, sb +} + +// BenchmarkRIOTunnelDelivered measures delivered throughput through the real +// bind with a windowed sender, and reports the loss rate. +func BenchmarkRIOTunnelDelivered(b *testing.B) { + sa, sb := rioPair(b) + + const payloadSize = 1420 + received := make(chan int, 4096) + go func() { + bufs := make([][]byte, sb.pc.BatchSize()) + for i := range bufs { + bufs[i] = make([]byte, 1600) + } + senders := make([]string, sb.pc.BatchSize()) + for { + n, err := sb.pc.ReadTunBatch(bufs, senders) + if err != nil { + return + } + received <- n + } + }() + + const sendGroup = 64 // what the connect.go sender typically aggregates + payloads := make([][]byte, sendGroup) + for i := range payloads { + pkt := make([]byte, payloadSize) + pkt[0] = 0x45 + payloads[i] = pkt + } + + b.SetBytes(payloadSize) + b.ResetTimer() + sent, got, timeouts := 0, 0, 0 + deadline := time.Now().Add(60 * time.Second) + for got < b.N { + for sent-got < 8*sendGroup && sent < b.N+16*sendGroup { + if err := sa.pc.SendTunBatch(sb.addr, payloads); err != nil { + b.Fatalf("SendTunBatch: %v", err) + } + sent += sendGroup + } + select { + case n := <-received: + got += n + case <-time.After(100 * time.Millisecond): + timeouts++ + if time.Now().After(deadline) { + b.Fatalf("stalled: sent %d, received %d of %d", sent, got, b.N) + } + sent = got // window resync after loss + } + } + b.ReportMetric(float64(timeouts), "loss-stalls") + b.ReportMetric(float64(sb.pc.BatchSize()), "bind-batch") +} + +// BenchmarkRIOTunnelBlast sends as fast as the tx path allows with no window +// and reports what fraction survives to the receiver — measures raw rx-side +// drop behavior of the real bind under burst load. +func BenchmarkRIOTunnelBlast(b *testing.B) { + sa, sb := rioPair(b) + + const payloadSize = 1420 + var got int64 + done := make(chan struct{}) + go func() { + bufs := make([][]byte, sb.pc.BatchSize()) + for i := range bufs { + bufs[i] = make([]byte, 1600) + } + senders := make([]string, sb.pc.BatchSize()) + for { + n, err := sb.pc.ReadTunBatch(bufs, senders) + if err != nil { + close(done) + return + } + got += int64(n) + } + }() + + const sendGroup = 64 + payloads := make([][]byte, sendGroup) + for i := range payloads { + pkt := make([]byte, payloadSize) + pkt[0] = 0x45 + payloads[i] = pkt + } + + b.SetBytes(payloadSize) + b.ResetTimer() + for sent := 0; sent < b.N; sent += sendGroup { + if err := sa.pc.SendTunBatch(sb.addr, payloads); err != nil { + b.Fatalf("SendTunBatch: %v", err) + } + } + elapsedDrain := time.Now().Add(500 * time.Millisecond) + for time.Now().Before(elapsedDrain) { + time.Sleep(50 * time.Millisecond) + } + b.StopTimer() + b.ReportMetric(float64(got)/float64(b.N)*100, "delivered-%") +} diff --git a/internal/network/transport.go b/internal/network/transport.go index 7bb0c92..88370ce 100644 --- a/internal/network/transport.go +++ b/internal/network/transport.go @@ -78,7 +78,17 @@ type Transport struct { bind wgconn.Bind recvFns []wgconn.ReceiveFunc port uint16 - batch int + // recvBatch is the number of datagrams one call to a receive function can + // return — the bind's own batch size. + recvBatch int + // batch is the pipeline batch size: rx queue sizing, consumer batches + // (ReadTunBatch), and send chunking. It is deliberately NOT tied to the + // bind's BatchSize: WinRingBind reports 1 (batching in and out of the RIO + // ring is an upstream TODO), which would otherwise disable the parallel + // seal/open and batched TUN writes entirely on Windows. Every bind's Send + // accepts arbitrarily long bufs slices, so sending IdealBatchSize-sized + // batches is always safe. + batch int } // OpenTransport opens the platform's default batched bind (RIO on Windows, @@ -90,7 +100,12 @@ func OpenTransport() (*Transport, error) { if err != nil { return nil, fmt.Errorf("opening UDP bind: %w", err) } - return &Transport{bind: bind, recvFns: fns, port: port, batch: bind.BatchSize()}, nil + recvBatch := bind.BatchSize() + batch := wgconn.IdealBatchSize + if recvBatch > batch { + batch = recvBatch + } + return &Transport{bind: bind, recvFns: fns, port: port, recvBatch: recvBatch, batch: batch}, nil } // Port returns the local UDP port the transport is bound to. @@ -103,10 +118,11 @@ func (t *Transport) Port() int { return int(t.port) } func WrapUDPConn(c *net.UDPConn) *Transport { b := &udpBind{conn: c} return &Transport{ - bind: b, - recvFns: []wgconn.ReceiveFunc{b.receive}, - port: uint16(c.LocalAddr().(*net.UDPAddr).Port), - batch: b.BatchSize(), + bind: b, + recvFns: []wgconn.ReceiveFunc{b.receive}, + port: uint16(c.LocalAddr().(*net.UDPAddr).Port), + recvBatch: 1, // receive fills exactly one datagram per call + batch: b.BatchSize(), } } diff --git a/test/connection_test.go b/test/connection_test.go index 95605d0..5055e15 100644 --- a/test/connection_test.go +++ b/test/connection_test.go @@ -355,6 +355,71 @@ func runConnectionTest(t *testing.T, sessionID, password string) { _ = connectedOnB } +// TestRekeyHealsAfterOneSidedTimeout covers the mid-transfer blackout bug: one +// side declares the peer dead (keepalive timeout) and re-arms its handshake +// while the other side still holds the established session. Without the +// CtrlRekey notice the still-established side answers every fresh msg1 with its +// stale cached msg2 and the two deadlock until the second side's own keepalive +// times out (~30s). With the notice, both sides must re-establish within a few +// seconds and pass traffic again. +func TestRekeyHealsAfterOneSidedTimeout(t *testing.T) { + sessionID, password := "rekey-session", "pass1234!" + + peerA := newTestPeer(t, sessionID, password) + defer peerA.conn.Close() + defer peerA.pc.Shutdown() + + peerB := newTestPeer(t, sessionID, password) + defer peerB.conn.Close() + defer peerB.pc.Shutdown() + + peerA.startReadLoop() + peerB.startReadLoop() + + peerB.pc.AddKnownPeer(mustResolve(t, peerA.addr), peerA.kp.PublicKey) + peerA.pc.AddKnownPeer(mustResolve(t, peerB.addr), peerB.kp.PublicKey) + + addrOnA := peerA.waitConnected(t, 5*time.Second) + peerB.waitConnected(t, 5*time.Second) + + // Simulate a keepalive timeout on A only. B still believes the session is + // established and healthy. + peerA.pc.TimeoutPeer(addrOnA) + select { + case <-peerA.pc.Dead: + case <-time.After(3 * time.Second): + t.Fatal("TimeoutPeer did not surface a Dead event on A") + } + + // Both sides must re-establish far faster than B's own 30s keepalive limit. + reconnectedOnA := peerA.waitConnected(t, 5*time.Second) + peerB.waitConnected(t, 5*time.Second) + + // The healed session must carry traffic both ways. + if err := peerA.pc.Send(reconnectedOnA, []byte("post-rekey A->B")); err != nil { + t.Fatalf("send A->B after rekey: %v", err) + } + select { + case got := <-peerB.recv: + if string(got.data) != "post-rekey A->B" { + t.Fatalf("peer B received %q after rekey", got.data) + } + if err := peerB.pc.Send(got.addr, []byte("post-rekey B->A")); err != nil { + t.Fatalf("send B->A after rekey: %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("peer B did not receive data after rekey") + } + select { + case got := <-peerA.recv: + if string(got.data) != "post-rekey B->A" { + t.Fatalf("peer A received %q after rekey", got.data) + } + case <-time.After(3 * time.Second): + t.Fatal("peer A did not receive data after rekey") + } +} + // TestConnectionNoPassword simulates `blindspot connect -s ` with two peers. func TestConnectionNoPassword(t *testing.T) { runConnectionTest(t, "test-session", "")