diff --git a/README.md b/README.md index 9342c8b..e3e47f5 100644 --- a/README.md +++ b/README.md @@ -215,16 +215,16 @@ Trust authorization order: ### NAT Traversal -- **STUN**: RFC 5389 Binding Request to discover public endpoint + NAT type -- **Hole punching**: Rendezvous-mediated UDP probing for cone NATs (4 concurrent, 5s timeout) -- **Relay**: Public-IP mesh member forwards WG ciphertext for symmetric NATs +- **STUN**: RFC 5389 Binding Requests across multiple servers to discover public endpoint + NAT mapping type +- **Hole punching**: Rendezvous-mediated, identity/session-bound UDP probing for cone NATs (4 concurrent, 5s timeout) +- **Relay**: Public-IP relay forwards only opaque WireGuard/Noise ciphertext for symmetric NATs ### Wire Protocol - Binary codec: type-tag-delimited, fixed-size fields, little-endian - SWIM: Ping (`0x01`), Ack (`0x03`), PingReq (`0x02`) - Handshake: Standard WireGuard Noise_IKpsk2 (Type 1, Type 2) -- NAT: HolepunchRequest (`0x33`), HolepunchResponse (`0x34`) +- NAT: RelayData (`0x31`), HolepunchRequest (`0x33`), HolepunchResponse (`0x34`) - Org: OrgAliasAnnounce (`0x41`), OrgCertRevoke (`0x42`), OrgTrustVouch (`0x43`) ## Benchmarks diff --git a/docs/concepts/nat-traversal.md b/docs/concepts/nat-traversal.md index 0207a29..9b78e97 100644 --- a/docs/concepts/nat-traversal.md +++ b/docs/concepts/nat-traversal.md @@ -17,13 +17,14 @@ Node STUN Server │ (XOR-MAPPED-ADDRESS) │ ``` -The response contains the **XOR-MAPPED-ADDRESS** attribute — the node's public IP and port as seen by the STUN server. By comparing the local and external ports: +The response contains the **XOR-MAPPED-ADDRESS** attribute — the node's public IP and port as seen by the STUN server. meshguard compares observations from up to two STUN servers: -| Local Port == External Port | NAT Type | Implication | -| --------------------------- | --------- | ------------------------------------------ | -| Yes | `public` | No NAT or full cone — direct connect works | -| No | `cone` | Port-mapped NAT — hole punch will work | -| _(STUN fails)_ | `unknown` | Likely symmetric NAT or firewall | +| Observation | NAT Type | Implication | +| -------------------------------------------- | ----------- | ------------------------------------------ | +| Stable mapping and local port matches | `public` | No NAT or directly reachable mapping | +| Stable mapping and local port differs | `cone` | Endpoint-independent mapping; punch first | +| Mapping changes between STUN servers | `symmetric` | Endpoint-dependent mapping; prefer relay | +| No STUN response | `unknown` | Firewall or undetermined | The discovered public endpoint is then shared via gossip, so other peers know how to reach this node. @@ -69,10 +70,10 @@ Node A (NATed) Rendezvous (Public) Node B (NATed) ### Details - **Rendezvous selection**: Any mutual public-IP peer in the membership table -- **Probe magic**: `MGHP` (`0x4D 0x47 0x48 0x50`) — 4-byte packet recognized by `Holepuncher.isProbe()` +- **Probe magic**: `MGHP` (`0x4D 0x47 0x48 0x50`) followed by the 16-byte punch token - **Probe timing**: Every 200ms, up to 25 probes (5-second timeout) - **Concurrency**: Up to 4 concurrent hole punch attempts -- **Token verification**: Random 16-byte nonce prevents spoofing +- **Token verification**: Random 16-byte nonce is bound to the initiator/target identities and echoed in probes The `meshguard connect` token-exchange command uses a separate coordinated punch path in `coordinated_punch.zig`. Its raw probe magic is `MGCP` and the @@ -84,22 +85,26 @@ Hole punching works for **endpoint-independent mapping** (cone NAT). It fails fo ## Tier 3: Relay Fallback -When hole punching fails, a **public-IP mesh member** serves as a relay: +When hole punching fails, a **public-IP relay** forwards an opaque relay frame: ``` -Node A (NATed) ←─WG─→ Relay (public) ←─WG─→ Node B (NATed) +Node A (NATed) -- RelayData(sender=A,target=B,payload=WG bytes) --> Relay +Relay -- RelayData(sender=A,target=B,payload=WG bytes) --> Node B ``` -Since WireGuard provides end-to-end encryption, the relay only handles ciphertext. No special relay protocol is needed — the relay is simply a WireGuard peer of both NATed nodes. +The relay frame carries routing metadata plus a WireGuard/Noise packet payload (message types 1-4). The relay validates that the payload is shaped like a WireGuard packet, rate-limits by identity, and forwards the bytes unchanged. It is never a WireGuard peer for the relayed tunnel and never receives plaintext or authorization authority. + +Hosted rendezvous registration is identity-authenticated: the relay issues a nonce, and the node signs `(identity pubkey, endpoint, nonce)` with its Ed25519 identity key before the relay stores the identity -> observed endpoint mapping. ### Relay Selection -The `relay.zig` module selects the best relay candidate: +The `relay.zig` module selects the best relay candidate and exposes the hosted relay/rendezvous core: 1. Must be **alive** in the membership table 2. Must be a **public** NAT type 3. Must be **relay-capable** (not at capacity — default max: 10 relay connections) 4. Prefer **lowest RTT** (measured by SWIM ping round-trips) +5. Prefer direct path, then hole punch, then relay ```zig pub fn selectRelay( @@ -108,6 +113,12 @@ pub fn selectRelay( ) ?*const Membership.Peer ``` +Relay frames use wire type `0x31`: + +``` +[0x31][32B sender pubkey][32B target pubkey][2B payload length][WireGuard packet bytes] +``` + ### NAT Type Classification Each node's NAT type is broadcast via gossip, so the mesh knows which peers need relaying: diff --git a/docs/concepts/wire-protocol.md b/docs/concepts/wire-protocol.md index 69d5abe..9aac68a 100644 --- a/docs/concepts/wire-protocol.md +++ b/docs/concepts/wire-protocol.md @@ -18,25 +18,26 @@ SWIM messages use a **1-byte type tag**: ## SWIM / Protocol Codec Tags -The 1-byte tags below are handled by `protocol/codec.zig` after packet -classification has ruled out WireGuard and STUN: +The 1-byte tags below are handled after packet classification has ruled out +WireGuard and STUN. SWIM and hole-punch messages use `protocol/codec.zig`; +`RelayData` uses the ciphertext-only helpers in `nat/relay.zig`. | Tag | Name | Category | Direction | | ------ | ----------------- | --------- | --------------- | | `0x01` | Ping | SWIM | A → B | | `0x02` | PingReq | SWIM | A → C (probe B) | | `0x03` | Ack | SWIM | B → A | +| `0x31` | RelayData | NAT | A → Relay → B | | `0x33` | HolepunchRequest | NAT | A → Rendezvous | | `0x34` | HolepunchResponse | NAT | B → Rendezvous | | `0x41` | OrgAliasAnnounce | Org Trust | Gossip | | `0x42` | OrgCertRevoke | Org Trust | Gossip | | `0x43` | OrgTrustVouch | Org Trust | Gossip | -`messages.zig` reserves additional enum values for future protocol messages, -but the codec currently decodes the tags listed above. WireGuard handshake, -cookie, and transport packets are classified by their 4-byte WireGuard type -(`1`-`4`), not by this 1-byte table. The FFI app-message path uses `0x50` -outside this codec. +`messages.zig` reserves additional enum values for future protocol messages. +WireGuard handshake, cookie, and transport packets are classified by their +4-byte WireGuard type (`1`-`4`), not by this 1-byte table. The FFI app-message +path uses `0x50` outside this codec. ## Ping @@ -119,6 +120,20 @@ Standard Noise_IKpsk2 response message. Total: **92 bytes**. +## RelayData + +Opaque relay frame for WireGuard/Noise packets when direct and punched paths are +unavailable. The relay routes by identity metadata and forwards the payload +unchanged; it is not a WireGuard peer for the relayed tunnel. + +``` +[0x31][32B sender_pubkey][32B target_pubkey][2B payload_len (BE)][N WireGuard packet bytes] +``` + +The payload must be shaped like WireGuard message type `1`, `2`, `3`, or `4` +using the standard little-endian WireGuard type field. Non-WireGuard payloads +are rejected by the relay frame decoder. + ## HolepunchRequest ``` @@ -135,6 +150,17 @@ Total: **101 bytes**. Total: **69 bytes**. +## HolepunchProbe + +Raw probe packet, not decoded by `protocol/codec.zig`: + +``` +["MGHP"][16B punch token] +``` + +The token is the same identity/session-bound nonce from the corresponding +HolepunchRequest/HolepunchResponse exchange. + ## OrgAliasAnnounce Propagated via gossip to claim a human-readable `*.name.mesh` domain for an org. diff --git a/docs/reference/modules.md b/docs/reference/modules.md index 2d092c6..3efda3d 100644 --- a/docs/reference/modules.md +++ b/docs/reference/modules.md @@ -47,9 +47,9 @@ Reference map of all source modules and their responsibilities. | File | Purpose | | ----------------------- | ---------------------------------------------------------------------------------------------------------------------- | -| `stun.zig` | STUN client (RFC 5389): Binding Request/Response encoding, XOR-MAPPED-ADDRESS parsing, NAT type detection | -| `holepunch.zig` | UDP hole punching: `Holepuncher` state machine, probe magic (`MGHP`), rendezvous-mediated exchange, 4 concurrent slots | -| `relay.zig` | Relay selection: best public-IP peer by RTT, capacity checking, `RelayInfo` struct | +| `stun.zig` | STUN client (RFC 5389): Binding Request/Response encoding, XOR-MAPPED-ADDRESS parsing, multi-server NAT classification | +| `holepunch.zig` | UDP hole punching: `Holepuncher` state machine, token-bound `MGHP` probes, rendezvous-mediated exchange | +| `relay.zig` | Ciphertext-only relay/rendezvous core: signed endpoint registration, relay frame codec, per-identity rate limiting | | `upnp.zig` | UPnP-IGD port forwarding: SSDP discovery, SOAP AddPortMapping, lease renewal | | `coordinated_punch.zig` | Token-based coordinated punch: `meshguard connect` token exchange for direct peer setup | diff --git a/src/discovery/swim.zig b/src/discovery/swim.zig index 43d7fcd..73d5aae 100644 --- a/src/discovery/swim.zig +++ b/src/discovery/swim.zig @@ -16,6 +16,7 @@ const keys = @import("../identity/keys.zig"); const Org = @import("../identity/org.zig"); const Udp = @import("../net/udp.zig"); const Holepuncher = @import("../nat/holepunch.zig").Holepuncher; +const Relay = @import("../nat/relay.zig"); const log = std.log.scoped(.swim); // Max amount a single (possibly forged, unauthenticated) gossip entry may advance @@ -75,6 +76,7 @@ pub const EventHandler = struct { onPeerPunched: ?*const fn (ctx: *anyopaque, peer: *const Membership.Peer, endpoint: messages.Endpoint) void = null, onAppMessage: ?*const fn (ctx: *anyopaque, data: []const u8) void = null, onWgPacket: ?*const fn (ctx: *anyopaque, data: []const u8, addr: [4]u8, port: u16) void = null, + onRelayWgPacket: ?*const fn (ctx: *anyopaque, data: []const u8, sender_pubkey: [32]u8, relay_endpoint: messages.Endpoint) void = null, /// Query whether a live WireGuard session currently exists for `wg_pubkey`. /// Used as an origin-authentication signal for restart detection (S2): SWIM /// ping/ack are not origin-authenticated, but holding a live tunnel proves the @@ -200,6 +202,7 @@ pub const SwimProtocol = struct { // Hole punching coordinator holepuncher: Holepuncher = .{}, + relay_limiter: Relay.IdentityRateLimiter = .{}, last_punch_check_ns: i128 = 0, last_gossip_ns: i128 = 0, @@ -406,6 +409,15 @@ pub const SwimProtocol = struct { return false; } + pub fn isKnownRelayParticipant(self: *const SwimProtocol, pubkey: [32]u8) bool { + if (std.mem.eql(u8, &pubkey, &self.our_pubkey)) return true; + return self.membership.peers.contains(pubkey); + } + + pub fn isAllowedRelayParticipant(self: *const SwimProtocol, pubkey: [32]u8) bool { + return self.isKnownRelayParticipant(pubkey) and self.isAuthorizedPeer(pubkey, null); + } + /// Check if a peer is authorized via org cert (checking peer's org cert against trusted orgs). pub fn isOrgAuthorizedPeer(self: *const SwimProtocol, org_pubkey: [32]u8) bool { for (self.trusted_orgs[0..self.trusted_org_count]) |trusted| { @@ -485,7 +497,7 @@ pub const SwimProtocol = struct { if (try self.socket.pollRead(poll_ms)) { // Drain all available messages while (true) { - var recv_buf: [1500]u8 = undefined; + var recv_buf: [Relay.MAX_RELAY_DATAGRAM]u8 = undefined; const result = try self.socket.recvFrom(&recv_buf); if (result == null) break; const recv = result.?; @@ -632,7 +644,7 @@ pub const SwimProtocol = struct { const poll_ms: i32 = 200; if (try self.socket.pollRead(poll_ms)) { while (true) { - var recv_buf: [1500]u8 = undefined; + var recv_buf: [Relay.MAX_RELAY_DATAGRAM]u8 = undefined; const result = try self.socket.recvFrom(&recv_buf); if (result == null) break; const recv = result.?; @@ -674,8 +686,13 @@ pub const SwimProtocol = struct { fn handleMessage(self: *SwimProtocol, data: []const u8, sender_endpoint: messages.Endpoint) void { // Check for holepunch probes (raw UDP with MGHP magic, not SWIM-encoded) - if (data.len >= 4 and Holepuncher.isProbe(data[0..4])) { - self.handleHolepunchProbe(sender_endpoint); + if (Holepuncher.decodeProbe(data)) |probe| { + self.handleHolepunchProbe(probe, sender_endpoint); + return; + } + + if (data.len > 0 and data[0] == @intFromEnum(messages.MessageType.relay_data)) { + self.handleRelayData(data, sender_endpoint); return; } @@ -738,6 +755,29 @@ pub const SwimProtocol = struct { } } + fn handleRelayData(self: *SwimProtocol, data: []const u8, sender_endpoint: messages.Endpoint) void { + const frame = Relay.decodeRelayData(data) catch return; + if (!self.isAllowedRelayParticipant(frame.sender_pubkey)) return; + if (!self.isAllowedRelayParticipant(frame.target_pubkey)) return; + if (!self.relay_limiter.allow(frame.sender_pubkey, nowNs())) return; + + if (std.mem.eql(u8, &frame.target_pubkey, &self.our_pubkey)) { + if (self.handler) |h| { + if (h.onRelayWgPacket) |cb| { + cb(h.ctx, frame.payload, frame.sender_pubkey, sender_endpoint); + } else if (h.onWgPacket) |cb| { + const msg_type = if (frame.payload.len >= 4) std.mem.readInt(u32, frame.payload[0..4], .little) else 0; + if (msg_type == 4) cb(h.ctx, frame.payload, sender_endpoint.addr, sender_endpoint.port); + } + } + return; + } + + const peer = self.membership.peers.get(frame.target_pubkey) orelse return; + const ep = peer.gossip_endpoint orelse peer.public_endpoint orelse return; + self.gossipSend(data, ep); + } + /// Handle an incoming 0x50 app message: deliver locally or relay. /// Wire: [0x50][32B dest_pubkey][32B sender_pubkey][12B nonce][N ciphertext][16B tag] fn handleAppMessage(self: *SwimProtocol, data: []const u8, sender_endpoint: messages.Endpoint) void { @@ -1010,14 +1050,7 @@ pub const SwimProtocol = struct { self.gossipSend(buf[0..written], sender_endpoint); // Also start probing the initiator's endpoint - _ = self.holepuncher.initiate(self.our_pubkey, req.sender_pubkey, our_ep); - if (self.holepuncher.handleResponse(messages.HolepunchResponse{ - .sender_pubkey = req.sender_pubkey, - .public_endpoint = req.public_endpoint, - .token_echo = req.token, - })) |_| { - // Probing will happen in tick() - } + _ = self.holepuncher.acceptRequest(self.our_pubkey, req, our_ep); } } else { // We're the rendezvous — forward to the target @@ -1039,26 +1072,19 @@ pub const SwimProtocol = struct { } } - fn handleHolepunchProbe(self: *SwimProtocol, sender_endpoint: messages.Endpoint) void { + fn handleHolepunchProbe(self: *SwimProtocol, probe: @import("../nat/holepunch.zig").Probe, sender_endpoint: messages.Endpoint) void { // A probe arrived — the hole is punched! - // Find which peer this corresponds to by checking active punches - // For now, check all peers with matching public endpoints - var iter = self.membership.peers.iterator(); - while (iter.next()) |entry| { - const peer = entry.value_ptr; - if (peer.public_endpoint) |pub_ep| { - if (pub_ep.eql(sender_endpoint)) { - // Found the peer — notify handler to configure WG endpoint - var ep_buf: [64]u8 = undefined; - std.debug.print(" [punch] hole punched with {x:0>2}{x:0>2}... at {s}\n", .{ - peer.pubkey[0], peer.pubkey[1], sender_endpoint.format(&ep_buf), - }); - if (self.handler) |h| { - if (h.onPeerPunched) |callback| { - callback(h.ctx, peer, sender_endpoint); - } + // Bind success to the active session token and expected peer endpoint. + if (self.holepuncher.markProbeSuccess(probe, sender_endpoint)) |success| { + if (self.membership.peers.getPtr(success.peer_pubkey)) |peer| { + var ep_buf: [64]u8 = undefined; + std.debug.print(" [punch] hole punched with {x:0>2}{x:0>2}... at {s}\n", .{ + peer.pubkey[0], peer.pubkey[1], success.endpoint.format(&ep_buf), + }); + if (self.handler) |h| { + if (h.onPeerPunched) |callback| { + callback(h.ctx, peer, success.endpoint); } - return; } } } @@ -1861,12 +1887,22 @@ const JoinCounter = struct { last_pubkey: ?[32]u8 = null, }; +const RelayDeliveryCounter = struct { + count: usize = 0, + payload: [64]u8 = .{0} ** 64, + payload_len: usize = 0, + addr: [4]u8 = .{0} ** 4, + port: u16 = 0, +}; + fn countPeerJoin(ctx: *anyopaque, peer: *const Membership.Peer) void { const counter: *JoinCounter = @ptrCast(@alignCast(ctx)); counter.count += 1; counter.last_pubkey = peer.pubkey; } +fn ignorePeerJoin(_: *anyopaque, _: *const Membership.Peer) void {} + fn ignorePeerDead(_: *anyopaque, _: [32]u8) void {} fn countPeerDead(ctx: *anyopaque, _: [32]u8) void { @@ -1874,6 +1910,15 @@ fn countPeerDead(ctx: *anyopaque, _: [32]u8) void { counter.dead_count += 1; } +fn countRelayWgPacket(ctx: *anyopaque, data: []const u8, addr: [4]u8, port: u16) void { + const counter: *RelayDeliveryCounter = @ptrCast(@alignCast(ctx)); + counter.count += 1; + counter.payload_len = @min(data.len, counter.payload.len); + @memcpy(counter.payload[0..counter.payload_len], data[0..counter.payload_len]); + counter.addr = addr; + counter.port = port; +} + fn inertTestSocket() Udp.UdpSocket { const fd: std.posix.socket_t = if (builtin.os.tag == .windows) @as(std.posix.socket_t, @ptrFromInt(std.math.maxInt(usize))) @@ -1903,6 +1948,75 @@ fn issueDelegatedCertV2Wire(org_kp: Org.OrgKeyPair, issuer_kp: Org.OrgKeyPair, n return cert_wire; } +test "relay data addressed to us is delivered as opaque WG packet" { + const allocator = std.testing.allocator; + var membership = Membership.MembershipTable.init(allocator, 5000); + defer membership.deinit(); + + var delivery = RelayDeliveryCounter{}; + const our_pubkey = [_]u8{0x11} ** 32; + const sender_pubkey = [_]u8{0x22} ** 32; + try membership.upsert(.{ + .pubkey = sender_pubkey, + .name = "", + .state = .alive, + .gossip_endpoint = messages.Endpoint.initV4(.{ 198, 51, 100, 9 }, 40000), + .wg_pubkey = null, + .mesh_ip = .{ 10, 99, 0x22, 0x22 }, + .mesh_ip6 = .{0} ** 16, + .wg_port = 51830, + .lamport = 1, + .last_seen_ns = nowNs(), + .suspected_at_ns = null, + .last_rtt_ns = null, + .handshake_complete = false, + }); + const handler = EventHandler{ + .ctx = &delivery, + .onPeerJoin = ignorePeerJoin, + .onPeerDead = ignorePeerDead, + .onWgPacket = countRelayWgPacket, + }; + var swim = SwimProtocol.init(&membership, inertTestSocket(), .{}, our_pubkey, [_]u8{0x33} ** 32, .{ 127, 0, 0, 1 }, 51821, handler); + + var wg_payload = [_]u8{0} ** 32; + std.mem.writeInt(u32, wg_payload[0..4], 4, .little); + var frame: [Relay.RELAY_FRAME_HEADER_SIZE + wg_payload.len]u8 = undefined; + const frame_len = try Relay.encodeRelayData(&frame, sender_pubkey, our_pubkey, &wg_payload); + + swim.feedPacketEndpoint(frame[0..frame_len], messages.Endpoint.initV4(.{ 198, 51, 100, 9 }, 40000)); + + try std.testing.expectEqual(@as(usize, 1), delivery.count); + try std.testing.expectEqualSlices(u8, &wg_payload, delivery.payload[0..delivery.payload_len]); + try std.testing.expectEqual([4]u8{ 198, 51, 100, 9 }, delivery.addr); + try std.testing.expectEqual(@as(u16, 40000), delivery.port); +} + +test "relay data from unknown sender is ignored" { + const allocator = std.testing.allocator; + var membership = Membership.MembershipTable.init(allocator, 5000); + defer membership.deinit(); + + var delivery = RelayDeliveryCounter{}; + const our_pubkey = [_]u8{0x11} ** 32; + const unknown_sender = [_]u8{0x44} ** 32; + const handler = EventHandler{ + .ctx = &delivery, + .onPeerJoin = ignorePeerJoin, + .onPeerDead = ignorePeerDead, + .onWgPacket = countRelayWgPacket, + }; + var swim = SwimProtocol.init(&membership, inertTestSocket(), .{}, our_pubkey, [_]u8{0x33} ** 32, .{ 127, 0, 0, 1 }, 51821, handler); + + var wg_payload = [_]u8{0} ** 32; + std.mem.writeInt(u32, wg_payload[0..4], 4, .little); + var frame: [Relay.RELAY_FRAME_HEADER_SIZE + wg_payload.len]u8 = undefined; + const frame_len = try Relay.encodeRelayData(&frame, unknown_sender, our_pubkey, &wg_payload); + + swim.feedPacketEndpoint(frame[0..frame_len], messages.Endpoint.initV4(.{ 198, 51, 100, 9 }, 40000)); + try std.testing.expectEqual(@as(usize, 0), delivery.count); +} + fn issueVouch(org_kp: Org.OrgKeyPair, node_pubkey: [32]u8, lamport: u64) messages.OrgTrustVouch { var vouch = messages.OrgTrustVouch{ .org_pubkey = org_kp.public_key.toBytes(), diff --git a/src/main.zig b/src/main.zig index e8dfa84..b741b7e 100644 --- a/src/main.zig +++ b/src/main.zig @@ -2623,6 +2623,7 @@ fn flushPeerTxRing( /// Used only when encrypt_workers == 0 (legacy mode). fn dataPlaneWorker( running: *const std.atomic.Value(bool), + swim: *lib.discovery.Swim.SwimProtocol, wg_dev: *lib.wireguard.Device.WgDevice, tun_fd: posix.fd_t, udp_fd: posix.fd_t, @@ -2634,6 +2635,7 @@ fn dataPlaneWorker( var tx = BatchUdp.BatchSender{}; var tun_buf: [Offload.VNET_HDR_LEN + 65535]u8 align(@alignOf(Offload.VirtioNetHdr)) = undefined; var encrypt_bufs: [BatchUdp.BATCH_SIZE][1600]u8 = undefined; + var relay_bufs: [BatchUdp.BATCH_SIZE][lib.nat.Relay.RELAY_FRAME_HEADER_SIZE + lib.nat.Relay.MAX_RELAY_PAYLOAD]u8 = undefined; var seg_bufs: [BatchUdp.BATCH_SIZE][1500]u8 = undefined; var seg_slices: [BatchUdp.BATCH_SIZE][]u8 = undefined; for (0..BatchUdp.BATCH_SIZE) |i| { @@ -2670,18 +2672,18 @@ fn dataPlaneWorker( ); for (0..seg_count) |s| { if (send_idx >= BatchUdp.BATCH_SIZE) break; - encryptAndQueue(wg_dev, seg_bufs[s][0..seg_sizes[s]], &encrypt_bufs[send_idx], &tx, &send_idx); + encryptAndQueue(swim, wg_dev, seg_bufs[s][0..seg_sizes[s]], &encrypt_bufs[send_idx], &relay_bufs[send_idx], &tx, &send_idx); } } else { if (ip_data.len < 20) continue; const mutable_data = tun_buf[Offload.VNET_HDR_LEN..n]; Offload.completeChecksum(vhdr.*, mutable_data); - encryptAndQueue(wg_dev, mutable_data, &encrypt_bufs[send_idx], &tx, &send_idx); + encryptAndQueue(swim, wg_dev, mutable_data, &encrypt_bufs[send_idx], &relay_bufs[send_idx], &tx, &send_idx); } } else { if (n < 20) continue; const ip_packet = tun_buf[0..n]; - encryptAndQueue(wg_dev, ip_packet, &encrypt_bufs[send_idx], &tx, &send_idx); + encryptAndQueue(swim, wg_dev, ip_packet, &encrypt_bufs[send_idx], &relay_bufs[send_idx], &tx, &send_idx); } } @@ -2692,9 +2694,11 @@ fn dataPlaneWorker( /// Encrypt an IP packet and queue it for batch sending. fn encryptAndQueue( + swim: *lib.discovery.Swim.SwimProtocol, wg_dev: *lib.wireguard.Device.WgDevice, ip_packet: []const u8, encrypt_buf: *[1600]u8, + relay_buf: *[lib.nat.Relay.RELAY_FRAME_HEADER_SIZE + lib.nat.Relay.MAX_RELAY_PAYLOAD]u8, tx: *lib.net.BatchUdp.BatchSender, send_idx: *usize, ) void { @@ -2702,16 +2706,50 @@ fn encryptAndQueue( const target_slot = wg_dev.lookupByMeshIp(dst_ip); if (target_slot) |slot| { if (wg_dev.encryptForPeer(slot, ip_packet, encrypt_buf)) |enc_len| { - if (wg_dev.peers[slot]) |peer| { - if (peer.endpoint_addr[0] != 0) { - tx.queue(encrypt_buf[0..enc_len], peer.endpoint_addr, peer.endpoint_port); - send_idx.* += 1; - } - } + queueWgCiphertextForSlot(swim, wg_dev, slot, encrypt_buf[0..enc_len], relay_buf, tx, send_idx); } else |_| {} } } +fn queueWgCiphertextForSlot( + swim: *lib.discovery.Swim.SwimProtocol, + wg_dev: *lib.wireguard.Device.WgDevice, + slot: usize, + payload: []const u8, + relay_buf: *[lib.nat.Relay.RELAY_FRAME_HEADER_SIZE + lib.nat.Relay.MAX_RELAY_PAYLOAD]u8, + tx: *lib.net.BatchUdp.BatchSender, + send_idx: *usize, +) void { + const peer = wg_dev.peers[slot] orelse return; + + var target_nat: lib.protocol.Messages.NatType = .unknown; + var relay_ep: ?@import("protocol/messages.zig").Endpoint = null; + swim.membership.lock.lockSharedUncancelable(zio()); + if (swim.membership.peers.get(peer.identity_key)) |member| target_nat = member.nat_type; + relay_ep = relayEndpointForPeer(swim, peer.identity_key); + swim.membership.lock.unlockShared(zio()); + + const relay_available = relay_ep != null; + if (peer.endpoint()) |direct_ep| { + const path = lib.nat.Relay.choosePath(swim.our_nat_type, target_nat, true, relay_available); + if (path == .direct or path == .holepunch or !relay_available) { + if (direct_ep.addr6 == null) { + tx.queue(payload, direct_ep.addr, direct_ep.port); + send_idx.* += 1; + return; + } + if (!relay_available) return; + } + } + + if (relay_ep) |ep| { + if (ep.addr6 != null) return; + const relay_len = lib.nat.Relay.encodeRelayData(relay_buf, swim.our_pubkey, peer.identity_key, payload) catch return; + tx.queue(relay_buf[0..relay_len], ep.addr, ep.port); + send_idx.* += 1; + } +} + /// Write decrypted packets to TUN, coalescing consecutive same-flow TCP /// segments into GSO super-packets to reduce TUN write syscalls. /// @@ -2845,6 +2883,126 @@ fn orgPubkeyLocked(membership: *lib.discovery.Membership.MembershipTable, identi return if (membership.peers.getPtr(identity_key)) |mp| mp.org_pubkey else null; } +fn relayEndpointForPeer( + swim: *lib.discovery.Swim.SwimProtocol, + target_pubkey: [32]u8, +) ?@import("protocol/messages.zig").Endpoint { + const relay = lib.nat.Relay.selectRelayForPair(&swim.membership.peers, swim.our_pubkey, target_pubkey) orelse return null; + return relay.gossip_endpoint orelse relay.public_endpoint; +} + +fn sendRelayFrameToEndpoint( + swim: *lib.discovery.Swim.SwimProtocol, + udp_sock: *lib.net.Udp.UdpSocket, + target_pubkey: [32]u8, + payload: []const u8, + relay_endpoint: @import("protocol/messages.zig").Endpoint, +) bool { + var relay_buf: [lib.nat.Relay.RELAY_FRAME_HEADER_SIZE + lib.nat.Relay.MAX_RELAY_PAYLOAD]u8 = undefined; + const relay_len = lib.nat.Relay.encodeRelayData(&relay_buf, swim.our_pubkey, target_pubkey, payload) catch return false; + _ = udp_sock.sendToEndpoint(relay_buf[0..relay_len], relay_endpoint) catch return false; + return true; +} + +fn sendWgCiphertextToPeer( + swim: *lib.discovery.Swim.SwimProtocol, + wg_dev: *lib.wireguard.Device.WgDevice, + udp_sock: *lib.net.Udp.UdpSocket, + target_pubkey: [32]u8, + payload: []const u8, + preferred_relay: ?@import("protocol/messages.zig").Endpoint, +) bool { + if (preferred_relay) |ep| { + return sendRelayFrameToEndpoint(swim, udp_sock, target_pubkey, payload, ep); + } + + const relay_ep = preferred_relay orelse relayEndpointForPeer(swim, target_pubkey); + const relay_available = relay_ep != null; + + if (wg_dev.findByIdentity(target_pubkey)) |slot| { + if (wg_dev.peers[slot]) |peer| { + if (peer.endpoint()) |direct_ep| { + const path: lib.nat.Relay.RelayPath = if (swim.membership.peers.get(target_pubkey)) |member| + lib.nat.Relay.choosePath(swim.our_nat_type, member.nat_type, true, relay_available) + else + .direct; + if (path == .direct or path == .holepunch or !relay_available) { + _ = udp_sock.sendToEndpoint(payload, direct_ep) catch return false; + return true; + } + } + } + } + + if (relay_ep) |ep| { + return sendRelayFrameToEndpoint(swim, udp_sock, target_pubkey, payload, ep); + } + return false; +} + +fn sendWgCiphertextForSlot( + swim: *lib.discovery.Swim.SwimProtocol, + wg_dev: *lib.wireguard.Device.WgDevice, + udp_sock: *lib.net.Udp.UdpSocket, + slot: usize, + payload: []const u8, +) bool { + const peer = wg_dev.peers[slot] orelse return false; + return sendWgCiphertextToPeer(swim, wg_dev, udp_sock, peer.identity_key, payload, null); +} + +fn processRelayedWgPacket( + frame: lib.nat.Relay.DecodedRelayData, + relay_endpoint: @import("protocol/messages.zig").Endpoint, + wg_dev: *lib.wireguard.Device.WgDevice, + swim: *lib.discovery.Swim.SwimProtocol, + udp_sock: *lib.net.Udp.UdpSocket, + stdout: std.Io.File, + decrypt_storage: *[64][1500]u8, + decrypt_lens: *[64]usize, + decrypt_slots: *[64]usize, + n_decrypted: *usize, + service_filter: *const lib.services.Policy.ServiceFilter, +) void { + if (!swim.isAllowedRelayParticipant(frame.sender_pubkey)) return; + if (!swim.isAllowedRelayParticipant(frame.target_pubkey)) return; + if (!swim.relay_limiter.allow(frame.sender_pubkey, nowAwakeNs())) return; + + const Device = lib.wireguard.Device; + switch (Device.PacketType.classify(frame.payload)) { + .wg_transport => { + if (n_decrypted.* < 64) { + if (wg_dev.decryptTransport(frame.payload, &decrypt_storage[n_decrypted.*])) |result| { + const org_pk = orgPubkeyLocked(swim.membership, result.identity_key); + if (!service_filter.allowPacket(result.identity_key, org_pk, decrypt_storage[n_decrypted.*][0..result.len])) return; + decrypt_lens[n_decrypted.*] = result.len; + decrypt_slots[n_decrypted.*] = result.slot; + n_decrypted.* += 1; + } else |_| {} + } + }, + .wg_handshake_init => { + if (frame.payload.len >= @sizeOf(lib.wireguard.Noise.HandshakeInitiation)) { + const msg: *const lib.wireguard.Noise.HandshakeInitiation = @ptrCast(@alignCast(frame.payload.ptr)); + if (wg_dev.handleInitiation(msg, relay_endpoint.addr)) |hs_result| { + const resp_bytes = std.mem.asBytes(&hs_result.response); + _ = sendWgCiphertextToPeer(swim, wg_dev, udp_sock, frame.sender_pubkey, resp_bytes, relay_endpoint); + writeFormatted(stdout, " WG handshake: responded via relay\n", .{}) catch {}; + } else |_| {} + } + }, + .wg_handshake_resp => { + if (frame.payload.len >= @sizeOf(lib.wireguard.Noise.HandshakeResponse)) { + const msg: *const lib.wireguard.Noise.HandshakeResponse = @ptrCast(@alignCast(frame.payload.ptr)); + if (wg_dev.handleResponse(msg)) |_| { + writeFormatted(stdout, " WG handshake: completed via relay\n", .{}) catch {}; + } else |_| {} + } + }, + .wg_cookie, .stun, .swim, .unknown => {}, + } +} + /// Parallel decrypt worker: pulls encrypted transport packets from the DecryptQueue, /// decrypts them using wg_dev.decryptTransport (thread-safe via replay_lock), and writes /// plaintext to TUN. This parallelizes the download path across N cores. @@ -2910,6 +3068,32 @@ fn processIncomingPacket( ) void { const Device = lib.wireguard.Device; + if (pkt.len > 0 and pkt[0] == @intFromEnum(lib.protocol.Messages.MessageType.relay_data)) { + const relay_endpoint = @import("protocol/messages.zig").Endpoint.initV4(sender_addr, sender_port); + if (lib.nat.Relay.decodeRelayData(pkt)) |frame| { + if (std.mem.eql(u8, &frame.target_pubkey, &swim.our_pubkey)) { + processRelayedWgPacket( + frame, + relay_endpoint, + wg_dev, + swim, + udp_sock, + stdout, + decrypt_storage, + decrypt_lens, + decrypt_slots, + n_decrypted, + service_filter, + ); + } else { + swim.feedPacket(pkt, sender_addr, sender_port); + } + } else |_| { + swim.feedPacket(pkt, sender_addr, sender_port); + } + return; + } + const pkt_type = Device.PacketType.classify(pkt); // Optimization: Extract dominant data-plane case to explicit if branch if (pkt_type == .wg_transport) { @@ -2974,12 +3158,11 @@ fn windowsEventLoop( service_filter: *const lib.services.Policy.ServiceFilter, control_socket: *lib.services.Control.ControlSocket, ) !void { - const Device = lib.wireguard.Device; - const Noise = lib.wireguard.Noise; - var tun_buf: [65536]u8 = undefined; - var udp_recv_buf: [2048]u8 = undefined; - var decrypt_buf: [1500]u8 = undefined; + var udp_recv_buf: [lib.nat.Relay.MAX_RELAY_DATAGRAM]u8 = undefined; + var decrypt_storage: [64][1500]u8 = undefined; + var decrypt_lens: [64]usize = undefined; + var decrypt_slots: [64]usize = undefined; var encrypt_buf: [2048]u8 = undefined; // 16B header + 1500B payload + 16B poly1305 tag var last_handshake_check_ns: i128 = 0; @@ -2987,52 +3170,28 @@ fn windowsEventLoop( // ─── 1. Process incoming UDP packets (WG + SWIM multiplexed) ─── // Drain up to 64 packets per iteration — fully non-blocking var udp_count: u32 = 0; + var n_decrypted: usize = 0; while (udp_count < 64) : (udp_count += 1) { const recv = (udp_sock.recvFrom(&udp_recv_buf) catch break) orelse break; const pkt = recv.data; - const pkt_type = Device.PacketType.classify(pkt); - // Optimization: Extract dominant data-plane case to explicit if branch - if (pkt_type == .wg_transport) { - // Decrypt WG transport → write plaintext to Wintun - if (wg_dev.decryptTransport(pkt, &decrypt_buf)) |result| { - // Apply service filter before writing to TUN (IPv4 + IPv6, M5). - { - const org_pk = orgPubkeyLocked(swim.membership, result.identity_key); - if (!service_filter.allowPacket(result.identity_key, org_pk, decrypt_buf[0..result.len])) continue; - } - tun_dev.write(decrypt_buf[0..result.len]) catch {}; - } else |_| {} - } else switch (pkt_type) { - .wg_handshake_init => { - if (pkt.len >= @sizeOf(Noise.HandshakeInitiation)) { - const msg: *const Noise.HandshakeInitiation = @ptrCast(@alignCast(pkt.ptr)); - if (wg_dev.handleInitiation(msg, recv.sender_addr)) |hs_result| { - const resp_bytes = std.mem.asBytes(&hs_result.response); - _ = udp_sock.sendTo(resp_bytes, recv.sender_addr, recv.sender_port) catch 0; - writeFormatted(stdout, " WG handshake: responded to initiation\n", .{}) catch {}; - } else |_| {} - } - }, - .wg_handshake_resp => { - if (pkt.len >= @sizeOf(Noise.HandshakeResponse)) { - const msg: *const Noise.HandshakeResponse = @ptrCast(@alignCast(pkt.ptr)); - if (wg_dev.handleResponse(msg)) |slot| { - if (wg_dev.peers[slot]) |*p| { - p.endpoint_addr = recv.sender_addr; - p.endpoint_port = recv.sender_port; - } - writeFormatted(stdout, " WG handshake: completed with peer\n", .{}) catch {}; - } else |_| {} - } - }, - .wg_transport => unreachable, - .wg_cookie => {}, - // SWIM and STUN packets: feed to SWIM via feedPacket (non-blocking) - .stun => swim.feedPacket(pkt, recv.sender_addr, recv.sender_port), - .swim => swim.feedPacket(pkt, recv.sender_addr, recv.sender_port), - .unknown => {}, - } + processIncomingPacket( + pkt, + recv.sender_addr, + recv.sender_port, + wg_dev, + swim, + udp_sock, + stdout, + &decrypt_storage, + &decrypt_lens, + &decrypt_slots, + &n_decrypted, + service_filter, + ); + } + for (0..n_decrypted) |i| { + tun_dev.write(decrypt_storage[i][0..decrypt_lens[i]]) catch {}; } // ─── 2. SWIM timers-only tick (gossip, failure detection, NAT) ─── @@ -3053,17 +3212,14 @@ fn windowsEventLoop( const dst_ip: [4]u8 = .{ ip_pkt[16], ip_pkt[17], ip_pkt[18], ip_pkt[19] }; if (wg_dev.lookupByMeshIp(dst_ip)) |slot| { - const peer = wg_dev.peers[slot] orelse continue; - if (peer.endpoint_port == 0) continue; // No endpoint yet - // Encrypt and send if (wg_dev.encryptForPeer(slot, ip_pkt, &encrypt_buf)) |enc_len| { - _ = udp_sock.sendTo(encrypt_buf[0..enc_len], peer.endpoint_addr, peer.endpoint_port) catch {}; + _ = sendWgCiphertextForSlot(swim, wg_dev, udp_sock, slot, encrypt_buf[0..enc_len]); } else |_| { // No tunnel — attempt handshake if due if (wg_dev.initiateHandshake(slot)) |init_msg| { const init_bytes = std.mem.asBytes(&init_msg); - _ = udp_sock.sendTo(init_bytes, peer.endpoint_addr, peer.endpoint_port) catch {}; + _ = sendWgCiphertextForSlot(swim, wg_dev, udp_sock, slot, init_bytes); } else |_| {} // rate-limited or other error } } @@ -3075,10 +3231,10 @@ fn windowsEventLoop( last_handshake_check_ns = now_ns; for (&wg_dev.peers, 0..) |*slot, i| { if (slot.*) |peer| { - if (peer.active_tunnel == null and peer.endpoint_port != 0) { + if (peer.active_tunnel == null) { if (wg_dev.initiateHandshake(i)) |init_msg| { const init_bytes = std.mem.asBytes(&init_msg); - _ = udp_sock.sendTo(init_bytes, peer.endpoint_addr, peer.endpoint_port) catch {}; + _ = sendWgCiphertextForSlot(swim, wg_dev, udp_sock, i, init_bytes); } else |_| {} } } @@ -3121,12 +3277,11 @@ fn macosEventLoop( service_filter: *const lib.services.Policy.ServiceFilter, control_socket: *lib.services.Control.ControlSocket, ) !void { - const Device = lib.wireguard.Device; - const Noise = lib.wireguard.Noise; - var tun_buf: [65536]u8 = undefined; - var udp_recv_buf: [2048]u8 = undefined; - var decrypt_buf: [1500]u8 = undefined; + var udp_recv_buf: [lib.nat.Relay.MAX_RELAY_DATAGRAM]u8 = undefined; + var decrypt_storage: [64][1500]u8 = undefined; + var decrypt_lens: [64]usize = undefined; + var decrypt_slots: [64]usize = undefined; var encrypt_buf: [2048]u8 = undefined; // 16B header + 1500B payload + 16B poly1305 tag var last_handshake_check_ns: i128 = 0; @@ -3136,51 +3291,28 @@ fn macosEventLoop( // ─── 2. Process incoming UDP packets (WG + SWIM multiplexed) ─── var udp_count: u32 = 0; + var n_decrypted: usize = 0; while (udp_count < 64) : (udp_count += 1) { const recv = (udp_sock.recvFrom(&udp_recv_buf) catch break) orelse break; const pkt = recv.data; - const pkt_type = Device.PacketType.classify(pkt); - // Optimization: Extract dominant data-plane case to explicit if branch - if (pkt_type == .wg_transport) { - // Decrypt WG transport → write plaintext to utun - if (wg_dev.decryptTransport(pkt, &decrypt_buf)) |result| { - // Apply service filter before writing to TUN (IPv4 + IPv6, M5). - { - const org_pk = orgPubkeyLocked(swim.membership, result.identity_key); - if (!service_filter.allowPacket(result.identity_key, org_pk, decrypt_buf[0..result.len])) continue; - } - tun_dev.write(decrypt_buf[0..result.len]) catch {}; - } else |_| {} - } else switch (pkt_type) { - .wg_handshake_init => { - if (pkt.len >= @sizeOf(Noise.HandshakeInitiation)) { - const msg: *const Noise.HandshakeInitiation = @ptrCast(@alignCast(pkt.ptr)); - if (wg_dev.handleInitiation(msg, recv.sender_addr)) |hs_result| { - const resp_bytes = std.mem.asBytes(&hs_result.response); - _ = udp_sock.sendTo(resp_bytes, recv.sender_addr, recv.sender_port) catch 0; - writeFormatted(stdout, " WG handshake: responded to initiation\n", .{}) catch {}; - } else |_| {} - } - }, - .wg_handshake_resp => { - if (pkt.len >= @sizeOf(Noise.HandshakeResponse)) { - const msg: *const Noise.HandshakeResponse = @ptrCast(@alignCast(pkt.ptr)); - if (wg_dev.handleResponse(msg)) |slot| { - if (wg_dev.peers[slot]) |*p| { - p.endpoint_addr = recv.sender_addr; - p.endpoint_port = recv.sender_port; - } - writeFormatted(stdout, " WG handshake: completed with peer\n", .{}) catch {}; - } else |_| {} - } - }, - .wg_transport => unreachable, - .wg_cookie => {}, - .stun => swim.feedPacket(pkt, recv.sender_addr, recv.sender_port), - .swim => swim.feedPacket(pkt, recv.sender_addr, recv.sender_port), - .unknown => {}, - } + processIncomingPacket( + pkt, + recv.sender_addr, + recv.sender_port, + wg_dev, + swim, + udp_sock, + stdout, + &decrypt_storage, + &decrypt_lens, + &decrypt_slots, + &n_decrypted, + service_filter, + ); + } + for (0..n_decrypted) |i| { + tun_dev.write(decrypt_storage[i][0..decrypt_lens[i]]) catch {}; } // ─── 3. Read utun → encrypt → send via UDP ─── @@ -3196,17 +3328,14 @@ fn macosEventLoop( const dst_ip: [4]u8 = .{ ip_pkt[16], ip_pkt[17], ip_pkt[18], ip_pkt[19] }; if (wg_dev.lookupByMeshIp(dst_ip)) |slot| { - const peer = wg_dev.peers[slot] orelse continue; - if (peer.endpoint_port == 0) continue; // No endpoint yet - // Encrypt and send if (wg_dev.encryptForPeer(slot, ip_pkt, &encrypt_buf)) |enc_len| { - _ = udp_sock.sendTo(encrypt_buf[0..enc_len], peer.endpoint_addr, peer.endpoint_port) catch {}; + _ = sendWgCiphertextForSlot(swim, wg_dev, udp_sock, slot, encrypt_buf[0..enc_len]); } else |_| { // No tunnel — attempt handshake if due if (wg_dev.initiateHandshake(slot)) |init_msg| { const init_bytes = std.mem.asBytes(&init_msg); - _ = udp_sock.sendTo(init_bytes, peer.endpoint_addr, peer.endpoint_port) catch {}; + _ = sendWgCiphertextForSlot(swim, wg_dev, udp_sock, slot, init_bytes); } else |_| {} } } @@ -3218,10 +3347,10 @@ fn macosEventLoop( last_handshake_check_ns = now_ns; for (&wg_dev.peers, 0..) |*slot, i| { if (slot.*) |peer| { - if (peer.active_tunnel == null and peer.endpoint_port != 0) { + if (peer.active_tunnel == null) { if (wg_dev.initiateHandshake(i)) |init_msg| { const init_bytes = std.mem.asBytes(&init_msg); - _ = udp_sock.sendTo(init_bytes, peer.endpoint_addr, peer.endpoint_port) catch {}; + _ = sendWgCiphertextForSlot(swim, wg_dev, udp_sock, i, init_bytes); } else |_| {} } } @@ -3399,6 +3528,7 @@ fn userspaceEventLoop( for (0..opened_workers) |w| { threads[spawned] = std.Thread.spawn(.{}, dataPlaneWorker, .{ &swim.running, + swim, wg_dev, tun_fds[w], udp_sock.fd, diff --git a/src/meshguard_ffi.zig b/src/meshguard_ffi.zig index cda9b96..f8e3846 100644 --- a/src/meshguard_ffi.zig +++ b/src/meshguard_ffi.zig @@ -358,6 +358,7 @@ export fn meshguard_join( .onPeerPunched = null, .onAppMessage = &onAppMessageCallback, .onWgPacket = &onWgPacketCallback, + .onRelayWgPacket = &onRelayWgPacketCallback, .hasActiveTunnel = &hasActiveTunnelCallback, .reinitiateHandshake = &reinitiateHandshakeCallback, }; @@ -417,6 +418,7 @@ export fn meshguard_join_ipv6( .onPeerPunched = null, .onAppMessage = &onAppMessageCallback, .onWgPacket = &onWgPacketCallback, + .onRelayWgPacket = &onRelayWgPacketCallback, .hasActiveTunnel = &hasActiveTunnelCallback, .reinitiateHandshake = &reinitiateHandshakeCallback, }; @@ -478,6 +480,7 @@ export fn meshguard_join_lan( .onPeerPunched = null, .onAppMessage = &onAppMessageCallback, .onWgPacket = &onWgPacketCallback, + .onRelayWgPacket = &onRelayWgPacketCallback, .hasActiveTunnel = &hasActiveTunnelCallback, .reinitiateHandshake = &reinitiateHandshakeCallback, }; @@ -1213,18 +1216,79 @@ fn reinitiateHandshakeCallback(raw_ctx: *anyopaque, peer: *const Membership.Peer if (std.mem.order(u8, &dev.static_public, &wg_key) != .lt) return; // only the initiator const ep: messages.Endpoint = if (peer.gossip_endpoint) |e| e else if (peer.public_endpoint) |pe| pe else return; if (dev.reinitiate(wg_key)) |init_msg| { - if (ctx.socket) |*sock| { - _ = sock.sendToEndpoint(std.mem.asBytes(&init_msg), ep) catch 0; - } + _ = sendWgCiphertextToPeer(ctx, peer.pubkey, std.mem.asBytes(&init_msg), ep, null); } else |_| {} } } +fn relayEndpointForPeer(ctx: *MeshguardContext, target_pubkey: [32]u8) ?messages.Endpoint { + if (ctx.swim != null) { + const relay = lib.nat.Relay.selectRelayForPair(&ctx.membership.peers, ctx.ed25519_public, target_pubkey) orelse return null; + return relay.gossip_endpoint orelse relay.public_endpoint; + } + return null; +} + +fn sendRelayFrame(ctx: *MeshguardContext, target_pubkey: [32]u8, data: []const u8, relay_endpoint: messages.Endpoint) bool { + var relay_buf: [lib.nat.Relay.MAX_RELAY_DATAGRAM]u8 = undefined; + const relay_len = lib.nat.Relay.encodeRelayData(&relay_buf, ctx.ed25519_public, target_pubkey, data) catch return false; + if (ctx.socket) |*sock| { + _ = sock.sendToEndpoint(relay_buf[0..relay_len], relay_endpoint) catch return false; + return true; + } + return false; +} + +fn sendWgCiphertextToPeer( + ctx: *MeshguardContext, + target_pubkey: [32]u8, + data: []const u8, + direct_endpoint: ?messages.Endpoint, + preferred_relay: ?messages.Endpoint, +) bool { + if (preferred_relay) |relay_ep| return sendRelayFrame(ctx, target_pubkey, data, relay_ep); + + var target_nat: messages.NatType = .unknown; + var relay_ep: ?messages.Endpoint = null; + if (ctx.swim) |*swim| { + ctx.membership.lock.lockSharedUncancelable(zio()); + if (ctx.membership.peers.get(target_pubkey)) |peer| target_nat = peer.nat_type; + relay_ep = relayEndpointForPeer(ctx, target_pubkey); + ctx.membership.lock.unlockShared(zio()); + + const relay_available = relay_ep != null; + if (direct_endpoint) |ep| { + const path = lib.nat.Relay.choosePath(swim.our_nat_type, target_nat, true, relay_available); + if (path == .direct or path == .holepunch or !relay_available) { + if (ctx.socket) |*sock| { + _ = sock.sendToEndpoint(data, ep) catch return false; + return true; + } + return false; + } + } + } else if (direct_endpoint) |ep| { + if (ctx.socket) |*sock| { + _ = sock.sendToEndpoint(data, ep) catch return false; + return true; + } + return false; + } + + if (relay_ep) |ep| return sendRelayFrame(ctx, target_pubkey, data, ep); + return false; +} + fn onWgPacketCallback(raw_ctx: *anyopaque, data: []const u8, sender_addr: [4]u8, sender_port: u16) void { const ctx: *MeshguardContext = @ptrCast(@alignCast(raw_ctx)); handleWgPacket(ctx, data, sender_addr, sender_port); } +fn onRelayWgPacketCallback(raw_ctx: *anyopaque, data: []const u8, sender_pubkey: [32]u8, relay_endpoint: messages.Endpoint) void { + const ctx: *MeshguardContext = @ptrCast(@alignCast(raw_ctx)); + handleRelayWgPacket(ctx, data, sender_pubkey, relay_endpoint); +} + /// Process incoming WireGuard packets (Type 1/2/4) for tunnel support. fn handleWgPacket(ctx: *MeshguardContext, data: []const u8, sender_addr: [4]u8, sender_port: u16) void { if (data.len < 4) return; @@ -1343,6 +1407,74 @@ fn handleWgPacket(ctx: *MeshguardContext, data: []const u8, sender_addr: [4]u8, } } +fn handleRelayWgPacket(ctx: *MeshguardContext, data: []const u8, sender_pubkey: [32]u8, relay_endpoint: messages.Endpoint) void { + if (data.len < 4) return; + if (ctx.wg_device == null) return; + + const pkt_type = Device.PacketType.classify(data); + switch (pkt_type) { + .wg_handshake_init => { + ctx.wg_lock.lockUncancelable(zio()); + defer ctx.wg_lock.unlock(zio()); + + var dev = &ctx.wg_device.?; + if (data.len < @sizeOf(noise.HandshakeInitiation)) return; + const msg: *const noise.HandshakeInitiation = @ptrCast(@alignCast(data.ptr)); + + const result = dev.handleInitiation(msg, relay_endpoint.addr) catch |err| blk: { + if (err != error.UnknownPeer) return; + + if (ctx.membership.peers.get(sender_pubkey)) |m_peer| { + if (m_peer.wg_pubkey) |wg_pk| { + const ep = m_peer.gossip_endpoint orelse m_peer.public_endpoint orelse relay_endpoint; + _ = dev.addPeerWithEndpoint( + m_peer.pubkey, + wg_pk, + ep, + m_peer.mesh_ip, + m_peer.mesh_ip6, + ) catch {}; + } + } + + break :blk dev.handleInitiationAdmitted(msg) catch return; + }; + + const resp_bytes: [*]const u8 = @ptrCast(&result.response); + _ = sendRelayFrame(ctx, sender_pubkey, resp_bytes[0..@sizeOf(noise.HandshakeResponse)], relay_endpoint); + + std.debug.print(" 🔑 WG handshake init handled via relay (slot {d})\n", .{result.slot}); + }, + .wg_handshake_resp => { + ctx.wg_lock.lockUncancelable(zio()); + defer ctx.wg_lock.unlock(zio()); + + var dev = &ctx.wg_device.?; + if (data.len < @sizeOf(noise.HandshakeResponse)) return; + const msg: *const noise.HandshakeResponse = @ptrCast(@alignCast(data.ptr)); + const slot = dev.handleResponse(msg) catch return; + + std.debug.print(" 🔑 WG handshake complete via relay (slot {d}), tunnel ready\n", .{slot}); + }, + .wg_transport => { + ctx.wg_lock.lockSharedUncancelable(zio()); + defer ctx.wg_lock.unlockShared(zio()); + + var dev = &ctx.wg_device.?; + var plaintext: [1500]u8 = undefined; + const result = dev.decryptTransport(data, &plaintext) catch return; + const sender_identity: [32]u8 = result.identity_key; + + if (result.len < 4) return; + if (plaintext[0] != 0 or plaintext[1] != 0) return; + const payload_len = std.mem.readInt(u16, plaintext[2..4], .little); + const payload_end = tunnelFramePayloadLen(result.len, payload_len) orelse return; + enqueueTunnelMessage(ctx, sender_identity, plaintext[4..][0..payload_end]); + }, + else => {}, + } +} + /// Enqueue a decrypted tunnel message to the tunnel inbox. /// Finding #3: Capacity check + atomic valid flag to prevent torn reads. fn enqueueTunnelMessage(ctx: *MeshguardContext, sender_identity: [32]u8, data: []const u8) void { @@ -1419,8 +1551,7 @@ export fn meshguard_tunnel_open( // Send handshake initiation packet to peer const init_bytes: [*]const u8 = @ptrCast(&init_msg); - const socket = c.socket orelse return -8; - _ = socket.sendTo(init_bytes[0..@sizeOf(noise.HandshakeInitiation)], ep.addr, ep.port) catch return -8; + if (!sendWgCiphertextToPeer(c, target_key, init_bytes[0..@sizeOf(noise.HandshakeInitiation)], ep, null)) return -8; std.debug.print(" \xf0\x9f\x94\x91 Tunnel handshake sent to {x:0>2}{x:0>2}...\n", .{ target_key[0], target_key[1] }); return 0; @@ -1495,15 +1626,14 @@ export fn meshguard_tunnel_send( } // Read endpoint while we have the lock - const ep_addr = peer.endpoint_addr; - const ep_port = peer.endpoint_port; + const direct_ep = peer.endpoint(); // Release shared lock before I/O c.wg_lock.unlockShared(zio()); // Send via UDP to peer's endpoint - const socket = c.socket orelse return -6; - _ = socket.sendTo(out_buf[0..encrypted_len], ep_addr, ep_port) catch return -7; + if (c.socket == null) return -6; + if (!sendWgCiphertextToPeer(c, target_key, out_buf[0..encrypted_len], direct_ep, null)) return -7; // Rekeying requires exclusive lock (modifies index_map) if (requires_rekey) { @@ -1513,7 +1643,7 @@ export fn meshguard_tunnel_send( var rekey_dev = wd; if (rekey_dev.initiateHandshake(slot)) |rekey_msg| { const rekey_bytes: [*]const u8 = @ptrCast(&rekey_msg); - _ = socket.sendTo(rekey_bytes[0..@sizeOf(noise.HandshakeInitiation)], ep_addr, ep_port) catch {}; + _ = sendWgCiphertextToPeer(c, target_key, rekey_bytes[0..@sizeOf(noise.HandshakeInitiation)], direct_ep, null); } else |_| {} } } diff --git a/src/nat/holepunch.zig b/src/nat/holepunch.zig index 2631997..664a7a0 100644 --- a/src/nat/holepunch.zig +++ b/src/nat/holepunch.zig @@ -69,6 +69,16 @@ const MAX_PROBES: u8 = 25; // 5 seconds / 200ms /// The receiver recognizes this as a hole punch probe and ignores it /// (it's not a valid SWIM message and won't be processed by the gossip handler). const PROBE_MAGIC: [4]u8 = .{ 0x4D, 0x47, 0x48, 0x50 }; // "MGHP" = MeshGuard HolePunch +const PROBE_SIZE: usize = PROBE_MAGIC.len + 16; + +pub const Probe = struct { + token: [16]u8, +}; + +pub const PunchSuccess = struct { + peer_pubkey: [32]u8, + endpoint: messages.Endpoint, +}; /// Hole punch coordinator — manages concurrent punch attempts. pub const Holepuncher = struct { @@ -118,7 +128,9 @@ pub const Holepuncher = struct { ) ?messages.Endpoint { for (&self.active) |*slot| { if (slot.*) |*state| { - if (std.mem.eql(u8, &state.token, &response.token_echo)) { + if (std.mem.eql(u8, &state.token, &response.token_echo) and + std.mem.eql(u8, &state.target_pubkey, &response.sender_pubkey)) + { state.target_public_endpoint = response.public_endpoint; state.response_received = true; return response.public_endpoint; @@ -128,6 +140,48 @@ pub const Holepuncher = struct { return null; // no matching punch attempt } + /// Accept an incoming request as the target peer and start a session keyed to + /// the initiator identity and token. This mirrors initiate(), but preserves + /// the initiator's token so later probes can be bound to the same session. + pub fn acceptRequest( + self: *Holepuncher, + our_pubkey: [32]u8, + request: messages.HolepunchRequest, + our_public_endpoint: messages.Endpoint, + ) bool { + for (&self.active) |*slot| { + if (slot.*) |*state| { + if (std.mem.eql(u8, &state.initiator_pubkey, &our_pubkey) and + std.mem.eql(u8, &state.target_pubkey, &request.sender_pubkey)) + { + state.our_public_endpoint = our_public_endpoint; + state.target_public_endpoint = request.public_endpoint; + state.token = request.token; + state.started_at_ns = nowNs(); + state.probes_sent = 0; + state.response_received = true; + return true; + } + } + } + + const slot = for (&self.active, 0..) |*s, i| { + if (s.* == null) break i; + } else return false; + + self.active[slot] = .{ + .initiator_pubkey = our_pubkey, + .target_pubkey = request.sender_pubkey, + .our_public_endpoint = our_public_endpoint, + .target_public_endpoint = request.public_endpoint, + .token = request.token, + .started_at_ns = nowNs(), + .probes_sent = 0, + .response_received = true, + }; + return true; + } + /// Build a HolepunchResponse for an incoming request. pub fn buildResponse( our_pubkey: [32]u8, @@ -162,7 +216,9 @@ pub const Holepuncher = struct { if (state.target_public_endpoint) |target_ep| { if (state.probes_sent < MAX_PROBES) { // Send probe packet - _ = socket.sendToEndpoint(&PROBE_MAGIC, target_ep) catch {}; + var probe_buf: [PROBE_SIZE]u8 = undefined; + encodeProbe(state.token, &probe_buf); + _ = socket.sendToEndpoint(&probe_buf, target_ep) catch {}; state.probes_sent += 1; } @@ -190,9 +246,39 @@ pub const Holepuncher = struct { return null; } + pub fn markProbeSuccess(self: *Holepuncher, probe: Probe, sender_endpoint: messages.Endpoint) ?PunchSuccess { + for (&self.active) |*slot| { + if (slot.*) |state| { + if (!std.mem.eql(u8, &state.token, &probe.token)) continue; + const target_ep = state.target_public_endpoint orelse continue; + if (!messages.Endpoint.eql(target_ep, sender_endpoint)) continue; + const success = PunchSuccess{ + .peer_pubkey = state.target_pubkey, + .endpoint = target_ep, + }; + slot.* = null; + return success; + } + } + return null; + } + + pub fn encodeProbe(token: [16]u8, out: *[PROBE_SIZE]u8) void { + @memcpy(out[0..PROBE_MAGIC.len], &PROBE_MAGIC); + @memcpy(out[PROBE_MAGIC.len..][0..16], &token); + } + + pub fn decodeProbe(data: []const u8) ?Probe { + if (data.len != PROBE_SIZE) return null; + if (!std.mem.eql(u8, data[0..PROBE_MAGIC.len], &PROBE_MAGIC)) return null; + var token: [16]u8 = undefined; + @memcpy(&token, data[PROBE_MAGIC.len..][0..16]); + return .{ .token = token }; + } + /// Check if a received packet is a hole punch probe. pub fn isProbe(data: []const u8) bool { - return data.len == PROBE_MAGIC.len and std.mem.eql(u8, data, &PROBE_MAGIC); + return decodeProbe(data) != null; } /// Count active punch attempts. @@ -240,7 +326,12 @@ test "initiate and respond" { } test "probe magic detection" { - try std.testing.expect(Holepuncher.isProbe(&PROBE_MAGIC)); + var probe_buf: [PROBE_SIZE]u8 = undefined; + Holepuncher.encodeProbe([_]u8{0x42} ** 16, &probe_buf); + + try std.testing.expect(Holepuncher.isProbe(&probe_buf)); + try std.testing.expectEqualSlices(u8, &([_]u8{0x42} ** 16), &Holepuncher.decodeProbe(&probe_buf).?.token); + try std.testing.expect(!Holepuncher.isProbe(&PROBE_MAGIC)); try std.testing.expect(!Holepuncher.isProbe(&.{ 0x01, 0x02, 0x03, 0x04 })); try std.testing.expect(!Holepuncher.isProbe(&.{0x01})); } @@ -260,3 +351,51 @@ test "max concurrent punches" { try std.testing.expect(puncher.initiate([_]u8{0xAA} ** 32, [_]u8{0xFF} ** 32, our_ep) == null); try std.testing.expectEqual(puncher.activeCount(), MAX_CONCURRENT_PUNCHES); } + +test "holepunch response is bound to token and peer identity" { + var puncher = Holepuncher{}; + const our_pubkey = [_]u8{0xAA} ** 32; + const target_pubkey = [_]u8{0xBB} ** 32; + const wrong_pubkey = [_]u8{0xCC} ** 32; + const our_ep = messages.Endpoint.initV4(.{ 1, 2, 3, 4 }, 51821); + const target_ep = messages.Endpoint.initV4(.{ 5, 6, 7, 8 }, 51821); + + const req = puncher.initiate(our_pubkey, target_pubkey, our_ep).?; + try std.testing.expect(puncher.handleResponse(.{ + .sender_pubkey = wrong_pubkey, + .public_endpoint = target_ep, + .token_echo = req.token, + }) == null); + try std.testing.expect(puncher.handleResponse(.{ + .sender_pubkey = target_pubkey, + .public_endpoint = target_ep, + .token_echo = [_]u8{0x11} ** 16, + }) == null); + try std.testing.expect(puncher.handleResponse(.{ + .sender_pubkey = target_pubkey, + .public_endpoint = target_ep, + .token_echo = req.token, + }) != null); +} + +test "target accepts request with initiator session token" { + var puncher = Holepuncher{}; + const our_pubkey = [_]u8{0xBB} ** 32; + const initiator_pubkey = [_]u8{0xAA} ** 32; + const our_ep = messages.Endpoint.initV4(.{ 5, 6, 7, 8 }, 51821); + const initiator_ep = messages.Endpoint.initV4(.{ 1, 2, 3, 4 }, 51821); + const token = [_]u8{0x99} ** 16; + + try std.testing.expect(puncher.acceptRequest(our_pubkey, .{ + .sender_pubkey = initiator_pubkey, + .target_pubkey = our_pubkey, + .public_endpoint = initiator_ep, + .token = token, + }, our_ep)); + try std.testing.expectEqual(@as(usize, 1), puncher.activeCount()); + + const success = puncher.markProbeSuccess(.{ .token = token }, initiator_ep).?; + try std.testing.expectEqualSlices(u8, &initiator_pubkey, &success.peer_pubkey); + try std.testing.expect(messages.Endpoint.eql(initiator_ep, success.endpoint)); + try std.testing.expectEqual(@as(usize, 0), puncher.activeCount()); +} diff --git a/src/nat/relay.zig b/src/nat/relay.zig index 5b299c4..f2f6bcd 100644 --- a/src/nat/relay.zig +++ b/src/nat/relay.zig @@ -14,6 +14,7 @@ const std = @import("std"); const messages = @import("../protocol/messages.zig"); const Membership = @import("../discovery/membership.zig"); +const keys = @import("../identity/keys.zig"); /// Relay node info. pub const RelayInfo = struct { @@ -27,6 +28,261 @@ pub const RelayInfo = struct { /// Maximum relay connections per node (default). pub const DEFAULT_MAX_RELAY_PEERS: u16 = 10; +pub const MAX_RENDEZVOUS_RECORDS: usize = 64; +pub const MAX_RELAY_PAYLOAD: usize = 2048; +pub const RELAY_FRAME_HEADER_SIZE: usize = 1 + 32 + 32 + 2; +pub const MAX_RELAY_DATAGRAM: usize = RELAY_FRAME_HEADER_SIZE + MAX_RELAY_PAYLOAD; +pub const REGISTRATION_NONCE_SIZE: usize = 32; +pub const REGISTRATION_SIGNED_SIZE: usize = 32 + 1 + 16 + 2 + REGISTRATION_NONCE_SIZE; +const CHALLENGE_TTL_NS: i128 = 60 * std.time.ns_per_s; +const REGISTER_MIN_INTERVAL_NS: i128 = std.time.ns_per_s; + +fn zio() std.Io { + return std.Io.Threaded.global_single_threaded.io(); +} + +pub const RelayError = error{ + BufferTooShort, + InvalidFrame, + PayloadTooLarge, + NotWireGuardCiphertext, + NoChallenge, + ChallengeExpired, + InvalidSignature, + RateLimited, + RegistryFull, +}; + +pub const RelayPath = enum { + direct, + holepunch, + relay, + unavailable, +}; + +pub const DecodedRelayData = struct { + sender_pubkey: [32]u8, + target_pubkey: [32]u8, + payload: []const u8, +}; + +pub const RendezvousRecord = struct { + pubkey: [32]u8, + endpoint: ?messages.Endpoint = null, + challenge: [REGISTRATION_NONCE_SIZE]u8 = .{0} ** REGISTRATION_NONCE_SIZE, + challenge_issued_ns: i128 = 0, + registered_at_ns: i128 = 0, +}; + +const RateEntry = struct { + pubkey: [32]u8, + tokens: f64, + last_refill_ns: i128, +}; + +pub const IdentityRateLimiter = struct { + pub const BURST_TOKENS: usize = 100; + const RATE_PER_SEC: f64 = 50.0; + const BURST: f64 = @floatFromInt(BURST_TOKENS); + + entries: [MAX_RENDEZVOUS_RECORDS]?RateEntry = .{null} ** MAX_RENDEZVOUS_RECORDS, + + pub fn allow(self: *IdentityRateLimiter, pubkey: [32]u8, now_ns: i128) bool { + const slot = self.findOrInsert(pubkey, now_ns) orelse return false; + var entry = &self.entries[slot].?; + const elapsed_ns = @max(now_ns - entry.last_refill_ns, 0); + const elapsed_s = @as(f64, @floatFromInt(elapsed_ns)) / @as(f64, @floatFromInt(std.time.ns_per_s)); + entry.tokens = @min(BURST, entry.tokens + elapsed_s * RATE_PER_SEC); + entry.last_refill_ns = now_ns; + if (entry.tokens < 1.0) return false; + entry.tokens -= 1.0; + return true; + } + + fn findOrInsert(self: *IdentityRateLimiter, pubkey: [32]u8, now_ns: i128) ?usize { + var empty: ?usize = null; + var oldest: ?usize = null; + var oldest_seen_ns: i128 = std.math.maxInt(i128); + for (&self.entries, 0..) |*entry, i| { + if (entry.*) |e| { + if (std.mem.eql(u8, &e.pubkey, &pubkey)) return i; + if (e.last_refill_ns < oldest_seen_ns) { + oldest_seen_ns = e.last_refill_ns; + oldest = i; + } + } else if (empty == null) { + empty = i; + } + } + const slot = empty orelse oldest orelse return null; + self.entries[slot] = .{ + .pubkey = pubkey, + .tokens = BURST, + .last_refill_ns = now_ns, + }; + return slot; + } +}; + +pub const RendezvousRegistry = struct { + records: [MAX_RENDEZVOUS_RECORDS]?RendezvousRecord = .{null} ** MAX_RENDEZVOUS_RECORDS, + + pub fn issueChallenge(self: *RendezvousRegistry, pubkey: [32]u8, now_ns: i128) RelayError![REGISTRATION_NONCE_SIZE]u8 { + const slot = self.findOrInsert(pubkey) orelse return error.RegistryFull; + var challenge: [REGISTRATION_NONCE_SIZE]u8 = undefined; + zio().random(&challenge); + if (self.records[slot]) |*record| { + record.challenge = challenge; + record.challenge_issued_ns = now_ns; + } else { + self.records[slot] = .{ + .pubkey = pubkey, + .challenge = challenge, + .challenge_issued_ns = now_ns, + }; + } + return challenge; + } + + pub fn register( + self: *RendezvousRegistry, + pubkey: [32]u8, + endpoint: messages.Endpoint, + challenge: [REGISTRATION_NONCE_SIZE]u8, + signature: [64]u8, + now_ns: i128, + ) RelayError!void { + const slot = self.findSlot(pubkey) orelse return error.NoChallenge; + var record = &self.records[slot].?; + if (!std.mem.eql(u8, &record.challenge, &challenge)) return error.NoChallenge; + if (now_ns - record.challenge_issued_ns > CHALLENGE_TTL_NS) return error.ChallengeExpired; + if (record.endpoint != null and now_ns - record.registered_at_ns < REGISTER_MIN_INTERVAL_NS) return error.RateLimited; + + const signed = registrationSignedBytes(pubkey, endpoint, challenge); + const pk = std.crypto.sign.Ed25519.PublicKey.fromBytes(pubkey) catch return error.InvalidSignature; + if (!keys.verify(&signed, signature, pk)) return error.InvalidSignature; + + record.endpoint = endpoint; + record.registered_at_ns = now_ns; + } + + pub fn lookup(self: *const RendezvousRegistry, pubkey: [32]u8) ?messages.Endpoint { + const slot = self.findSlot(pubkey) orelse return null; + return self.records[slot].?.endpoint; + } + + fn findSlot(self: *const RendezvousRegistry, pubkey: [32]u8) ?usize { + for (self.records, 0..) |record, i| { + if (record) |r| { + if (std.mem.eql(u8, &r.pubkey, &pubkey)) return i; + } + } + return null; + } + + fn findOrInsert(self: *RendezvousRegistry, pubkey: [32]u8) ?usize { + var empty: ?usize = null; + for (&self.records, 0..) |*record, i| { + if (record.*) |r| { + if (std.mem.eql(u8, &r.pubkey, &pubkey)) return i; + } else if (empty == null) { + empty = i; + } + } + return empty; + } +}; + +pub fn registrationSignedBytes( + pubkey: [32]u8, + endpoint: messages.Endpoint, + challenge: [REGISTRATION_NONCE_SIZE]u8, +) [REGISTRATION_SIGNED_SIZE]u8 { + var out: [REGISTRATION_SIGNED_SIZE]u8 = undefined; + var pos: usize = 0; + @memcpy(out[pos..][0..32], &pubkey); + pos += 32; + if (endpoint.addr6) |addr6| { + out[pos] = 6; + pos += 1; + @memcpy(out[pos..][0..16], &addr6); + } else { + out[pos] = 4; + pos += 1; + @memset(out[pos..][0..16], 0); + @memcpy(out[pos..][0..4], &endpoint.addr); + } + pos += 16; + std.mem.writeInt(u16, out[pos..][0..2], endpoint.port, .big); + pos += 2; + @memcpy(out[pos..][0..REGISTRATION_NONCE_SIZE], &challenge); + return out; +} + +pub fn isWireGuardCiphertext(payload: []const u8) bool { + if (payload.len < 4) return false; + const msg_type = std.mem.readInt(u32, payload[0..4], .little); + return msg_type >= 1 and msg_type <= 4; +} + +pub fn encodeRelayData( + buf: []u8, + sender_pubkey: [32]u8, + target_pubkey: [32]u8, + payload: []const u8, +) RelayError!usize { + if (!isWireGuardCiphertext(payload)) return error.NotWireGuardCiphertext; + if (payload.len > MAX_RELAY_PAYLOAD or payload.len > std.math.maxInt(u16)) return error.PayloadTooLarge; + const required = RELAY_FRAME_HEADER_SIZE + payload.len; + if (buf.len < required) return error.BufferTooShort; + + var pos: usize = 0; + buf[pos] = @intFromEnum(messages.MessageType.relay_data); + pos += 1; + @memcpy(buf[pos..][0..32], &sender_pubkey); + pos += 32; + @memcpy(buf[pos..][0..32], &target_pubkey); + pos += 32; + std.mem.writeInt(u16, buf[pos..][0..2], @intCast(payload.len), .big); + pos += 2; + @memcpy(buf[pos..][0..payload.len], payload); + return required; +} + +pub fn decodeRelayData(data: []const u8) RelayError!DecodedRelayData { + if (data.len < RELAY_FRAME_HEADER_SIZE) return error.BufferTooShort; + if (data[0] != @intFromEnum(messages.MessageType.relay_data)) return error.InvalidFrame; + var pos: usize = 1; + var sender_pubkey: [32]u8 = undefined; + @memcpy(&sender_pubkey, data[pos..][0..32]); + pos += 32; + var target_pubkey: [32]u8 = undefined; + @memcpy(&target_pubkey, data[pos..][0..32]); + pos += 32; + const payload_len = std.mem.readInt(u16, data[pos..][0..2], .big); + pos += 2; + if (payload_len > MAX_RELAY_PAYLOAD) return error.PayloadTooLarge; + if (data.len != pos + payload_len) return error.InvalidFrame; + const payload = data[pos..][0..payload_len]; + if (!isWireGuardCiphertext(payload)) return error.NotWireGuardCiphertext; + return .{ + .sender_pubkey = sender_pubkey, + .target_pubkey = target_pubkey, + .payload = payload, + }; +} + +pub fn choosePath( + our_nat: messages.NatType, + peer_nat: messages.NatType, + has_peer_endpoint: bool, + relay_available: bool, +) RelayPath { + if (has_peer_endpoint and (our_nat == .public or peer_nat == .public)) return .direct; + if (has_peer_endpoint and our_nat == .cone and peer_nat == .cone) return .holepunch; + if (relay_available) return .relay; + return .unavailable; +} /// Check if this node can serve as a relay. pub fn selfRelayInfo(nat_type: messages.NatType, active_count: u16, max_peers: u16) RelayInfo { @@ -45,6 +301,14 @@ pub fn selfRelayInfo(nat_type: messages.NatType, active_count: u16, max_peers: u pub fn selectRelay( peers: *std.AutoHashMap([32]u8, Membership.Peer), exclude: ?[32]u8, +) ?*const Membership.Peer { + return selectRelayForPair(peers, exclude, null); +} + +pub fn selectRelayForPair( + peers: *std.AutoHashMap([32]u8, Membership.Peer), + exclude_a: ?[32]u8, + exclude_b: ?[32]u8, ) ?*const Membership.Peer { var best: ?*const Membership.Peer = null; var best_rtt: u64 = std.math.maxInt(u64); @@ -53,8 +317,11 @@ pub fn selectRelay( while (iter.next()) |entry| { const peer = entry.value_ptr; - // Skip excluded peer (usually ourselves) - if (exclude) |ex| { + // Skip endpoints involved in the relayed pair. + if (exclude_a) |ex| { + if (std.mem.eql(u8, &peer.pubkey, &ex)) continue; + } + if (exclude_b) |ex| { if (std.mem.eql(u8, &peer.pubkey, &ex)) continue; } @@ -95,3 +362,77 @@ test "selfRelayInfo at capacity" { const info = selfRelayInfo(.public, 10, 10); try std.testing.expect(!info.is_relay_capable); } + +test "choosePath prefers direct then punch then relay" { + try std.testing.expectEqual(RelayPath.direct, choosePath(.public, .cone, true, true)); + try std.testing.expectEqual(RelayPath.direct, choosePath(.cone, .public, true, true)); + try std.testing.expectEqual(RelayPath.holepunch, choosePath(.cone, .cone, true, true)); + try std.testing.expectEqual(RelayPath.relay, choosePath(.symmetric, .symmetric, true, true)); + try std.testing.expectEqual(RelayPath.relay, choosePath(.unknown, .cone, false, true)); + try std.testing.expectEqual(RelayPath.unavailable, choosePath(.symmetric, .symmetric, true, false)); +} + +test "relay frame carries only opaque WireGuard packet types" { + const sender = [_]u8{0xAA} ** 32; + const target = [_]u8{0xBB} ** 32; + var wg_payload = [_]u8{0} ** 32; + std.mem.writeInt(u32, wg_payload[0..4], 4, .little); + + var buf: [RELAY_FRAME_HEADER_SIZE + wg_payload.len]u8 = undefined; + const len = try encodeRelayData(&buf, sender, target, &wg_payload); + const decoded = try decodeRelayData(buf[0..len]); + try std.testing.expectEqualSlices(u8, &sender, &decoded.sender_pubkey); + try std.testing.expectEqualSlices(u8, &target, &decoded.target_pubkey); + try std.testing.expectEqualSlices(u8, &wg_payload, decoded.payload); + + var plaintext = [_]u8{ 0x50, 0x4c, 0x41, 0x49 }; // "PLAI", not WG type 1-4 + try std.testing.expectError(error.NotWireGuardCiphertext, encodeRelayData(&buf, sender, target, &plaintext)); +} + +test "rendezvous registration requires signed challenge" { + const kp = keys.generate(); + const pubkey = kp.public_key.toBytes(); + const endpoint = messages.Endpoint.initV4(.{ 203, 0, 113, 7 }, 51821); + + var registry = RendezvousRegistry{}; + const challenge = try registry.issueChallenge(pubkey, 1); + const signed = registrationSignedBytes(pubkey, endpoint, challenge); + const signature = try keys.sign(&signed, kp.secret_key); + + try registry.register(pubkey, endpoint, challenge, signature, 2); + try std.testing.expect(messages.Endpoint.eql(endpoint, registry.lookup(pubkey).?)); + + var bad_sig = signature; + bad_sig[0] ^= 0xff; + try std.testing.expectError(error.RateLimited, registry.register(pubkey, endpoint, challenge, signature, 3)); + + const challenge2 = try registry.issueChallenge(pubkey, REGISTER_MIN_INTERVAL_NS + 10); + const signed2 = registrationSignedBytes(pubkey, endpoint, challenge2); + try std.testing.expectError(error.InvalidSignature, registry.register(pubkey, endpoint, challenge2, bad_sig, REGISTER_MIN_INTERVAL_NS + 11)); + try registry.register(pubkey, endpoint, challenge2, try keys.sign(&signed2, kp.secret_key), REGISTER_MIN_INTERVAL_NS + 11); +} + +test "identity rate limiter is per identity" { + var limiter = IdentityRateLimiter{}; + const a = [_]u8{0xAA} ** 32; + const b = [_]u8{0xBB} ** 32; + + for (0..IdentityRateLimiter.BURST_TOKENS) |_| { + try std.testing.expect(limiter.allow(a, 0)); + } + try std.testing.expect(!limiter.allow(a, 0)); + try std.testing.expect(limiter.allow(b, 0)); + try std.testing.expect(limiter.allow(a, std.time.ns_per_s)); +} + +test "identity rate limiter evicts oldest identity when full" { + var limiter = IdentityRateLimiter{}; + for (0..MAX_RENDEZVOUS_RECORDS) |i| { + var pubkey = [_]u8{0} ** 32; + pubkey[31] = @intCast(i); + try std.testing.expect(limiter.allow(pubkey, @intCast(i))); + } + + const newcomer = [_]u8{0xFF} ** 32; + try std.testing.expect(limiter.allow(newcomer, 10_000)); +} diff --git a/src/nat/stun.zig b/src/nat/stun.zig index 2d77bb2..7ea637b 100644 --- a/src/nat/stun.zig +++ b/src/nat/stun.zig @@ -270,26 +270,54 @@ pub fn discoverPublicEndpoint(socket: *Udp.UdpSocket, server: StunServer) !Exter }; } +pub fn externalAddressEql(a: ExternalAddress, b: ExternalAddress) bool { + return a.port == b.port and std.mem.eql(u8, &a.addr, &b.addr); +} + +/// Classify NAT mapping behavior from one or more STUN observations. +/// +/// A single server can only distinguish "same mapped port" from "mapped port"; +/// two different server observations are needed to identify endpoint-dependent +/// mappings. If the external mapping changes across servers, peers should stop +/// spending time on hole punching and move to relay fallback. +pub fn classifyMapping(local_port: u16, observations: []const ExternalAddress) NatType { + if (observations.len == 0) return .unknown; + const first = observations[0]; + for (observations[1..]) |obs| { + if (!externalAddressEql(first, obs)) return .symmetric; + } + return if (first.port == local_port) .public else .cone; +} + /// Try multiple STUN servers and determine our NAT type. /// `local_port` is the port we're bound to locally. pub fn discover(socket: *Udp.UdpSocket, local_port: u16, servers: []const StunServer) StunResult { + var observations: [MAX_CONFIGURED_STUN_SERVERS]ExternalAddress = undefined; + var observation_count: usize = 0; + var first_server: ?StunServer = null; + for (servers) |server| { const ext = discoverPublicEndpoint(socket, server) catch continue; + if (observation_count == 0) first_server = server; + if (observation_count < observations.len) { + observations[observation_count] = ext; + observation_count += 1; + } + if (observation_count >= 2) break; + } - // Determine NAT type by comparing ports - const nat_type: NatType = if (ext.port == local_port) - .public // same port → likely no NAT or full cone - else - .cone; // different port → some NAT present - - return .{ .external = ext, .nat_type = nat_type, .server = server }; + if (observation_count == 0) { + return .{ + .external = .{ .addr = .{ 0, 0, 0, 0 }, .port = 0 }, + .nat_type = .unknown, + .server = null, + }; } - // All servers failed return .{ - .external = .{ .addr = .{ 0, 0, 0, 0 }, .port = 0 }, - .nat_type = .unknown, - .server = null, + .external = observations[0], + .nat_type = classifyMapping(local_port, observations[0..observation_count]), + .server = first_server, }; } @@ -434,3 +462,19 @@ test "configured STUN servers fall back when none resolve" { try std.testing.expectEqual(DEFAULT_STUN_SERVERS[0].host, resolved[0].host); try std.testing.expectEqual(DEFAULT_STUN_SERVERS[0].port, resolved[0].port); } + +test "classify mapping detects endpoint dependent symmetric NAT" { + const a = ExternalAddress{ .addr = .{ 203, 0, 113, 10 }, .port = 40000 }; + const b = ExternalAddress{ .addr = .{ 203, 0, 113, 10 }, .port = 40001 }; + try std.testing.expectEqual(NatType.symmetric, classifyMapping(51821, &.{ a, b })); +} + +test "classify mapping keeps stable mappings direct or cone" { + const public = ExternalAddress{ .addr = .{ 203, 0, 113, 10 }, .port = 51821 }; + try std.testing.expectEqual(NatType.public, classifyMapping(51821, &.{public})); + try std.testing.expectEqual(NatType.public, classifyMapping(51821, &.{ public, public })); + + const cone = ExternalAddress{ .addr = .{ 203, 0, 113, 10 }, .port = 40000 }; + try std.testing.expectEqual(NatType.cone, classifyMapping(51821, &.{cone})); + try std.testing.expectEqual(NatType.cone, classifyMapping(51821, &.{ cone, cone })); +} diff --git a/src/net/batch_udp.zig b/src/net/batch_udp.zig index c445fea..178b5c4 100644 --- a/src/net/batch_udp.zig +++ b/src/net/batch_udp.zig @@ -7,7 +7,8 @@ const linux = std.os.linux; const posix = std.posix; pub const BATCH_SIZE: usize = 64; -pub const MAX_PACKET: usize = 2048; +/// Fits the largest current relay datagram: 67-byte RelayData header + 2048-byte payload. +pub const MAX_PACKET: usize = 2115; // Linux iovec — must match kernel ABI exactly const Iovec = extern struct { diff --git a/src/net/io_uring.zig b/src/net/io_uring.zig index 3f9fd89..1408481 100644 --- a/src/net/io_uring.zig +++ b/src/net/io_uring.zig @@ -134,7 +134,8 @@ pub const UdpRing = struct { // Keep spare SQEs beyond recv+send slots for completions/resubmits during bursts. pub const RING_DEPTH: u16 = 128; pub const RECV_BUF_SIZE: usize = 65536; - pub const SEND_BUF_SIZE: usize = 2048; + /// Fits the largest current relay datagram: 67-byte RelayData header + 2048-byte payload. + pub const SEND_BUF_SIZE: usize = 2115; pub const RecvCompletion = struct { data: []const u8, diff --git a/src/protocol/messages.zig b/src/protocol/messages.zig index d8018a1..8bdc15f 100644 --- a/src/protocol/messages.zig +++ b/src/protocol/messages.zig @@ -195,6 +195,13 @@ pub const RelayRequest = struct { target_pubkey: [32]u8, }; +/// Relay data: opaque WireGuard/Noise packet forwarded by identity metadata. +pub const RelayData = struct { + sender_pubkey: [32]u8, + target_pubkey: [32]u8, + payload: []const u8, +}; + /// Endpoint update: broadcast when a node's external IP changes. pub const EndpointUpdate = struct { sender_pubkey: [32]u8, @@ -251,6 +258,7 @@ test "message type values" { try std.testing.expectEqual(@as(u8, 0x01), @intFromEnum(MessageType.ping)); try std.testing.expectEqual(@as(u8, 0x10), @intFromEnum(MessageType.handshake_init)); try std.testing.expectEqual(@as(u8, 0x30), @intFromEnum(MessageType.relay_request)); + try std.testing.expectEqual(@as(u8, 0x31), @intFromEnum(MessageType.relay_data)); try std.testing.expectEqual(@as(u8, 0x33), @intFromEnum(MessageType.holepunch_request)); try std.testing.expectEqual(@as(u8, 0x34), @intFromEnum(MessageType.holepunch_response)); try std.testing.expectEqual(@as(u8, 0x43), @intFromEnum(MessageType.org_trust_vouch));