diff --git a/conn/bind_std.go b/conn/bind_std.go index 0a15de0..eb27e10 100644 --- a/conn/bind_std.go +++ b/conn/bind_std.go @@ -272,7 +272,7 @@ again: return 0, err } sizes[0] = dataLength - if dataLength > 3 { + if dataLength > 3 && s.hasReserved() { // lx: SPEC 026 — gate reserved-clear on the egress receive path too, so a small-padding AmneziaWG magic in bytes 1-3 survives when no WARP reserved value is set common.ClearArray(bufs[0][1:4]) } endpoints[0] = &StdNetEndpoint{AddrPort: source} @@ -359,7 +359,7 @@ func (s *StdNetBind) receiveIP( if sizes[i] == 0 { continue } - if msg.N > 3 { + if msg.N > 3 && s.hasReserved() { // lx: only strip reserved bytes for WARP (see hasReserved) common.ClearArray(bufs[i][1:4]) } ep := &StdNetEndpoint{AddrPort: M.AddrPortFromNet(msg.Addr)} // TODO: remove allocation @@ -539,6 +539,20 @@ func (s *StdNetBind) SetReservedForEndpoint(destination netip.AddrPort, reserved s.reservedForEndpoint[destination] = reserved } +// lx: hasReserved reports whether any Cloudflare "reserved" value is set. The +// receive path must only zero bytes 1-3 when a reserved value exists (WARP); +// otherwise an AmneziaWG magic header that lands in bytes 1-3 (small s1/s2/s4 +// padding) would be corrupted and the packet dropped. The send path already +// gates its stamp on a per-endpoint `loaded` check, so no change is needed there. +func (s *StdNetBind) hasReserved() bool { + for _, reserved := range s.reservedForEndpoint { + if reserved != [3]uint8{} { + return true + } + } + return false +} + func (s *StdNetBind) send(conn *net.UDPConn, pc batchWriter, msgs []ipv6.Message) error { var ( n int diff --git a/conn/bind_windows.go b/conn/bind_windows.go index 121079f..c31bb35 100644 --- a/conn/bind_windows.go +++ b/conn/bind_windows.go @@ -461,7 +461,7 @@ func (bind *WinRingBind) receiveIPv4(bufs [][]byte, sizes []int, eps []Endpoint) bind.mu.RLock() defer bind.mu.RUnlock() n, ep, err := bind.v4.Receive(bufs[0], &bind.isOpen) - if n > 3 { + if n > 3 && bind.hasReserved() { // lx: only strip reserved bytes for WARP (see hasReserved) common.ClearArray(bufs[0][1:4]) } sizes[0] = n @@ -473,7 +473,7 @@ func (bind *WinRingBind) receiveIPv6(bufs [][]byte, sizes []int, eps []Endpoint) bind.mu.RLock() defer bind.mu.RUnlock() n, ep, err := bind.v6.Receive(bufs[0], &bind.isOpen) - if n > 3 { + if n > 3 && bind.hasReserved() { // lx: only strip reserved bytes for WARP (see hasReserved) common.ClearArray(bufs[0][1:4]) } sizes[0] = n @@ -576,6 +576,18 @@ func (bind *WinRingBind) SetReservedForEndpoint(destination netip.AddrPort, rese bind.reservedForEndpoint[*endpoint.(*WinRingEndpoint)] = reserved } +// lx: hasReserved reports whether any Cloudflare "reserved" value is set. See +// the StdNetBind.hasReserved comment — the unconditional receive clear would +// corrupt an AmneziaWG magic header sitting in bytes 1-3 (small padding). +func (bind *WinRingBind) hasReserved() bool { + for _, reserved := range bind.reservedForEndpoint { + if reserved != [3]uint8{} { + return true + } + } + return false +} + func (s *StdNetBind) BindSocketToInterface4(interfaceIndex uint32, blackhole bool) error { s.mu.Lock() defer s.mu.Unlock() diff --git a/conn/msgx_darwin.go b/conn/msgx_darwin.go index 138c8a8..da9bb07 100644 --- a/conn/msgx_darwin.go +++ b/conn/msgx_darwin.go @@ -234,7 +234,7 @@ func (s *StdNetBind) receiveSingle(conn *net.UDPConn, bufs [][]byte, sizes []int return 0, err } sizes[0] = n - if n > 3 { + if n > 3 && s.hasReserved() { // lx: only strip reserved bytes for WARP (see hasReserved) bufs[0][1] = 0 bufs[0][2] = 0 bufs[0][3] = 0 @@ -299,7 +299,7 @@ func (s *StdNetBind) makeReceiveMsgX(conn *net.UDPConn, isV6 bool) (ReceiveFunc, numMsgs := int(n) for i := 0; i < numMsgs; i++ { sizes[i] = int(state.hdrs[i].DataLen) - if sizes[i] > 3 { + if sizes[i] > 3 && s.hasReserved() { // lx: only strip reserved bytes for WARP (see hasReserved) bufs[i][1] = 0 bufs[i][2] = 0 bufs[i][3] = 0 diff --git a/conn/reserved_gate_lx_test.go b/conn/reserved_gate_lx_test.go new file mode 100644 index 0000000..3502a79 --- /dev/null +++ b/conn/reserved_gate_lx_test.go @@ -0,0 +1,56 @@ +/* SPDX-License-Identifier: MIT + * + * lx: unit coverage for the StdNetBind.hasReserved() gate that guards the + * receive-side reserved-clear. receiveIP zeroes bytes 1-3 (Cloudflare WARP + * "reserved") only when a non-zero reserved value is set for some endpoint; + * otherwise an AmneziaWG magic header landing in bytes 1-3 (small s1/s2/s4 + * padding) would be corrupted and the packet dropped. This test pins the gate + * itself; the end-to-end handshake proof lives in the device package. + */ + +package conn + +import ( + "net/netip" + "testing" +) + +func stdNetBindForTest(t *testing.T) *StdNetBind { + t.Helper() + b, ok := NewStdNetBind(nil).(*StdNetBind) + if !ok { + t.Fatalf("NewStdNetBind did not return *StdNetBind") + } + return b +} + +func TestStdNetBindHasReserved(t *testing.T) { + b := stdNetBindForTest(t) + if b.hasReserved() { + t.Fatal("fresh bind must report no reserved value") + } + + ep := netip.MustParseAddrPort("127.0.0.1:51820") + + // An all-zero reserved value is indistinguishable from "unset" and must + // not arm the clear. + b.SetReservedForEndpoint(ep, [3]byte{0, 0, 0}) + if b.hasReserved() { + t.Fatal("all-zero reserved must not count as reserved") + } + + // Any non-zero byte (WARP anycast tag) arms the clear. + b.SetReservedForEndpoint(ep, [3]byte{0, 0, 1}) + if !b.hasReserved() { + t.Fatal("non-zero reserved (byte 3) must count as reserved") + } + + // A second endpoint's non-zero value must also be seen. + b2 := stdNetBindForTest(t) + ep2 := netip.MustParseAddrPort("192.0.2.1:2408") + b2.SetReservedForEndpoint(ep, [3]byte{0, 0, 0}) + b2.SetReservedForEndpoint(ep2, [3]byte{0xAB, 0, 0}) + if !b2.hasReserved() { + t.Fatal("non-zero reserved on any endpoint must count as reserved") + } +} diff --git a/device/awg_stdnetbind_reserved_lx_test.go b/device/awg_stdnetbind_reserved_lx_test.go new file mode 100644 index 0000000..0d431e1 --- /dev/null +++ b/device/awg_stdnetbind_reserved_lx_test.go @@ -0,0 +1,190 @@ +/* SPDX-License-Identifier: MIT + * + * lx: e2e regression for the reserved-clear vs AWG magic-header collision, + * exercised over the StdNetBind path (no detour) with real loopback UDP. + * + * Bug model. On receive, StdNetBind.receiveIP unconditionally zeroed bytes + * 1-3 of every datagram >3 bytes (the Cloudflare WARP "reserved" field). + * AmneziaWG reads its magic header as LittleEndian.Uint32(packet[padding:]), + * where padding is s1/s2 (handshake) or s4 (transport). With small padding + * (0..3) the 4-byte magic overlaps bytes 1-3, so the unconditional clear + * corrupts it: the value falls outside the ranged h1-h4 window, the packet is + * classified MessageUnknownType and dropped. WARP was never configured on + * these binds (no SetReservedForEndpoint), so the clear was pure collateral. + * + * The fix gates the clear behind StdNetBind.hasReserved(): bytes 1-3 are only + * zeroed when a non-zero reserved value is actually set for some endpoint. + * With no reserved value the magic survives and the handshake completes. + * + * This test provokes the worst case: padding = 0 (no s1/s2/s4 at all), so the + * h1 initiation magic sits in bytes [0..3] and its high bytes (1-3) are the + * ones the clear would destroy. The h1-h4 ranges are chosen entirely above + * 0x10000000, so after zeroing bytes 1-3 the surviving value is <= 255 and can + * never land back inside any range -> guaranteed drop on the buggy tree. + * + * GREEN on the fixed tree. To see RED, temporarily restore the unconditional + * clear in conn/bind_std.go receiveIP: + * if msg.N > 3 { + * common.ClearArray(bufs[i][1:4]) + * } + * and the handshake times out (init magic zeroed in bytes 1-3). + */ + +package device + +import ( + "bufio" + "bytes" + "context" + "encoding/hex" + "fmt" + "strconv" + "strings" + "testing" + "time" + + "github.com/sagernet/wireguard-go/conn" +) + +// magic header ranges kept entirely above 0x10000000 (268435456). Any value +// the sender picks therefore has a non-zero byte among positions 1-3; zeroing +// those bytes collapses the value to <= 0xFF, which is below every range start, +// so a corrupted magic can never validate. Distinct windows per message type. +const ( + lxH1Lo, lxH1Hi = 268500000, 268600000 // init + lxH2Lo, lxH2Hi = 300000000, 300100000 // response + lxH3Lo, lxH3Hi = 400000000, 400100000 // cookie + lxH4Lo, lxH4Hi = 500000000, 500100000 // transport +) + +// lxReadListenPort parses listen_port= out of a device's IpcGet dump. +func lxReadListenPort(t *testing.T, dev *Device) uint16 { + t.Helper() + dump, err := dev.IpcGet() + if err != nil { + t.Fatalf("IpcGet: %v", err) + } + scanner := bufio.NewScanner(strings.NewReader(dump)) + for scanner.Scan() { + line := scanner.Text() + if v, ok := strings.CutPrefix(line, "listen_port="); ok { + p, err := strconv.Atoi(v) + if err != nil { + t.Fatalf("parse listen_port %q: %v", v, err) + } + return uint16(p) + } + } + t.Fatalf("listen_port not found in dump:\n%s", dump) + return 0 +} + +// newStdNetPaddedPair builds two Up()'d Devices peered over real loopback UDP +// (NewStdNetBind), configured with ranged h1-h4 + junk and *no* s1/s2/s4 +// (padding = 0). Endpoints are wired after Up, once the ephemeral ports are +// known. No reserved value is ever set, so hasReserved() is false. +func newStdNetPaddedPair(t *testing.T) (devA, devB *Device, tunA, tunB *chanTun) { + 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() + + tunA = newChanTun() + tunB = newChanTun() + + devA = NewDevice(context.Background(), tunA, conn.NewStdNetBind(nil), NewLogger(LogLevelError, "devA: "), 1) + devB = NewDevice(context.Background(), tunB, conn.NewStdNetBind(nil), NewLogger(LogLevelError, "devB: "), 1) + t.Cleanup(devA.Close) + t.Cleanup(devB.Close) + + // obfuscation shared by both ends. Ranged magic headers, junk packets, + // and deliberately no s1/s2/s4 so padding stays 0 for every message type. + obf := fmt.Sprintf( + "jc=3\njmin=8\njmax=16\n"+ + "h1=%d-%d\nh2=%d-%d\nh3=%d-%d\nh4=%d-%d\n", + lxH1Lo, lxH1Hi, lxH2Lo, lxH2Hi, lxH3Lo, lxH3Hi, lxH4Lo, lxH4Hi) + + // Bring both up on an ephemeral port (listen_port=0), no endpoint yet. + cfgA := fmt.Sprintf("private_key=%s\nlisten_port=0\n%sreplace_peers=true\npublic_key=%s\nallowed_ip=%s/32\n", + hex.EncodeToString(skA[:]), obf, hex.EncodeToString(pkB[:]), testIPB) + cfgB := fmt.Sprintf("private_key=%s\nlisten_port=0\n%sreplace_peers=true\npublic_key=%s\nallowed_ip=%s/32\n", + hex.EncodeToString(skB[:]), obf, 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 devA.paddings.init != 0 || devA.paddings.response != 0 || devA.paddings.transport != 0 { + t.Fatalf("padding must be 0 for this test: init=%d resp=%d transport=%d", + devA.paddings.init, devA.paddings.response, devA.paddings.transport) + } + + if err := devA.Up(); err != nil { + t.Fatalf("Up A: %v", err) + } + if err := devB.Up(); err != nil { + t.Fatalf("Up B: %v", err) + } + + portA := lxReadListenPort(t, devA) + portB := lxReadListenPort(t, devB) + if portA == 0 || portB == 0 { + t.Fatalf("ephemeral ports not assigned: A=%d B=%d", portA, portB) + } + + // Now that ports are known, point each peer at the other over loopback. + if err := devA.IpcSet(fmt.Sprintf("public_key=%s\nupdate_only=true\nendpoint=127.0.0.1:%d\n", + hex.EncodeToString(pkB[:]), portB)); err != nil { + t.Fatalf("set endpoint A->B: %v", err) + } + if err := devB.IpcSet(fmt.Sprintf("public_key=%s\nupdate_only=true\nendpoint=127.0.0.1:%d\n", + hex.EncodeToString(pkA[:]), portA)); err != nil { + t.Fatalf("set endpoint B->A: %v", err) + } + + return devA, devB, tunA, tunB +} + +// TestStdNetBindReservedClearVsMagic_ZeroPadding drives a real handshake and a +// data packet A->B over loopback UDP through StdNetBind, with padding=0 so the +// h1 magic overlaps the reserved bytes 1-3. It passes only when receive does +// not blindly clear those bytes (the fix). +func TestStdNetBindReservedClearVsMagic_ZeroPadding(t *testing.T) { + devA, _, tunA, tunB := newStdNetPaddedPair(t) + _ = devA + + pkt := buildIPv4Packet(testIPA, testIPB, 8) + + // Re-inject periodically: the first packet triggers the handshake and may + // be dropped until keys are established. + send := func() { tunA.toDevice <- pkt } + send() + + deadline := time.After(15 * time.Second) + retry := time.NewTicker(500 * time.Millisecond) + defer retry.Stop() + for { + select { + case got := <-tunB.fromDevice: + if bytes.Equal(got, pkt) { + return // delivered end to end: magic survived, handshake ok + } + t.Logf("ignoring unexpected packet len=%d", len(got)) + case <-retry.C: + send() + case <-deadline: + t.Fatal("timed out waiting for packet on peer tun " + + "(handshake never completed: reserved-clear likely corrupted the h1 magic)") + } + } +} diff --git a/device/cookie.go b/device/cookie.go index a093c8b..6a0463c 100644 --- a/device/cookie.go +++ b/device/cookie.go @@ -118,6 +118,7 @@ func (st *CookieChecker) CreateReply( msg []byte, recv uint32, src []byte, + msgType uint32, ) (*MessageCookieReply, error) { st.RLock() @@ -153,7 +154,7 @@ func (st *CookieChecker) CreateReply( smac1 := smac2 - blake2s.Size128 reply := new(MessageCookieReply) - reply.Type = MessageCookieReplyType + reply.Type = msgType reply.Receiver = recv _, err := rand.Read(reply.Nonce[:]) diff --git a/device/device.go b/device/device.go index a826c6d..fe11b7a 100644 --- a/device/device.go +++ b/device/device.go @@ -64,7 +64,8 @@ type Device struct { lookupFunc PeerLookupFunc // or nil if unused } - peerStateFn atomic.Pointer[PeerSessionStateFunc] // observes peer session state changes, nil if unset + peerStateFn atomic.Pointer[PeerSessionStateFunc] // observes peer session state changes, nil if unset + priorityMsgFn atomic.Pointer[PeerPriorityMessageFunc] // returns a priority message to be sent around session establishment, nil if unset rate struct { underLoadUntil atomic.Int64 @@ -98,6 +99,43 @@ type Device struct { closed chan struct{} log *Logger pauseManager pause.Manager + + // lx: AmneziaWG obfuscation state (grafted from amneziawg-go). + junk struct { + min int + max int + count int + } + + headers struct { + init *magicHeader + cookie *magicHeader + response *magicHeader + transport *magicHeader + } + + paddings struct { + init int + response int + cookie int + transport int + } + + ipackets [5]*obfChain + + // lx: SPEC 041 — passive self-heal state: reopen the bind once (fresh + // ephemeral port when freshPort is set) and immediately re-initiate, to + // heal 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. The whole mechanism — three triggers (giveup / + // early / nudge) sharing this state and its debounce — lives in + // lx_giveup_rebind.go. 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. @@ -171,7 +209,8 @@ func (device *Device) changeState(want deviceState) (err error) { err = errDown } } - device.log.Verbosef("Interface state was %s, requested %s, now %s", old, want, device.deviceState()) + device.log.Verbosef( + "Interface state was %s, requested %s, now %s", old, want, device.deviceState()) return } @@ -301,6 +340,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 @@ -316,6 +356,11 @@ func NewDevice(ctx context.Context, tunDevice tun.Device, bind conn.Bind, logger device.rate.limiter.Init() device.indexTable.Init() + device.headers.init = &magicHeader{start: MessageInitiationType, end: MessageInitiationType} + device.headers.response = &magicHeader{start: MessageResponseType, end: MessageResponseType} + device.headers.cookie = &magicHeader{start: MessageCookieReplyType, end: MessageCookieReplyType} + device.headers.transport = &magicHeader{start: MessageTransportType, end: MessageTransportType} + device.PopulatePools() // create queues @@ -553,6 +598,38 @@ func (device *Device) SetSessionStateFunc(f PeerSessionStateFunc) { device.peerStateFn.Store(&f) } +// MaxPriorityMessageContentSize is the maximum size of a message returned by a +// [PeerPriorityMessageFunc]. It's a power of 2 that leaves significant space +// when accounting for all WireGuard overhead and encapsulating network protocol +// headers. Future adjustments to this value should consider all these overheads +// and any [conn.Bind] implementation limitations. +const MaxPriorityMessageContentSize = 512 + +// PeerPriorityMessageFunc is called when a peer's WireGuard session keypair is +// established (or re-keyed) for forward data transmission. +// +// The returned message is transmitted to the peer in priority fashion. Priority +// means it cannot be evicted from the staged packet queue by non-priority +// (read from [tun.Device]) packets. It avoids the staged queue altogether. +// +// The callback must be cheap and must not call back into [Device]. A zero length +// message or a message whose length exceeds [MaxPriorityMessageContentSize] will +// be silently dropped. Message should start with an IPv4 or IPv6 header as it +// is subject to allowed IPs lookup on the receiver, same as any other transport +// message. +type PeerPriorityMessageFunc func(peer NoisePublicKey) (msg []byte) + +// SetPriorityMessageOnEstablishmentFunc sets a function to be used for sending +// a priority message around session establishment. See [PeerPriorityMessageFunc] +// docs for more details. A nil value clears any previously set value. +func (device *Device) SetPriorityMessageOnEstablishmentFunc(f PeerPriorityMessageFunc) { + if f == nil { + device.priorityMsgFn.Store(nil) + return + } + device.priorityMsgFn.Store(&f) +} + func (device *Device) Close() { device.state.Lock() defer device.state.Unlock() diff --git a/device/lx_early_rebind_test.go b/device/lx_early_rebind_test.go new file mode 100644 index 0000000..30910c1 --- /dev/null +++ b/device/lx_early_rebind_test.go @@ -0,0 +1,58 @@ +/* SPDX-License-Identifier: MIT + * + * lx: SPEC 041 v2 — behavioural red/green test for the EARLY give-up rebind. + * Field failure mode (the v1 field leftover, dump 2026-08-01): the v1 give-up + * rebind heals a dead 5-tuple, but only at ~90s after the first demand — while + * the user pings within the first 5-35s after device wake and sees every node + * in ERR. The early trigger fires from the retry branch once >=3 initiations + * went unanswered AND the session is provably dead (no live keypair, or last + * handshake older than RejectAfterTime), shrinking the ERR window to ~15-20s. + * + * The test reuses the v1 harness (gateBind: first socket generation blackholes + * every send) and drives the peer to the 3rd retry expiry — the state a real + * ~15s of unanswered retries ends in. Post-fix the retry branch rebinds and + * the tunnel comes up; pre-fix (v1 base) the retry keeps dying in the first + * socket generation and only the ~90s give-up would heal, so the packet never + * arrives within the test window. + * + * This file deliberately uses NO post-fix API, so it compiles and runs RED on + * the v1 base commit. + */ + +package device + +import ( + "testing" + "time" +) + +// TestEarlyRebindSelfHeal: dead first socket, cold session (no keypair yet), +// 3rd retry expiry fires — the tunnel must come up and deliver traffic without +// waiting out the full 90s give-up cycle. +func TestEarlyRebindSelfHeal(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. + send() + + pair.devA.peers.RLock() + peer := pair.devA.peers.keyMap[pair.pkB] + pair.devA.peers.RUnlock() + if peer == nil { + t.Fatal("peer not found") + } + + // Simulate reaching the 3rd retry expiry (~15s in the field): two retries + // already counted, the last initiation older than the retransmit timeout. + peer.handshake.mutex.Lock() + peer.handshake.lastSentHandshake = time.Now().Add(-2 * RekeyTimeout) + peer.handshake.mutex.Unlock() + peer.timers.handshakeAttempts.Store(2) + expiredRetransmitHandshake(peer) + + awaitPacket(t, pair.tunB, pkt, send) +} diff --git a/device/lx_giveup_rebind.go b/device/lx_giveup_rebind.go new file mode 100644 index 0000000..60d6684 --- /dev/null +++ b/device/lx_giveup_rebind.go @@ -0,0 +1,143 @@ +/* SPDX-License-Identifier: MIT + * + * lx: SPEC 041 — passive self-heal for a dead per-flow path (an expired NAT + * mapping or a poisoned DPI flow entry that pins every retry to the same dead + * 5-tuple until a manual reconnect). One mechanism — reopen the bind (fresh + * ephemeral port when allowed) and immediately re-initiate — with three + * triggers sharing one debounce window: + * + * giveup — the handshake retry cycle exhausted (~90s of unanswered + * initiations under traffic demand); safety net, covers every path; + * early — >=3 unanswered initiations against a provably dead session + * (see sessionProvablyDead): no point waiting out the rest of the + * cycle, rebind at ~15s instead of ~90s; + * nudge — the consumer reports "device woke up" via + * Device.RebindIfSessionStale (wired through sing-box libbox); + * heals without waiting for traffic demand at all. + * + * Zero cost while healthy: no timers, no goroutines — triggers 1-2 live in + * the existing retry cycle, trigger 3 is paid by the caller. The state lives + * in Device.giveUpRebind (device.go); enabled defaults to true in NewDevice, + * sing-box decides freshPort from whether the user pinned listen_port. + */ + +package device + +import "time" + +// earlyGiveUpMinAttempts is the number of unanswered initiations (retry timer +// expiries) after which a provably dead session is rebound early instead of +// waiting out the full RekeyAttemptTime cycle: ~15s at RekeyTimeout=5s. +const earlyGiveUpMinAttempts = 3 + +// SetGiveUpRebind configures the self-heal (see the giveUpRebind field +// comment in device.go). 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) +} + +// handleHandshakeGiveUp is invoked from the give-up branch of +// expiredRetransmitHandshake: ~90s of initiations went unanswered, so the +// current socket's 5-tuple is proven dead. +func (device *Device) handleHandshakeGiveUp(peer *Peer) { + device.selfHealRebind("giveup", peer) +} + +// maybeEarlyGiveUpRebind is invoked from the RETRY branch of +// expiredRetransmitHandshake. Once enough initiations went unanswered AND the +// session is provably dead there is nothing left to protect — rebind now, at +// ~15s instead of ~90s. The retry cycle itself continues untouched: this only +// moves the socket under it. A live session with transient packet loss fails +// sessionProvablyDead and keeps byte-for-byte upstream behaviour; the shared +// debounce means this also suppresses the giveup rebind of the same series. +func (device *Device) maybeEarlyGiveUpRebind(peer *Peer) { + if peer.timers.handshakeAttempts.Load() < earlyGiveUpMinAttempts { + return + } + if !device.sessionProvablyDead(peer) { + return + } + device.selfHealRebind("early", peer) +} + +// sessionProvablyDead reports whether the peer's session is beyond saving: no +// live keypair, or the last successful handshake is older than +// RejectAfterTime (the keys are invalid after that, so a rebind loses +// nothing). The stale predicate shared by the early and nudge triggers. +func (device *Device) sessionProvablyDead(peer *Peer) bool { + if peer.keypairs.Current() == nil { + return true + } + return time.Since(time.Unix(0, peer.lastHandshakeNano.Load())) > RejectAfterTime +} + +// RebindIfSessionStale is the wake-nudge entry (trigger 3): the consumer +// observed a device wake-up and asks for an immediate heal instead of waiting +// for traffic demand to walk the retry cycle. If any running peer's session +// is provably dead the bind is reopened once (shared debounce) and every such +// peer re-initiates immediately; a healthy device is a no-op. Returns whether +// a rebind was actually scheduled. Never blocks on the rebind itself — the +// heavy part runs in a goroutine (see selfHealRebind). On a down or closed +// device it is a no-op, so callers racing idle-suspend or Close are safe. +func (device *Device) RebindIfSessionStale() bool { + if !device.giveUpRebind.enabled.Load() || !device.isUp() { + return false + } + var stale []*Peer + device.peers.RLock() + for _, peer := range device.peers.keyMap { + if peer.isRunning.Load() && device.sessionProvablyDead(peer) { + stale = append(stale, peer) + } + } + device.peers.RUnlock() + if len(stale) == 0 { + return false + } + return device.selfHealRebind("nudge", stale...) +} + +// selfHealRebind is the shared action behind all three triggers. Runs the +// heavy part in a goroutine so a timer callback (or a nudge caller) never +// blocks on BindUpdate's worker drain. Debounced to one rebind per +// RekeyAttemptTime per device across ALL triggers (CAS on `last` settles +// concurrent multi-peer races): an early rebind at ~15s suppresses the giveup +// rebind of the same failed series at ~90s. 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) selfHealRebind(trigger string, peers ...*Peer) bool { + if !device.giveUpRebind.enabled.Load() { + return false + } + if device.isClosed() { + return false + } + now := time.Now().Unix() + last := device.giveUpRebind.last.Load() + if now-last < int64(RekeyAttemptTime/time.Second) { + return false + } + if !device.giveUpRebind.last.CompareAndSwap(last, now) { + return false + } + 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 self-heal rebind (trigger=%s): %v", peers[0], trigger, err) + return + } + device.log.Verbosef("%v - Rebound socket for self-heal (trigger=%s, fresh port=%v)", peers[0], trigger, fresh) + for _, peer := range peers { + peer.SendHandshakeInitiation(false) + } + }() + return true +} 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/lx_ipcget_awg_test.go b/device/lx_ipcget_awg_test.go new file mode 100644 index 0000000..a5954e6 --- /dev/null +++ b/device/lx_ipcget_awg_test.go @@ -0,0 +1,63 @@ +/* SPDX-License-Identifier: MIT + * + * Pins the AWG get path: IpcGet must report every obfuscation parameter it + * accepted, including i1..i5. The I-slots are emitted by an `i%d=` loop rather + * than literal per-key sendf calls, which makes them easy to miss when auditing + * introspection parity against amneziawg-go by grep alone. + */ + +package device + +import ( + "context" + "encoding/hex" + "strings" + "testing" +) + +func TestIpcGetReportsAWGParams(t *testing.T) { + sk, err := newPrivateKey() + if err != nil { + t.Fatalf("newPrivateKey: %v", err) + } + + bind, _ := newChanBindPair() + dev := NewDevice(context.Background(), newChanTun(), bind, NewLogger(LogLevelError, "dev: "), 1) + t.Cleanup(dev.Close) + + set := strings.Join([]string{ + "private_key=" + hex.EncodeToString(sk[:]), + "jc=4", "jmin=40", "jmax=70", + "s1=15", "s2=20", "s3=25", "s4=30", + "h1=1", "h2=2", "h3=3", "h4=100-200", + "i1=", "i3=", "i5=", + "", + }, "\n") + if err := dev.IpcSet(set); err != nil { + t.Fatalf("IpcSet: %v", err) + } + + got, err := dev.IpcGet() + if err != nil { + t.Fatalf("IpcGet: %v", err) + } + t.Logf("IpcGet:\n%s", got) + + for _, want := range []string{ + "jc=4", "jmin=40", "jmax=70", + "s1=15", "s2=20", "s3=25", "s4=30", + "h1=1", "h2=2", "h3=3", "h4=100-200", + "i1=", "i3=", "i5=", + } { + if !strings.Contains(got, want) { + t.Errorf("IpcGet missing %q", want) + } + } + + // Unset I-slots must stay absent, not surface as empty values. + for _, absent := range []string{"i2=", "i4="} { + if strings.Contains(got, absent) { + t.Errorf("IpcGet reported unset %q", absent) + } + } +} diff --git a/device/lx_stale_rebind_test.go b/device/lx_stale_rebind_test.go new file mode 100644 index 0000000..91b8a31 --- /dev/null +++ b/device/lx_stale_rebind_test.go @@ -0,0 +1,228 @@ +/* SPDX-License-Identifier: MIT + * + * lx: SPEC 041 v2 — unit tests for the stale predicate, the wake nudge + * (RebindIfSessionStale) and the shared debounce across triggers, on top of + * the v1 harness (lx_giveup_selfheal_test.go). These use the post-fix API and + * are NOT expected to compile on the pre-fix base. + */ + +package device + +import ( + "sync" + "testing" + "time" +) + +// establishTunnel completes a real handshake over a healthy pair so the peer +// holds a live keypair and a fresh lastHandshakeNano. +func establishTunnel(t *testing.T, pair *giveUpPair) { + t.Helper() + pkt := buildIPv4Packet(testIPA, testIPB, 8) + send := func() { pair.tunA.toDevice <- pkt } + send() + awaitPacket(t, pair.tunB, pkt, send) +} + +func peerOf(t *testing.T, dev *Device, pk NoisePublicKey) *Peer { + t.Helper() + dev.peers.RLock() + peer := dev.peers.keyMap[pk] + dev.peers.RUnlock() + if peer == nil { + t.Fatal("peer not found") + } + return peer +} + +// A cold peer (no keypair yet, nothing to lose): the nudge must rebind and +// immediately initiate — the tunnel comes up without any traffic demand. +func TestNudgeRebindsStaleSession(t *testing.T) { + pair := newGiveUpPair(t, true) + + if !pair.devA.RebindIfSessionStale() { + t.Fatal("nudge on a cold (keypair-less) session must rebind") + } + waitOpens(t, pair.bindA, 2) + + // The immediate initiation must bring the tunnel up: traffic sent only + // AFTER the nudge flows end to end. + pkt := buildIPv4Packet(testIPA, testIPB, 8) + send := func() { pair.tunA.toDevice <- pkt } + send() + awaitPacket(t, pair.tunB, pkt, send) +} + +// A healthy session (live keypair, fresh handshake) must be a strict no-op. +func TestNudgeHealthySessionNoop(t *testing.T) { + pair := newGiveUpPair(t, false) + establishTunnel(t, pair) + + opens := pair.bindA.openCount() + if pair.devA.RebindIfSessionStale() { + t.Fatal("nudge on a healthy session must not rebind") + } + time.Sleep(100 * time.Millisecond) + if got := pair.bindA.openCount(); got != opens { + t.Fatalf("healthy nudge reopened the bind: %d opens, want %d", got, opens) + } +} + +// A live keypair whose last handshake is older than RejectAfterTime is +// provably dead (the keys are invalid): the nudge must rebind. +func TestNudgeExpiredHandshakeIsStale(t *testing.T) { + pair := newGiveUpPair(t, false) + establishTunnel(t, pair) + + peer := peerOf(t, pair.devA, pair.pkB) + peer.lastHandshakeNano.Store(time.Now().Add(-RejectAfterTime - time.Second).UnixNano()) + + if !pair.devA.RebindIfSessionStale() { + t.Fatal("nudge on an expired session must rebind") + } + waitOpens(t, pair.bindA, 2) +} + +// A down device (how SPEC 020 idle-suspend leaves it) must be a no-op — the +// nudge never wakes sleepers. +func TestNudgeDownDeviceNoop(t *testing.T) { + pair := newGiveUpPair(t, true) + + if err := pair.devA.Down(); err != nil { + t.Fatalf("Down: %v", err) + } + if pair.devA.RebindIfSessionStale() { + t.Fatal("nudge on a down device must be a no-op") + } +} + +// Pinned listen_port survives the nudge rebind. +func TestNudgePinnedPortPreserved(t *testing.T) { + pair := newGiveUpPair(t, true) + pair.devA.SetGiveUpRebind(true, false) + + if !pair.devA.RebindIfSessionStale() { + t.Fatal("nudge must rebind a cold session") + } + waitOpens(t, pair.bindA, 2) + + ports := pair.bindA.portsSnapshot() + if ports[1] != 1 { + t.Fatalf("nudge rebind requested port %d, want 1 (pinned)", ports[1]) + } +} + +// The debounce window is SHARED across triggers: an early/nudge rebind +// suppresses the give-up rebind of the same failed series, and a later series +// (window elapsed) heals again — sliding window, not a latch. +func TestSharedDebounceAcrossTriggers(t *testing.T) { + pair := newGiveUpPair(t, false) + + // First trigger of the series: nudge. + if !pair.devA.RebindIfSessionStale() { + t.Fatal("first nudge must rebind") + } + waitOpens(t, pair.bindA, 2) + + // The give-up of the same series lands inside the window: suppressed. + triggerGiveUp(t, pair.devA, pair.pkB) + time.Sleep(300 * time.Millisecond) + if got := pair.bindA.openCount(); got != 2 { + t.Fatalf("give-up inside the shared window rebound: %d opens, want 2", got) + } + + // Next series: age the window as wall clocks would — heals again. + pair.devA.giveUpRebind.last.Store(time.Now().Add(-RekeyAttemptTime - time.Second).Unix()) + triggerGiveUp(t, pair.devA, pair.pkB) + waitOpens(t, pair.bindA, 3) +} + +// The early trigger must NOT fire before enough initiations went unanswered, +// even against a provably dead session. +func TestEarlyRebindNeedsMinAttempts(t *testing.T) { + pair := newGiveUpPair(t, true) + peer := peerOf(t, pair.devA, pair.pkB) + + peer.handshake.mutex.Lock() + peer.handshake.lastSentHandshake = time.Now().Add(-2 * RekeyTimeout) + peer.handshake.mutex.Unlock() + peer.timers.handshakeAttempts.Store(0) // this expiry brings it to 1 (< min) + expiredRetransmitHandshake(peer) + + time.Sleep(300 * time.Millisecond) + if got := pair.bindA.openCount(); got != 1 { + t.Fatalf("early rebind fired below the attempt floor: %d opens, want 1", got) + } +} + +// A fresh session (live keypair, recent handshake) must keep the retry branch +// byte-for-byte upstream even past the attempt floor: no rebind. +func TestEarlyRebindFreshSessionNoop(t *testing.T) { + pair := newGiveUpPair(t, false) + establishTunnel(t, pair) + peer := peerOf(t, pair.devA, pair.pkB) + + opens := pair.bindA.openCount() + peer.handshake.mutex.Lock() + peer.handshake.lastSentHandshake = time.Now().Add(-2 * RekeyTimeout) + peer.handshake.mutex.Unlock() + peer.timers.handshakeAttempts.Store(earlyGiveUpMinAttempts) + expiredRetransmitHandshake(peer) + + time.Sleep(300 * time.Millisecond) + if got := pair.bindA.openCount(); got != opens { + t.Fatalf("early rebind fired on a fresh session: %d opens, want %d", got, opens) + } +} + +// Nudge racing Close: no panic, no deadlock, no race-detector report. +func TestNudgeRacesClose(t *testing.T) { + for i := 0; i < 25; i++ { + pair := newGiveUpPair(t, false) + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + pair.devA.RebindIfSessionStale() + }() + go func() { + defer wg.Done() + pair.devA.Close() + }() + wg.Wait() + } +} + +// Nudge racing Down/Up (the SPEC 020 suspend/resume shape): the device must +// end consistent — up, with a live bind. +func TestNudgeRacesSuspend(t *testing.T) { + for i := 0; i < 25; i++ { + pair := newGiveUpPair(t, false) + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + pair.devA.RebindIfSessionStale() + }() + go func() { + defer wg.Done() + if err := pair.devA.Down(); err != nil { + t.Errorf("Down: %v", err) + } + if err := pair.devA.Up(); err != nil { + t.Errorf("Up: %v", err) + } + }() + wg.Wait() + + pair.devA.net.RLock() + bindAlive := pair.devA.net.bind != nil + pair.devA.net.RUnlock() + if !bindAlive || !pair.devA.isUp() { + t.Fatalf("iteration %d: device inconsistent after nudge/suspend race (bind=%v up=%v)", + i, bindAlive, pair.devA.isUp()) + } + pair.devA.Close() + pair.devB.Close() + } +} diff --git a/device/magic-header.go b/device/magic-header.go new file mode 100644 index 0000000..6ea0ce5 --- /dev/null +++ b/device/magic-header.go @@ -0,0 +1,65 @@ +package device + +import ( + "crypto/rand" + "errors" + "fmt" + "math/big" + "strconv" + "strings" +) + +type magicHeader struct { + start uint32 + end uint32 +} + +func newMagicHeader(spec string) (*magicHeader, error) { + parts := strings.Split(spec, "-") + if len(parts) < 1 || len(parts) > 2 { + return nil, errors.New("bad format") + } + + start, err := strconv.ParseUint(parts[0], 10, 32) + if err != nil { + return nil, fmt.Errorf("failed to parse %s: %w", parts[0], err) + } + + var end uint64 + if len(parts) > 1 { + end, err = strconv.ParseUint(parts[1], 10, 32) + if err != nil { + return nil, fmt.Errorf("failed to parse %s: %w", parts[1], err) + } + } else { + end = start + } + + if end < start { + return nil, errors.New("wrong range specified") + } + + return &magicHeader{ + start: uint32(start), + end: uint32(end), + }, nil +} + +func (h *magicHeader) GenSpec() string { + if h.start == h.end { + return fmt.Sprintf("%d", h.start) + } + return fmt.Sprintf("%d-%d", h.start, h.end) +} + +func (h *magicHeader) Validate(val uint32) bool { + return h.start <= val && val <= h.end +} + +func (h *magicHeader) Generate() uint32 { + // Widen before arithmetic: end-start+1 in uint32 wraps to 0 for the + // full 0..2^32-1 range, which would panic rand.Int (bound <= 0). + high := int64(h.end) - int64(h.start) + 1 + r, _ := rand.Int(rand.Reader, big.NewInt(high)) + return h.start + uint32(r.Int64()) +} diff --git a/device/noise-protocol.go b/device/noise-protocol.go index d72bb25..75fa025 100644 --- a/device/noise-protocol.go +++ b/device/noise-protocol.go @@ -54,10 +54,11 @@ const ( ) const ( - MessageInitiationType = 1 - MessageResponseType = 2 - MessageCookieReplyType = 3 - MessageTransportType = 4 + MessageUnknownType uint32 = 0 + MessageInitiationType uint32 = 1 + MessageResponseType uint32 = 2 + MessageCookieReplyType uint32 = 3 + MessageTransportType uint32 = 4 ) const ( @@ -65,7 +66,7 @@ const ( MessageResponseSize = 92 // size of response message MessageCookieReplySize = 64 // size of cookie reply message MessageTransportHeaderSize = 16 // size of data preceding content in transport message - MessageEncapsulatingTransportSize = 8 // size of optional, free (for use by conn.Bind.Send()) space preceding the transport header + MessageEncapsulatingTransportSize = 0 // lx: zeroed so AmneziaWG obfuscation composes without sagernet headroom (AWG path doesn't use the Bind.Send prepend) MessageTransportSize = MessageTransportHeaderSize + poly1305.TagSize // size of empty transport MessageKeepaliveSize = MessageTransportSize // size of keepalive MessageHandshakeSize = MessageInitiationSize // size of largest handshake related message @@ -218,7 +219,7 @@ type Handshake struct { localEphemeral NoisePrivateKey // ephemeral secret key localIndex uint32 // used to clear hash-table remoteIndex uint32 // index for sending - remoteStatic NoisePublicKey // long term key + remoteStatic NoisePublicKey // long term key, never changes, can be accessed without mutex remoteEphemeral NoisePublicKey // ephemeral public key precomputedStaticStatic [NoisePublicKeySize]byte // precomputed shared secret lastTimestamp tai64n.Timestamp @@ -287,8 +288,10 @@ func (device *Device) CreateMessageInitiation(peer *Peer) (*MessageInitiation, e handshake.mixHash(handshake.remoteStatic[:]) + msgType := device.headers.init.Generate() + msg := MessageInitiation{ - Type: MessageInitiationType, + Type: msgType, Ephemeral: handshake.localEphemeral.publicKey(), } @@ -471,7 +474,7 @@ func (device *Device) CreateMessageResponse(peer *Peer) (*MessageResponse, error } var msg MessageResponse - msg.Type = MessageResponseType + msg.Type = device.headers.response.Generate() msg.Sender = handshake.localIndex msg.Receiver = handshake.remoteIndex diff --git a/device/obf.go b/device/obf.go new file mode 100644 index 0000000..269007c --- /dev/null +++ b/device/obf.go @@ -0,0 +1,155 @@ +package device + +import ( + "errors" + "fmt" + "strconv" + "strings" +) + +type obfBuilder func(val string) (obf, error) + +// parseObfLen parses and bounds an obfuscator length argument: a negative +// value would panic slice bounds in obfChain.Obfuscate, a huge one would +// OOM in the handshake-time make (SendHandshakeInitiation). +func parseObfLen(val string) (int, error) { + length, err := strconv.Atoi(val) + if err != nil { + return 0, err + } + if length < 0 || length > MaxMessageSize { + return 0, fmt.Errorf("obfuscator length %d out of range [0, %d]", length, MaxMessageSize) + } + return length, nil +} + +var obfBuilders = map[string]obfBuilder{ + "b": newBytesObf, + "t": newTimestampObf, + "r": newRandObf, + "rc": newRandCharObf, + "rd": newRandDigitsObf, + "d": newDataObf, + "ds": newDataStringObf, + "dz": newDataSizeObf, +} + +type obf interface { + Obfuscate(dst, src []byte) + Deobfuscate(dst, src []byte) bool + ObfuscatedLen(srcLen int) int + DeobfuscatedLen(srcLen int) int +} + +type obfChain struct { + Spec string + obfs []obf +} + +func newObfChain(spec string) (*obfChain, error) { + var ( + obfs []obf + errs []error + ) + + remaining := spec[:] + for { + start := strings.IndexByte(remaining, '<') + if start == -1 { + break + } + + end := strings.IndexByte(remaining[start:], '>') + if end == -1 { + return nil, errors.New("missing enclosing >") + } + end += start + + tag := remaining[start+1 : end] + parts := strings.Fields(tag) + if len(parts) == 0 { + errs = append(errs, errors.New("empty tag")) + remaining = remaining[end+1:] + continue + } + + key := parts[0] + builder, ok := obfBuilders[key] + if !ok { + errs = append(errs, fmt.Errorf("unknown tag <%s>", key)) + remaining = remaining[end+1:] + continue + } + + val := "" + if len(parts) > 1 { + val = parts[1] + } + + o, err := builder(val) + if err != nil { + errs = append(errs, fmt.Errorf("failed to build <%s>: %w", key, err)) + remaining = remaining[end+1:] + continue + } + + obfs = append(obfs, o) + remaining = remaining[end+1:] + } + + if len(errs) > 0 { + return nil, errors.Join(errs...) + } + + return &obfChain{ + Spec: spec, + obfs: obfs, + }, nil +} + +func (c *obfChain) Obfuscate(dst, src []byte) { + written := 0 + for _, o := range c.obfs { + obfLen := o.ObfuscatedLen(len(src)) + o.Obfuscate(dst[written:written+obfLen], src) + written += obfLen + } +} + +func (c *obfChain) Deobfuscate(dst, src []byte) bool { + dynamicLen := len(src) - c.ObfuscatedLen(0) + + written, read := 0, 0 + + for _, o := range c.obfs { + deobfLen := o.DeobfuscatedLen(dynamicLen) + obfLen := o.ObfuscatedLen(deobfLen) + + if !o.Deobfuscate(dst[written:written+deobfLen], src[read:read+obfLen]) { + return false + } + + written += deobfLen + read += obfLen + } + + return true +} + +func (c *obfChain) ObfuscatedLen(n int) int { + total := 0 + for _, o := range c.obfs { + total += o.ObfuscatedLen(n) + } + return total +} + +func (c *obfChain) DeobfuscatedLen(n int) int { + dynamicLen := n - c.ObfuscatedLen(0) + + total := 0 + for _, o := range c.obfs { + total += o.DeobfuscatedLen(dynamicLen) + } + return total +} diff --git a/device/obf_bytes.go b/device/obf_bytes.go new file mode 100644 index 0000000..68d722b --- /dev/null +++ b/device/obf_bytes.go @@ -0,0 +1,47 @@ +package device + +import ( + "bytes" + "encoding/hex" + "errors" + "strings" +) + +func newBytesObf(val string) (obf, error) { + val = strings.TrimPrefix(val, "0x") + + if len(val) == 0 { + return nil, errors.New("empty argument") + } + + if len(val)%2 != 0 { + return nil, errors.New("odd amount of symbols") + } + + bytes, err := hex.DecodeString(val) + if err != nil { + return nil, err + } + + return &bytesObf{data: bytes}, nil +} + +type bytesObf struct { + data []byte +} + +func (o *bytesObf) Obfuscate(dst, src []byte) { + copy(dst, o.data) +} + +func (o *bytesObf) Deobfuscate(dst, src []byte) bool { + return bytes.Equal(o.data, src[:o.ObfuscatedLen(0)]) +} + +func (o *bytesObf) ObfuscatedLen(srcLen int) int { + return len(o.data) +} + +func (o *bytesObf) DeobfuscatedLen(srcLen int) int { + return 0 +} diff --git a/device/obf_data.go b/device/obf_data.go new file mode 100644 index 0000000..42d3f65 --- /dev/null +++ b/device/obf_data.go @@ -0,0 +1,25 @@ +package device + +func newDataObf(val string) (obf, error) { + return &dataObf{}, nil +} + +type dataObf struct { +} + +func (obf *dataObf) Obfuscate(dst, src []byte) { + copy(dst, src) +} + +func (obf *dataObf) Deobfuscate(dst, src []byte) bool { + copy(dst, src) + return true +} + +func (o *dataObf) ObfuscatedLen(n int) int { + return n +} + +func (o *dataObf) DeobfuscatedLen(n int) int { + return n +} diff --git a/device/obf_datasize.go b/device/obf_datasize.go new file mode 100644 index 0000000..8ad1a71 --- /dev/null +++ b/device/obf_datasize.go @@ -0,0 +1,36 @@ +package device + +func newDataSizeObf(val string) (obf, error) { + length, err := parseObfLen(val) + if err != nil { + return nil, err + } + + return &dataSizeObf{ + length: length, + }, nil +} + +type dataSizeObf struct { + length int +} + +func (o *dataSizeObf) Obfuscate(dst, src []byte) { + srcLen := len(src) + for i := o.length - 1; i >= 0; i-- { + dst[i] = byte(srcLen & 0xFF) + srcLen >>= 8 + } +} + +func (o *dataSizeObf) Deobfuscate(dst, src []byte) bool { + return true +} + +func (o *dataSizeObf) ObfuscatedLen(srcLen int) int { + return o.length +} + +func (o *dataSizeObf) DeobfuscatedLen(srcLen int) int { + return 0 +} diff --git a/device/obf_datastring.go b/device/obf_datastring.go new file mode 100644 index 0000000..2701e95 --- /dev/null +++ b/device/obf_datastring.go @@ -0,0 +1,29 @@ +package device + +import ( + "encoding/base64" +) + +func newDataStringObf(val string) (obf, error) { + return &dataStringObf{}, nil +} + +type dataStringObf struct { +} + +func (o *dataStringObf) Obfuscate(dst, src []byte) { + base64.RawStdEncoding.Encode(dst, src) +} + +func (o *dataStringObf) Deobfuscate(dst, src []byte) bool { + base64.RawStdEncoding.Decode(dst, src) + return true +} + +func (o *dataStringObf) ObfuscatedLen(n int) int { + return base64.RawStdEncoding.EncodedLen(n) +} + +func (o *dataStringObf) DeobfuscatedLen(n int) int { + return base64.RawStdEncoding.DecodedLen(n) +} diff --git a/device/obf_guards_test.go b/device/obf_guards_test.go new file mode 100644 index 0000000..5080196 --- /dev/null +++ b/device/obf_guards_test.go @@ -0,0 +1,108 @@ +/* SPDX-License-Identifier: MIT + * + * Guards around AWG obfuscation config values: these tests pin the + * crash-on-config-value fixes (swapped jmin/jmax, out-of-range obfuscator + * lengths, full-range magic headers). + */ + +package device + +import ( + "context" + "encoding/hex" + "fmt" + "testing" +) + +func TestParseObfLen(t *testing.T) { + cases := []struct { + val string + want int + wantErr bool + }{ + {"0", 0, false}, + {"100", 100, false}, + {fmt.Sprintf("%d", MaxMessageSize), MaxMessageSize, false}, + {"-1", 0, true}, // would panic slice bounds in Obfuscate + {fmt.Sprintf("%d", MaxMessageSize+1), 0, true}, // would OOM the handshake make + {"2000000000", 0, true}, + {"abc", 0, true}, + } + for _, c := range cases { + got, err := parseObfLen(c.val) + if c.wantErr != (err != nil) { + t.Errorf("parseObfLen(%q): err = %v, wantErr = %v", c.val, err, c.wantErr) + } + if err == nil && got != c.want { + t.Errorf("parseObfLen(%q) = %d, want %d", c.val, got, c.want) + } + } +} + +func TestMagicHeaderGenerateFullRange(t *testing.T) { + // end-start+1 computed in uint32 wraps to 0 for the full range and + // panics rand.Int; the fix widens to int64 before the arithmetic. + h := &magicHeader{start: 0, end: ^uint32(0)} + for i := 0; i < 8; i++ { + v := h.Generate() + if !h.Validate(v) { + t.Fatalf("generated value %d outside range", v) + } + } +} + +// TestJunkSwappedBounds brings up a device pair whose junk config has +// jmin > jmax (passes per-field UAPI validation); without the swap guard +// the first handshake panics rand.Int with a non-positive bound. +func TestJunkSwappedBounds(t *testing.T) { + 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() + + bindA, bindB := newChanBindPair() + tunA := newChanTun() + tunB := newChanTun() + + devA := NewDevice(context.Background(), tunA, bindA, NewLogger(LogLevelError, "devA: "), 1) + devB := NewDevice(context.Background(), tunB, bindB, NewLogger(LogLevelError, "devB: "), 1) + t.Cleanup(devA.Close) + t.Cleanup(devB.Close) + + // jmin deliberately greater than jmax: each field alone is valid. + junk := "jc=2\njmin=100\njmax=50\n" + cfgA := fmt.Sprintf( + "private_key=%s\n%sreplace_peers=true\npublic_key=%s\nendpoint=127.0.0.1:2\nallowed_ip=%s/32\n", + hex.EncodeToString(skA[:]), junk, hex.EncodeToString(pkB[:]), testIPB) + cfgB := fmt.Sprintf( + "private_key=%s\n%sreplace_peers=true\npublic_key=%s\nendpoint=127.0.0.1:1\nallowed_ip=%s/32\n", + hex.EncodeToString(skB[:]), junk, 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) + } + + // Drive a packet end-to-end: the handshake (junk packets included) + // must complete without panicking the process. + pkt := buildIPv4Packet(testIPA, testIPB, 28) + devA.InputPacket(testIPB.AsSlice(), [][]byte{pkt}) + awaitPacket(t, tunB, pkt, func() { + devA.InputPacket(testIPB.AsSlice(), [][]byte{pkt}) + }) +} diff --git a/device/obf_rand.go b/device/obf_rand.go new file mode 100644 index 0000000..1560460 --- /dev/null +++ b/device/obf_rand.go @@ -0,0 +1,38 @@ +package device + +import ( + "crypto/rand" +) + +func newRandObf(val string) (obf, error) { + length, err := parseObfLen(val) + if err != nil { + return nil, err + } + + return &randObf{ + length: length, + }, nil +} + +type randObf struct { + length int +} + +func (o *randObf) Obfuscate(dst, src []byte) { + rand.Read(dst[:o.length]) +} + +func (o *randObf) Deobfuscate(dst, src []byte) bool { + // there is no way to validate randomness :) + // assume that it is always true + return true +} + +func (o *randObf) ObfuscatedLen(n int) int { + return o.length +} + +func (o *randObf) DeobfuscatedLen(n int) int { + return 0 +} diff --git a/device/obf_randchars.go b/device/obf_randchars.go new file mode 100644 index 0000000..470ca6f --- /dev/null +++ b/device/obf_randchars.go @@ -0,0 +1,47 @@ +package device + +import ( + "crypto/rand" + "unicode" +) + +const chars52 = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ" + +func newRandCharObf(val string) (obf, error) { + length, err := parseObfLen(val) + if err != nil { + return nil, err + } + + return &randCharObf{ + length: length, + }, nil +} + +type randCharObf struct { + length int +} + +func (o *randCharObf) Obfuscate(dst, src []byte) { + rand.Read(dst[:o.length]) + for i := range dst[:o.length] { + dst[i] = chars52[dst[i]%52] + } +} + +func (o *randCharObf) Deobfuscate(dst, src []byte) bool { + for _, b := range src[:o.length] { + if !unicode.IsLetter(rune(b)) { + return false + } + } + return true +} + +func (o *randCharObf) ObfuscatedLen(n int) int { + return o.length +} + +func (o *randCharObf) DeobfuscatedLen(n int) int { + return 0 +} diff --git a/device/obf_randdigits.go b/device/obf_randdigits.go new file mode 100644 index 0000000..d3585a0 --- /dev/null +++ b/device/obf_randdigits.go @@ -0,0 +1,47 @@ +package device + +import ( + "crypto/rand" + "unicode" +) + +const digits10 = "0123456789" + +func newRandDigitsObf(val string) (obf, error) { + length, err := parseObfLen(val) + if err != nil { + return nil, err + } + + return &randDigitObf{ + length: length, + }, nil +} + +type randDigitObf struct { + length int +} + +func (o *randDigitObf) Obfuscate(dst, src []byte) { + rand.Read(dst[:o.length]) + for i := range dst[:o.length] { + dst[i] = digits10[dst[i]%10] + } +} + +func (o *randDigitObf) Deobfuscate(dst, src []byte) bool { + for _, b := range src[:o.length] { + if !unicode.IsDigit(rune(b)) { + return false + } + } + return true +} + +func (o *randDigitObf) ObfuscatedLen(n int) int { + return o.length +} + +func (o *randDigitObf) DeobfuscatedLen(n int) int { + return 0 +} diff --git a/device/obf_timestamp.go b/device/obf_timestamp.go new file mode 100644 index 0000000..0a8180b --- /dev/null +++ b/device/obf_timestamp.go @@ -0,0 +1,31 @@ +package device + +import ( + "encoding/binary" + "time" +) + +func newTimestampObf(_ string) (obf, error) { + return ×tampObf{}, nil +} + +type timestampObf struct{} + +func (o *timestampObf) Obfuscate(dst, src []byte) { + t := uint32(time.Now().Unix()) + binary.BigEndian.PutUint32(dst, t) +} + +func (o *timestampObf) Deobfuscate(dst, src []byte) bool { + // replay attack check? + // requires time to be always synchronized + return true +} + +func (o *timestampObf) ObfuscatedLen(n int) int { + return 4 +} + +func (o *timestampObf) DeobfuscatedLen(n int) int { + return 0 +} diff --git a/device/receive.go b/device/receive.go index e11e30a..8064f61 100644 --- a/device/receive.go +++ b/device/receive.go @@ -76,7 +76,10 @@ func (peer *Peer) keepKeyFreshReceiving() { * Every time the bind is updated a new routine is started for * IPv4 and IPv6 (separately) */ -func (device *Device) RoutineReceiveIncoming(maxBatchSize int, recv conn.ReceiveFunc) { +func (device *Device) RoutineReceiveIncoming( + maxBatchSize int, + recv conn.ReceiveFunc, +) { recvName := recv.PrettyName() defer func() { device.log.Verbosef("Routine: receive incoming %s - stopped", recvName) @@ -139,9 +142,14 @@ func (device *Device) RoutineReceiveIncoming(maxBatchSize int, recv conn.Receive } // check size of packet - packet := bufsArrs[i][:size] - msgType := binary.LittleEndian.Uint32(packet[:4]) + + // get message padding and type based on information from S1-S4 and H1-H4 + msgType, padding := device.DeterminePacketTypeAndPadding(packet, MessageUnknownType) + if padding > 0 { + copy(packet, packet[padding:]) + packet = packet[:len(packet)-padding] + } switch msgType { @@ -283,7 +291,6 @@ func (device *Device) RoutineHandshake(id int) { device.log.Verbosef("Routine: handshake worker %d - started", id) for elem := range device.queue.handshake.c { - // handle cookie fields and ratelimiting switch elem.msgType { @@ -310,9 +317,14 @@ func (device *Device) RoutineHandshake(id int) { // consume reply if peer := entry.peer; peer.isRunning.Load() { - device.log.Verbosef("Receiving cookie response from %s", elem.endpoint.DstToString()) + device.log.Verbosef( + "Receiving cookie response from %s", + elem.endpoint.DstToString(), + ) if !peer.cookieGenerator.ConsumeReply(&reply) { - device.log.Verbosef("Could not decrypt invalid cookie response") + device.log.Verbosef( + "Could not decrypt invalid cookie response", + ) } } @@ -354,9 +366,7 @@ func (device *Device) RoutineHandshake(id int) { switch elem.msgType { case MessageInitiationType: - // unmarshal - var msg MessageInitiation err := msg.unmarshal(elem.packet) if err != nil { @@ -364,7 +374,8 @@ func (device *Device) RoutineHandshake(id int) { goto skip } - // consume initiation + // have to reassign msgType for ranged msgType to work + msg.Type = elem.msgType peer := device.ConsumeMessageInitiation(&msg, elem.endpoint) if peer == nil { @@ -396,6 +407,9 @@ func (device *Device) RoutineHandshake(id int) { goto skip } + // have to reassign msgType for ranged msgType to work + msg.Type = elem.msgType + // consume response peer := device.ConsumeMessageResponse(&msg) @@ -425,6 +439,7 @@ func (device *Device) RoutineHandshake(id int) { peer.timersSessionDerived() peer.timersHandshakeComplete() + peer.SendPriorityMessage() peer.SendKeepalive() } skip: @@ -493,6 +508,7 @@ func (peer *Peer) processInboundContainer(elemsContainer *QueueInboundElementsCo if peer.ReceivedWithKeypair(elem.keypair) { peer.SetEndpointFromPacket(elem.endpoint) peer.timersHandshakeComplete() + peer.SendPriorityMessage() peer.SendStagedPackets() } if ep, ok := elem.endpoint.(conn.PeerAwareEndpoint); ok { @@ -571,3 +587,57 @@ func (peer *Peer) processInboundContainer(elemsContainer *QueueInboundElementsCo device.PutInboundElement(elem) } } + +func (device *Device) DeterminePacketTypeAndPadding(packet []byte, expectedType uint32) (uint32, int) { + size := len(packet) + + if expectedType == MessageUnknownType || expectedType == MessageInitiationType { + padding := device.paddings.init + header := device.headers.init + + if size == padding+MessageInitiationSize { + data := packet[padding:] + if header.Validate(binary.LittleEndian.Uint32(data)) { + return MessageInitiationType, padding + } + } + } + + if expectedType == MessageUnknownType || expectedType == MessageResponseType { + padding := device.paddings.response + header := device.headers.response + + if size == padding+MessageResponseSize { + data := packet[padding:] + if header.Validate(binary.LittleEndian.Uint32(data)) { + return MessageResponseType, padding + } + } + } + + if expectedType == MessageUnknownType || expectedType == MessageCookieReplyType { + padding := device.paddings.cookie + header := device.headers.cookie + + if size == padding+MessageCookieReplySize { + data := packet[padding:] + if header.Validate(binary.LittleEndian.Uint32(data)) { + return MessageCookieReplyType, padding + } + } + } + + if expectedType == MessageUnknownType || expectedType == MessageTransportType { + padding := device.paddings.transport + header := device.headers.transport + + if size >= padding+MessageTransportHeaderSize { + data := packet[padding:] + if header.Validate(binary.LittleEndian.Uint32(data)) { + return MessageTransportType, padding + } + } + } + + return MessageUnknownType, 0 +} diff --git a/device/send.go b/device/send.go index fae5c41..982d61f 100644 --- a/device/send.go +++ b/device/send.go @@ -6,9 +6,12 @@ package device import ( + "bytes" + "crypto/rand" "encoding/binary" "errors" "fmt" + "math/big" "net" "net/netip" "os" @@ -107,6 +110,70 @@ func (peer *Peer) SendKeepalive() { peer.SendStagedPackets() } +// SendPriorityMessage invokes the [PeerPriorityMessageFunc] callback if one is +// set, and queues the returned message for encryption and transmission if the +// current keypair is valid. +func (peer *Peer) SendPriorityMessage() { + f := peer.device.priorityMsgFn.Load() + if f == nil { + return + } + keypair := peer.keypairs.Current() + if keypair == nil || keypair.sendNonce.Load() >= RejectAfterMessages || time.Since(keypair.created) >= RejectAfterTime { + // SendStagedPackets initializes a handshake when the keypair is invalid, + // but we explicitly avoid that here. A priority message is only intended + // to flow around symmetric session establishment, but it should never + // trigger a new session. Reaching this branch due to nonce exhaustion + // or keypair expiration is highly unlikely considering where + // SendPriorityMessage is called (at current keypair establishment). + return + } + + // get plaintext message to send + msg := (*f)(peer.handshake.remoteStatic) + if len(msg) == 0 { + return + } + if len(msg) > MaxPriorityMessageContentSize { + peer.device.log.Verbosef("%v - Failed to queue priority message due to size", peer) + return + } + + // get pooled elements + elem := peer.device.NewOutboundElement() + elemsContainer := peer.device.GetOutboundElementsContainer() + elemsContainer.elems = append(elemsContainer.elems, elem) + packetQueued := false + defer func() { + if !packetQueued { + peer.device.PutOutboundBuffer(elem.buffer) + peer.device.PutOutboundElement(elem) + peer.device.PutOutboundElementsContainer(elemsContainer) + } + }() + + // initialize outbound element + const offset = MessageEncapsulatingTransportSize + MessageTransportHeaderSize + n := copy(elem.buffer[offset:], msg) + elem.packet = elem.buffer[offset : offset+n] + elem.peer = peer + elem.nonce = keypair.sendNonce.Add(1) - 1 + if elem.nonce >= RejectAfterMessages { + keypair.sendNonce.Store(RejectAfterMessages) + return + } + elem.keypair = keypair + + // add to parallel and sequential queue + if peer.isRunning.Load() { + elemsContainer.filling.Add(1) + peer.queuedOutboundPackets.Add(1) + peer.queue.outbound.c <- elemsContainer + peer.device.queue.encryption.c <- elemsContainer + packetQueued = true + } +} + func (peer *Peer) SendHandshakeInitiation(isRetry bool) error { if !isRetry { peer.timers.handshakeAttempts.Store(0) @@ -135,15 +202,53 @@ func (peer *Peer) SendHandshakeInitiation(isRetry bool) error { return err } - buf := make([]byte, MessageEncapsulatingTransportSize+MessageInitiationSize) - packet := buf[MessageEncapsulatingTransportSize:] - _ = msg.marshal(packet) + var sendBuffer [][]byte + + for _, ipacket := range peer.device.ipackets { + if ipacket != nil { + buf := make([]byte, ipacket.ObfuscatedLen(0)) + ipacket.Obfuscate(buf, nil) + sendBuffer = append(sendBuffer, buf) + } + } + + jc := peer.device.junk.count + jmin := peer.device.junk.min + jmax := peer.device.junk.max + if jmax < jmin { + // UAPI validates jmin/jmax only individually; a swapped pair + // would panic rand.Int below with a non-positive bound. + jmin, jmax = jmax, jmin + } + + for i := 0; i < jc; i++ { + nBig, _ := rand.Int(rand.Reader, big.NewInt(int64(jmax-jmin+1))) + n := int(nBig.Int64()) + jmin + + buf := make([]byte, n) + rand.Read(buf) + sendBuffer = append(sendBuffer, buf) + } + + var buf [MessageInitiationSize]byte + writer := bytes.NewBuffer(buf[:0]) + binary.Write(writer, binary.LittleEndian, msg) + packet := writer.Bytes() peer.cookieGenerator.AddMacs(packet) peer.timersAnyAuthenticatedPacketTraversal() peer.timersAnyAuthenticatedPacketSent() - err = peer.SendBuffers([][]byte{buf}) + if padding := peer.device.paddings.init; padding > 0 { + buf := make([]byte, padding+len(packet)) + rand.Read(buf[:padding]) + copy(buf[padding:], packet) + packet = buf + } + + sendBuffer = append(sendBuffer, packet) + + err = peer.SendBuffers(sendBuffer) if err != nil { peer.device.log.Errorf("%v - Failed to send handshake initiation: %v", peer, err) } @@ -165,9 +270,11 @@ func (peer *Peer) SendHandshakeResponse() error { return err } - buf := make([]byte, MessageEncapsulatingTransportSize+MessageResponseSize) - packet := buf[MessageEncapsulatingTransportSize:] - _ = response.marshal(packet) + var buf [MessageResponseSize]byte + writer := bytes.NewBuffer(buf[:0]) + + binary.Write(writer, binary.LittleEndian, response) + packet := writer.Bytes() peer.cookieGenerator.AddMacs(packet) err = peer.BeginSymmetricSession() @@ -180,8 +287,15 @@ func (peer *Peer) SendHandshakeResponse() error { peer.timersAnyAuthenticatedPacketTraversal() peer.timersAnyAuthenticatedPacketSent() + if padding := peer.device.paddings.response; padding > 0 { + buf := make([]byte, padding+len(packet)) + rand.Read(buf[:padding]) + copy(buf[padding:], packet) + packet = buf + } + // TODO: allocation could be avoided - err = peer.SendBuffers([][]byte{buf}) + err = peer.SendBuffers([][]byte{packet}) if err != nil { peer.device.log.Errorf("%v - Failed to send handshake response: %v", peer, err) } @@ -192,18 +306,33 @@ func (device *Device) SendHandshakeCookie(initiatingElem *QueueHandshakeElement) device.log.Verbosef("Sending cookie response for denied handshake message for %v", initiatingElem.endpoint.DstToString()) sender := binary.LittleEndian.Uint32(initiatingElem.packet[4:8]) - reply, err := device.cookieChecker.CreateReply(initiatingElem.packet, sender, initiatingElem.endpoint.DstToBytes()) + msgType := device.headers.cookie.Generate() + + reply, err := device.cookieChecker.CreateReply( + initiatingElem.packet, + sender, + initiatingElem.endpoint.DstToBytes(), + msgType, + ) if err != nil { device.log.Errorf("Failed to create cookie reply: %v", err) return err } - buf := make([]byte, MessageEncapsulatingTransportSize+MessageCookieReplySize) - packet := buf[MessageEncapsulatingTransportSize:] - _ = reply.marshal(packet) - // TODO: allocation could be avoided - device.net.bind.Send([][]byte{buf}, initiatingElem.endpoint, MessageEncapsulatingTransportSize) + var buf [MessageCookieReplySize]byte + writer := bytes.NewBuffer(buf[:0]) + binary.Write(writer, binary.LittleEndian, reply) + packet := writer.Bytes() + if padding := device.paddings.cookie; padding > 0 { + buf := make([]byte, padding+len(packet)) + rand.Read(buf[:padding]) + copy(buf[padding:], packet) + packet = buf + } + + // TODO: allocation could be avoided + device.net.bind.Send([][]byte{packet}, initiatingElem.endpoint, 0) return nil } @@ -339,8 +468,51 @@ func (device *Device) RoutineReadFromTUN() { // each), so without this cap a flood is buffered instead of dropped. const maxQueuedInputPackets = 2048 +func (device *Device) inputPacketPeer(destination []byte, packetSlices [][]byte) *Peer { + var src, dst netip.Addr + switch len(destination) { + case net.IPv4len: + dst = netip.AddrFrom4([4]byte(destination)) + var srcBytes [net.IPv4len]byte + if !gatherPacketBytes(packetSlices, IPv4offsetSrc, srcBytes[:]) { + return nil + } + src = netip.AddrFrom4(srcBytes) + case net.IPv6len: + dst = netip.AddrFrom16([16]byte(destination)) + var srcBytes [net.IPv6len]byte + if !gatherPacketBytes(packetSlices, IPv6offsetSrc, srcBytes[:]) { + return nil + } + src = netip.AddrFrom16(srcBytes) + default: + return nil + } + var ipPkt []byte + if len(packetSlices) == 1 { + ipPkt = packetSlices[0] + } + return device.allowedips.LookupFromPacket(src, dst, ipPkt) +} + +func gatherPacketBytes(packetSlices [][]byte, offset int, destination []byte) bool { + for _, packetSlice := range packetSlices { + if offset >= len(packetSlice) { + offset -= len(packetSlice) + continue + } + n := copy(destination, packetSlice[offset:]) + destination = destination[n:] + offset = 0 + if len(destination) == 0 { + return true + } + } + return false +} + func (device *Device) InputPacket(destination []byte, packetSlices [][]byte) { - peer := device.allowedips.Lookup(destination) + peer := device.inputPacketPeer(destination, packetSlices) if peer == nil { return } @@ -351,7 +523,9 @@ func (device *Device) InputPacket(destination []byte, packetSlices [][]byte) { for _, packetSlice := range packetSlices { totalLength += len(packetSlice) } - allocLength := MessageEncapsulatingTransportSize + MessageTransportHeaderSize + totalLength + PaddingMultiple + chacha20poly1305.Overhead + // paddings.transport (AWG s4) is prepended in-buffer by + // RoutineSequentialSender; reserve headroom for the shift. + allocLength := MessageEncapsulatingTransportSize + MessageTransportHeaderSize + totalLength + PaddingMultiple + chacha20poly1305.Overhead + device.paddings.transport if allocLength > MaxMessageSize { return } @@ -385,7 +559,7 @@ func (device *Device) InputPackets(packets []*InputPacketRef) []*InputPacketRef var unmatched []*InputPacketRef elemsByPeer := make(map[*Peer][]*QueueOutboundElementsContainer, len(packets)) for _, packetRef := range packets { - peer := device.allowedips.Lookup(packetRef.Destination) + peer := device.inputPacketPeer(packetRef.Destination, packetRef.PacketSlices) if peer == nil { unmatched = append(unmatched, packetRef) continue @@ -397,7 +571,9 @@ func (device *Device) InputPackets(packets []*InputPacketRef) []*InputPacketRef for _, packetSlice := range packetRef.PacketSlices { totalLength += len(packetSlice) } - allocLength := MessageEncapsulatingTransportSize + MessageTransportHeaderSize + totalLength + PaddingMultiple + chacha20poly1305.Overhead + // paddings.transport (AWG s4) is prepended in-buffer by + // RoutineSequentialSender; reserve headroom for the shift. + allocLength := MessageEncapsulatingTransportSize + MessageTransportHeaderSize + totalLength + PaddingMultiple + chacha20poly1305.Overhead + device.paddings.transport if allocLength > MaxMessageSize { continue } @@ -580,7 +756,9 @@ func (device *Device) RoutineEncryption(id int) { fieldReceiver := header[4:8] fieldNonce := header[8:16] - binary.LittleEndian.PutUint32(fieldType, MessageTransportType) + msgType := device.headers.transport.Generate() + + binary.LittleEndian.PutUint32(fieldType, msgType) binary.LittleEndian.PutUint32(fieldReceiver, elem.keypair.remoteIndex) binary.LittleEndian.PutUint64(fieldNonce, elem.nonce) @@ -597,9 +775,6 @@ func (device *Device) RoutineEncryption(id int) { elem.packet, nil, ) - - // re-slice packet to include encapsulating transport space - elem.packet = elem.buffer[:MessageEncapsulatingTransportSize+len(elem.packet)] } elemsContainer.filling.Done() } @@ -613,7 +788,7 @@ func (peer *Peer) RoutineSequentialSender(maxBatchSize int) { }() device.log.Verbosef("%v - Routine: sequential sender - started", peer) - bufs := make([][]byte, 0, maxBatchSize) + bufs := make([][]byte, 0, max(maxBatchSize, conn.IdealBatchSize)) for elemsContainer := range peer.queue.outbound.c { if elemsContainer == nil { @@ -668,6 +843,19 @@ func (peer *Peer) processOutboundContainer(elemsContainer *QueueOutboundElements if len(elem.packet[MessageEncapsulatingTransportSize:]) != MessageKeepaliveSize { dataSent = true } + // lx:begin awg (SPEC 025 — AmneziaWG transport padding, S4) + // Prepend `transport` random bytes ahead of the transport header. The AWG + // path zeroes MessageEncapsulatingTransportSize (see noise-protocol.go), so + // elem.packet starts at buffer offset 0 and this shift is what creates the + // prefix; the buffer is allocated with PaddingMultiple headroom. + if padding := device.paddings.transport; padding > 0 { + for i := len(elem.packet) - 1; i >= 0; i-- { + elem.buffer[i+padding] = elem.buffer[i] + } + rand.Read(elem.buffer[:padding]) + elem.packet = elem.buffer[:padding+len(elem.packet)] + } + // lx:end awg scratch = append(scratch, elem.packet) } diff --git a/device/timers.go b/device/timers.go index d30f26b..9ec3d18 100644 --- a/device/timers.go +++ b/device/timers.go @@ -99,10 +99,25 @@ 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) + /* lx: SPEC 041 v2 — early self-heal: >=3 unanswered initiations against + * a provably dead session (no live keypair, or last handshake older + * than RejectAfterTime) prove the 5-tuple dead without waiting out the + * full cycle. Rebind now (debounced with the give-up trigger below); + * the retry cycle itself continues untouched. */ + peer.device.maybeEarlyGiveUpRebind(peer) + /* We clear the endpoint address src address, in case this is the cause of trouble. */ peer.markEndpointSrcForClearing() diff --git a/device/transport_padding_test.go b/device/transport_padding_test.go new file mode 100644 index 0000000..c235846 --- /dev/null +++ b/device/transport_padding_test.go @@ -0,0 +1,361 @@ +/* SPDX-License-Identifier: MIT + * + * Regression test for the AWG transport-padding (uapi "s4") out-of-bounds + * crash: RoutineSequentialSender shifts elem.packet right by + * device.paddings.transport bytes inside elem.buffer to prepend a random + * padding prefix, but the injection paths (InputPacket/InputPackets) + * allocated elem.buffer tightly, without headroom for that shift: + * + * panic: runtime error: index out of range [123] with length 76 + * device.(*Peer).RoutineSequentialSender + * + * (payload 28 bytes -> allocLength 76, sealed packet 64, s4=60 -> 63+60=123). + * + * The tests spin up two real Devices wired together through an in-memory + * conn.Bind (Go channels) and an in-memory tun.Device, configure s4=60 on + * both sides, and pass a small IPv4 packet end to end via both outbound + * paths: Device.InputPacket (the crashing path) and the regular tun read + * loop (in-place shift path). + */ + +package device + +import ( + "bytes" + "context" + "encoding/binary" + "encoding/hex" + "fmt" + "net" + "net/netip" + "os" + "sync" + "testing" + "time" + + "github.com/sagernet/wireguard-go/conn" + "github.com/sagernet/wireguard-go/tun" + + "golang.org/x/net/ipv4" +) + +const testTransportPadding = 60 // uapi s4, matches the on-device crash + +// --------------------------------------------------------------------------- +// In-memory conn.Bind over Go channels (minimal bindtest.ChannelBind clone). +// --------------------------------------------------------------------------- + +type chanEndpoint uint16 + +func (e chanEndpoint) ClearSrc() {} +func (e chanEndpoint) SrcToString() string { return "" } +func (e chanEndpoint) DstToString() string { return fmt.Sprintf("127.0.0.1:%d", uint16(e)) } +func (e chanEndpoint) DstToBytes() []byte { return []byte{byte(e), byte(e >> 8)} } +func (e chanEndpoint) DstIP() netip.Addr { return netip.AddrFrom4([4]byte{127, 0, 0, 1}) } +func (e chanEndpoint) SrcIP() netip.Addr { return netip.Addr{} } + +type chanBind struct { + rx, tx chan []byte + source chanEndpoint // "port" this bind listens on + target chanEndpoint // endpoint of the opposite bind + + mu sync.Mutex + closeSignal chan struct{} // recreated on every Open (BindUpdate closes+reopens) +} + +// newChanBindPair returns two Binds whose Send/Receive are cross-wired. +func newChanBindPair() (*chanBind, *chanBind) { + aToB := make(chan []byte, 1024) + bToA := make(chan []byte, 1024) + a := &chanBind{rx: bToA, tx: aToB, source: 1, target: 2} + b := &chanBind{rx: aToB, tx: bToA, source: 2, target: 1} + return a, b +} + +func (b *chanBind) currentCloseSignal() chan struct{} { + b.mu.Lock() + defer b.mu.Unlock() + return b.closeSignal +} + +func (b *chanBind) Open(port uint16) ([]conn.ReceiveFunc, uint16, error) { + b.mu.Lock() + b.closeSignal = make(chan struct{}) + closeSignal := b.closeSignal + b.mu.Unlock() + fn := func(packets [][]byte, sizes []int, eps []conn.Endpoint) (int, error) { + select { + case <-closeSignal: + // Must be net.ErrClosed: RoutineReceiveIncoming treats anything + // else as a transient error and death-spirals before exiting. + return 0, net.ErrClosed + case pkt, ok := <-b.rx: + if !ok { + return 0, net.ErrClosed + } + sizes[0] = copy(packets[0], pkt) + eps[0] = b.target + return 1, nil + } + } + return []conn.ReceiveFunc{fn}, uint16(b.source), nil +} + +func (b *chanBind) Close() error { + b.mu.Lock() + defer b.mu.Unlock() + if b.closeSignal != nil { + select { + case <-b.closeSignal: + default: + close(b.closeSignal) + } + } + return nil +} + +func (b *chanBind) SetMark(mark uint32) error { return nil } + +func (b *chanBind) Send(bufs [][]byte, ep conn.Endpoint, offset int) error { + closeSignal := b.currentCloseSignal() + if closeSignal == nil { + return net.ErrClosed + } + for _, buf := range bufs { + pkt := make([]byte, len(buf)-offset) + copy(pkt, buf[offset:]) + select { + case <-closeSignal: + return net.ErrClosed + case b.tx <- pkt: + } + } + return nil +} + +func (b *chanBind) ParseEndpoint(s string) (conn.Endpoint, error) { return b.target, nil } + +func (b *chanBind) BatchSize() int { return 1 } + +func (b *chanBind) SetReservedForEndpoint(destination netip.AddrPort, reserved [3]byte) {} + +// --------------------------------------------------------------------------- +// In-memory tun.Device over Go channels (minimal tuntest.ChannelTUN clone). +// --------------------------------------------------------------------------- + +type chanTun struct { + toDevice chan []byte // packets the device Reads (outbound plaintext) + fromDevice chan []byte // packets the device Writes (inbound plaintext) + events chan tun.Event + closed chan struct{} + closeOnce sync.Once +} + +func newChanTun() *chanTun { + return &chanTun{ + toDevice: make(chan []byte, 1024), + fromDevice: make(chan []byte, 1024), + events: make(chan tun.Event, 4), + closed: make(chan struct{}), + } +} + +func (t *chanTun) File() *os.File { return nil } + +func (t *chanTun) Read(bufs [][]byte, sizes []int, offset int) (int, error) { + select { + case <-t.closed: + return 0, os.ErrClosed + case pkt, ok := <-t.toDevice: + if !ok { + return 0, os.ErrClosed + } + sizes[0] = copy(bufs[0][offset:], pkt) + return 1, nil + } +} + +func (t *chanTun) Write(bufs [][]byte, offset int) (int, error) { + for _, buf := range bufs { + pkt := make([]byte, len(buf)-offset) + copy(pkt, buf[offset:]) + select { + case <-t.closed: + return 0, os.ErrClosed + case t.fromDevice <- pkt: + } + } + return len(bufs), nil +} + +func (t *chanTun) MTU() (int, error) { return DefaultMTU, nil } +func (t *chanTun) Name() (string, error) { return "chantun", nil } +func (t *chanTun) Events() <-chan tun.Event { return t.events } +func (t *chanTun) BatchSize() int { return 1 } + +func (t *chanTun) Close() error { + t.closeOnce.Do(func() { + close(t.closed) + close(t.events) + }) + return nil +} + +// --------------------------------------------------------------------------- +// Test scaffolding. +// --------------------------------------------------------------------------- + +var ( + testIPA = netip.AddrFrom4([4]byte{10, 0, 0, 1}) + testIPB = netip.AddrFrom4([4]byte{10, 0, 0, 2}) +) + +// buildIPv4Packet builds a minimal, routable IPv4/UDP packet whose header +// fields satisfy the receive-side validation in RoutineSequentialReceiver +// (version, total-length field, allowed source address). +func buildIPv4Packet(src, dst netip.Addr, payloadLen int) []byte { + total := ipv4.HeaderLen + payloadLen + pkt := make([]byte, total) + pkt[0] = 0x45 // version 4, IHL 5 + binary.BigEndian.PutUint16(pkt[IPv4offsetTotalLength:IPv4offsetTotalLength+2], uint16(total)) + pkt[8] = 64 // TTL + pkt[9] = 17 // protocol: UDP + copy(pkt[IPv4offsetSrc:], src.AsSlice()) + copy(pkt[IPv4offsetDst:], dst.AsSlice()) + for i := ipv4.HeaderLen; i < total; i++ { + pkt[i] = byte(i) // deterministic payload + } + return pkt +} + +type paddedPair struct { + devA, devB *Device + tunA, tunB *chanTun +} + +// newPaddedDevicePair builds two Up()'d devices peered with each other over +// the channel bind, both configured with s4 (transport padding) enabled. +func newPaddedDevicePair(t *testing.T) *paddedPair { + 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() + + bindA, bindB := newChanBindPair() + tunA := newChanTun() + tunB := newChanTun() + + devA := NewDevice(context.Background(), tunA, bindA, NewLogger(LogLevelError, "devA: "), 1) + devB := NewDevice(context.Background(), tunB, bindB, NewLogger(LogLevelError, "devB: "), 1) + t.Cleanup(devA.Close) + t.Cleanup(devB.Close) + + cfgA := fmt.Sprintf( + "private_key=%s\ns4=%d\nreplace_peers=true\npublic_key=%s\nendpoint=127.0.0.1:2\nallowed_ip=%s/32\n", + hex.EncodeToString(skA[:]), testTransportPadding, hex.EncodeToString(pkB[:]), testIPB) + cfgB := fmt.Sprintf( + "private_key=%s\ns4=%d\nreplace_peers=true\npublic_key=%s\nendpoint=127.0.0.1:1\nallowed_ip=%s/32\n", + hex.EncodeToString(skB[:]), testTransportPadding, 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 devA.paddings.transport != testTransportPadding { + t.Fatalf("s4 not applied: paddings.transport = %d", devA.paddings.transport) + } + + 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 &paddedPair{devA: devA, devB: devB, tunA: tunA, tunB: tunB} +} + +// awaitPacket waits for want to arrive on the receiving tun, periodically +// re-sending via resend (injection has no delivery guarantee before the +// handshake completes). +func awaitPacket(t *testing.T, from *chanTun, want []byte, resend func()) { + t.Helper() + deadline := time.After(20 * time.Second) + retry := time.NewTicker(1 * time.Second) + defer retry.Stop() + for { + select { + case got := <-from.fromDevice: + if bytes.Equal(got, want) { + return + } + t.Logf("ignoring unexpected packet, len=%d", len(got)) + case <-retry.C: + resend() + case <-deadline: + t.Fatal("timed out waiting for packet on peer tun") + } + } +} + +// --------------------------------------------------------------------------- +// Tests. +// --------------------------------------------------------------------------- + +// TestTransportPaddingInputPacket exercises the exact crash path: an injected +// packet (Device.InputPacket) whose buffer was allocated by payload size. +// With s4=60 and a 28-byte IPv4 packet the pre-fix buffer was 76 bytes and +// the padding shift indexed [123] -> index out of range. +func TestTransportPaddingInputPacket(t *testing.T) { + pair := newPaddedDevicePair(t) + + // 20-byte header + 8-byte payload = 28 bytes, the on-device crash size. + pkt := buildIPv4Packet(testIPA, testIPB, 8) + dst := testIPB.AsSlice() + + send := func() { pair.devA.InputPacket(dst, [][]byte{pkt}) } + send() + awaitPacket(t, pair.tunB, pkt, send) +} + +// TestTransportPaddingInputPackets covers the batched injection path +// (Device.InputPackets), which had the same tight allocation. +func TestTransportPaddingInputPackets(t *testing.T) { + pair := newPaddedDevicePair(t) + + pkt := buildIPv4Packet(testIPA, testIPB, 8) + refs := []*InputPacketRef{{ + Destination: testIPB.AsSlice(), + PacketSlices: [][]byte{pkt[:12], pkt[12:]}, // multi-slice on purpose + }} + + send := func() { + if unmatched := pair.devA.InputPackets(refs); len(unmatched) != 0 { + t.Fatalf("InputPackets returned %d unmatched refs", len(unmatched)) + } + } + send() + awaitPacket(t, pair.tunB, pkt, send) +} + +// TestTransportPaddingTunPath covers the regular outbound path (tun read +// loop), whose MaxMessageSize buffers take the in-place shift branch. +func TestTransportPaddingTunPath(t *testing.T) { + pair := newPaddedDevicePair(t) + + pkt := buildIPv4Packet(testIPA, testIPB, 8) + + send := func() { pair.tunA.toDevice <- pkt } + send() + awaitPacket(t, pair.tunB, pkt, send) +} diff --git a/device/uapi.go b/device/uapi.go index cba371d..4f295c1 100644 --- a/device/uapi.go +++ b/device/uapi.go @@ -97,6 +97,56 @@ func (device *Device) IpcGetOperation(w io.Writer) error { sendf("fwmark=%d", device.net.fwmark) } + if device.junk.count != 0 { + sendf("jc=%d", device.junk.count) + } + + if device.junk.min != 0 { + sendf("jmin=%d", device.junk.min) + } + + if device.junk.max != 0 { + sendf("jmax=%d", device.junk.max) + } + + if device.paddings.init != 0 { + sendf("s1=%d", device.paddings.init) + } + + if device.paddings.response != 0 { + sendf("s2=%d", device.paddings.response) + } + + if device.paddings.cookie != 0 { + sendf("s3=%d", device.paddings.cookie) + } + + if device.paddings.transport != 0 { + sendf("s4=%d", device.paddings.transport) + } + + if device.headers.init != nil { + sendf("h1=%s", device.headers.init.GenSpec()) + } + + if device.headers.response != nil { + sendf("h2=%s", device.headers.response.GenSpec()) + } + + if device.headers.cookie != nil { + sendf("h3=%s", device.headers.cookie.GenSpec()) + } + + if device.headers.transport != nil { + sendf("h4=%s", device.headers.transport.GenSpec()) + } + + for i, ipacket := range device.ipackets { + if ipacket != nil { + sendf("i%d=%s", i+1, ipacket.Spec) + } + } + for _, peer := range device.peers.keyMap { // Serialize peer state. peer.handshake.mutex.RLock() @@ -147,6 +197,7 @@ func (device *Device) IpcSetOperation(r io.Reader) (err error) { } }() + ipcDev := new(ipcSetDevice) peer := new(ipcSetPeer) deviceConfig := true @@ -155,12 +206,20 @@ func (device *Device) IpcSetOperation(r io.Reader) (err error) { line := scanner.Text() if line == "" { // Blank line means terminate operation. + err := ipcDev.mergeWithDevice(device) + if err != nil { + return ipcErrorf(ipc.IpcErrorInvalid, "failed to merge with device: %w", err) + } peer.handlePostConfig() return nil } key, value, ok := strings.Cut(line, "=") if !ok { - return ipcErrorf(ipc.IpcErrorProtocol, "failed to parse line %q", line) + return ipcErrorf( + ipc.IpcErrorProtocol, + "failed to parse line %q", + line, + ) } if key == "public_key" { @@ -186,6 +245,10 @@ func (device *Device) IpcSetOperation(r io.Reader) (err error) { return err } } + err = ipcDev.mergeWithDevice(device) + if err != nil { + return ipcErrorf(ipc.IpcErrorInvalid, "failed to merge with device: %w", err) + } peer.handlePostConfig() if err := scanner.Err(); err != nil { @@ -235,11 +298,155 @@ func (device *Device) handleDeviceLine(key, value string) error { case "replace_peers": if value != "true" { - return ipcErrorf(ipc.IpcErrorInvalid, "failed to set replace_peers, invalid value: %v", value) + return ipcErrorf( + ipc.IpcErrorInvalid, + "failed to set replace_peers, invalid value: %v", + value, + ) } device.log.Verbosef("UAPI: Removing all peers") device.RemoveAllPeers() + case "jc": + jc, err := strconv.Atoi(value) + if err != nil { + return ipcErrorf(ipc.IpcErrorInvalid, "failed to parse jc: %w", err) + } + if jc <= 0 { + return ipcErrorf(ipc.IpcErrorInvalid, "jc must be a positive value") + } + device.log.Verbosef("UAPI: Updating junk count") + device.junk.count = jc + + case "jmin": + jmin, err := strconv.Atoi(value) + if err != nil { + return ipcErrorf(ipc.IpcErrorInvalid, "failed to parse jmin: %w", err) + } + if jmin <= 0 { + return ipcErrorf(ipc.IpcErrorInvalid, "jmin must be a positive value") + } + device.log.Verbosef("UAPI: Updating junk min") + device.junk.min = jmin + + case "jmax": + jmax, err := strconv.Atoi(value) + if err != nil { + return ipcErrorf(ipc.IpcErrorInvalid, "failed to parse jmax: %w", err) + } + if jmax <= 0 { + return ipcErrorf(ipc.IpcErrorInvalid, "jmax must be a positive value") + } + device.log.Verbosef("UAPI: Updating junk max") + device.junk.max = jmax + + case "s1": + padding, err := strconv.Atoi(value) + if err != nil { + return ipcErrorf(ipc.IpcErrorInvalid, "failed to parse s1: %w", err) + } + if padding < 0 { + return ipcErrorf(ipc.IpcErrorInvalid, "s1 must be non-negative") + } + device.log.Verbosef("UAPI: Updating s1 padding") + device.paddings.init = padding + + case "s2": + padding, err := strconv.Atoi(value) + if err != nil { + return ipcErrorf(ipc.IpcErrorInvalid, "failed to parse s2: %w", err) + } + if padding < 0 { + return ipcErrorf(ipc.IpcErrorInvalid, "s2 must be non-negative") + } + device.log.Verbosef("UAPI: Updating s2 padding") + device.paddings.response = padding + + case "s3": + padding, err := strconv.Atoi(value) + if err != nil { + return ipcErrorf(ipc.IpcErrorInvalid, "failed to parse s3: %w", err) + } + if padding < 0 { + return ipcErrorf(ipc.IpcErrorInvalid, "s3 must be non-negative") + } + device.log.Verbosef("UAPI: Updating s3 padding") + device.paddings.cookie = padding + + case "s4": + padding, err := strconv.Atoi(value) + if err != nil { + return ipcErrorf(ipc.IpcErrorInvalid, "failed to parse s4: %w", err) + } + if padding < 0 { + return ipcErrorf(ipc.IpcErrorInvalid, "s4 must be non-negative") + } + device.log.Verbosef("UAPI: Updating s4 padding") + device.paddings.transport = padding + + case "h1": + header, err := newMagicHeader(value) + if err != nil { + return ipcErrorf(ipc.IpcErrorInvalid, "failed to parse H1: %w", err) + } + device.headers.init = header + + case "h2": + header, err := newMagicHeader(value) + if err != nil { + return ipcErrorf(ipc.IpcErrorInvalid, "failed to parse H2: %w", err) + } + device.headers.response = header + + case "h3": + header, err := newMagicHeader(value) + if err != nil { + return ipcErrorf(ipc.IpcErrorInvalid, "failed to parse H3: %w", err) + } + device.headers.cookie = header + + case "h4": + header, err := newMagicHeader(value) + if err != nil { + return ipcErrorf(ipc.IpcErrorInvalid, "failed to parse H4: %w", err) + } + device.headers.transport = header + + case "i1": + chain, err := newObfChain(value) + if err != nil { + return ipcErrorf(ipc.IpcErrorInvalid, "failed to parse I1: %w", err) + } + device.ipackets[0] = chain + + case "i2": + chain, err := newObfChain(value) + if err != nil { + return ipcErrorf(ipc.IpcErrorInvalid, "failed to parse I2: %w", err) + } + device.ipackets[1] = chain + + case "i3": + chain, err := newObfChain(value) + if err != nil { + return ipcErrorf(ipc.IpcErrorInvalid, "failed to parse I3: %w", err) + } + device.ipackets[2] = chain + + case "i4": + chain, err := newObfChain(value) + if err != nil { + return ipcErrorf(ipc.IpcErrorInvalid, "failed to parse I4: %w", err) + } + device.ipackets[3] = chain + + case "i5": + chain, err := newObfChain(value) + if err != nil { + return ipcErrorf(ipc.IpcErrorInvalid, "failed to parse I5: %w", err) + } + device.ipackets[4] = chain + default: return ipcErrorf(ipc.IpcErrorInvalid, "invalid UAPI device key: %v", key) } @@ -271,7 +478,10 @@ func (peer *ipcSetPeer) handlePostConfig() { } } -func (device *Device) handlePublicKeyLine(peer *ipcSetPeer, value string) error { +func (device *Device) handlePublicKeyLine( + peer *ipcSetPeer, + value string, +) error { // Load/create the peer we are configuring. var publicKey NoisePublicKey err := publicKey.FromHex(value) @@ -301,12 +511,19 @@ func (device *Device) handlePublicKeyLine(peer *ipcSetPeer, value string) error return nil } -func (device *Device) handlePeerLine(peer *ipcSetPeer, key, value string) error { +func (device *Device) handlePeerLine( + peer *ipcSetPeer, + key, value string, +) error { switch key { case "update_only": // allow disabling of creation if value != "true" { - return ipcErrorf(ipc.IpcErrorInvalid, "failed to set update only, invalid value: %v", value) + return ipcErrorf( + ipc.IpcErrorInvalid, + "failed to set update only, invalid value: %v", + value, + ) } if peer.created && !peer.dummy { device.RemovePeer(peer.handshake.remoteStatic) @@ -352,7 +569,11 @@ func (device *Device) handlePeerLine(peer *ipcSetPeer, key, value string) error secs, err := strconv.ParseUint(value, 10, 16) if err != nil { - return ipcErrorf(ipc.IpcErrorInvalid, "failed to set persistent keepalive interval: %w", err) + return ipcErrorf( + ipc.IpcErrorInvalid, + "failed to set persistent keepalive interval: %w", + err, + ) } old := peer.persistentKeepaliveInterval.Swap(uint32(secs)) @@ -363,7 +584,11 @@ func (device *Device) handlePeerLine(peer *ipcSetPeer, key, value string) error case "replace_allowed_ips": device.log.Verbosef("%v - UAPI: Removing all allowedips", peer.Peer) if value != "true" { - return ipcErrorf(ipc.IpcErrorInvalid, "failed to replace allowedips, invalid value: %v", value) + return ipcErrorf( + ipc.IpcErrorInvalid, + "failed to replace allowedips, invalid value: %v", + value, + ) } if peer.dummy { return nil @@ -442,7 +667,11 @@ func (device *Device) IpcHandle(socket net.Conn) { return } if nextByte != '\n' { - err = ipcErrorf(ipc.IpcErrorInvalid, "trailing character in UAPI get: %q", nextByte) + err = ipcErrorf( + ipc.IpcErrorInvalid, + "trailing character in UAPI get: %q", + nextByte, + ) break } err = device.IpcGetOperation(buffered.Writer) @@ -466,3 +695,49 @@ func (device *Device) IpcHandle(socket net.Conn) { buffered.Flush() } } + +type ipcSetDevice struct { + headers struct { + init *magicHeader + response *magicHeader + cookie *magicHeader + transport *magicHeader + } +} + +func (d *ipcSetDevice) mergeWithDevice(device *Device) error { + if d.headers.init == nil { + d.headers.init = device.headers.init + } + + if d.headers.response == nil { + d.headers.response = device.headers.response + } + + if d.headers.cookie == nil { + d.headers.cookie = device.headers.cookie + } + + if d.headers.transport == nil { + d.headers.transport = device.headers.transport + } + + headers := []*magicHeader{d.headers.init, d.headers.response, d.headers.cookie, d.headers.transport} + for i := 0; i < len(headers); i++ { + for j := i + 1; j < len(headers); j++ { + left := headers[i] + right := headers[j] + + if left.start <= right.end && right.start <= left.end { + return errors.New("headers must not overlap") + } + } + } + + device.headers.init = d.headers.init + device.headers.response = d.headers.response + device.headers.cookie = d.headers.cookie + device.headers.transport = d.headers.transport + + return nil +}