Merge branch 'aig' into dev

peers/manager.go

@@ -44,7 +44,11 @@ type PeerManager struct {
 	peerMonitor   *monitor.PeerMonitor
 	dnsProxy      *dns.DNSProxy
 	interfaceName string
-	privateKey    wgtypes.Key
+	// localIP is our own address on the site tunnel (as opposed to any exit
+	// node's secondary address that may also be present on the interface).
+	// Routes for site peers are pinned to it on darwin - see AddRoutesWithSource.
+	localIP    string
+	privateKey wgtypes.Key
 	// allowedIPOwners tracks which peer currently "owns" each allowed IP in WireGuard
 	// key is the CIDR string, value is the siteId that has it configured in WG
 	allowedIPOwners map[string]int
@@ -88,6 +92,7 @@ func NewPeerManager(config PeerManagerConfig) *PeerManager {
 		peers:           make(map[int]SiteConfig),
 		dnsProxy:        config.DNSProxy,
 		interfaceName:   config.InterfaceName,
+		localIP:         config.LocalIP,
 		privateKey:      config.PrivateKey,
 		allowedIPOwners: make(map[string]int),
 		allowedIPClaims: make(map[string]map[int]bool),
@@ -127,6 +132,27 @@ func (pm *PeerManager) GetPeerMonitor() *monitor.PeerMonitor {
 	return pm.peerMonitor
 }
 
+// SetExitNode starts (or updates) ICMP connectivity monitoring of the given exit node.
+// tunnelIP is the secondary address assigned to us for this exit node, which the ping
+// probe must be sourced from since the exit node's WireGuard peer entry only accepts
+// traffic from that address.
+func (pm *PeerManager) SetExitNode(serverIP, tunnelIP string) {
+	pm.mu.RLock()
+	defer pm.mu.RUnlock()
+	if pm.peerMonitor != nil {
+		pm.peerMonitor.SetExitNode(serverIP, tunnelIP)
+	}
+}
+
+// ClearExitNode stops ICMP connectivity monitoring of the exit node
+func (pm *PeerManager) ClearExitNode() {
+	pm.mu.RLock()
+	defer pm.mu.RUnlock()
+	if pm.peerMonitor != nil {
+		pm.peerMonitor.ClearExitNode()
+	}
+}
+
 // SetPublicDNS replaces the DNS servers used to resolve WireGuard peer
 // endpoints and hole-punch targets.  The servers must be in "host:port" format
 // (e.g. "8.8.8.8:53").  The change takes effect for all future peer
@@ -195,10 +221,10 @@ func (pm *PeerManager) AddPeer(siteConfig SiteConfig) error {
 		return err
 	}
 
-	if err := network.AddRouteForServerIP(siteConfig.ServerIP, pm.interfaceName); err != nil {
+	if err := network.AddRouteForServerIPWithSource(siteConfig.ServerIP, pm.interfaceName, pm.localIP); err != nil {
 		logger.Error("Failed to add route for server IP: %v", err)
 	}
-	if err := network.AddRoutes(siteConfig.RemoteSubnets, pm.interfaceName); err != nil {
+	if err := network.AddRoutesWithSource(siteConfig.RemoteSubnets, pm.interfaceName, pm.localIP); err != nil {
 		logger.Error("Failed to add routes for remote subnets: %v", err)
 	}
 
@@ -259,7 +285,7 @@ func (pm *PeerManager) RemovePeer(siteId int) error {
 		return err
 	}
 
-	if err := network.RemoveRouteForServerIP(peer.ServerIP, pm.interfaceName); err != nil {
+	if err := network.RemoveRouteForServerIPWithSource(peer.ServerIP, pm.interfaceName, pm.localIP); err != nil {
 		logger.Error("Failed to remove route for server IP: %v", err)
 	}
 
@@ -495,7 +521,7 @@ func (pm *PeerManager) UpdatePeer(siteConfig SiteConfig) error {
 
 	// Add routes for added subnets
 	if len(addedSubnets) > 0 {
-		if err := network.AddRoutes(addedSubnets, pm.interfaceName); err != nil {
+		if err := network.AddRoutesWithSource(addedSubnets, pm.interfaceName, pm.localIP); err != nil {
 			logger.Error("Failed to add routes: %v", err)
 		}
 	}
@@ -696,7 +722,7 @@ func (pm *PeerManager) AddRemoteSubnet(siteId int, cidr string) error {
 	}
 
 	// Add route
-	if err := network.AddRoutes([]string{cidr}, pm.interfaceName); err != nil {
+	if err := network.AddRoutesWithSource([]string{cidr}, pm.interfaceName, pm.localIP); err != nil {
 		return err
 	}
 

peers/monitor/exitnode.go

@@ -0,0 +1,279 @@
+package monitor
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net/netip"
+	"time"
+
+	"github.com/fosrl/newt/logger"
+	"golang.org/x/net/icmp"
+	xipv4 "golang.org/x/net/ipv4"
+	"gvisor.dev/gvisor/pkg/tcpip"
+	gipv4 "gvisor.dev/gvisor/pkg/tcpip/network/ipv4"
+	gstack "gvisor.dev/gvisor/pkg/tcpip/stack"
+	gicmp "gvisor.dev/gvisor/pkg/tcpip/transport/icmp"
+	"gvisor.dev/gvisor/pkg/waiter"
+)
+
+const (
+	exitNodePingInterval    = 3 * time.Second
+	exitNodePingTimeout     = 1 * time.Second
+	exitNodePingMaxAttempts = 3
+)
+
+// SetExitNode starts (or, if the exit node changed, restarts) ICMP
+// connectivity monitoring of the exit node at serverIP. tunnelIP is the
+// secondary address assigned to us for this exit node (ExitNodeConfig.TunnelIP) -
+// the exit node's WireGuard peer entry only accepts traffic sourced from that
+// address, so probes must be sourced from it rather than the site tunnel IP.
+// Both serverIP and tunnelIP must be bare IP addresses (no CIDR suffix).
+func (pm *PeerMonitor) SetExitNode(serverIP, tunnelIP string) {
+	pm.exitNodeMu.Lock()
+	if pm.exitNodeCancel != nil && pm.exitNodeServerIP == serverIP && pm.exitNodeTunnelIP == tunnelIP {
+		pm.exitNodeMu.Unlock()
+		return
+	}
+	prevTunnelIP := pm.exitNodeTunnelIP
+	if pm.exitNodeCancel != nil {
+		pm.exitNodeCancel()
+	}
+	pm.exitNodeServerIP = serverIP
+	pm.exitNodeTunnelIP = tunnelIP
+	ctx, cancel := context.WithCancel(context.Background())
+	pm.exitNodeCancel = cancel
+	pm.exitNodeMu.Unlock()
+
+	if prevTunnelIP != "" && prevTunnelIP != tunnelIP {
+		pm.removeExitNodeAddress(prevTunnelIP)
+	}
+	if tunnelIP != prevTunnelIP {
+		if err := pm.addExitNodeAddress(tunnelIP); err != nil {
+			logger.Error("Failed to register exit node tunnel address %s: %v", tunnelIP, err)
+		}
+	}
+
+	logger.Info("Started exit node connectivity monitor for %s (via %s)", serverIP, tunnelIP)
+	go pm.runExitNodeMonitor(ctx, serverIP, tunnelIP)
+}
+
+// ClearExitNode stops ICMP monitoring of the exit node and clears its status
+// from the API.
+func (pm *PeerMonitor) ClearExitNode() {
+	pm.exitNodeMu.Lock()
+	if pm.exitNodeCancel != nil {
+		pm.exitNodeCancel()
+		pm.exitNodeCancel = nil
+	}
+	tunnelIP := pm.exitNodeTunnelIP
+	pm.exitNodeServerIP = ""
+	pm.exitNodeTunnelIP = ""
+	pm.exitNodeMu.Unlock()
+
+	if tunnelIP != "" {
+		pm.removeExitNodeAddress(tunnelIP)
+	}
+
+	if pm.apiServer != nil {
+		pm.apiServer.ClearExitNodeStatus()
+	}
+
+	logger.Info("Stopped exit node connectivity monitor")
+}
+
+// addExitNodeAddress registers tunnelIP as a protocol address on the peer
+// monitor's netstack NIC and adds a MiddleDevice rule so ICMP replies destined
+// to it are intercepted and redirected into the netstack instead of being
+// delivered to the host TUN device.
+func (pm *PeerMonitor) addExitNodeAddress(tunnelIP string) error {
+	pm.mutex.Lock()
+	st := pm.stack
+	pm.mutex.Unlock()
+
+	if st == nil {
+		return fmt.Errorf("netstack not initialized")
+	}
+
+	addr, err := netip.ParseAddr(tunnelIP)
+	if err != nil {
+		return fmt.Errorf("invalid tunnel IP: %w", err)
+	}
+
+	protoAddr := tcpip.ProtocolAddress{
+		Protocol:          gipv4.ProtocolNumber,
+		AddressWithPrefix: tcpip.AddrFrom4(addr.As4()).WithPrefix(),
+	}
+	if tcpipErr := st.AddProtocolAddress(1, protoAddr, gstack.AddressProperties{}); tcpipErr != nil {
+		return fmt.Errorf("failed to add protocol address: %s", tcpipErr)
+	}
+
+	pm.middleDev.AddRule(addr, pm.handlePacket)
+	return nil
+}
+
+// removeExitNodeAddress undoes addExitNodeAddress.
+func (pm *PeerMonitor) removeExitNodeAddress(tunnelIP string) {
+	addr, err := netip.ParseAddr(tunnelIP)
+	if err != nil {
+		return
+	}
+
+	pm.middleDev.RemoveRule(addr)
+
+	pm.mutex.Lock()
+	st := pm.stack
+	pm.mutex.Unlock()
+
+	if st != nil {
+		st.RemoveAddress(1, tcpip.AddrFrom4(addr.As4()))
+	}
+}
+
+// runExitNodeMonitor periodically pings the exit node and reports its status
+// to the API server until ctx is cancelled.
+func (pm *PeerMonitor) runExitNodeMonitor(ctx context.Context, serverIP, tunnelIP string) {
+	check := func() {
+		var (
+			connected bool
+			rtt       time.Duration
+		)
+		for attempt := 0; attempt < exitNodePingMaxAttempts; attempt++ {
+			if d, err := pm.pingExitNode(serverIP, tunnelIP, exitNodePingTimeout); err == nil {
+				connected = true
+				rtt = d
+				break
+			} else {
+				logger.Debug("Exit node ping attempt %d/%d to %s failed: %v", attempt+1, exitNodePingMaxAttempts, serverIP, err)
+			}
+			select {
+			case <-ctx.Done():
+				return
+			default:
+			}
+		}
+		if pm.apiServer != nil {
+			pm.apiServer.SetExitNodeStatus(connected, rtt, serverIP)
+		}
+	}
+
+	check()
+
+	ticker := time.NewTicker(exitNodePingInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			check()
+		}
+	}
+}
+
+// pingExitNode sends a single ICMP echo request from localTunnelIP to dst and
+// waits up to timeout for the matching reply. The request is built and read
+// directly on the peer monitor's gvisor netstack, so it's injected into (and
+// intercepted from) the WireGuard device via MiddleDevice - it never touches
+// the host's real network stack, matching how the UDP peer tests above work.
+func (pm *PeerMonitor) pingExitNode(dst, localTunnelIP string, timeout time.Duration) (time.Duration, error) {
+	pm.mutex.Lock()
+	st := pm.stack
+	pm.mutex.Unlock()
+
+	if st == nil {
+		return 0, fmt.Errorf("netstack not initialized")
+	}
+
+	dstAddr, err := netip.ParseAddr(dst)
+	if err != nil {
+		return 0, fmt.Errorf("invalid destination address: %w", err)
+	}
+	localAddr, err := netip.ParseAddr(localTunnelIP)
+	if err != nil {
+		return 0, fmt.Errorf("invalid local address: %w", err)
+	}
+
+	var wq waiter.Queue
+	ep, tcpipErr := st.NewEndpoint(gicmp.ProtocolNumber4, gipv4.ProtocolNumber, &wq)
+	if tcpipErr != nil {
+		return 0, fmt.Errorf("failed to create ICMP endpoint: %s", tcpipErr)
+	}
+	defer ep.Close()
+
+	if tcpipErr := ep.Bind(tcpip.FullAddress{NIC: 1, Addr: tcpip.AddrFromSlice(localAddr.AsSlice())}); tcpipErr != nil {
+		return 0, fmt.Errorf("failed to bind ICMP endpoint: %s", tcpipErr)
+	}
+
+	// gvisor's ICMP endpoint overwrites whatever Identifier we put in the outgoing
+	// echo with its own bound "port" (assigned above by Bind), and demuxes incoming
+	// Echo Replies by that same value - so we must use it, not one we generate
+	// ourselves, both for the outgoing message and to register with handlePacket's
+	// filter below.
+	laddr, tcpipErr := ep.GetLocalAddress()
+	if tcpipErr != nil {
+		return 0, fmt.Errorf("failed to get local ICMP endpoint address: %s", tcpipErr)
+	}
+	echoID := int(laddr.Port)
+
+	pm.portsLock.Lock()
+	pm.activeICMPIdents[laddr.Port] = true
+	pm.portsLock.Unlock()
+	defer func() {
+		pm.portsLock.Lock()
+		delete(pm.activeICMPIdents, laddr.Port)
+		pm.portsLock.Unlock()
+	}()
+
+	if tcpipErr := ep.Connect(tcpip.FullAddress{NIC: 1, Addr: tcpip.AddrFromSlice(dstAddr.AsSlice())}); tcpipErr != nil {
+		return 0, fmt.Errorf("failed to connect ICMP endpoint: %s", tcpipErr)
+	}
+
+	requestPing := icmp.Echo{
+		ID:   echoID,
+		Seq:  1,
+		Data: []byte("olmping"),
+	}
+	icmpBytes, err := (&icmp.Message{Type: xipv4.ICMPTypeEcho, Code: 0, Body: &requestPing}).Marshal(nil)
+	if err != nil {
+		return 0, fmt.Errorf("failed to marshal ICMP message: %w", err)
+	}
+
+	waitEntry, notifyCh := waiter.NewChannelEntry(waiter.EventIn)
+	wq.EventRegister(&waitEntry)
+	defer wq.EventUnregister(&waitEntry)
+
+	start := time.Now()
+
+	if _, tcpipErr := ep.Write(bytes.NewReader(icmpBytes), tcpip.WriteOptions{}); tcpipErr != nil {
+		return 0, fmt.Errorf("failed to write ICMP echo request: %s", tcpipErr)
+	}
+
+	deadline := time.NewTimer(timeout)
+	defer deadline.Stop()
+
+	readBuf := make([]byte, 1500)
+	for {
+		select {
+		case <-deadline.C:
+			return 0, fmt.Errorf("ping to %s timed out", dst)
+		case <-notifyCh:
+			w := tcpip.SliceWriter(readBuf)
+			res, tcpipErr := ep.Read(&w, tcpip.ReadOptions{})
+			if tcpipErr != nil {
+				continue
+			}
+
+			reply, err := icmp.ParseMessage(1, readBuf[:res.Count])
+			if err != nil {
+				continue
+			}
+			replyEcho, ok := reply.Body.(*icmp.Echo)
+			if !ok || replyEcho.ID != echoID || replyEcho.Seq != requestPing.Seq {
+				continue
+			}
+
+			return time.Since(start), nil
+		}
+	}
+}

peers/monitor/monitor.go

@@ -3,6 +3,7 @@ package monitor
 import (
 	"context"
 	"crypto/rand"
+	"encoding/binary"
 	"encoding/hex"
 	"fmt"
 	"net"
@@ -25,6 +26,7 @@ import (
 	"gvisor.dev/gvisor/pkg/tcpip/network/ipv4"
 	"gvisor.dev/gvisor/pkg/tcpip/network/ipv6"
 	"gvisor.dev/gvisor/pkg/tcpip/stack"
+	"gvisor.dev/gvisor/pkg/tcpip/transport/icmp"
 	"gvisor.dev/gvisor/pkg/tcpip/transport/udp"
 )
 
@@ -105,6 +107,20 @@ type PeerMonitor struct {
 	wgConnectionStatus   map[int]bool          // siteID -> WG connected status
 	wgConnectionRTT      map[int]time.Duration // siteID -> last known RTT
 	statusChangeCallback func(siteId int)      // called when any peer's connection status changes
+
+	// Exit node ICMP monitoring fields. The exit node is a single peer (not a
+	// site), pinged over the same gvisor netstack used for the peer UDP tests
+	// above, so the probe never touches the host's real network stack - it's
+	// injected directly into the WireGuard device via MiddleDevice.
+	exitNodeMu       sync.Mutex
+	exitNodeServerIP string
+	exitNodeTunnelIP string
+	exitNodeCancel   context.CancelFunc
+
+	// activeICMPIdents tracks the ICMP identifiers of our own in-flight exit-node
+	// ping probes (guarded by portsLock, alongside activePorts), so handlePacket
+	// only intercepts Echo Replies that are actually ours.
+	activeICMPIdents map[uint16]bool
 }
 
 // NewPeerMonitor creates a new peer monitor with the given callback
@@ -125,6 +141,7 @@ func NewPeerMonitor(wsClient *websocket.Client, middleDev *middleDevice.MiddleDe
 		localIP:              localIP,
 		publicDNS:            publicDNS,
 		activePorts:          make(map[uint16]bool),
+		activeICMPIdents:     make(map[uint16]bool),
 		nsCtx:                ctx,
 		nsCancel:             cancel,
 		sharedBind:           sharedBind,
@@ -1152,6 +1169,14 @@ func (pm *PeerMonitor) Close() {
 	// Stop holepunch monitor first (outside of mutex to avoid deadlock)
 	pm.stopHolepunchMonitor()
 
+	// Stop exit node ICMP monitor, if running
+	pm.exitNodeMu.Lock()
+	if pm.exitNodeCancel != nil {
+		pm.exitNodeCancel()
+		pm.exitNodeCancel = nil
+	}
+	pm.exitNodeMu.Unlock()
+
 	// Stop all pending relay senders
 	pm.relaySendMu.Lock()
 	for chainId, stop := range pm.relaySends {
@@ -1288,7 +1313,7 @@ func (pm *PeerMonitor) initNetstack() error {
 	// Create gvisor netstack
 	stackOpts := stack.Options{
 		NetworkProtocols:   []stack.NetworkProtocolFactory{ipv4.NewProtocol, ipv6.NewProtocol},
-		TransportProtocols: []stack.TransportProtocolFactory{udp.NewProtocol},
+		TransportProtocols: []stack.TransportProtocolFactory{udp.NewProtocol, icmp.NewProtocol4, icmp.NewProtocol6},
 		HandleLocal:        true,
 	}
 
@@ -1329,26 +1354,64 @@ func (pm *PeerMonitor) initNetstack() error {
 	return nil
 }
 
+// icmpv4EchoReplyIdent returns the ICMP identifier of packet if it is an IPv4
+// ICMP Echo Reply (type 0), so it can be matched against our own in-flight
+// exit-node ping probes before being pulled off the host's real traffic path.
+func icmpv4EchoReplyIdent(packet []byte) (uint16, bool) {
+	if len(packet) < 20 || packet[0]>>4 != 4 {
+		return 0, false
+	}
+	ihl := int(packet[0]&0x0f) * 4
+	if ihl < 20 || len(packet) < ihl+8 {
+		return 0, false
+	}
+	const icmpEchoReply = 0
+	if packet[ihl] != icmpEchoReply {
+		return 0, false
+	}
+	return binary.BigEndian.Uint16(packet[ihl+4 : ihl+6]), true
+}
+
 // handlePacket is called by MiddleDevice when a packet arrives for our IP
 func (pm *PeerMonitor) handlePacket(packet []byte) bool {
-	// Check if it's UDP
 	proto, ok := util.GetProtocol(packet)
-	if !ok || proto != 17 { // UDP
-		return false
-	}
-
-	// Check destination port
-	port, ok := util.GetDestPort(packet)
 	if !ok {
 		return false
 	}
 
-	// Check if we are listening on this port
-	pm.portsLock.RLock()
-	active := pm.activePorts[uint16(port)]
-	pm.portsLock.RUnlock()
+	switch proto {
+	case 1: // ICMPv4 - only intercept Echo Replies matching one of our own active
+		// exit-node ping probes, identified by the ICMP identifier field. Anything
+		// else (including real ICMP traffic to/from the host, e.g. `ping`) must be
+		// left alone so it reaches the host TUN normally.
+		ident, ok := icmpv4EchoReplyIdent(packet)
+		if !ok {
+			return false
+		}
 
-	if !active {
+		pm.portsLock.RLock()
+		active := pm.activeICMPIdents[ident]
+		pm.portsLock.RUnlock()
+
+		if !active {
+			return false
+		}
+	case 17: // UDP
+		// Check destination port
+		port, ok := util.GetDestPort(packet)
+		if !ok {
+			return false
+		}
+
+		// Check if we are listening on this port
+		pm.portsLock.RLock()
+		active := pm.activePorts[uint16(port)]
+		pm.portsLock.RUnlock()
+
+		if !active {
+			return false
+		}
+	default:
 		return false
 	}