Merge pull request #473 from fosrl/fix-471

Fix #471

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

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 {

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))
 	}

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)

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) {

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")
+	}
+}