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/device/nat.go b/device/nat.go new file mode 100644 index 0000000..74ce59a --- /dev/null +++ b/device/nat.go @@ -0,0 +1,266 @@ +package device + +import ( + "encoding/binary" + "sync" + "time" + + "github.com/fosrl/newt/logger" +) + +const ( + ipv4SrcOffset = 12 + ipv4DstOffset = 16 +) + +// FixIPv4Source rewrites an IPv4 packet's source address to correctSrc if it +// doesn't already match. It returns whether a rewrite happened. +func FixIPv4Source(packet []byte, correctSrc [4]byte) bool { + return fixIPv4Address(packet, ipv4SrcOffset, correctSrc) +} + +// FixIPv4Dest rewrites an IPv4 packet's destination address to correctDst if +// it doesn't already match. It returns whether a rewrite happened. +func FixIPv4Dest(packet []byte, correctDst [4]byte) bool { + return fixIPv4Address(packet, ipv4DstOffset, correctDst) +} + +// fixIPv4Address rewrites the IPv4 address at the given header offset (source +// or destination) to newAddr if it doesn't already match, incrementally +// fixing up the IPv4 header checksum and (for TCP/UDP) the transport +// checksum so the packet stays valid. +// +// The common case - address already correct - is a single 4-byte comparison +// and nothing else, so this is safe to call unconditionally on every packet +// matched by a MiddleDevice rule. When a rewrite is needed, checksums are +// updated via the RFC 1624 incremental method (add the delta of the changed +// 16-bit words) rather than a full recompute over the packet, since only the +// address field changed. The formula is agnostic to which field (source or +// destination) changed - both are covered by the IPv4 header checksum and +// the TCP/UDP pseudo-header checksum identically. ICMP has no pseudo-header +// dependency on the IP addresses, so its checksum is left untouched. +// Non-IPv4 or malformed packets are left untouched. +func fixIPv4Address(packet []byte, offset int, newAddr [4]byte) bool { + if len(packet) < 20 || packet[0]>>4 != 4 { + return false + } + + if packet[offset] == newAddr[0] && packet[offset+1] == newAddr[1] && + packet[offset+2] == newAddr[2] && packet[offset+3] == newAddr[3] { + return false + } + + ihl := int(packet[0]&0x0f) * 4 + if ihl < 20 || len(packet) < ihl { + return false + } + + old := [4]byte{packet[offset], packet[offset+1], packet[offset+2], packet[offset+3]} + + ipChecksum := binary.BigEndian.Uint16(packet[10:12]) + binary.BigEndian.PutUint16(packet[10:12], checksumAdjust(ipChecksum, old[:], newAddr[:])) + + switch packet[9] { + case 6: // TCP + if len(packet) >= ihl+20 { + off := ihl + 16 + c := binary.BigEndian.Uint16(packet[off : off+2]) + binary.BigEndian.PutUint16(packet[off:off+2], checksumAdjust(c, old[:], newAddr[:])) + } + case 17: // UDP + if len(packet) >= ihl+8 { + off := ihl + 6 + c := binary.BigEndian.Uint16(packet[off : off+2]) + if c != 0 { // zero means checksum not used - must stay zero + binary.BigEndian.PutUint16(packet[off:off+2], checksumAdjust(c, old[:], newAddr[:])) + } + } + } + + copy(packet[offset:offset+4], newAddr[:]) + return true +} + +// checksumAdjust incrementally updates a ones-complement checksum after some +// of the bytes it covers changed from old to new (RFC 1624), avoiding a full +// recompute over the packet. old and new must be the same (even) length. +func checksumAdjust(checksum uint16, old, new []byte) uint16 { + sum := uint32(^checksum) + + for i := 0; i+1 < len(old); i += 2 { + sum += uint32(^binary.BigEndian.Uint16(old[i:i+2])) & 0xffff + } + for i := 0; i+1 < len(new); i += 2 { + sum += uint32(binary.BigEndian.Uint16(new[i : i+2])) + } + + for sum>>16 != 0 { + sum = (sum & 0xffff) + (sum >> 16) + } + + return ^uint16(sum) +} + +// ipv4L4Ports extracts the TCP/UDP source and destination ports from an IPv4 +// packet. ok is false for anything else (non-IPv4, non-TCP/UDP, malformed). +func ipv4L4Ports(packet []byte) (proto uint8, srcPort, dstPort uint16, ok bool) { + if len(packet) < 20 || packet[0]>>4 != 4 { + return 0, 0, 0, false + } + proto = packet[9] + if proto != 6 && proto != 17 { + return 0, 0, 0, false + } + ihl := int(packet[0]&0x0f) * 4 + if ihl < 20 || len(packet) < ihl+4 { + return 0, 0, 0, false + } + srcPort = binary.BigEndian.Uint16(packet[ihl : ihl+2]) + dstPort = binary.BigEndian.Uint16(packet[ihl+2 : ihl+4]) + return proto, srcPort, dstPort, true +} + +// IPv4SourceEquals reports whether packet's IPv4 source address equals addr. +func IPv4SourceEquals(packet []byte, addr [4]byte) bool { + return len(packet) >= 16 && packet[0]>>4 == 4 && + packet[12] == addr[0] && packet[13] == addr[1] && packet[14] == addr[2] && packet[15] == addr[3] +} + +// natEntryTTL bounds how long an ExitNodeNAT entry is honored without being +// refreshed by further traffic on the same port. It's a var rather than a +// const so tests can shrink it. Chosen generously relative to typical +// request/response traffic - the only cost of expiring too early is the +// original bug reappearing for that one flow, not corruption of anything +// else, so this errs on the long side. +var natEntryTTL = 5 * time.Minute + +// natRefreshInterval bounds how often a busy flow's entry timestamp actually +// gets rewritten. A saturating connection (e.g. iperf) calls FixOutboundSource +// or FixInboundDest on every single packet - refreshing on every one of them +// would mean a map write (and, for new entries, a full-table prune) at line +// rate instead of at most once per interval. natEntryTTL is minutes, so +// resolution at this granularity costs nothing. +const natRefreshInterval = time.Second + +type natKey struct { + proto uint8 + port uint16 +} + +// ExitNodeNAT tracks which local (protocol, port) pairs had their outbound +// source address corrected by FixOutboundSource, so FixInboundDest can +// translate the destination of the matching inbound reply back to the +// address the local OS socket actually expects. +// +// This statefulness exists because rewriting the outbound packet's source +// only changes what goes out on the wire - it does not change the local +// kernel's own record of the connection's local address, which was already +// selected and cached (in the socket's own connection state) at connect()/ +// send() time, before this packet ever reached this interception point. +// Without also translating the reply's destination back, the OS can't match +// the exit node's response to the socket waiting for it, and the request +// hangs even though the corrected outbound packet reached the server fine. +// +// Entries are keyed by local port only (not the full flow), refreshed on +// every match, and expire after natEntryTTL of inactivity - both so a later, +// unrelated connection that happens to reuse the same ephemeral port isn't +// wrongly treated as needing translation (e.g. one that was never affected +// because it bound explicitly to the correct address), and so the table +// doesn't grow unbounded over a long-lived tunnel. +type ExitNodeNAT struct { + mu sync.Mutex + seen map[natKey]time.Time +} + +func NewExitNodeNAT() *ExitNodeNAT { + return &ExitNodeNAT{seen: make(map[natKey]time.Time)} +} + +// FixOutboundSource rewrites packet's source to correctSrc (see +// FixIPv4Source) and, if a rewrite was needed, remembers the packet's source +// port so FixInboundDest knows to translate the reply back. +func (n *ExitNodeNAT) FixOutboundSource(packet []byte, correctSrc [4]byte) { + if !FixIPv4Source(packet, correctSrc) { + return + } + + proto, srcPort, _, ok := ipv4L4Ports(packet) + if !ok { + return + } + + key := natKey{proto, srcPort} + now := time.Now() + + n.mu.Lock() + t, existed := n.seen[key] + if existed && now.Sub(t) < natRefreshInterval { + // Already recorded recently enough - skip the write entirely. This is + // the common case for a busy flow: every packet gets here, but only + // one per interval needs to touch the map. + n.mu.Unlock() + return + } + n.seen[key] = now + if !existed { + // Only prune when the table is actually growing (a new connection), + // not on every packet - this is an O(map size) scan and the map only + // ever gains entries here. + n.prune() + } + n.mu.Unlock() + + if !existed { + logger.Debug("ExitNodeNAT: corrected outbound source for proto=%d port=%d", proto, srcPort) + } +} + +// FixInboundDest rewrites packet's destination to wrongDst, but only if its +// destination port matches an outbound flow FixOutboundSource actually +// corrected - otherwise this connection was never affected by the bug (e.g. +// a socket explicitly bound to the correct address already) and must be +// left alone. +func (n *ExitNodeNAT) FixInboundDest(packet []byte, wrongDst [4]byte) { + proto, _, dstPort, ok := ipv4L4Ports(packet) + if !ok { + return + } + + key := natKey{proto, dstPort} + now := time.Now() + + n.mu.Lock() + t, tracked := n.seen[key] + expired := tracked && now.Sub(t) > natEntryTTL + if tracked { + if expired { + delete(n.seen, key) + tracked = false + } else if now.Sub(t) >= natRefreshInterval { + n.seen[key] = now + } + } + n.mu.Unlock() + + if expired { + logger.Warn("ExitNodeNAT: entry for proto=%d port=%d expired before a reply arrived on it - that flow's replies will be dropped by the OS from here on", proto, dstPort) + } + + if !tracked { + return + } + + FixIPv4Dest(packet, wrongDst) +} + +// prune removes expired entries. Called with n.mu held, only from +// FixOutboundSource so the cost is amortized over new outbound connections +// rather than paid on every packet. +func (n *ExitNodeNAT) prune() { + now := time.Now() + for k, t := range n.seen { + if now.Sub(t) > natEntryTTL { + delete(n.seen, k) + } + } +} diff --git a/device/nat_test.go b/device/nat_test.go new file mode 100644 index 0000000..18b0fa9 --- /dev/null +++ b/device/nat_test.go @@ -0,0 +1,383 @@ +package device + +import ( + "bytes" + "encoding/binary" + "testing" + "time" +) + +// onesComplementSum computes an RFC 1071 ones-complement checksum from +// scratch, independent of checksumAdjust, so it can be used to verify +// FixIPv4Source's incremental updates rather than tautologically re-deriving +// them with the same formula. +func onesComplementSum(data []byte) uint16 { + var sum uint32 + n := len(data) + for i := 0; i+1 < n; i += 2 { + sum += uint32(data[i])<<8 | uint32(data[i+1]) + } + if n%2 == 1 { + sum += uint32(data[n-1]) << 8 + } + for sum>>16 != 0 { + sum = (sum & 0xffff) + (sum >> 16) + } + return ^uint16(sum) +} + +func buildIPv4Header(src, dst [4]byte, proto byte, payloadLen int) []byte { + h := make([]byte, 20) + h[0] = 0x45 + binary.BigEndian.PutUint16(h[2:4], uint16(20+payloadLen)) + h[6] = 0x40 // DF + h[8] = 64 // TTL + h[9] = proto + copy(h[12:16], src[:]) + copy(h[16:20], dst[:]) + binary.BigEndian.PutUint16(h[10:12], onesComplementSum(h)) + return h +} + +func buildUDPSegment(src, dst [4]byte, payload []byte) []byte { + udpLen := 8 + len(payload) + seg := make([]byte, udpLen) + binary.BigEndian.PutUint16(seg[0:2], 12345) + binary.BigEndian.PutUint16(seg[2:4], 53) + binary.BigEndian.PutUint16(seg[4:6], uint16(udpLen)) + copy(seg[8:], payload) + + pseudo := make([]byte, 12+udpLen) + copy(pseudo[0:4], src[:]) + copy(pseudo[4:8], dst[:]) + pseudo[9] = 17 + binary.BigEndian.PutUint16(pseudo[10:12], uint16(udpLen)) + copy(pseudo[12:], seg) + csum := onesComplementSum(pseudo) + if csum == 0 { + csum = 0xffff + } + binary.BigEndian.PutUint16(seg[6:8], csum) + return seg +} + +func buildTCPSegment(src, dst [4]byte, payload []byte) []byte { + tcpLen := 20 + len(payload) + seg := make([]byte, tcpLen) + binary.BigEndian.PutUint16(seg[0:2], 54321) + binary.BigEndian.PutUint16(seg[2:4], 443) + seg[12] = 0x50 // data offset 5 + copy(seg[20:], payload) + + pseudo := make([]byte, 12+tcpLen) + copy(pseudo[0:4], src[:]) + copy(pseudo[4:8], dst[:]) + pseudo[9] = 6 + binary.BigEndian.PutUint16(pseudo[10:12], uint16(tcpLen)) + copy(pseudo[12:], seg) + binary.BigEndian.PutUint16(seg[16:18], onesComplementSum(pseudo)) + return seg +} + +func verifyIPv4HeaderChecksum(t *testing.T, packet []byte) { + t.Helper() + header := append([]byte(nil), packet[:20]...) + binary.BigEndian.PutUint16(header[10:12], 0) + want := onesComplementSum(header) + got := binary.BigEndian.Uint16(packet[10:12]) + if got != want { + t.Errorf("IPv4 header checksum = %#04x, want %#04x", got, want) + } +} + +func verifyUDPChecksum(t *testing.T, packet []byte, src, dst [4]byte) { + t.Helper() + seg := append([]byte(nil), packet[20:]...) + binary.BigEndian.PutUint16(seg[6:8], 0) + pseudo := make([]byte, 12+len(seg)) + copy(pseudo[0:4], src[:]) + copy(pseudo[4:8], dst[:]) + pseudo[9] = 17 + binary.BigEndian.PutUint16(pseudo[10:12], uint16(len(seg))) + copy(pseudo[12:], seg) + want := onesComplementSum(pseudo) + if want == 0 { + want = 0xffff + } + got := binary.BigEndian.Uint16(packet[26:28]) + if got != want { + t.Errorf("UDP checksum = %#04x, want %#04x", got, want) + } +} + +func verifyTCPChecksum(t *testing.T, packet []byte, src, dst [4]byte) { + t.Helper() + seg := append([]byte(nil), packet[20:]...) + binary.BigEndian.PutUint16(seg[16:18], 0) + pseudo := make([]byte, 12+len(seg)) + copy(pseudo[0:4], src[:]) + copy(pseudo[4:8], dst[:]) + pseudo[9] = 6 + binary.BigEndian.PutUint16(pseudo[10:12], uint16(len(seg))) + copy(pseudo[12:], seg) + want := onesComplementSum(pseudo) + got := binary.BigEndian.Uint16(packet[36:38]) + if got != want { + t.Errorf("TCP checksum = %#04x, want %#04x", got, want) + } +} + +func TestFixIPv4SourceUDP(t *testing.T) { + wrongSrc := [4]byte{10, 0, 0, 1} + correctSrc := [4]byte{10, 0, 0, 2} + dst := [4]byte{192, 168, 1, 1} + payload := []byte("hello world") + + udp := buildUDPSegment(wrongSrc, dst, payload) + ip := buildIPv4Header(wrongSrc, dst, 17, len(udp)) + packet := append(ip, udp...) + + FixIPv4Source(packet, correctSrc) + + if got := [4]byte{packet[12], packet[13], packet[14], packet[15]}; got != correctSrc { + t.Fatalf("source = %v, want %v", got, correctSrc) + } + verifyIPv4HeaderChecksum(t, packet) + verifyUDPChecksum(t, packet, correctSrc, dst) + if !bytes.Equal(packet[28:], payload) { + t.Errorf("UDP payload was mutated: got %q, want %q", packet[28:], payload) + } +} + +func TestFixIPv4SourceTCP(t *testing.T) { + wrongSrc := [4]byte{172, 16, 0, 5} + correctSrc := [4]byte{172, 16, 0, 9} + dst := [4]byte{8, 8, 8, 8} + payload := []byte("GET / HTTP/1.1") + + tcp := buildTCPSegment(wrongSrc, dst, payload) + ip := buildIPv4Header(wrongSrc, dst, 6, len(tcp)) + packet := append(ip, tcp...) + + FixIPv4Source(packet, correctSrc) + + if got := [4]byte{packet[12], packet[13], packet[14], packet[15]}; got != correctSrc { + t.Fatalf("source = %v, want %v", got, correctSrc) + } + verifyIPv4HeaderChecksum(t, packet) + verifyTCPChecksum(t, packet, correctSrc, dst) +} + +func TestFixIPv4SourceAlreadyCorrect(t *testing.T) { + correctSrc := [4]byte{10, 0, 0, 2} + dst := [4]byte{192, 168, 1, 1} + udp := buildUDPSegment(correctSrc, dst, []byte("payload")) + ip := buildIPv4Header(correctSrc, dst, 17, len(udp)) + packet := append(ip, udp...) + + original := append([]byte(nil), packet...) + FixIPv4Source(packet, correctSrc) + + if !bytes.Equal(packet, original) { + t.Errorf("fast path mutated an already-correct packet: got %x, want %x", packet, original) + } +} + +func TestFixIPv4SourceICMPChecksumUntouched(t *testing.T) { + wrongSrc := [4]byte{10, 0, 0, 1} + correctSrc := [4]byte{10, 0, 0, 2} + dst := [4]byte{192, 168, 1, 1} + + // Minimal ICMP echo request: type=8, code=0, checksum, id, seq. + icmp := []byte{8, 0, 0xf7, 0xfd, 0x00, 0x01, 0x00, 0x01} + originalICMP := append([]byte(nil), icmp...) + ip := buildIPv4Header(wrongSrc, dst, 1, len(icmp)) + packet := append(ip, icmp...) + + FixIPv4Source(packet, correctSrc) + + if got := [4]byte{packet[12], packet[13], packet[14], packet[15]}; got != correctSrc { + t.Fatalf("source = %v, want %v", got, correctSrc) + } + verifyIPv4HeaderChecksum(t, packet) + if !bytes.Equal(packet[20:], originalICMP) { + t.Errorf("ICMP body was mutated: got %x, want %x", packet[20:], originalICMP) + } +} + +func TestFixIPv4SourceMalformedPacketNoPanic(t *testing.T) { + correctSrc := [4]byte{10, 0, 0, 2} + + defer func() { + if r := recover(); r != nil { + t.Fatalf("FixIPv4Source panicked: %v", r) + } + }() + + FixIPv4Source(nil, correctSrc) + FixIPv4Source([]byte{}, correctSrc) + FixIPv4Source([]byte{0x45, 0x00, 0x00}, correctSrc) + FixIPv4Source([]byte{0x60, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}, correctSrc) // IPv6 version nibble +} + +func TestFixIPv4DestUDP(t *testing.T) { + src := [4]byte{192, 168, 1, 1} + wrongDst := [4]byte{10, 0, 0, 1} + correctDst := [4]byte{10, 0, 0, 2} + payload := []byte("reply") + + udp := buildUDPSegment(src, wrongDst, payload) + ip := buildIPv4Header(src, wrongDst, 17, len(udp)) + packet := append(ip, udp...) + + if !FixIPv4Dest(packet, correctDst) { + t.Fatal("expected FixIPv4Dest to report a rewrite") + } + if got := [4]byte{packet[16], packet[17], packet[18], packet[19]}; got != correctDst { + t.Fatalf("dest = %v, want %v", got, correctDst) + } + verifyIPv4HeaderChecksum(t, packet) + verifyUDPChecksum(t, packet, src, correctDst) +} + +// exitNodeNATTestPacket builds a minimal IPv4/UDP packet with the given +// addresses and ports, for exercising ExitNodeNAT's port-based tracking. +func exitNodeNATTestPacket(src, dst [4]byte, srcPort, dstPort uint16) []byte { + seg := make([]byte, 8) + binary.BigEndian.PutUint16(seg[0:2], srcPort) + binary.BigEndian.PutUint16(seg[2:4], dstPort) + binary.BigEndian.PutUint16(seg[4:6], uint16(len(seg))) + + pseudo := make([]byte, 12+len(seg)) + copy(pseudo[0:4], src[:]) + copy(pseudo[4:8], dst[:]) + pseudo[9] = 17 + binary.BigEndian.PutUint16(pseudo[10:12], uint16(len(seg))) + copy(pseudo[12:], seg) + csum := onesComplementSum(pseudo) + if csum == 0 { + csum = 0xffff + } + binary.BigEndian.PutUint16(seg[6:8], csum) + + ip := buildIPv4Header(src, dst, 17, len(seg)) + return append(ip, seg...) +} + +func TestExitNodeNATRoundTrip(t *testing.T) { + wrongSrc := [4]byte{100, 89, 128, 9} // primary/site tunnel IP (the bug's default pick) + correctSrc := [4]byte{100, 89, 128, 4} // exit node's secondary tunnel IP + serverIP := [4]byte{100, 89, 128, 1} + const localPort = 52746 + + nat := NewExitNodeNAT() + + // Outbound: kernel picked the wrong source; our fix rewrites it and should + // remember the local port so the reply gets translated. + outbound := exitNodeNATTestPacket(wrongSrc, serverIP, localPort, 80) + nat.FixOutboundSource(outbound, correctSrc) + if got := [4]byte{outbound[12], outbound[13], outbound[14], outbound[15]}; got != correctSrc { + t.Fatalf("outbound source = %v, want %v", got, correctSrc) + } + + // Inbound reply: correctly addressed to correctSrc (the exit node saw the + // fixed source), but the OS's own connection state still expects wrongSrc. + reply := exitNodeNATTestPacket(serverIP, correctSrc, 80, localPort) + nat.FixInboundDest(reply, wrongSrc) + if got := [4]byte{reply[16], reply[17], reply[18], reply[19]}; got != wrongSrc { + t.Fatalf("reply dest = %v, want %v (translated back for the OS to match the socket)", got, wrongSrc) + } + verifyIPv4HeaderChecksum(t, reply) +} + +func TestExitNodeNATUntrackedPortPassesThrough(t *testing.T) { + wrongSrc := [4]byte{100, 89, 128, 9} + correctSrc := [4]byte{100, 89, 128, 4} + serverIP := [4]byte{100, 89, 128, 1} + const localPort = 55555 // never seen by FixOutboundSource + + nat := NewExitNodeNAT() + + // A socket that was already, legitimately bound to correctSrc: its reply + // must not be touched, since translating it would misroute it away from + // the socket that's actually expecting it. + reply := exitNodeNATTestPacket(serverIP, correctSrc, 80, localPort) + original := append([]byte(nil), reply...) + nat.FixInboundDest(reply, wrongSrc) + + if !bytes.Equal(reply, original) { + t.Errorf("untracked port was translated: got %x, want unchanged %x", reply, original) + } +} + +func TestExitNodeNATEntryExpires(t *testing.T) { + origTTL := natEntryTTL + natEntryTTL = 10 * time.Millisecond + defer func() { natEntryTTL = origTTL }() + + wrongSrc := [4]byte{100, 89, 128, 9} + correctSrc := [4]byte{100, 89, 128, 4} + serverIP := [4]byte{100, 89, 128, 1} + const localPort = 52746 + + nat := NewExitNodeNAT() + + outbound := exitNodeNATTestPacket(wrongSrc, serverIP, localPort, 80) + nat.FixOutboundSource(outbound, correctSrc) + + time.Sleep(50 * time.Millisecond) + + reply := exitNodeNATTestPacket(serverIP, correctSrc, 80, localPort) + original := append([]byte(nil), reply...) + nat.FixInboundDest(reply, wrongSrc) + + if !bytes.Equal(reply, original) { + t.Errorf("expired entry was still translated: got %x, want unchanged %x", reply, original) + } +} + +// BenchmarkExitNodeNATSteadyStateOutbound simulates a single busy flow (e.g. +// an iperf upload) hammering FixOutboundSource, as happens for real since the +// OS keeps stamping every packet of an affected socket with the wrong source +// for the connection's whole lifetime, not just its first packet. Before the +// refresh-throttling/prune-on-insert-only fix, every call here paid for a +// map write plus a full-table prune; steady state should now be a single +// lock/lookup/compare with no write and no allocation. +func BenchmarkExitNodeNATSteadyStateOutbound(b *testing.B) { + wrongSrc := [4]byte{100, 89, 128, 9} + correctSrc := [4]byte{100, 89, 128, 4} + serverIP := [4]byte{100, 89, 128, 1} + + nat := NewExitNodeNAT() + packet := exitNodeNATTestPacket(wrongSrc, serverIP, 52746, 80) + nat.FixOutboundSource(packet, correctSrc) // prime the entry + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + // FixOutboundSource rewrites in place, so re-derive a wrong-source + // packet each iteration rather than measuring the already-correct + // (no-op) fast path. + packet := exitNodeNATTestPacket(wrongSrc, serverIP, 52746, 80) + nat.FixOutboundSource(packet, correctSrc) + } +} + +// BenchmarkExitNodeNATSteadyStateInbound is BenchmarkExitNodeNATSteadyStateOutbound's +// counterpart for the download direction / ACK stream. +func BenchmarkExitNodeNATSteadyStateInbound(b *testing.B) { + wrongSrc := [4]byte{100, 89, 128, 9} + correctSrc := [4]byte{100, 89, 128, 4} + serverIP := [4]byte{100, 89, 128, 1} + + nat := NewExitNodeNAT() + outbound := exitNodeNATTestPacket(wrongSrc, serverIP, 52746, 80) + nat.FixOutboundSource(outbound, correctSrc) // establish the tracked port + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + reply := exitNodeNATTestPacket(serverIP, correctSrc, 80, 52746) + nat.FixInboundDest(reply, wrongSrc) + } +} diff --git a/go.mod b/go.mod index 96a3c23..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 @@ -32,4 +32,4 @@ require ( ) // To be used ONLY for local development -// replace github.com/fosrl/newt => ../newt +replace github.com/fosrl/newt => ../newt diff --git a/go.sum b/go.sum index a46f567..ce3ee42 100644 --- a/go.sum +++ b/go.sum @@ -1,7 +1,5 @@ github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= -github.com/fosrl/newt v1.15.0 h1:WpL0whZM1FMjUe2Vy5jSH1bgbxm1O9k1qCyF/mqZT+s= -github.com/fosrl/newt v1.15.0/go.mod h1:l6kWoZPSaXT+ZRUjiyPgwflRqZWYaXpUj9oQ0sOPh4o= github.com/godbus/dbus/v5 v5.2.2 h1:TUR3TgtSVDmjiXOgAAyaZbYmIeP3DPkld3jgKGV8mXQ= github.com/godbus/dbus/v5 v5.2.2/go.mod h1:3AAv2+hPq5rdnr5txxxRwiGjPXamgoIHgz9FPBfOp3c= github.com/google/btree v1.1.3 h1:CVpQJjYgC4VbzxeGVHfvZrv1ctoYCAI8vbl07Fcxlyg= diff --git a/olm/connect.go b/olm/connect.go index ff2b5f4..e828dc9 100644 --- a/olm/connect.go +++ b/olm/connect.go @@ -3,6 +3,7 @@ package olm import ( "encoding/json" "fmt" + "net/netip" "os" "runtime" "strconv" @@ -11,6 +12,7 @@ import ( "github.com/fosrl/newt/logger" "github.com/fosrl/newt/network" + "github.com/fosrl/newt/util" olmDevice "github.com/fosrl/olm/device" "github.com/fosrl/olm/dns" dnsOverride "github.com/fosrl/olm/dns/override" @@ -51,6 +53,11 @@ func (o *Olm) handleConnect(msg websocket.WSMessage) { o.updateRegister = nil } + if o.stopPingRequest != nil { + o.stopPingRequest() + o.stopPingRequest = nil + } + // if there is an existing tunnel then close it if o.dev != nil { logger.Info("Got new message. Closing existing tunnel!") @@ -68,6 +75,17 @@ func (o *Olm) handleConnect(msg websocket.WSMessage) { return } + // When handed an already-open FD (mobile/NetworkExtension platforms), the + // TUN device's addresses and routes are owned and reconciled by the host + // platform from NetworkSettings (e.g. Apple's NEPacketTunnelProvider via + // setTunnelNetworkSettings) - our own ifconfig/route subprocess calls must + // not also run against the same interface, or the two end up installing + // competing routes to the same destination. On macOS specifically this + // package's darwin code paths would otherwise run for real here (the NE + // build shares GOOS=darwin with the CLI), unlike iOS where they're already + // no-ops via a GOOS check. + network.NativeConfigDisabled = o.tunnelConfig.FileDescriptorTun != 0 + o.tdev, err = func() (tun.Device, error) { if o.tunnelConfig.FileDescriptorTun != 0 { return olmDevice.CreateTUNFromFD(o.tunnelConfig.FileDescriptorTun, o.tunnelConfig.MTU) @@ -138,11 +156,24 @@ func (o *Olm) handleConnect(msg websocket.WSMessage) { logger.Error("Failed to bring up WireGuard device: %v", err) } + // Set the private key unconditionally, since it's otherwise only ever set as a + // side effect of configuring a site peer (see peers.ConfigurePeer) - if there are + // no sites (e.g. an exit-node-only connection), the interface would otherwise be + // brought up with no private key configured at all. + if err := o.dev.IpcSet(fmt.Sprintf("private_key=%s\n", util.FixKey(o.privateKey.String()))); err != nil { + logger.Error("Failed to set private key on WireGuard device: %v", err) + } + // Extract interface IP (strip CIDR notation if present) interfaceIP := wgData.TunnelIP if strings.Contains(interfaceIP, "/") { interfaceIP = strings.Split(interfaceIP, "/")[0] } + if addr, err := netip.ParseAddr(interfaceIP); err == nil { + o.primaryTunnelIP = addr + } else { + logger.Warn("Failed to parse tunnel IP %q: %v", interfaceIP, err) + } // Create and start DNS proxy o.dnsProxy, err = dns.NewDNSProxy(o.middleDev, o.tunnelConfig.MTU, wgData.UtilitySubnet, o.tunnelConfig.UpstreamDNS, o.tunnelConfig.TunnelDNS, interfaceIP, o.tunnelConfig.MatchDomains, o.tunnelConfig.PublicDNS) @@ -162,7 +193,7 @@ func (o *Olm) handleConnect(msg websocket.WSMessage) { logger.Error("Failed to o.tunnelConfigure interface: %v", err) } - if network.AddRoutes([]string{wgData.UtilitySubnet}, o.tunnelConfig.InterfaceName); err != nil { // also route the utility subnet + if err := network.AddRoutesWithSource([]string{wgData.UtilitySubnet}, o.tunnelConfig.InterfaceName, interfaceIP); err != nil { // also route the utility subnet logger.Error("Failed to add route for utility subnet: %v", err) } @@ -257,6 +288,14 @@ func (o *Olm) handleConnect(msg websocket.WSMessage) { network.SetDNSServers([]string{o.dnsProxy.GetProxyIP().String()}) } + if wgData.ExitNode != nil && wgData.ExitNode.Connect { + if err := o.connectExitNode(*wgData.ExitNode); err != nil { + logger.Error("Failed to connect to exit node: %v", err) + } + } else { + logger.Debug("No exit node to connect to (not provided, or connect flag is false)") + } + o.apiServer.SetRegistered(true) o.registered = true diff --git a/olm/data.go b/olm/data.go index 879e61e..587d102 100644 --- a/olm/data.go +++ b/olm/data.go @@ -202,6 +202,10 @@ func (o *Olm) handleSync(msg websocket.WSMessage) { // Sync exit nodes for hole punching o.syncExitNodes(syncData.ExitNodes) + // Reconcile the client's own exit node connection (connect/switch/update + // aliases/disconnect), same as what the initial olm/wg/connect message does + o.syncExitNodeConnection(syncData.ExitNode) + // Build a map of expected peers from the incoming data expectedPeers := make(map[int]peers.SiteConfig) for _, site := range syncData.Sites { diff --git a/olm/exitnode.go b/olm/exitnode.go new file mode 100644 index 0000000..26ae59c --- /dev/null +++ b/olm/exitnode.go @@ -0,0 +1,490 @@ +package olm + +import ( + "encoding/json" + "fmt" + "net" + "net/netip" + "strings" + + "github.com/fosrl/newt/logger" + "github.com/fosrl/newt/network" + "github.com/fosrl/newt/util" + olmDevice "github.com/fosrl/olm/device" + "github.com/fosrl/olm/peers" + "github.com/fosrl/olm/websocket" +) + +// exitNodeAliasSiteId is the sentinel siteId used when registering exit node +// aliases with the DNS proxy. It is not a real site, and the JIT handler +// treats siteId 0 as "no JIT lookup", which is correct here since the exit +// node is connected directly rather than on demand. +const exitNodeAliasSiteId = 0 + +// connectExitNode configures a WireGuard peer connection to an exit node, on the +// same interface and WireGuard device already used for site peers. The exit node +// lives in a different address space than the site tunnel, so a secondary address +// (ExitNodeConfig.TunnelIP) is added to the interface for it - the exit node's own +// WireGuard peer entry only accepts traffic sourced from that address. Nothing here +// is persisted; it's purely in-memory WireGuard/routing state, same as site peers. +func (o *Olm) connectExitNode(cfg ExitNodeConfig) error { + if !o.tunnelRunning { + return fmt.Errorf("tunnel not running") + } + if cfg.PublicKey == "" || cfg.Endpoint == "" || cfg.ServerIP == "" || cfg.TunnelIP == "" { + return fmt.Errorf("incomplete exit node configuration") + } + + o.exitNodeMu.Lock() + defer o.exitNodeMu.Unlock() + + dev := o.dev + if dev == nil { + return fmt.Errorf("wireguard device not initialized") + } + + if o.exitNode != nil && o.exitNode.PublicKey == cfg.PublicKey && + o.exitNode.Endpoint == cfg.Endpoint && o.exitNode.ServerIP == cfg.ServerIP && + o.exitNode.TunnelIP == cfg.TunnelIP { + if !slicesEqual(o.exitNode.Aliases, cfg.Aliases) { + logger.Info("Already connected to exit node %s, updating aliases", cfg.PublicKey) + o.updateExitNodeAliasesLocked(cfg.Aliases) + } else { + logger.Info("Already connected to exit node %s, ignoring duplicate connect message", cfg.PublicKey) + } + return nil + } + + if o.exitNode != nil && o.exitNode.PublicKey != cfg.PublicKey { + logger.Info("Switching exit nodes, removing previous exit node peer") + if err := o.removeExitNodePeerLocked(); err != nil { + logger.Warn("Failed to remove previous exit node peer: %v", err) + } + } + + endpoint := cfg.Endpoint + if !strings.Contains(endpoint, ":") { + relayPort := cfg.RelayPort + if relayPort == 0 { + relayPort = 21820 + } + endpoint = fmt.Sprintf("%s:%d", endpoint, relayPort) + } + + resolvedEndpoint, err := util.ResolveDomain(endpoint) + if err != nil { + return fmt.Errorf("failed to resolve exit node endpoint: %w", err) + } + + interfaceName := o.tunnelConfig.InterfaceName + tunnelIP := cfg.TunnelIP + if !strings.Contains(tunnelIP, "/") { + tunnelIP += "/32" + } + // Add the secondary address before configuring the peer or route below, and + // fail closed if it doesn't succeed: AddSecondaryAddress (via AddIPv4Address) + // refuses to add when no primary address is configured yet, which would + // otherwise silently make the exit node's address the interface's primary + // one on mobile platforms (array order is what determines primary there). + // Bailing out here before touching the WireGuard device at all means there's + // never a half-configured peer left behind to roll back. + if err := network.AddSecondaryAddress(interfaceName, tunnelIP); err != nil { + return fmt.Errorf("failed to add secondary address %s for exit node: %w", tunnelIP, err) + } + + persistentKeepalive := 0 + if pm := o.getPeerManager(); pm != nil { + persistentKeepalive = pm.PersistentKeepalive + } + + allowedIP := strings.Split(cfg.ServerIP, "/")[0] + "/32" + + wgConfig := fmt.Sprintf(`public_key=%s +allowed_ip=%s +endpoint=%s +persistent_keepalive_interval=%d`, util.FixKey(cfg.PublicKey), allowedIP, resolvedEndpoint, persistentKeepalive) + + if err := dev.IpcSet(wgConfig); err != nil { + return fmt.Errorf("failed to configure exit node peer: %w", err) + } + + // ServerIP arrives as a bare IP with no CIDR suffix, but AddRouteForServerIP + // parses it as a CIDR on darwin (to explicitly route the subnet up the tunnel, + // since unlike Linux, adding the address to the interface does not implicitly + // create a route for it) - without a mask that parse fails and the route (and + // its corresponding NetworkSettings entry, which is what surfaces it via the + // API) is silently never added. + serverIPForRoute := strings.Split(cfg.ServerIP, "/")[0] + "/32" + // The route must also be pinned to our exit node tunnel address as its source + // (darwin route(8) -ifa): the interface carries a second address for the site + // tunnel too, and without an explicit source darwin picks that one instead, + // which the exit node's WireGuard AllowedIPs filtering then silently drops. + tunnelIPForRoute := strings.Split(cfg.TunnelIP, "/")[0] + if err := network.AddRouteForServerIPWithSource(serverIPForRoute, interfaceName, tunnelIPForRoute); err != nil { + logger.Warn("Failed to add route for exit node server IP: %v", err) + } + + // On macOS/iOS NetworkExtension, the OS can't reliably pin an outbound socket's + // source address to this interface's secondary address the way BSD route(8) + // -ifa does for the CLI path above - unbound sockets still get the primary + // (site tunnel) address stamped as source even for traffic destined to the + // exit node, which the exit node's WireGuard AllowedIPs filtering then + // silently drops. Fix it in-tunnel: intercept outbound packets addressed to + // the exit node and rewrite their source back to tunnelIP before WireGuard + // encrypts them. + // + // That alone isn't enough for anything that expects a reply (TCP, or any + // request/response over UDP): rewriting the outbound packet only changes + // what goes out on the wire - it doesn't change the OS's own connection + // state, which already recorded the *wrong* (primary) address as this + // socket's local address at connect()/send() time, before the packet ever + // reached this interception point. When the exit node's reply comes back + // correctly addressed to tunnelIP, the OS can't match it to a socket whose + // local address it thinks is the primary tunnel IP, and silently drops it - + // the connection hangs even though the corrected request reached the server + // fine. So also intercept inbound replies from the exit node and translate + // their destination back to the primary address, but only for flows we + // actually corrected outbound (tracked by ExitNodeNAT) - a socket that + // happened to already be bound to the correct address must be left alone. + // + // The fast path (address already correct) is cheap enough - a 4-byte + // comparison and nothing else when no rewrite is needed - to just leave + // this on unconditionally rather than gate it per-GOOS. Every platform's + // route-based source pinning (AddRouteForServerIPWithSource and friends) + // is a best-effort hint to the OS, not a guarantee: Android's + // VpnService.Builder only supports plain destination/prefix routes with + // no source/gateway at all, and even where the OS route can carry a + // source, an unbound socket's address selection is the OS's call, not + // ours. Keeping this rule active everywhere means any platform that gets + // the source wrong for any reason - not just the ones we've already hit + // this bug on - self-corrects instead of silently blackholing exit node + // traffic. + if o.middleDev != nil { + serverAddr, errS := netip.ParseAddr(strings.Split(cfg.ServerIP, "/")[0]) + correctAddr, errC := netip.ParseAddr(tunnelIPForRoute) + switch { + case errS != nil || errC != nil || !correctAddr.Is4(): + logger.Warn("Exit node NAT: skipping source-NAT setup, invalid address (server=%v tunnel=%v)", errS, errC) + case !o.primaryTunnelIP.IsValid() || !o.primaryTunnelIP.Is4(): + logger.Warn("Exit node NAT: skipping source-NAT setup, no primary tunnel IP recorded") + default: + correctSrc := correctAddr.As4() + wrongSrc := o.primaryTunnelIP.As4() + serverSrc := serverAddr.As4() + nat := olmDevice.NewExitNodeNAT() + + o.middleDev.AddRule(serverAddr, func(packet []byte) bool { + nat.FixOutboundSource(packet, correctSrc) + return false + }) + o.middleDev.AddRule(correctAddr, func(packet []byte) bool { + // Only packets that actually came from the exit node's own + // peer should ever be translated - this rule's key (tunnelIP) + // is also used by the ICMP connectivity monitor's own address, + // so a defensive source check keeps this from ever touching + // unrelated traffic that happens to be addressed to tunnelIP. + if olmDevice.IPv4SourceEquals(packet, serverSrc) { + nat.FixInboundDest(packet, wrongSrc) + } + return false + }) + + logger.Debug("Exit node NAT: intercepting traffic to %s, translating source/dest between %s (primary) and %s (exit node secondary)", serverAddr, o.primaryTunnelIP, correctAddr) + } + } + + cfgCopy := cfg + o.exitNode = &cfgCopy + + if o.dnsProxy != nil { + serverIP := net.ParseIP(cfg.ServerIP) + if serverIP != nil { + for _, alias := range cfg.Aliases { + logger.Debug("Adding alias %s to the edit node", alias) + if err := o.dnsProxy.AddDNSRecord(alias, serverIP, exitNodeAliasSiteId); err != nil { + logger.Warn("Failed to add DNS record for exit node alias %s: %v", alias, err) + } + } + } + } + + if pm := o.getPeerManager(); pm != nil { + pm.SetExitNode(strings.Split(cfg.ServerIP, "/")[0], strings.Split(cfg.TunnelIP, "/")[0]) + } + + logger.Info("Connected to exit node at %s", resolvedEndpoint) + return nil +} + +// disconnectExitNode tears down the current exit node peer connection, if any. +func (o *Olm) disconnectExitNode() error { + o.exitNodeMu.Lock() + defer o.exitNodeMu.Unlock() + + return o.removeExitNodePeerLocked() +} + +// removeExitNodePeerLocked removes the current exit node peer, its secondary +// interface address, and its server IP route. Must be called with exitNodeMu held. +func (o *Olm) removeExitNodePeerLocked() error { + if o.exitNode == nil { + return nil + } + 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 { + for _, alias := range cfg.Aliases { + o.dnsProxy.RemoveDNSRecordForSite(alias, serverIP, exitNodeAliasSiteId) + } + } + } + + if o.dev != nil { + if err := peers.RemovePeer(o.dev, 0, cfg.PublicKey); err != nil { + logger.Warn("Failed to remove exit node peer: %v", err) + } + } + + interfaceName := o.tunnelConfig.InterfaceName + + if o.middleDev != nil { + if serverAddr, err := netip.ParseAddr(strings.Split(cfg.ServerIP, "/")[0]); err == nil { + o.middleDev.RemoveRule(serverAddr) + } + // Also removes the ICMP connectivity monitor's own rule under this same + // key (pm.ClearExitNode, called just above, already does this too - see + // RemoveRule's doc comment on it clearing every rule for a key rather + // than being handler-specific), so this call is normally a harmless + // no-op by the time it runs; kept for defensiveness/independence from + // that other subsystem's cleanup ordering. + if tunnelAddr, err := netip.ParseAddr(strings.Split(cfg.TunnelIP, "/")[0]); err == nil { + o.middleDev.RemoveRule(tunnelAddr) + } + } + + serverIPForRoute := strings.Split(cfg.ServerIP, "/")[0] + "/32" + tunnelIPForRoute := strings.Split(cfg.TunnelIP, "/")[0] + if err := network.RemoveRouteForServerIPWithSource(serverIPForRoute, interfaceName, tunnelIPForRoute); err != nil { + logger.Warn("Failed to remove route for exit node server IP: %v", err) + } + + tunnelIP := cfg.TunnelIP + if !strings.Contains(tunnelIP, "/") { + tunnelIP += "/32" + } + if err := network.RemoveSecondaryAddress(interfaceName, tunnelIP); err != nil { + logger.Warn("Failed to remove secondary address %s for exit node: %v", tunnelIP, err) + } + + logger.Info("Disconnected from exit node") + return nil +} + +// syncExitNodeConnection reconciles the client's own exit node connection (used +// for site resources hosted on the exit node) with the desired state sent in a +// sync message - connecting, switching, updating aliases, or disconnecting as +// needed. This mirrors what the initial "olm/wg/connect" message does, so a +// client that reconnects with a stale exit node assignment (or none at all) +// converges without needing to fully re-register. +func (o *Olm) syncExitNodeConnection(cfg *ExitNodeConfig) { + if !o.tunnelRunning { + logger.Debug("Tunnel stopped, ignoring exit node sync") + return + } + + if cfg == nil || !cfg.Connect { + if err := o.disconnectExitNode(); err != nil { + logger.Error("Sync: Failed to disconnect from exit node: %v", err) + } + return + } + + if err := o.connectExitNode(*cfg); err != nil { + logger.Error("Sync: Failed to connect to exit node: %v", err) + } +} + +// updateExitNodeAliasesLocked reconciles the currently connected exit node's +// aliases with newAliases, adding new ones before removing stale ones so a +// rename never has a gap in resolution. Must be called with exitNodeMu held. +func (o *Olm) updateExitNodeAliasesLocked(newAliases []string) { + if o.exitNode == nil { + return + } + + added := stringSliceDiff(newAliases, o.exitNode.Aliases) + removed := stringSliceDiff(o.exitNode.Aliases, newAliases) + + serverIP := net.ParseIP(o.exitNode.ServerIP) + if o.dnsProxy != nil && serverIP != nil { + for _, alias := range added { + if err := o.dnsProxy.AddDNSRecord(alias, serverIP, exitNodeAliasSiteId); err != nil { + logger.Warn("Failed to add DNS record for exit node alias %s: %v", alias, err) + } + } + for _, alias := range removed { + o.dnsProxy.RemoveDNSRecordForSite(alias, serverIP, exitNodeAliasSiteId) + } + } + + o.exitNode.Aliases = applyStringListUpdate(o.exitNode.Aliases, removed, added) +} + +// stringSliceDiff returns the elements of a that are not present in b. +func stringSliceDiff(a, b []string) []string { + inB := make(map[string]struct{}, len(b)) + for _, s := range b { + inB[s] = struct{}{} + } + diff := make([]string, 0, len(a)) + for _, s := range a { + if _, ok := inB[s]; !ok { + diff = append(diff, s) + } + } + return diff +} + +// handleExitNodeConnect handles a server-initiated request to connect to (or switch to) +// an exit node, delivered as a full ExitNodeConfig payload. +func (o *Olm) handleExitNodeConnect(msg websocket.WSMessage) { + logger.Debug("Received exit node connect message: %v", msg.Data) + + if !o.tunnelRunning { + logger.Debug("Tunnel stopped, ignoring exit node connect message") + return + } + + // The primary tunnel interface must already be configured before an exit + // node's secondary address can safely be added (see the ordering + // enforcement in connectExitNode) - o.registered is only set true after + // that happens in handleConnect. This guards against a stray/early + // message reaching connectExitNode before then. + if !o.registered { + logger.Debug("Not yet registered, ignoring exit node connect message") + return + } + + jsonData, err := json.Marshal(msg.Data) + if err != nil { + logger.Error("Error marshaling exit node connect data: %v", err) + return + } + + var cfg ExitNodeConfig + if err := json.Unmarshal(jsonData, &cfg); err != nil { + logger.Error("Error unmarshaling exit node connect data: %v", err) + return + } + + if !cfg.Connect { + logger.Debug("Exit node connect message has connect=false, disconnecting instead") + if err := o.disconnectExitNode(); err != nil { + logger.Error("Failed to disconnect from exit node: %v", err) + } + return + } + + if err := o.connectExitNode(cfg); err != nil { + logger.Error("Failed to connect to exit node: %v", err) + } +} + +// handleExitNodeDisconnect handles a server-initiated request to disconnect from the +// currently connected exit node. +func (o *Olm) handleExitNodeDisconnect(msg websocket.WSMessage) { + logger.Debug("Received exit node disconnect message: %v", msg.Data) + + if !o.tunnelRunning { + logger.Debug("Tunnel stopped, ignoring exit node disconnect message") + return + } + + if err := o.disconnectExitNode(); err != nil { + logger.Error("Failed to disconnect from exit node: %v", err) + } +} + +// handleExitNodeUpdateData handles a server-initiated request to change data +// associated with the currently connected exit node, such as its aliases (e.g. a +// resource was renamed). Unlike site aliases, there is no per-alias address to +// track since every exit node alias resolves to the exit node's own ServerIP. +func (o *Olm) handleExitNodeUpdateData(msg websocket.WSMessage) { + logger.Debug("Received exit node update data message: %v", msg.Data) + + if !o.tunnelRunning { + logger.Debug("Tunnel stopped, ignoring exit node update data message") + return + } + + jsonData, err := json.Marshal(msg.Data) + if err != nil { + logger.Error("Error marshaling exit node update data: %v", err) + return + } + + var update ExitNodeUpdateData + if err := json.Unmarshal(jsonData, &update); err != nil { + logger.Error("Error unmarshaling exit node update data: %v", err) + return + } + + o.exitNodeMu.Lock() + defer o.exitNodeMu.Unlock() + + if o.exitNode == nil { + logger.Debug("Ignoring exit node update data message: no exit node connected") + return + } + + serverIP := net.ParseIP(o.exitNode.ServerIP) + + // Add new aliases BEFORE removing old ones, same as site aliases, so a rename + // that keeps the same underlying address never has a gap in resolution. + if o.dnsProxy != nil && serverIP != nil { + for _, alias := range update.NewAliases { + if err := o.dnsProxy.AddDNSRecord(alias, serverIP, exitNodeAliasSiteId); err != nil { + logger.Warn("Failed to add DNS record for exit node alias %s: %v", alias, err) + } + } + } + + if o.dnsProxy != nil && serverIP != nil { + for _, alias := range update.OldAliases { + o.dnsProxy.RemoveDNSRecordForSite(alias, serverIP, exitNodeAliasSiteId) + } + } + + o.exitNode.Aliases = applyStringListUpdate(o.exitNode.Aliases, update.OldAliases, update.NewAliases) + + logger.Info("Successfully updated exit node data") +} + +// applyStringListUpdate returns list with every entry in removed dropped and every +// entry in added appended, preserving the add-before-remove semantics of the caller. +func applyStringListUpdate(list, removed, added []string) []string { + next := make([]string, 0, len(list)+len(added)) + next = append(next, list...) + next = append(next, added...) + + removedSet := make(map[string]struct{}, len(removed)) + for _, alias := range removed { + removedSet[alias] = struct{}{} + } + + filtered := next[:0] + for _, alias := range next { + if _, ok := removedSet[alias]; ok { + continue + } + filtered = append(filtered, alias) + } + return filtered +} diff --git a/olm/olm.go b/olm/olm.go index fa638be..8df0abe 100644 --- a/olm/olm.go +++ b/olm/olm.go @@ -4,9 +4,11 @@ import ( "context" "crypto/rand" "encoding/hex" + "encoding/json" "fmt" "net" "net/http" + "net/netip" _ "net/http/pprof" "os" "os/exec" @@ -15,6 +17,7 @@ import ( "github.com/fosrl/newt/bind" "github.com/fosrl/newt/clients/permissions" + "github.com/fosrl/newt/exitnode" "github.com/fosrl/newt/holepunch" "github.com/fosrl/newt/logger" "github.com/fosrl/newt/network" @@ -55,6 +58,20 @@ type Olm struct { holePunchManager *holepunch.Manager peerManager *peers.PeerManager peerManagerMu sync.RWMutex + + // exitNode tracks the currently connected exit node peer, if any. It lives on a + // secondary address on the same interface/WireGuard device as the site peers. + exitNode *ExitNodeConfig + exitNodeMu sync.Mutex + + // primaryTunnelIP is the site tunnel's own address (wgData.TunnelIP), set once + // per connect in handleConnect. It's the interface's first/primary address - + // on macOS/iOS NetworkExtension, an unbound outbound socket's source gets + // stamped with this address by default even when the traffic should use an + // exit node's secondary address instead (see connectExitNode's NAT setup), + // and inbound replies need to be translated back to it for the OS to match + // them to the socket that's waiting. + primaryTunnelIP netip.Addr // Power mode management currentPowerMode string powerModeMu sync.Mutex @@ -75,6 +92,11 @@ type Olm struct { stopRegister func() updateRegister func(newData any) + // Exit node ping dance, run before registration so the server can pick + // the best exit node (mirrors newt's newt/ping/request flow). + stopPingRequest func() + pendingPingChainId string + stopPeerSends map[string]func() stopPeerInits map[string]func() jitPendingSites map[int]string // siteId -> chainId for in-flight JIT requests @@ -550,6 +572,66 @@ func (o *Olm) StartTunnel(config TunnelConfig) { o.websocket.RegisterHandler("olm/wg/peer/chain/cancel", o.handleCancelChain) o.websocket.RegisterHandler("olm/sync", o.handleSync) + // Handlers for the server to direct connecting/disconnecting an exit node after registration + o.websocket.RegisterHandler("olm/wg/exitnode/connect", o.handleExitNodeConnect) + o.websocket.RegisterHandler("olm/wg/exitnode/disconnect", o.handleExitNodeDisconnect) + o.websocket.RegisterHandler("olm/wg/exitnode/data/update", o.handleExitNodeUpdateData) + + o.websocket.RegisterHandler("olm/ping/exitNodes", func(msg websocket.WSMessage) { + logger.Debug("Received exit node ping request") + + if o.stopPingRequest != nil { + o.stopPingRequest() + o.stopPingRequest = nil + } + + if !o.tunnelRunning { + logger.Debug("Tunnel is no longer running, skipping exit node ping") + return + } + + var exitNodeData exitnode.ExitNodeData + jsonData, err := json.Marshal(msg.Data) + if err != nil { + logger.Error("Error marshaling exit node data: %v", err) + return + } + if err := json.Unmarshal(jsonData, &exitNodeData); err != nil { + logger.Error("Error unmarshaling exit node data: %v", err) + return + } + + if exitNodeData.ChainId != "" { + if exitNodeData.ChainId != o.pendingPingChainId { + logger.Debug("Discarding duplicate/stale olm/ping/exitNodes (chainId=%s, expected=%s)", exitNodeData.ChainId, o.pendingPingChainId) + return + } + o.pendingPingChainId = "" + } + + if len(exitNodeData.ExitNodes) == 0 { + logger.Info("No exit nodes provided") + return + } + + pingResults := exitnode.PingExitNodes(exitNodeData.ExitNodes, "", false) + + publicKey := o.privateKey.PublicKey() + logger.Debug("Sending registration message to server with public key: %s, relay: %v, pingResults: %+v", publicKey, !config.Holepunch, pingResults) + o.stopRegister, o.updateRegister = o.websocket.SendMessageInterval("olm/wg/register", map[string]any{ + "publicKey": publicKey.String(), + "relay": !config.Holepunch, + "olmVersion": o.olmConfig.Version, + "olmAgent": o.olmConfig.Agent, + "orgId": config.OrgID, + "userToken": userToken, + "fingerprint": o.fingerprint, + "postures": o.postures, + "pingResults": pingResults, + "chainId": generateChainId(), // use a random chainId for registration updates - it won't be used for cancellation since registration is a one-time message but for tracking the session + }, 2*time.Second, 20) // after 18 tries on the server side we send the error so dont change this without changing that + }) + o.websocket.OnConnect(func() error { logger.Info("Websocket Connected") @@ -568,8 +650,6 @@ func (o *Olm) StartTunnel(config TunnelConfig) { return nil } - publicKey := o.privateKey.PublicKey() - // delay for 500ms to allow for time for the hp to get processed time.Sleep(500 * time.Millisecond) @@ -579,19 +659,36 @@ func (o *Olm) StartTunnel(config TunnelConfig) { return nil } - if o.stopRegister == nil { - logger.Debug("Sending registration message to server with public key: %s and relay: %v", publicKey, !config.Holepunch) - o.stopRegister, o.updateRegister = o.websocket.SendMessageInterval("olm/wg/register", map[string]any{ - "publicKey": publicKey.String(), - "relay": !config.Holepunch, - "olmVersion": o.olmConfig.Version, - "olmAgent": o.olmConfig.Agent, - "orgId": config.OrgID, - "userToken": userToken, - "fingerprint": o.fingerprint, - "postures": o.postures, - "chainId": generateChainId(), // use a random chainId for registration updates - it won't be used for cancellation since registration is a one-time message but for tracking the session - }, 2*time.Second, 20) // after 18 tries on the server side we send the error so dont change this without changing that + if o.stopRegister == nil && o.stopPingRequest == nil { + publicKey := o.privateKey.PublicKey() + + pingChainId := generateChainId() + o.pendingPingChainId = pingChainId + logger.Debug("Requesting exit nodes from server for ping selection") + o.stopPingRequest, _ = o.websocket.SendMessageInterval("olm/ping/request", map[string]any{ + "chainId": pingChainId, + }, 3*time.Second, 10) + + // Backwards-compatible one-shot registration, with no pingResults, + // for servers that predate the exit node ping dance. Servers that + // support it ignore backwardsCompatible register messages (see + // handleOlmRegisterMessage server-side) and wait for the real + // registration sent from the olm/ping/exitNodes handler above. + bcChainId := generateChainId() + if err := o.websocket.SendMessage("olm/wg/register", map[string]any{ + "publicKey": publicKey.String(), + "relay": !config.Holepunch, + "olmVersion": o.olmConfig.Version, + "olmAgent": o.olmConfig.Agent, + "orgId": config.OrgID, + "userToken": userToken, + "fingerprint": o.fingerprint, + "postures": o.postures, + "backwardsCompatible": true, + "chainId": bcChainId, + }); err != nil { + logger.Error("Failed to send registration message: %v", err) + } // Invoke onRegistered callback if configured if o.olmConfig.OnRegistered != nil { @@ -688,6 +785,12 @@ func (o *Olm) Close() { o.stopRegister = nil } + if o.stopPingRequest != nil { + logger.Debug("Stopping exit node ping request interval") + o.stopPingRequest() + o.stopPingRequest = nil + } + // Stop all pending peer init and send senders before closing websocket o.peerSendMu.Lock() for _, stop := range o.stopPeerInits { @@ -744,6 +847,13 @@ func (o *Olm) Close() { } o.peerManagerMu.Unlock() + // The WireGuard device and TUN interface are being torn down below, which takes + // the exit node peer and its secondary address with them - just clear the + // in-memory record so a stale config isn't reused on the next connect. + o.exitNodeMu.Lock() + o.exitNode = nil + o.exitNodeMu.Unlock() + if o.uapiListener != nil { _ = o.uapiListener.Close() o.uapiListener = nil diff --git a/olm/types.go b/olm/types.go index b5b3a5c..a51f9e9 100644 --- a/olm/types.go +++ b/olm/types.go @@ -10,11 +10,44 @@ type WgData struct { Sites []peers.SiteConfig `json:"sites"` TunnelIP string `json:"tunnelIP"` UtilitySubnet string `json:"utilitySubnet"` // this is for things like the DNS server, and alias addresses + ExitNode *ExitNodeConfig `json:"exitNode,omitempty"` +} + +// ExitNodeConfig describes an exit node the olm client can connect to for +// resources (e.g. inference) hosted on that node, separate from the site +// peers. It lives in a different address space than the site tunnel - the +// client is assigned TunnelIP (within the exit node's subnet) to reach the +// node at ServerIP. It arrives on the initial "olm/wg/connect" message and can +// also be sent later via "olm/wg/exitnode/connect" / "olm/wg/exitnode/disconnect" +// so the server can direct a client to connect/disconnect after registration. +type ExitNodeConfig struct { + Connect bool `json:"connect"` + Endpoint string `json:"endpoint"` + RelayPort uint16 `json:"relayPort"` + PublicKey string `json:"publicKey"` + ServerIP string `json:"serverIP"` + TunnelIP string `json:"tunnelIP"` + Aliases []string `json:"aliases,omitempty"` +} + +// ExitNodeUpdateData describes a change to data associated with the currently +// connected exit node, e.g. when a resource's alias is renamed on the server. +// Aliases have no per-alias address here since every exit node alias resolves +// to the exit node's own ServerIP. More fields can be added here in the +// future as other exit node data becomes updatable. +type ExitNodeUpdateData struct { + OldAliases []string `json:"oldAliases,omitempty"` + NewAliases []string `json:"newAliases,omitempty"` } type SyncData struct { Sites []peers.SiteConfig `json:"sites"` ExitNodes []SyncExitNode `json:"exitNodes"` + // ExitNode is the exit node the client itself is assigned to (for site + // resources hosted on it, e.g. inference), mirroring the ExitNode field + // on WgData sent at registration. It is separate from ExitNodes above, + // which is only the set of exit nodes used for hole punching. + ExitNode *ExitNodeConfig `json:"exitNode,omitempty"` } type SyncExitNode struct { diff --git a/peers/manager.go b/peers/manager.go index 507b3a8..71c0507 100644 --- a/peers/manager.go +++ b/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 } diff --git a/peers/monitor/exitnode.go b/peers/monitor/exitnode.go new file mode 100644 index 0000000..dd0340c --- /dev/null +++ b/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 + } + } +} diff --git a/peers/monitor/monitor.go b/peers/monitor/monitor.go index b4b1fcb..2796b41 100644 --- a/peers/monitor/monitor.go +++ b/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 }