From 202606917c1f02a1d3aa7279213001c03c80dee1 Mon Sep 17 00:00:00 2001 From: Owen Date: Wed, 5 Aug 2026 10:33:43 -0400 Subject: [PATCH] icmp monitor the exit node for connectivity and provide in api status --- api/api.go | 46 +++++++-- go.mod | 2 +- olm/exitnode.go | 8 ++ peers/manager.go | 18 ++++ peers/monitor/exitnode.go | 196 ++++++++++++++++++++++++++++++++++++++ peers/monitor/monitor.go | 49 +++++++--- 6 files changed, 298 insertions(+), 21 deletions(-) create mode 100644 peers/monitor/exitnode.go diff --git a/api/api.go b/api/api.go index b8f4e56..b2811fd 100644 --- a/api/api.go +++ b/api/api.go @@ -62,6 +62,15 @@ type OlmError struct { Message string `json:"message"` } +// ExitNodeStatus represents the connectivity status of the client's own exit +// node connection (used for site resources hosted on the exit node). +type ExitNodeStatus struct { + Connected bool `json:"connected"` + RTT time.Duration `json:"rtt"` + LastSeen time.Time `json:"lastSeen"` + Endpoint string `json:"endpoint,omitempty"` +} + // StatusResponse is returned by the status endpoint type StatusResponse struct { Connected bool `json:"connected"` @@ -73,6 +82,7 @@ type StatusResponse struct { OrgID string `json:"orgId,omitempty"` PeerStatuses map[int]*PeerStatus `json:"peers,omitempty"` NetworkSettings network.NetworkSettings `json:"networkSettings,omitempty"` + ExitNodeStatus *ExitNodeStatus `json:"exitNode,omitempty"` } type MetadataChangeRequest struct { @@ -103,13 +113,14 @@ type API struct { onPowerMode func(PowerModeRequest) error onJITConnect func(JITConnectionRequest) error - statusMu sync.RWMutex - peerStatuses map[int]*PeerStatus - connectedAt time.Time - isConnected bool - isRegistered bool - isTerminated bool - olmError *OlmError + statusMu sync.RWMutex + peerStatuses map[int]*PeerStatus + exitNodeStatus *ExitNodeStatus + connectedAt time.Time + isConnected bool + isRegistered bool + isTerminated bool + olmError *OlmError version string agent string @@ -409,6 +420,25 @@ func (s *API) UpdatePeerHolepunchStatus(siteID int, holepunchConnected bool) { status.HolepunchConnected = holepunchConnected } +// SetExitNodeStatus sets the connectivity status of the client's own exit node connection +func (s *API) SetExitNodeStatus(connected bool, rtt time.Duration, endpoint string) { + s.statusMu.Lock() + defer s.statusMu.Unlock() + s.exitNodeStatus = &ExitNodeStatus{ + Connected: connected, + RTT: rtt, + LastSeen: time.Now(), + Endpoint: endpoint, + } +} + +// ClearExitNodeStatus removes the exit node status, e.g. when disconnecting from it +func (s *API) ClearExitNodeStatus() { + s.statusMu.Lock() + defer s.statusMu.Unlock() + s.exitNodeStatus = nil +} + // handleConnect handles the /connect endpoint func (s *API) handleConnect(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { @@ -473,6 +503,7 @@ func (s *API) handleStatus(w http.ResponseWriter, r *http.Request) { OrgID: s.orgID, PeerStatuses: s.peerStatuses, NetworkSettings: network.GetSettings(), + ExitNodeStatus: s.exitNodeStatus, } s.statusMu.RUnlock() @@ -640,6 +671,7 @@ func (s *API) GetStatus() StatusResponse { OrgID: s.orgID, PeerStatuses: s.peerStatuses, NetworkSettings: network.GetSettings(), + ExitNodeStatus: s.exitNodeStatus, } } diff --git a/go.mod b/go.mod index 7ea9c61..83b175a 100644 --- a/go.mod +++ b/go.mod @@ -8,6 +8,7 @@ require ( github.com/godbus/dbus/v5 v5.2.2 github.com/gorilla/websocket v1.5.3 github.com/miekg/dns v1.1.70 + golang.org/x/net v0.56.0 golang.org/x/sys v0.46.0 golang.zx2c4.com/wireguard v0.0.0-20250521234502-f333402bd9cb golang.zx2c4.com/wireguard/wgctrl v0.0.0-20241231184526-a9ab2273dd10 @@ -23,7 +24,6 @@ require ( golang.org/x/crypto v0.53.0 // indirect golang.org/x/exp v0.0.0-20251113190631-e25ba8c21ef6 // indirect golang.org/x/mod v0.34.0 // indirect - golang.org/x/net v0.56.0 // indirect golang.org/x/sync v0.20.0 // indirect golang.org/x/time v0.12.0 // indirect golang.org/x/tools v0.43.0 // indirect diff --git a/olm/exitnode.go b/olm/exitnode.go index c1ca808..d0e3aeb 100644 --- a/olm/exitnode.go +++ b/olm/exitnode.go @@ -118,6 +118,10 @@ persistent_keepalive_interval=%d`, util.FixKey(cfg.PublicKey), allowedIP, resolv } } + if pm := o.getPeerManager(); pm != nil { + pm.SetExitNode(strings.Split(cfg.ServerIP, "/")[0]) + } + logger.Info("Connected to exit node at %s", resolvedEndpoint) return nil } @@ -139,6 +143,10 @@ func (o *Olm) removeExitNodePeerLocked() error { cfg := o.exitNode o.exitNode = nil + if pm := o.getPeerManager(); pm != nil { + pm.ClearExitNode() + } + if o.dnsProxy != nil { serverIP := net.ParseIP(cfg.ServerIP) if serverIP != nil { diff --git a/peers/manager.go b/peers/manager.go index 76f1fd8..6a84a5e 100644 --- a/peers/manager.go +++ b/peers/manager.go @@ -127,6 +127,24 @@ func (pm *PeerManager) GetPeerMonitor() *monitor.PeerMonitor { return pm.peerMonitor } +// SetExitNode starts (or updates) ICMP connectivity monitoring of the given exit node +func (pm *PeerManager) SetExitNode(serverIP string) { + pm.mu.RLock() + defer pm.mu.RUnlock() + if pm.peerMonitor != nil { + pm.peerMonitor.SetExitNode(serverIP) + } +} + +// 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 diff --git a/peers/monitor/exitnode.go b/peers/monitor/exitnode.go new file mode 100644 index 0000000..20ae8c3 --- /dev/null +++ b/peers/monitor/exitnode.go @@ -0,0 +1,196 @@ +package monitor + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/binary" + "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" + 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 server address changed, restarts) ICMP +// connectivity monitoring of the exit node at serverIP. serverIP must be a +// bare IP address (no CIDR suffix). +func (pm *PeerMonitor) SetExitNode(serverIP string) { + pm.exitNodeMu.Lock() + if pm.exitNodeCancel != nil && pm.exitNodeServerIP == serverIP { + pm.exitNodeMu.Unlock() + return + } + if pm.exitNodeCancel != nil { + pm.exitNodeCancel() + } + pm.exitNodeServerIP = serverIP + ctx, cancel := context.WithCancel(context.Background()) + pm.exitNodeCancel = cancel + pm.exitNodeMu.Unlock() + + logger.Info("Started exit node connectivity monitor for %s", serverIP) + go pm.runExitNodeMonitor(ctx, serverIP) +} + +// 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 + } + pm.exitNodeServerIP = "" + pm.exitNodeMu.Unlock() + + if pm.apiServer != nil { + pm.apiServer.ClearExitNodeStatus() + } + + logger.Info("Stopped exit node connectivity monitor") +} + +// 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 string) { + check := func() { + var ( + connected bool + rtt time.Duration + ) + for attempt := 0; attempt < exitNodePingMaxAttempts; attempt++ { + if d, err := pm.pingExitNode(serverIP, exitNodePingTimeout); err == nil { + connected = true + rtt = d + break + } + 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 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 string, timeout time.Duration) (time.Duration, error) { + pm.mutex.Lock() + st := pm.stack + localIPStr := pm.localIP + 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(localIPStr) + 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) + } + 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) + } + + var idBuf [2]byte + if _, err := rand.Read(idBuf[:]); err != nil { + return 0, fmt.Errorf("failed to generate echo ID: %w", err) + } + echoID := int(binary.BigEndian.Uint16(idBuf[:])) + + 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 + } + } +} diff --git a/peers/monitor/monitor.go b/peers/monitor/monitor.go index b4b1fcb..edabd3c 100644 --- a/peers/monitor/monitor.go +++ b/peers/monitor/monitor.go @@ -25,6 +25,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 +106,14 @@ 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 + exitNodeCancel context.CancelFunc } // NewPeerMonitor creates a new peer monitor with the given callback @@ -1152,6 +1161,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 +1305,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, } @@ -1331,24 +1348,30 @@ func (pm *PeerMonitor) initNetstack() error { // 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, 58: // ICMPv4, ICMPv6 - always ours, used only by the exit node ping probe + // no per-port filtering needed + case 17: // UDP + // Check destination port + port, ok := util.GetDestPort(packet) + if !ok { + return false + } - if !active { + // 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 }