feat(marketplace): realise a node's tunnel on the route server

ADMIN_API_ENDPOINTS.md

@@ -3898,7 +3898,7 @@ Body:
 Where tunnel inner addresses are allocated from — the tunnel equivalent of an
 `ip_range`. A pool names the route server that terminates the peers, the region
 whose nodes may use it, the socket the data plane listens on, and the blocks
-that point-to-point links are carved out of.
+that node addresses are carved out of.
 
 **LNVPS manages the interface.** Creating a pool generates its WireGuard
 keypair, stores the private key encrypted, and configures the interface on the
@@ -3928,10 +3928,20 @@ Required Permission: `router::view`
 Query Parameters: `limit` (default 50, max 100), `offset`, `region_id`.
 
 Returns a paginated list of `AdminTunnelPoolInfo`. `links_used` is how many
-links are already carved out; `links_total` is what the **smaller** of the two
-blocks can supply, because a dual-stack pool hands out one link of each family
-together — reporting the roomier block would promise capacity that cannot be
-allocated. `endpoint` is derived from `listen_addr` and `listen_port` (IPv6
+nodes are already placed; `links_total` is what the **smaller** of the two
+blocks can supply, because a dual-stack pool hands out one address of each
+family together — reporting the roomier block would promise capacity that cannot
+be allocated.
+
+A node takes a single address (`/32` or `/128`), not a link: WireGuard is layer
+3 and point-to-point, so the node needs no gateway on its side and a
+point-to-point link would spend two addresses describing something that needs
+one. The route server holds **one** address for the whole pool, carrying the
+block's own prefix so every node in it is on-link — an address per node would
+put thousands of addresses on a single interface for a /16 pool. That block's
+network address, the route server's address after it, and (on IPv4) its
+broadcast address are reserved, which is why a /24 places 253 nodes rather than
+256. `endpoint` is derived from `listen_addr` and `listen_port` (IPv6
 bracketed) so what a peer is told to dial cannot disagree with the socket that
 was configured.
 
@@ -4007,6 +4017,19 @@ or port has drifted is re-applied, and a pool whose stored keypair does not
 agree with itself is refused rather than configured into something no node could
 connect to.
 
+The sync then **reconciles the pool's peers**: every tunnel allocated from the
+pool is configured on the interface with its `AllowedIPs`, the route server gets
+its own address on the pool, and each guest address is routed down
+the interface. This always runs, and it matters most right after the interface
+was created or re-applied — that is a fresh interface with no peers at all, and
+every node on it is cut until they are put back.
+
+Peers are reconciled again on the routine router poll, so a peer wiped by a
+reboot, a stale key nobody removed, or a guest address assigned since the last
+push is corrected without an admin doing anything. A peer on an LNVPS-managed
+interface that no allocation accounts for is **removed**: `wgln*` interfaces are
+LNVPS's, so an unclaimed key there is either a revoked node or somebody else's.
+
 ### Reports
 
 #### Time Series Report

API_CHANGELOG.md

@@ -56,7 +56,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
 ### Added
 
-- **Marketplace nodes get a data plane** — an approved node can now ask for the WireGuard tunnel that carries its guests' traffic back to an LNVPS route server. `POST /api/v1/node/tunnel` (node token) presents the node's WireGuard **public** key and returns everything needed to bring the tunnel up: its own inner addresses, the route server's addresses on the same links, the server's public key, the `host:port` to dial, the keepalive and the MTU; `GET /api/v1/node/tunnel` reads the same back after a restart. The private key is generated on the node and never leaves it, which is why the key is presented rather than issued. Idempotent: retrying returns the allocation already made, and a node that presents a *new* key (restored from backup, state directory lost) is re-pinned in place without its addresses moving — refusing would leave a machine that can never be reached again. Addresses come from a new **tunnel pool**, administered at `GET/POST /api/admin/v1/tunnel_pools`, `GET/PATCH/DELETE /api/admin/v1/tunnel_pools/{id}` and `POST /api/admin/v1/tunnel_pools/{id}/sync` under the `router` resource: a route server, a region, the socket the data plane listens on (`listen_addr` + `listen_port`, unique per route server because a WireGuard interface listens on every local address at its port) and the v4/v6 blocks that /31 and /127 links are carved from. **LNVPS manages the interface end to end**: creating a pool generates its keypair, stores the private key encrypted and configures the interface on the route server; editing re-applies it; deleting removes it. The interface is named `wgln<id>` from the pool's id and is not settable — a route server also carries interfaces LNVPS did not configure, and a fixed prefix keeps a managed one from being confused with, or clobbering, those. An admin never pastes key material for an interface they set up by hand, so standing up a new route server is an API call rather than a manual job with a database row bolted on afterwards — an existing interface can still be adopted by supplying its private key, and the public half is always derived, never accepted. The private key is never returned by any endpoint. Allocation also fills in the blank control address that node approval leaves on the backing host — the node's control API is only reachable through the tunnel. The host stays **disabled**: an allocation is paperwork, not a working tunnel, and the peer is not configured on the route server until the next increment.
+- **A node takes one address, not a point-to-point link** — `address4`/`address6` in the tunnel response are now a `/32` and a `/128`, and `gateway4`/`gateway6` are one address shared by every node on the pool rather than a per-node link address. WireGuard is layer 3 and point-to-point: the node needs no gateway on its own side (`ip route add default dev wg0` suffices), so a `/31` spent two addresses describing something that needs one — and forced the route server to carry one address per node on a single interface, thousands of them on a /16 pool. Pool capacity is reported accordingly: a /24 places 253 nodes (256 less the block's network address, the route server's address after it, and the broadcast address) where it previously reported 128 links.
+
+- **A marketplace node's tunnel is now actually configured on the route server** — allocating a tunnel (`POST /api/v1/node/tunnel`) previously wrote down addresses and a key and stopped there; the node's peer is now pushed to the route server, so the tunnel carries traffic instead of only existing on paper. The peer's `AllowedIPs` is the node's own inner addresses plus exactly the guest addresses LNVPS assigned to it, which is also the anti-spoof boundary: WireGuard drops an inbound packet whose source is not on that list, so a node cannot source traffic as another node's customer. The route server takes an address on each point-to-point link, and each guest address is routed down the interface — `AllowedIPs` decides which *peer* a packet already headed for the tunnel belongs to, it does not put the packet there. Peers are also reconciled on the routine router poll, so an address assigned to a guest since the last push, a peer wiped by a route-server reboot, and a stale key nobody removed are all corrected without an admin doing anything; the reconcile reports what had drifted rather than quietly fixing it. The backing host stays **disabled** — a configured peer is still not a proven path, and the health gate that enables it comes with the node-side data plane.
+
+- **Marketplace nodes get a data plane** — an approved node can now ask for the WireGuard tunnel that carries its guests' traffic back to an LNVPS route server. `POST /api/v1/node/tunnel` (node token) presents the node's WireGuard **public** key and returns everything needed to bring the tunnel up: its own inner addresses (a `/32` and a `/128`), the route server's address on the pool, the server's public key, the `host:port` to dial, the keepalive and the MTU; `GET /api/v1/node/tunnel` reads the same back after a restart. The private key is generated on the node and never leaves it, which is why the key is presented rather than issued. Idempotent: retrying returns the allocation already made, and a node that presents a *new* key (restored from backup, state directory lost) is re-pinned in place without its addresses moving — refusing would leave a machine that can never be reached again. Addresses come from a new **tunnel pool**, administered at `GET/POST /api/admin/v1/tunnel_pools`, `GET/PATCH/DELETE /api/admin/v1/tunnel_pools/{id}` and `POST /api/admin/v1/tunnel_pools/{id}/sync` under the `router` resource: a route server, a region, the socket the data plane listens on (`listen_addr` + `listen_port`, unique per route server because a WireGuard interface listens on every local address at its port) and the v4/v6 blocks that node addresses are carved from. **LNVPS manages the interface end to end**: creating a pool generates its keypair, stores the private key encrypted and configures the interface on the route server; editing re-applies it; deleting removes it. The interface is named `wgln<id>` from the pool's id and is not settable — a route server also carries interfaces LNVPS did not configure, and a fixed prefix keeps a managed one from being confused with, or clobbering, those. An admin never pastes key material for an interface they set up by hand, so standing up a new route server is an API call rather than a manual job with a database row bolted on afterwards — an existing interface can still be adopted by supplying its private key, and the public half is always derived, never accepted. The private key is never returned by any endpoint. Allocation also fills in the blank control address that node approval leaves on the backing host — the node's control API is only reachable through the tunnel. The host stays **disabled**: an allocation is paperwork, not a working tunnel, and the peer is not configured on the route server until the next increment.
 
 - **Admin approval for marketplace nodes** — operator-registered hardware can now be reviewed and admitted to the fleet. `GET /api/admin/v1/marketplace/nodes` (filterable by `status`/`operator_id`) is the review queue, `POST /api/admin/v1/marketplace/nodes/{id}/approve` admits a node, `PATCH /api/admin/v1/marketplace/nodes/{id}` suspends, drains or re-tiers it, and `DELETE /api/admin/v1/marketplace/nodes/{id}` rejects a registration outright — there is no `rejected` state, so an operator can fix what was wrong and register the same machine again. Approval is the **only** transition into `approved` and the only place the gates are checked: a node with no pinned TLS certificate cannot be reached and is refused, and where the region's company charges a listing fee, that fee must be paid *to that company* — a fee paid to a cheaper company does not list hardware elsewhere. Approving a node creates its backing host **disabled and with an empty control address**, because the node is reachable only over a data-plane tunnel that does not exist yet; capacity defaults to zero rather than being guessed, since real figures arrive with node telemetry. Suspending or draining also disables that host, so placement stops immediately instead of depending on the scheduler learning about node status. Two new permission resources, deliberately separate: `marketplace_node` (placement state) and `marketplace_operator` (revenue-share override and payout configuration, via `GET`/`PATCH /api/admin/v1/marketplace/operators[/{id}]`) — stopping a misbehaving node should not require the ability to change what someone is paid.
 

lnvps_api/src/api/marketplace.rs

@@ -18,8 +18,8 @@ use serde::{Deserialize, Serialize};
 
 use chrono::Utc;
 use lnvps_api_common::{
-    ApiData, ApiError, ApiResult, NODE_TOKEN_TTL_SECS, Nip98Auth, NodeAuth, issue_node_token,
-    session_auth_enabled,
+    ApiData, ApiError, ApiResult, NODE_TOKEN_TTL_SECS, Nip98Auth, NodeAuth, WorkJob,
+    issue_node_token, session_auth_enabled,
 };
 use lnvps_db::{
     IntervalType, MarketplaceNode, MarketplaceNodeStatus, MarketplaceOperator,
@@ -574,6 +574,25 @@ async fn v1_node_request_tunnel(
     let allocation = crate::provisioner::allocate_node_tunnel(&this.db, &auth.node, &key)
         .await
         .map_err(|e| ApiError::new(e.to_string()))?;
+
+    // Realise the peer on the route server. Queued rather than awaited: the
+    // node has what it needs to configure its own end either way, and making
+    // the answer wait on an SSH round trip to a route server would fail the
+    // request for something the node cannot fix. A failure to queue is logged
+    // and left to the periodic reconcile, which pushes the same peer.
+    if let Err(e) = this
+        .work_sender
+        .send(WorkJob::SyncNodeTunnel {
+            tunnel_id: allocation.tunnel.id,
+        })
+        .await
+    {
+        log::error!(
+            "Allocated tunnel {} for node {} but could not queue its peer push: {e}",
+            allocation.tunnel.id,
+            auth.node.id
+        );
+    }
     ApiData::ok(allocation.into())
 }
 
@@ -1002,12 +1021,14 @@ mod tests {
         let allocation = crate::provisioner::NodeTunnel {
             tunnel: lnvps_db::Tunnel {
                 id: 1,
-                address4: Some("10.66.0.1/31".to_string()),
-                address6: Some("fd00:66::1/127".to_string()),
+                address4: Some("10.66.0.2/32".to_string()),
+                address6: Some("fd00:66::2/128".to_string()),
                 keepalive: Some(25),
                 ..Default::default()
             },
             pool: lnvps_db::TunnelPool {
+                cidr4: Some("10.66.0.0/24".to_string()),
+                cidr6: Some("fd00:66::/64".to_string()),
                 public_key: vec![0x33; 32],
                 listen_addr: "rs.example".to_string(),
                 listen_port: 51820,
@@ -1017,12 +1038,12 @@ mod tests {
         };
 
         let api: ApiNodeTunnel = allocation.into();
-        assert_eq!(api.address4.as_deref(), Some("10.66.0.1/31"));
-        assert_eq!(api.address6.as_deref(), Some("fd00:66::1/127"));
-        // The gateway is derived from the link rather than stored, so a second
-        // copy cannot disagree with the address it is paired with.
-        assert_eq!(api.gateway4.as_deref(), Some("10.66.0.0"));
-        assert_eq!(api.gateway6.as_deref(), Some("fd00:66::"));
+        assert_eq!(api.address4.as_deref(), Some("10.66.0.2/32"));
+        assert_eq!(api.address6.as_deref(), Some("fd00:66::2/128"));
+        // The gateway is derived from the pool's block rather than stored, and
+        // is the same one address for every node on the pool.
+        assert_eq!(api.gateway4.as_deref(), Some("10.66.0.1"));
+        assert_eq!(api.gateway6.as_deref(), Some("fd00:66::1"));
         assert_eq!(api.server_public_key, hex::encode([0x33; 32]));
         assert_eq!(api.endpoint, "rs.example:51820");
         assert_eq!(api.keepalive, Some(25));

lnvps_api/src/mocks.rs

@@ -52,6 +52,10 @@ pub struct MockRouter {
     tunnels: Arc<Mutex<HashMap<String, Tunnel>>>,
     sessions: Arc<Mutex<HashMap<String, BgpSession>>>,
     default_route: Arc<Mutex<Option<BgpRoute>>>,
+    /// Addresses configured per tunnel interface
+    addresses: Arc<Mutex<HashMap<String, Vec<String>>>>,
+    /// Routes pointing down each tunnel interface
+    routes: Arc<Mutex<HashMap<String, Vec<String>>>>,
 }
 
 impl Default for MockRouter {
@@ -75,6 +79,10 @@ impl MockRouter {
                 Arc::new(Mutex::new(HashMap::new()));
             static TL_SESSIONS: Arc<Mutex<HashMap<String, BgpSession>>> =
                 Arc::new(Mutex::new(HashMap::new()));
+            static TL_ADDRESSES: Arc<Mutex<HashMap<String, Vec<String>>>> =
+                Arc::new(Mutex::new(HashMap::new()));
+            static TL_ROUTES: Arc<Mutex<HashMap<String, Vec<String>>>> =
+                Arc::new(Mutex::new(HashMap::new()));
             static TL_DEFAULT_ROUTE: Arc<Mutex<Option<BgpRoute>>> =
                 Arc::new(Mutex::new(Some(BgpRoute {
                     prefix: "0.0.0.0/0".to_string(),
@@ -87,6 +95,8 @@ impl MockRouter {
             tunnels: TL_TUNNELS.with(|t| t.clone()),
             sessions: TL_SESSIONS.with(|s| s.clone()),
             default_route: TL_DEFAULT_ROUTE.with(|d| d.clone()),
+            addresses: TL_ADDRESSES.with(|a| a.clone()),
+            routes: TL_ROUTES.with(|r| r.clone()),
         }
     }
 
@@ -98,6 +108,31 @@ impl MockRouter {
         tunnels.clear();
         let mut sessions = self.sessions.lock().await;
         sessions.clear();
+        let mut addresses = self.addresses.lock().await;
+        addresses.clear();
+        let mut routes = self.routes.lock().await;
+        routes.clear();
+    }
+
+    /// Addresses configured on a tunnel interface
+    pub async fn interface_addresses(&self, interface: &str) -> Vec<String> {
+        let addresses = self.addresses.lock().await;
+        addresses.get(interface).cloned().unwrap_or_default()
+    }
+
+    /// Routes pointing down a tunnel interface
+    pub async fn interface_routes(&self, interface: &str) -> Vec<String> {
+        let routes = self.routes.lock().await;
+        routes.get(interface).cloned().unwrap_or_default()
+    }
+
+    /// Peers configured on a tunnel interface
+    pub async fn peers(&self, interface: &str) -> Vec<crate::router::WireguardPeer> {
+        let tunnels = self.tunnels.lock().await;
+        match tunnels.get(interface).map(|t| &t.config) {
+            Some(crate::router::TunnelConfig::Wireguard(c)) => c.peers.clone(),
+            _ => vec![],
+        }
     }
 
     /// Seed a BGP session for tests
@@ -257,6 +292,48 @@ impl TunnelRouter for MockRouter {
         Ok(())
     }
 
+    async fn set_tunnel_peer(
+        &self,
+        interface: &str,
+        peer: &crate::router::WireguardPeer,
+    ) -> OpResult<()> {
+        let mut tunnels = self.tunnels.lock().await;
+        let Some(t) = tunnels.get_mut(interface) else {
+            return Err(OpError::Fatal(anyhow::anyhow!(
+                "No such tunnel interface: {interface}"
+            )));
+        };
+        if let crate::router::TunnelConfig::Wireguard(c) = &mut t.config {
+            // Keyed by public key, like the real thing: pushing the same peer
+            // twice updates it rather than duplicating it.
+            c.peers.retain(|p| p.public_key != peer.public_key);
+            c.peers.push(peer.clone());
+        }
+        Ok(())
+    }
+
+    async fn remove_tunnel_peer(&self, interface: &str, public_key: &str) -> OpResult<()> {
+        let mut tunnels = self.tunnels.lock().await;
+        if let Some(t) = tunnels.get_mut(interface)
+            && let crate::router::TunnelConfig::Wireguard(c) = &mut t.config
+        {
+            c.peers.retain(|p| p.public_key != public_key);
+        }
+        Ok(())
+    }
+
+    async fn sync_tunnel_addresses(&self, interface: &str, addresses: &[String]) -> OpResult<()> {
+        let mut map = self.addresses.lock().await;
+        map.insert(interface.to_string(), addresses.to_vec());
+        Ok(())
+    }
+
+    async fn sync_tunnel_routes(&self, interface: &str, prefixes: &[String]) -> OpResult<()> {
+        let mut map = self.routes.lock().await;
+        map.insert(interface.to_string(), prefixes.to_vec());
+        Ok(())
+    }
+
     async fn tunnel_traffic(&self) -> OpResult<Vec<TunnelTraffic>> {
         let tunnels = self.tunnels.lock().await;
         Ok(tunnels