Milestone 2b: rendezvous + NAT traversal + relay

.github/workflows/integration.yml

@@ -123,10 +123,17 @@ jobs:
           # arq_recovers_bulk_loss drives a real Bulk flow and needs a release yipd
           # (debug RaptorQ is ~75x slower).
           cargo build --release -p yipd
+          # relay_path_ping and hole_punch_ping need the yip-rendezvous server
+          # binary (a debug build is fine: these are ping-only tests). It lives
+          # in a different package, so cargo does not build it as a side effect
+          # of the tunnel_netns test build above; the test locates it at
+          # target/debug/yip-rendezvous and SKIPs with instructions if absent,
+          # so build it here to keep this job honest (not vacuously green).
+          cargo build -p yip-rendezvous-bin
           echo "running netns tunnel tests under sudo: $BIN"
           # --exact so ping_across_yipd_tunnel does not also match _under_loss.
           for mode in poll uring; do
-            for t in ping_across_yipd_tunnel ping_across_yipd_tunnel_under_loss arq_recovers_bulk_loss l2_tap_ping_or_arp_across_tunnel triangle_full_mesh_ping; do
+            for t in ping_across_yipd_tunnel ping_across_yipd_tunnel_under_loss arq_recovers_bulk_loss l2_tap_ping_or_arp_across_tunnel triangle_full_mesh_ping relay_path_ping hole_punch_ping; do
               LOG="/tmp/netns-$mode-$t.log"
               if [ "$mode" = "uring" ]; then
                 echo "running $t with UringDriver (opt-in: YIP_USE_URING=1)"

bin/yip-rendezvous/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "yip-rendezvous-bin"
+version = "0.1.0"
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+
+[[bin]]
+name = "yip-rendezvous"
+path = "src/main.rs"
+
+[dependencies]
+yip-rendezvous = { path = "../../crates/yip-rendezvous" }
+
+[lints]
+workspace = true

bin/yip-rendezvous/src/main.rs

@@ -0,0 +1,68 @@
+//! The yip rendezvous + blind relay server. Binds one UDP socket, drives the
+//! pure `RendezvousServer` state machine, and sweeps expired registrations on a
+//! read-timeout cadence. No TUN, no tunnel keys, no unsafe.
+#![forbid(unsafe_code)]
+
+use std::net::UdpSocket;
+use std::time::{Duration, Instant};
+
+use yip_rendezvous::{decode, encode, RendezvousServer};
+
+const SWEEP_INTERVAL: Duration = Duration::from_secs(5);
+
+fn main() -> std::io::Result<()> {
+    let mut args = std::env::args();
+    let _prog = args.next();
+    let listen = match args.next().as_deref() {
+        Some("--version") | Some("-V") => {
+            println!("yip-rendezvous {}", env!("CARGO_PKG_VERSION"));
+            return Ok(());
+        }
+        Some(addr) => addr.to_string(),
+        None => {
+            eprintln!("usage: yip-rendezvous <listen-addr>   e.g. 0.0.0.0:51821");
+            std::process::exit(2);
+        }
+    };
+
+    let sock = UdpSocket::bind(&listen)?;
+    sock.set_read_timeout(Some(SWEEP_INTERVAL))?;
+    eprintln!("yip-rendezvous listening on {listen}");
+
+    // Millisecond clock from a monotonic base (Instant), so `now_ms` never goes
+    // backwards and needs no wall clock.
+    let base = Instant::now();
+    let now_ms =
+        |base: Instant| -> u64 { u64::try_from(base.elapsed().as_millis()).unwrap_or(u64::MAX) };
+
+    let mut server = RendezvousServer::new(now_ms(base));
+    let mut last_sweep = Instant::now();
+    let mut rx = [0u8; 2048];
+    let mut out = Vec::new();
+
+    loop {
+        match sock.recv_from(&mut rx) {
+            Ok((n, src)) => {
+                if let Some(msg) = decode(&rx[..n]) {
+                    for (dst, reply) in server.handle(src, msg, now_ms(base)) {
+                        out.clear();
+                        encode(&reply, &mut out);
+                        let _ = sock.send_to(&out, dst); // best-effort; drop on error
+                    }
+                }
+            }
+            Err(e)
+                if e.kind() == std::io::ErrorKind::WouldBlock
+                    || e.kind() == std::io::ErrorKind::TimedOut => {}
+            Err(e) => return Err(e),
+        }
+        if last_sweep.elapsed() >= SWEEP_INTERVAL {
+            server.sweep(now_ms(base));
+            last_sweep = Instant::now();
+            // Lets the netns money tests (and operators) grep stderr for the
+            // final relay-forward count to assert *which path* carried
+            // traffic, without needing any extra IPC/metrics surface.
+            eprintln!("relay-forwarded={}", server.forwarded_count());
+        }
+    }
+}

bin/yip-rendezvous/tests/smoke.rs

@@ -0,0 +1,85 @@
+//! Socket-level smoke: spawn the server, register from one socket, look up from
+//! another, and relay a payload — asserting the observed reflexive addr and the
+//! blind forward both work over real UDP.
+use std::net::UdpSocket;
+use std::process::{Child, Command};
+use std::time::Duration;
+
+use yip_rendezvous::{decode, encode, node_id, Message};
+
+fn spawn_server(listen: &str) -> Child {
+    Command::new(env!("CARGO_BIN_EXE_yip-rendezvous"))
+        .arg(listen)
+        .spawn()
+        .expect("spawn server")
+}
+
+#[test]
+fn register_lookup_relay_over_udp() {
+    let listen = "127.0.0.1:51821";
+    let mut server = spawn_server(listen);
+    std::thread::sleep(Duration::from_millis(300)); // let it bind
+
+    let a = UdpSocket::bind("127.0.0.1:0").unwrap();
+    a.set_read_timeout(Some(Duration::from_secs(2))).unwrap();
+    let b = UdpSocket::bind("127.0.0.1:0").unwrap();
+    b.set_read_timeout(Some(Duration::from_secs(2))).unwrap();
+
+    let a_id = node_id(&[1u8; 32]);
+    let b_id = node_id(&[2u8; 32]);
+
+    // A registers.
+    let mut buf = Vec::new();
+    encode(&Message::Register { node: a_id }, &mut buf);
+    a.send_to(&buf, listen).unwrap();
+    std::thread::sleep(Duration::from_millis(100));
+
+    // B looks up A -> expects PeerInfo(A, A's reflexive addr).
+    buf.clear();
+    encode(&Message::Lookup { node: a_id }, &mut buf);
+    b.send_to(&buf, listen).unwrap();
+    let mut rx = [0u8; 2048];
+    let (n, _) = b.recv_from(&mut rx).expect("B receives PeerInfo");
+    match decode(&rx[..n]) {
+        Some(Message::PeerInfo { node, reflexive }) => {
+            assert_eq!(node, a_id);
+            assert_eq!(reflexive, a.local_addr().unwrap());
+        }
+        other => panic!("expected PeerInfo, got {other:?}"),
+    }
+
+    // B's Lookup above also caused the server to send A a PunchHint (the
+    // simultaneous-open trigger): A is told to punch toward B's reflexive addr.
+    let (n, _) = a
+        .recv_from(&mut rx)
+        .expect("A receives PunchHint from the lookup");
+    match decode(&rx[..n]) {
+        Some(Message::PunchHint { reflexive, .. }) => {
+            assert_eq!(reflexive, b.local_addr().unwrap());
+        }
+        other => panic!("expected PunchHint, got {other:?}"),
+    }
+
+    // B relays a payload to A -> A receives RelayDeliver{src=B, payload}.
+    buf.clear();
+    encode(
+        &Message::RelaySend {
+            src: b_id,
+            dst: a_id,
+            payload: vec![7, 7, 7],
+        },
+        &mut buf,
+    );
+    b.send_to(&buf, listen).unwrap();
+    let (n, _) = a.recv_from(&mut rx).expect("A receives RelayDeliver");
+    match decode(&rx[..n]) {
+        Some(Message::RelayDeliver { src, payload }) => {
+            assert_eq!(src, b_id);
+            assert_eq!(payload, vec![7, 7, 7]);
+        }
+        other => panic!("expected RelayDeliver, got {other:?}"),
+    }
+
+    let _ = server.kill();
+    let _ = server.wait(); // reap the child so it doesn't linger as a zombie
+}

bin/yipd/Cargo.toml

@@ -11,10 +11,11 @@
 yip-crypto = { path = "../../crates/yip-crypto" }
 yip-transport = { path = "../../crates/yip-transport" }
 yip-device = { path = "../../crates/yip-device" }
+yip-rendezvous = { path = "../../crates/yip-rendezvous" }
 blake2 = { workspace = true }
 
 [lints]
 workspace = true
 
 [package.metadata.cargo-shear]
 ignored = ["yip-io"]

bin/yipd/src/config.rs

@@ -13,7 +13,10 @@ use crate::mode::TunnelMode;
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct PeerConfig {
     pub public_key: [u8; 32],
-    pub endpoint: SocketAddr,
+    /// This peer's known direct UDP endpoint, or `None` if the peer is known
+    /// only by public key (reachable only via rendezvous/relay once Task 6
+    /// wires that path in).
+    pub endpoint: Option<SocketAddr>,
 }
 
 /// Static configuration for one yip tunnel endpoint.
@@ -32,6 +35,10 @@ pub struct Config {
     pub device: String,
     /// Tunnel mode selected from `device_kind=tun|tap` (`tun` by default).
     pub device_kind: TunnelMode,
+    /// Configured rendezvous+relay server, if any (`rendezvous=<IP:port>`).
+    /// Enables lazy Direct→Punch→Relay peer bring-up in `PeerManager` via
+    /// `ConfiguredServerRendezvous`.
+    pub rendezvous: Option<SocketAddr>,
 }
 
 // ── hex decode helper ─────────────────────────────────────────────────────────
@@ -82,16 +89,21 @@ fn flush_peer_block(
     cur_ep: Option<SocketAddr>,
     peers: &mut Vec<PeerConfig>,
 ) -> io::Result<()> {
-    if let (Some(pk), Some(ep)) = (cur_pk, cur_ep) {
-        peers.push(PeerConfig {
+    match cur_pk {
+        // `endpoint` is optional: a peer known only by public key is
+        // rendezvous-only (unreachable directly until Task 6 supplies a
+        // candidate).
+        Some(pk) => peers.push(PeerConfig {
             public_key: pk,
-            endpoint: ep,
-        });
-    } else if cur_pk.is_some() || cur_ep.is_some() {
-        return Err(io::Error::new(
-            io::ErrorKind::InvalidData,
-            "peer block missing public_key or endpoint".to_string(),
-        ));
+            endpoint: cur_ep,
+        }),
+        None if cur_ep.is_some() => {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "peer block has an endpoint but no public_key".to_string(),
+            ));
+        }
+        None => {}
     }
     Ok(())
 }
@@ -116,6 +128,7 @@ impl Config {
         let mut listen: Option<SocketAddr> = None;
         let mut device: Option<String> = None;
         let mut device_kind = TunnelMode::default();
+        let mut rendezvous: Option<SocketAddr> = None;
 
         for line in text.lines() {
             let line = line.trim();
@@ -164,6 +177,12 @@ impl Config {
                 }
                 "device" => device = Some(val.to_owned()),
                 "device_kind" => device_kind = TunnelMode::parse_device_kind(val)?,
+                "rendezvous" => {
+                    rendezvous =
+                        Some(val.parse::<SocketAddr>().map_err(|e| {
+                            io::Error::new(io::ErrorKind::InvalidData, e.to_string())
+                        })?)
+                }
                 // Silently ignore unknown keys for forward-compatibility. The
                 // netns config files still contain `initiate=true|false` from
                 // before Task 5 removed the field; this is intentional so
@@ -182,7 +201,7 @@ impl Config {
             if let (Some(pk), Some(ep)) = (legacy_peer_public, legacy_peer_endpoint) {
                 peers.push(PeerConfig {
                     public_key: pk,
-                    endpoint: ep,
+                    endpoint: Some(ep),
                 });
             }
         }
@@ -203,6 +222,7 @@ impl Config {
             listen: listen.ok_or_else(|| missing("listen"))?,
             device: device.ok_or_else(|| missing("device"))?,
             device_kind,
+            rendezvous,
         })
     }
 }
@@ -421,7 +441,10 @@ peer_public=0000000000000000000000000000000000000000000000000000000000000003
                     [peer]\npublic_key=00000000000000000000000000000000000000000000000000000000000000b2\nendpoint=10.0.0.3:51820\n";
         let cfg = Config::parse(text).expect("parses");
         assert_eq!(cfg.peers.len(), 2);
-        assert_eq!(cfg.peers[0].endpoint, "10.0.0.2:51820".parse().unwrap());
+        assert_eq!(
+            cfg.peers[0].endpoint,
+            Some("10.0.0.2:51820".parse().unwrap())
+        );
         assert_eq!(cfg.peers[1].public_key[31], 0xb2);
     }
 
@@ -435,4 +458,33 @@ peer_public=0000000000000000000000000000000000000000000000000000000000000003
         assert_eq!(cfg.peers.len(), 1);
         assert_eq!(cfg.peers[0].public_key[31], 0xbb);
     }
+
+    #[test]
+    fn parses_rendezvous_and_optional_endpoint() {
+        let text = "local_private=00000000000000000000000000000000000000000000000000000000000000ff\n\
+                    local_public=000000000000000000000000000000000000000000000000000000000000aa01\n\
+                    listen=0.0.0.0:51820\ndevice=yip0\nrendezvous=203.0.113.1:51821\n\
+                    [peer]\npublic_key=00000000000000000000000000000000000000000000000000000000000000b1\n";
+        let cfg = Config::parse(text).expect("parses");
+        assert_eq!(cfg.rendezvous, Some("203.0.113.1:51821".parse().unwrap()));
+        assert_eq!(cfg.peers.len(), 1);
+        assert_eq!(
+            cfg.peers[0].endpoint, None,
+            "peer with no endpoint is rendezvous-only"
+        );
+    }
+
+    #[test]
+    fn rendezvous_absent_is_none() {
+        let text = "local_private=00000000000000000000000000000000000000000000000000000000000000ff\n\
+                    local_public=000000000000000000000000000000000000000000000000000000000000aa01\n\
+                    listen=0.0.0.0:51820\ndevice=yip0\n\
+                    [peer]\npublic_key=00000000000000000000000000000000000000000000000000000000000000b1\nendpoint=10.0.0.2:51820\n";
+        let cfg = Config::parse(text).unwrap();
+        assert_eq!(cfg.rendezvous, None);
+        assert_eq!(
+            cfg.peers[0].endpoint,
+            Some("10.0.0.2:51820".parse().unwrap())
+        );
+    }
 }

bin/yipd/src/main.rs

@@ -9,7 +9,9 @@ mod dataplane;
 mod handshake;
 mod mac_table;
 mod mode;
+mod path;
 mod peer_manager;
+mod rendezvous;
 mod tunnel;
 mod wire_glue;