From c4e0bcf7683aa576c891bc2243955a781bde691e Mon Sep 17 00:00:00 2001 From: Leadaxe <247031499+Leadaxe@users.noreply.github.com> Date: Fri, 31 Jul 2026 23:49:57 +0300 Subject: [PATCH] lx: rebind socket on handshake give-up (sing-box-lx SPEC 041 self-heal) After ~90s of unanswered handshake initiations (the give-up branch of expiredRetransmitHandshake) the socket's 5-tuple is proven dead (expired NAT mapping / poisoned DPI flow entry after device sleep) and upstream retries into it forever; only a manual reconnect healed the peer. Reopen the bind once per give-up cycle (fresh ephemeral port unless the user pinned listen_port), then re-initiate immediately. Debounced via CAS; no timers or goroutines while healthy; a rebind racing Down()/Close() degrades to a no-op inside BindUpdate. Red/green e2e + unit tests. --- device/device.go | 67 ++++++++++++ device/lx_giveup_rebind_test.go | 83 ++++++++++++++ device/lx_giveup_selfheal_test.go | 172 ++++++++++++++++++++++++++++++ device/timers.go | 8 ++ 4 files changed, 330 insertions(+) create mode 100644 device/lx_giveup_rebind_test.go create mode 100644 device/lx_giveup_selfheal_test.go diff --git a/device/device.go b/device/device.go index 654447a..4358980 100644 --- a/device/device.go +++ b/device/device.go @@ -122,6 +122,23 @@ type Device struct { } ipackets [5]*obfChain + + // lx: SPEC 041 — passive self-heal on handshake give-up. When a peer's + // handshake retry cycle exhausts (the give-up branch of + // expiredRetransmitHandshake), the device reopens its bind once — with a + // fresh ephemeral port when freshPort is set — and immediately + // re-initiates. Heals dead per-flow path state (an expired NAT mapping or + // a poisoned DPI flow entry) that otherwise pins every retry to the same + // dead 5-tuple until a manual reconnect. Zero cost while healthy: no + // timers, no goroutines — the trigger is the existing give-up event, + // which only fires under traffic demand after ~90s of unanswered + // initiations. Enabled by default; sing-box decides freshPort from + // whether the user pinned listen_port. + giveUpRebind struct { + enabled atomic.Bool + freshPort atomic.Bool + last atomic.Int64 // unix seconds of the last rebind (debounce) + } } // deviceState represents the state of a Device. @@ -326,6 +343,7 @@ func (device *Device) SetPrivateKey(sk NoisePrivateKey) error { func NewDevice(ctx context.Context, tunDevice tun.Device, bind conn.Bind, logger *Logger, workers int) *Device { device := new(Device) device.pauseManager = service.FromContext[pause.Manager](ctx) + device.giveUpRebind.enabled.Store(true) // lx: SPEC 041 — self-heal on by default device.state.state.Store(uint32(deviceStateDown)) device.closed = make(chan struct{}) device.log = logger @@ -785,6 +803,55 @@ func (device *Device) BindUpdate() error { return nil } +// lx: SPEC 041 — configure the handshake give-up self-heal (see the +// giveUpRebind field comment). freshPort must be false when the user pinned +// an explicit listen_port: the pinned port is preserved, at the cost of the +// rebind not changing the 5-tuple. +func (device *Device) SetGiveUpRebind(enabled, freshPort bool) { + device.giveUpRebind.enabled.Store(enabled) + device.giveUpRebind.freshPort.Store(freshPort) +} + +// lx: SPEC 041 — invoked from the give-up branch of +// expiredRetransmitHandshake: ~90s of initiations went unanswered, so the +// current socket's 5-tuple is proven dead. Reopen the bind (fresh ephemeral +// port when allowed) and kick a new handshake cycle immediately. Runs the +// heavy part in a goroutine so the timer callback never blocks on +// BindUpdate's worker drain. Debounced to one rebind per RekeyAttemptTime +// per device (CAS on `last` settles concurrent multi-peer give-ups). On a +// down or closed device BindUpdate does not reopen the socket, so a rebind +// racing idle-suspend (SPEC 020) or Close degrades to a no-op. +func (device *Device) handleHandshakeGiveUp(peer *Peer) { + if !device.giveUpRebind.enabled.Load() { + return + } + if device.isClosed() { + return + } + now := time.Now().Unix() + last := device.giveUpRebind.last.Load() + if now-last < int64(RekeyAttemptTime/time.Second) { + return + } + if !device.giveUpRebind.last.CompareAndSwap(last, now) { + return + } + fresh := device.giveUpRebind.freshPort.Load() + go func() { + if fresh { + device.net.Lock() + device.net.port = 0 + device.net.Unlock() + } + if err := device.BindUpdate(); err != nil { + device.log.Errorf("%v - Failed to rebind after handshake give-up: %v", peer, err) + return + } + device.log.Verbosef("%v - Rebound socket after handshake give-up (fresh port=%v)", peer, fresh) + peer.SendHandshakeInitiation(false) + }() +} + func (device *Device) BindClose() error { device.net.Lock() err := closeBindLocked(device) diff --git a/device/lx_giveup_rebind_test.go b/device/lx_giveup_rebind_test.go new file mode 100644 index 0000000..bbd4706 --- /dev/null +++ b/device/lx_giveup_rebind_test.go @@ -0,0 +1,83 @@ +/* SPDX-License-Identifier: MIT + * + * lx: SPEC 041 — unit tests for the give-up rebind mechanics on top of the + * self-heal harness (lx_giveup_selfheal_test.go): fresh vs pinned port, + * debounce, and disabled = upstream parity. These use the post-fix API + * (SetGiveUpRebind) and are NOT expected to compile on the pre-fix base. + */ + +package device + +import ( + "testing" + "time" +) + +func waitOpens(t *testing.T, bind *gateBind, want int) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for bind.openCount() < want { + if time.Now().After(deadline) { + t.Fatalf("bind reopened %d times, want %d", bind.openCount(), want) + } + time.Sleep(10 * time.Millisecond) + } +} + +// Fresh mode (listen_port not pinned): the rebind must ask the OS for a new +// ephemeral port — Open is called with port 0. +func TestGiveUpRebindFreshPort(t *testing.T) { + pair := newGiveUpPair(t, false) + pair.devA.SetGiveUpRebind(true, true) + + triggerGiveUp(t, pair.devA, pair.pkB) + waitOpens(t, pair.bindA, 2) + + ports := pair.bindA.portsSnapshot() + if ports[1] != 0 { + t.Fatalf("rebind requested port %d, want 0 (fresh ephemeral)", ports[1]) + } +} + +// Pinned mode (explicit listen_port): the rebind must keep the current port. +// chanBind.Open reports its source id (1) as the actual port, so the device +// stores net.port=1 after the first Open and must reuse it. +func TestGiveUpRebindPinnedPortPreserved(t *testing.T) { + pair := newGiveUpPair(t, false) + pair.devA.SetGiveUpRebind(true, false) + + triggerGiveUp(t, pair.devA, pair.pkB) + waitOpens(t, pair.bindA, 2) + + ports := pair.bindA.portsSnapshot() + if ports[1] != 1 { + t.Fatalf("rebind requested port %d, want 1 (pinned)", ports[1]) + } +} + +// A second give-up inside the debounce window must not rebind again. +func TestGiveUpRebindDebounce(t *testing.T) { + pair := newGiveUpPair(t, false) + + triggerGiveUp(t, pair.devA, pair.pkB) + waitOpens(t, pair.bindA, 2) + + triggerGiveUp(t, pair.devA, pair.pkB) + time.Sleep(300 * time.Millisecond) + if got := pair.bindA.openCount(); got != 2 { + t.Fatalf("debounce failed: bind opened %d times, want 2", got) + } +} + +// Disabled: the give-up branch must behave exactly like upstream — flush and +// stop, no rebind. +func TestGiveUpRebindDisabled(t *testing.T) { + pair := newGiveUpPair(t, false) + pair.devA.SetGiveUpRebind(false, false) + + triggerGiveUp(t, pair.devA, pair.pkB) + time.Sleep(300 * time.Millisecond) + if got := pair.bindA.openCount(); got != 1 { + t.Fatalf("disabled mechanism still rebound: %d opens, want 1", got) + } +} diff --git a/device/lx_giveup_selfheal_test.go b/device/lx_giveup_selfheal_test.go new file mode 100644 index 0000000..2f59471 --- /dev/null +++ b/device/lx_giveup_selfheal_test.go @@ -0,0 +1,172 @@ +/* SPDX-License-Identifier: MIT + * + * lx: SPEC 041 — behavioural red/green test for the handshake give-up + * self-heal. Field failure mode (WARP/AWG after device sleep): the per-flow + * path state of the socket's 5-tuple dies (expired NAT mapping / poisoned DPI + * flow entry), every packet sent from the old socket vanishes, and upstream + * wireguard-go retries into that dead socket forever — only a manual + * reconnect (new socket, new ephemeral port) heals the peer. + * + * The test models the dead 5-tuple with a bind whose FIRST socket generation + * silently swallows every send; any socket opened after a rebind delivers + * normally. It then drives the peer into the give-up branch of + * expiredRetransmitHandshake and expects traffic to flow end to end without + * any reconnect: + * + * - pre-fix (base): give-up only flushes staged packets, the bind is never + * reopened, every retry keeps dying in the first generation -> timeout; + * - post-fix: give-up rebinds the socket and re-initiates -> tunnel comes + * up and the packet arrives. + * + * This file deliberately uses NO post-fix API, so it compiles and runs RED on + * the pre-fix base commit. Reuses the chanBind/chanTun harness from + * transport_padding_test.go. + */ + +package device + +import ( + "context" + "encoding/hex" + "fmt" + "sync" + "testing" + "time" + + "github.com/sagernet/wireguard-go/conn" +) + +// gateBind wraps chanBind: it records every Open (the port argument the +// device asked for) and silently swallows sends while the socket generation +// is at most dropOpens — modelling a dead 5-tuple whose packets vanish on the +// path without any local error. +type gateBind struct { + *chanBind + mu sync.Mutex + openPorts []uint16 + opens int + dropOpens int // swallow sends while opens <= dropOpens +} + +func (b *gateBind) Open(port uint16) ([]conn.ReceiveFunc, uint16, error) { + fns, actual, err := b.chanBind.Open(port) + b.mu.Lock() + b.opens++ + b.openPorts = append(b.openPorts, port) + b.mu.Unlock() + return fns, actual, err +} + +func (b *gateBind) Send(bufs [][]byte, ep conn.Endpoint, offset int) error { + b.mu.Lock() + drop := b.opens <= b.dropOpens + b.mu.Unlock() + if drop { + return nil // the dead 5-tuple: no local error, the packet just vanishes + } + return b.chanBind.Send(bufs, ep, offset) +} + +func (b *gateBind) openCount() int { + b.mu.Lock() + defer b.mu.Unlock() + return b.opens +} + +func (b *gateBind) portsSnapshot() []uint16 { + b.mu.Lock() + defer b.mu.Unlock() + return append([]uint16(nil), b.openPorts...) +} + +type giveUpPair struct { + devA, devB *Device + tunA, tunB *chanTun + bindA *gateBind + pkB NoisePublicKey +} + +// newGiveUpPair builds two Up()'d peered devices; devA sits on a gateBind +// whose first socket generation optionally blackholes all sends. +func newGiveUpPair(t *testing.T, dropFirstOpen bool) *giveUpPair { + t.Helper() + + skA, err := newPrivateKey() + if err != nil { + t.Fatalf("newPrivateKey A: %v", err) + } + skB, err := newPrivateKey() + if err != nil { + t.Fatalf("newPrivateKey B: %v", err) + } + pkA := skA.publicKey() + pkB := skB.publicKey() + + rawA, rawB := newChanBindPair() + bindA := &gateBind{chanBind: rawA} + if dropFirstOpen { + bindA.dropOpens = 1 + } + tunA := newChanTun() + tunB := newChanTun() + + devA := NewDevice(context.Background(), tunA, bindA, NewLogger(LogLevelError, "devA: "), 1) + devB := NewDevice(context.Background(), tunB, rawB, NewLogger(LogLevelError, "devB: "), 1) + t.Cleanup(devA.Close) + t.Cleanup(devB.Close) + + cfgA := fmt.Sprintf( + "private_key=%s\nreplace_peers=true\npublic_key=%s\nendpoint=127.0.0.1:2\nallowed_ip=%s/32\n", + hex.EncodeToString(skA[:]), hex.EncodeToString(pkB[:]), testIPB) + cfgB := fmt.Sprintf( + "private_key=%s\nreplace_peers=true\npublic_key=%s\nendpoint=127.0.0.1:1\nallowed_ip=%s/32\n", + hex.EncodeToString(skB[:]), hex.EncodeToString(pkA[:]), testIPA) + + if err := devA.IpcSet(cfgA); err != nil { + t.Fatalf("IpcSet A: %v", err) + } + if err := devB.IpcSet(cfgB); err != nil { + t.Fatalf("IpcSet B: %v", err) + } + if err := devA.Up(); err != nil { + t.Fatalf("Up A: %v", err) + } + if err := devB.Up(); err != nil { + t.Fatalf("Up B: %v", err) + } + + return &giveUpPair{devA: devA, devB: devB, tunA: tunA, tunB: tunB, bindA: bindA, pkB: pkB} +} + +// triggerGiveUp drives dev's peer into the give-up branch of +// expiredRetransmitHandshake exactly the way 90s of unanswered retries would: +// attempts past the limit, last initiation older than RekeyTimeout. +func triggerGiveUp(t *testing.T, dev *Device, pk NoisePublicKey) { + t.Helper() + dev.peers.RLock() + peer := dev.peers.keyMap[pk] + dev.peers.RUnlock() + if peer == nil { + t.Fatal("peer not found") + } + peer.handshake.mutex.Lock() + peer.handshake.lastSentHandshake = time.Now().Add(-2 * RekeyTimeout) + peer.handshake.mutex.Unlock() + peer.timers.handshakeAttempts.Store(MaxTimerHandshakes + 1) + expiredRetransmitHandshake(peer) +} + +// TestHandshakeGiveUpSelfHeal: dead first socket, give-up fires — the tunnel +// must come up and deliver traffic without any reconnect. +func TestHandshakeGiveUpSelfHeal(t *testing.T) { + pair := newGiveUpPair(t, true) + + pkt := buildIPv4Packet(testIPA, testIPB, 8) + send := func() { pair.tunA.toDevice <- pkt } + + // Traffic demand: stages the packet and sends the first (blackholed) + // initiation, exactly the state a real give-up cycle ends in. + send() + triggerGiveUp(t, pair.devA, pair.pkB) + awaitPacket(t, pair.tunB, pkt, send) +} diff --git a/device/timers.go b/device/timers.go index d30f26b..05a5abd 100644 --- a/device/timers.go +++ b/device/timers.go @@ -99,6 +99,14 @@ func expiredRetransmitHandshake(peer *Peer) { peer.timers.zeroKeyMaterial.Mod(RejectAfterTime * 3) } peer.noteSessionHandshakeStopped() + + /* lx: SPEC 041 — the exhausted cycle just proved the current socket's + * 5-tuple dead (90s of initiations, zero replies). Rebind once and + * re-initiate, so a stale NAT mapping / poisoned DPI flow entry cannot + * pin this peer to a dead socket until a manual reconnect. Runs after + * the session-state notification so a consumer sees "handshake stopped" + * before the socket is recreated. */ + peer.device.handleHandshakeGiveUp(peer) } else { peer.timers.handshakeAttempts.Add(1) peer.device.log.Verbosef("%s - Handshake did not complete after %d seconds, retrying (try %d)", peer, int(RekeyTimeout.Seconds()), peer.timers.handshakeAttempts.Load()+1)