From ec3dc8993f5e0a49a8b9349fa19cec018dad0fff Mon Sep 17 00:00:00 2001 From: Owen Date: Thu, 22 Oct 2026 15:01:17 -0400 Subject: [PATCH 1/2] Add keepalive messages --- holepunch/holepunch.go | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/holepunch/holepunch.go b/holepunch/holepunch.go index 99828d7..9432e92 100644 --- a/holepunch/holepunch.go +++ b/holepunch/holepunch.go @@ -56,6 +56,19 @@ type Manager struct { const defaultSendHolepunchIntervalMax = 60 * time.Second const defaultSendHolepunchIntervalMin = 1 * time.Second +// natKeepaliveInterval is how often a keepalive is sent to each exit node +// between hole punches once the hole punch interval has backed off past it. +// It is under the 30s timeout many NATs (e.g. Linux conntrack) give UDP flows +// that never see a reply, so the NAT mapping - and with it the endpoint the +// server recorded from the last hole punch - stays the same between punches. +const natKeepaliveInterval = 20 * time.Second + +// natKeepalivePacket only exists to keep the NAT mapping alive, so it is much +// cheaper for the exit node than a hole punch. Gerbil drops it on sight; older +// gerbils fail to parse the ephemeral key and drop it silently before doing +// any crypto. +var natKeepalivePacket = []byte(`{"ephemeralPublicKey":"keepalive"}`) + // NewManager creates a new hole punch manager func NewManager(sharedBind *bind.SharedBind, ID string, clientType string, publicKey string, publicDNS []string) *Manager { return &Manager{ @@ -514,6 +527,9 @@ func (m *Manager) runMultipleExitNodes() { ticker := time.NewTicker(m.sendHolepunchInterval) defer ticker.Stop() + keepaliveTicker := time.NewTicker(natKeepaliveInterval) + defer keepaliveTicker.Stop() + for { select { case <-m.stopChan: @@ -539,6 +555,22 @@ func (m *Manager) runMultipleExitNodes() { logger.Debug("Failed to send hole punch to %s: %v", node.endpointName, err) } } + case <-keepaliveTicker.C: + // Only needed once hole punches are further apart than the + // keepalive interval. Custom longer intervals (olm low power mode) + // are there to save battery, so let the mapping lapse then. + m.mu.Lock() + needKeepalive := m.sendHolepunchInterval > natKeepaliveInterval && + m.sendHolepunchIntervalMax <= m.defaultIntervalMax + m.mu.Unlock() + if !needKeepalive { + continue + } + for _, node := range resolvedNodes { + if _, err := m.sharedBind.WriteToUDP(natKeepalivePacket, node.remoteAddr); err != nil { + logger.Debug("Failed to send NAT keepalive to %s: %v", node.endpointName, err) + } + } case <-ticker.C: // Send hole punch to all exit nodes (if any are available) if len(resolvedNodes) > 0 { From 7f06977f6b54403f2382a456964497ac4161ea1e Mon Sep 17 00:00:00 2001 From: Owen Date: Tue, 27 Oct 2026 15:31:41 -0400 Subject: [PATCH 2/2] Fix #471 --- go.mod | 2 ++ netstack2/handlers.go | 10 +++---- netstack2/proxy.go | 10 ++++--- netstack2/tun.go | 52 +++++++++++++++++++++++++--------- netstack2/tun_close_test.go | 56 +++++++++++++++++++++++++++++++++++++ 5 files changed, 108 insertions(+), 22 deletions(-) create mode 100644 netstack2/tun_close_test.go diff --git a/go.mod b/go.mod index 7452462..4be3531 100644 --- a/go.mod +++ b/go.mod @@ -2,6 +2,8 @@ module github.com/fosrl/newt go 1.26.0 +toolchain go1.26.10 + require ( github.com/coder/websocket v1.8.15 github.com/creack/pty v1.1.24 diff --git a/netstack2/handlers.go b/netstack2/handlers.go index e8035ab..ace8b76 100644 --- a/netstack2/handlers.go +++ b/netstack2/handlers.go @@ -414,12 +414,12 @@ func pipeUDP(origin, remote net.PacketConn, serverAddr, clientAddr net.Addr, tim func unidirectionalPacketStream(dst, src net.PacketConn, to net.Addr, dir string, wg *sync.WaitGroup, timeout time.Duration) { defer wg.Done() - logger.Info("UDP %s: Starting packet stream (to=%v)", dir, to) + logger.Debug("UDP %s: Starting packet stream (to=%v)", dir, to) err := copyPacketData(dst, src, to, timeout) if err != nil { - logger.Info("UDP %s: Stream ended with error: %v", dir, err) + logger.Debug("UDP %s: Stream ended with error: %v", dir, err) } else { - logger.Info("UDP %s: Stream ended (timeout)", dir) + logger.Debug("UDP %s: Stream ended (timeout)", dir) } } @@ -438,7 +438,7 @@ func copyPacketData(dst, src net.PacketConn, to net.Addr, timeout time.Duration) return err } - logger.Info("UDP copyPacketData: Read %d bytes from %v", n, srcAddr) + logger.Debug("UDP copyPacketData: Read %d bytes from %v", n, srcAddr) // Determine write destination writeAddr := to @@ -452,7 +452,7 @@ func copyPacketData(dst, src net.PacketConn, to net.Addr, timeout time.Duration) logger.Info("UDP copyPacketData: Write error to %v: %v", writeAddr, err) return err } - logger.Info("UDP copyPacketData: Wrote %d bytes to %v", written, writeAddr) + logger.Debug("UDP copyPacketData: Wrote %d bytes to %v", written, writeAddr) dst.SetReadDeadline(time.Now().Add(timeout)) } diff --git a/netstack2/proxy.go b/netstack2/proxy.go index 963eb98..888688b 100644 --- a/netstack2/proxy.go +++ b/netstack2/proxy.go @@ -674,6 +674,7 @@ func (p *ProxyHandler) HandleIncomingPacket(packet []byte) bool { Payload: buffer.MakeWithData(packet), }) p.proxyEp.InjectInbound(header.IPv4ProtocolNumber, pkb) + pkb.DecRef() return true } @@ -733,6 +734,8 @@ func (p *ProxyHandler) HandleIncomingPacket(packet []byte) bool { Payload: buffer.MakeWithData(packet), }) p.proxyEp.InjectInbound(header.IPv4ProtocolNumber, pkb) + // The stack clones anything it retains; release our reference. + pkb.DecRef() logger.Debug("HandleIncomingPacket: Injected packet into proxy stack (proto=%d)", protocol) return true } @@ -980,10 +983,9 @@ func (p *ProxyHandler) Close() error { p.httpHandler.Close() } - // Close ICMP replies channel - if p.icmpReplies != nil { - close(p.icmpReplies) - } + // icmpReplies is intentionally left open: proxyPing goroutines may still + // call QueueICMPReply after Close, and sending on a closed channel panics. + // It is garbage collected with the handler. if p.proxyStack != nil { p.proxyStack.RemoveNIC(1) diff --git a/netstack2/tun.go b/netstack2/tun.go index 40274b1..0b43c6e 100644 --- a/netstack2/tun.go +++ b/netstack2/tun.go @@ -14,6 +14,7 @@ import ( "regexp" "strconv" "strings" + "sync" "syscall" "time" @@ -40,9 +41,14 @@ type netTun struct { events chan tun.Event notifyHandle *channel.NotificationHandle incomingPacket chan *buffer.View - mtu int - dnsServers []netip.Addr - hasV4, hasV6 bool + // closed is closed by Close. incomingPacket itself is never closed: + // WriteNotify sends on it from gVisor goroutines that can still be + // running during and after Close, and a send on a closed channel panics. + closed chan struct{} + closeOnce sync.Once + mtu int + dnsServers []netip.Addr + hasV4, hasV6 bool // TODO: LETS NOT KEEP THIS ON THE TUN AND MOVE IT BUT WE CAN KEEP IT FOR NOW proxyHandler *ProxyHandler // Handles promiscuous mode packet processing } @@ -77,6 +83,7 @@ func CreateNetTUNWithOptions(localAddresses, dnsServers []netip.Addr, mtu int, o stack: stack.New(stackOpts), events: make(chan tun.Event, 10), incomingPacket: make(chan *buffer.View), + closed: make(chan struct{}), dnsServers: dnsServers, mtu: mtu, } @@ -166,8 +173,10 @@ func (tun *netTun) Events() <-chan tun.Event { } func (tun *netTun) Read(buf [][]byte, sizes []int, offset int) (int, error) { - view, ok := <-tun.incomingPacket - if !ok { + var view *buffer.View + select { + case view = <-tun.incomingPacket: + case <-tun.closed: return 0, os.ErrClosed } @@ -201,8 +210,12 @@ func (tun *netTun) Write(buf [][]byte, offset int) (int, error) { case 6: tun.ep.InjectInbound(header.IPv6ProtocolNumber, pkb) default: + pkb.DecRef() return 0, syscall.EAFNOSUPPORT } + // The stack clones anything it retains, so release our reference so + // the buffer goes back to gVisor's pool instead of leaking per packet. + pkb.DecRef() } return len(buf), nil } @@ -213,7 +226,7 @@ func (tun *netTun) WriteNotify() { if pkt != nil { view := pkt.ToView() pkt.DecRef() - tun.incomingPacket <- view + tun.sendIncoming(view) return } @@ -222,13 +235,32 @@ func (tun *netTun) WriteNotify() { if tun.proxyHandler != nil { view := tun.proxyHandler.ReadOutgoingPacket() if view != nil { - tun.incomingPacket <- view + tun.sendIncoming(view) return } } } +// sendIncoming hands a packet to Read, or drops it once the device is closed +// so gVisor goroutines never block forever on a reader that has gone away. +func (tun *netTun) sendIncoming(view *buffer.View) { + select { + case tun.incomingPacket <- view: + case <-tun.closed: + view.Release() + } +} + func (tun *netTun) Close() error { + tun.closeOnce.Do(tun.close) + return nil +} + +func (tun *netTun) close() { + // Signal first so WriteNotify calls made while tearing down the stack + // below return instead of blocking on Read. + close(tun.closed) + tun.stack.RemoveNIC(1) tun.stack.Close() @@ -243,12 +275,6 @@ func (tun *netTun) Close() error { if tun.events != nil { close(tun.events) } - - if tun.incomingPacket != nil { - close(tun.incomingPacket) - } - - return nil } func (tun *netTun) MTU() (int, error) { diff --git a/netstack2/tun_close_test.go b/netstack2/tun_close_test.go new file mode 100644 index 0000000..cb454e8 --- /dev/null +++ b/netstack2/tun_close_test.go @@ -0,0 +1,56 @@ +package netstack2 + +import ( + "errors" + "net/netip" + "os" + "testing" + "time" + + "gvisor.dev/gvisor/pkg/buffer" +) + +// After Close, packets the stack emits must be dropped rather than blocking +// the gVisor goroutine or panicking with "send on closed channel", and Read +// must report os.ErrClosed. +func TestNetTunCloseUnblocksSendersAndReaders(t *testing.T) { + dev, _, err := CreateNetTUN([]netip.Addr{netip.MustParseAddr("100.90.0.1")}, nil, 1280) + if err != nil { + t.Fatalf("CreateNetTUN: %v", err) + } + tn := dev.(*netTun) + + readErr := make(chan error, 1) + go func() { + _, err := tn.Read([][]byte{make([]byte, 1500)}, []int{0}, 0) + readErr <- err + }() + + if err := tn.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + if err := tn.Close(); err != nil { + t.Fatalf("second Close: %v", err) + } + + select { + case err := <-readErr: + if !errors.Is(err, os.ErrClosed) { + t.Fatalf("Read after Close: got %v, want os.ErrClosed", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Read did not return after Close") + } + + sent := make(chan struct{}) + go func() { + tn.sendIncoming(buffer.NewViewWithData([]byte{0x45})) + tn.proxyHandler.QueueICMPReply([]byte{0x45}) + close(sent) + }() + select { + case <-sent: + case <-time.After(5 * time.Second): + t.Fatal("sending a packet after Close blocked") + } +}