From 3074291fb90d2e61dac875095993e43abc5062d6 Mon Sep 17 00:00:00 2001 From: v0l Date: Fri, 7 Aug 2026 12:47:14 +0100 Subject: [PATCH] feat(marketplace): a node proves it can carry a customer before it takes one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Increment 4c3b, and the end of 4c. Approving a node no longer enables it. Everything between an admin approving hardware they cannot see and a customer's VM working — a tunnel that handshakes, a route server that routes, a bridge, a packet filter, a forwarding knob — is machinery nobody has tested on that particular machine. The gate tests it, on the customer's own path: an address from a real customer range in the node's region, sent to the node as a guest, pinged **from the route server**. That path is the point. A VM's address is statically routed from the core network to the node's tunnel address, forwarded across the guest bridge, and answered back out the node's default route, while the guest routes to the core router's address that the node answers for by proxy ARP. A probe on any other address, or from anywhere else, would test a path no customer takes — and the failure worth catching is the node that believes it is fine. The gate runs when a node's tunnel is allocated, which is the first moment it can run at all, and on demand from the admin API for the cases after that: an operator who has fixed their firewall, a node whose route server was down. Decisions worth the words: - **The run is recorded before it starts.** A gate that only wrote a row once it had taken an address would tell the operator whose node never handshook precisely nothing — and that is the most common failure. - **The address is released in the same statement as the verdict.** A gate that recorded its result and then failed to give the address back would leak one address per attempt out of a range customers are waiting for. - **The allocator can see held probe addresses.** They are not `vm_ip_assignment` rows, so without that a VM could be handed the address a gate is proving a node with — two machines answering for one address, which is the failure this increment exists to prevent, not cause. - **IPv4 ranges first.** A guest's IPv6 address is normally derived from its MAC and a probe has no guest, so the v4 path is the one a probe stands in for exactly. A v6-only region is still gated, from the first free address. - **The node is told to apply the document rather than left to its heartbeat**, through a new `POST /api/v1/dataplane/refresh` on the node control API. Nothing about the document is sent: the node fetches it itself, with its own credential. This only says *when* — an approval that took a minute to conclude would be a minute of an operator watching nothing happen. - **`probe_address` echoes its verdict instead of using ping's exit code.** To the SSH transport a non-zero exit means "the command failed", which is right everywhere else and wrong here: "no reply" is the answer the gate asked for, and it has to stay distinguishable from a route server that cannot be reached at all. Those two results need different people. - **A failure never retries forever.** A failed gate is a recorded verdict, not a job to repeat; `Err` is reserved for not being able to *ask*, which is LNVPS's problem rather than the operator's. Caught while writing the tests: the node's guest list (`node_guests`) and the route server's plan (`guest_addresses`) are two separate functions over the same idea. Adding the probe to one and not the other would have produced a node holding an address the route server never routed — a gate failing for a reason that had nothing to do with the node. A failure leaves the node approved and unusable with the failing step named, which is the safe direction: a node that never carries a customer is a support conversation, and one that carries a customer badly is an outage. `GET /marketplace/nodes/{id}` gains `host_enabled` and `probe`, which together answer "why is my node approved and empty?". --- ADMIN_API_ENDPOINTS.md | 48 ++ API_CHANGELOG.md | 2 + lnvps_api/src/api/marketplace.rs | 18 + lnvps_api/src/mocks.rs | 30 +- lnvps_api/src/provisioner/tunnel.rs | 41 ++ lnvps_api/src/router/linux_ssh.rs | 87 +++ lnvps_api/src/router/mod.rs | 49 ++ lnvps_api/src/settings.rs | 8 +- lnvps_api/src/worker.rs | 643 +++++++++++++++++- lnvps_api_admin/src/admin/marketplace.rs | 71 +- lnvps_api_common/src/mock.rs | 124 +++- lnvps_api_common/src/network.rs | 12 + lnvps_api_common/src/node_control.rs | 73 +- lnvps_api_common/src/work/mod.rs | 12 + .../20260810120000_marketplace_node_probe.sql | 75 ++ lnvps_db/src/lib.rs | 43 ++ lnvps_db/src/model.rs | 45 ++ lnvps_db/src/mysql.rs | 104 ++- lnvps_node/src/control.rs | 66 +- lnvps_node/src/main.rs | 41 +- work/marketplace.md | 31 +- 21 files changed, 1589 insertions(+), 34 deletions(-) create mode 100644 lnvps_db/migrations/20260810120000_marketplace_node_probe.sql diff --git a/ADMIN_API_ENDPOINTS.md b/ADMIN_API_ENDPOINTS.md index ccdd48f9..11ce6820 100644 --- a/ADMIN_API_ENDPOINTS.md +++ b/ADMIN_API_ENDPOINTS.md @@ -3764,6 +3764,28 @@ carries `fee_paid` (whether the one-off listing fee has settled), reached) and `host_id` (the backing host, `null` until approval creates it). Node tokens are never returned — LNVPS keeps no copy of them. +Each node carries `host_enabled` and `probe`. Approval alone does not enable a +node: the health gate does, so a node reading `approved` with `host_enabled: +false` is waiting on (or has failed) its gate, and `probe.detail` says which +step. That pair answers "why is my node approved and empty?". + +```json +{ + "host_enabled": false, + "probe": { + "status": "failed", + // running|passed|failed + "ip": null, + // The address held while the gate runs; released when it finishes + "ip_range_id": 4, + // null when the run was refused before an address was worth taking + "detail": "The route server could not reach 203.0.113.9 on this node, so a customer's VM would not be reachable either", + "created": "2026-08-10T09:00:00Z", + "finished": "2026-08-10T09:00:12Z" + } +} +``` + #### Get Node ``` @@ -3772,6 +3794,32 @@ GET /api/admin/v1/marketplace/nodes/{id} Required Permission: `marketplace_node::view` +#### Run the Health Gate + +``` +POST /api/admin/v1/marketplace/nodes/{id}/health_check +``` + +Required Permission: `marketplace_node::update` + +Queues a health-gate run. The gate takes an address from a real customer range +in the node's region, sends it to the node as a guest, and pings it **from the +route server** — which is the path a customer's traffic takes: the address is +statically routed from the core network to the node's tunnel address, forwarded +across the guest bridge, and answered back out the node's default route. On a +pass the backing host is **enabled**; on a failure it is left approved and +disabled, with the failing step recorded. + +Queued rather than run inline: it waits on a route server and a tunnel, which is +not work to hold an HTTP request open for. The verdict lands on the node's +`probe` field, returned by `GET /marketplace/nodes/{id}`. + +The gate also runs automatically when a node's tunnel is allocated, which is the +first moment it can run at all. This endpoint is for the cases after that — an +operator who has fixed their firewall, a node whose route server was down. + +Errors: `400` when the node has no host yet (not approved). + #### Get Live Node Status ``` diff --git a/API_CHANGELOG.md b/API_CHANGELOG.md index 2dd8fc1d..35a648fe 100644 --- a/API_CHANGELOG.md +++ b/API_CHANGELOG.md @@ -58,6 +58,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). - **The route server now routes its tunnel pool's own blocks.** An address on a point-to-point interface does not route the rest of its prefix, so a route server holding `10.66.0.1/16` answered "network is unreachable" for every node in that pool — the peers were configured correctly and unreachable. Found by an end-to-end harness that builds both ends of a tunnel and sends real packets, not by review. +- **A marketplace node has to prove it can carry a customer before it takes one** — approving a node no longer enables it. A health gate takes an address from a real customer range in the node's region, sends it to the node as a guest, and pings it from the route server; only then is the backing host enabled. That is the production path exactly, which is the point: a VM's address is statically routed from the core network to the node's tunnel address, forwarded across the guest bridge, and answered back out the node's default route, while the guest routes to the core router's address that the node answers for by proxy ARP. Asking the node how it thinks it is doing tests none of that, and the failure worth catching is the node that believes it is fine. It runs automatically when a node's tunnel is allocated — the first moment it can — and on demand via `POST /api/admin/v1/marketplace/nodes/{id}/health_check`. A failure leaves the node approved and unusable with the failing step named, which is the safe direction: a node that never carries a customer is a support conversation, and one that carries a customer badly is an outage. `GET /api/admin/v1/marketplace/nodes/{id}` gains `host_enabled` and `probe`, which together answer "why is my node approved and empty?". + - **LNVPS can call a node** — `GET /api/admin/v1/marketplace/nodes/{id}/status` (admin, `marketplace_node::view`) returns what the node says about itself right now: daemon version, tunnel state and handshake age, bridge, forwarding, routed guest count, and the packet filter. Signed with LNVPS's existing nostr identity (`nostr.nsec` — the account customers DM for support, `npub1lnvps32qq2nvg75cqwflq4y6cmnzn55d26ypzjakpkp3khqcx2ns7t7vjj`), which the admin API now also reads; without it the endpoint answers that this deployment runs no marketplace rather than failing obscurely. One identity rather than a control key of its own: a separate secret would have to be generated, distributed to whoever builds the node binaries and kept in step with the value compiled into them, whereas this one is already published — an operator can check the key their node obeys against an account that publicly answers. Both ends of the call are authenticated — the request is a NIP-98 event signed with LNVPS's control key, which the node checks against a public key compiled into its binary, and the node's TLS certificate is checked against the fingerprint it registered. There is no CA: the node is self-signed and its name is an address inside a tunnel, so what is verified is the pin. Without server authentication anything able to answer on that address — a guest that grabbed the IP, a mistake on the route server — could report that a VM is running when it is not. - **A node's status now reports the packet filter around its guests** — `GET /status` on the node control API gains `dataplane.firewall`: whether the machine has a working nftables, whether LNVPS's ruleset is loaded, whether guests are isolated from each other at layer 2, how many guest bindings are enforced, which ruleset the *kernel* is running (a tag read back out of a rule comment, not remembered by the daemon), and how many packets have been dropped for claiming an address the guest was not assigned. That last number is the only one here that says something about a customer rather than a node: a guest that is spoofing is either compromised or hostile, and LNVPS would rather learn it from a counter than from an upstream abuse report. A node whose filter is not loaded now reports itself **unhealthy** — an unfiltered node is one where any guest can be any other, which is worse than a node carrying nobody. diff --git a/lnvps_api/src/api/marketplace.rs b/lnvps_api/src/api/marketplace.rs index 75387689..6cd2d473 100644 --- a/lnvps_api/src/api/marketplace.rs +++ b/lnvps_api/src/api/marketplace.rs @@ -594,6 +594,24 @@ async fn v1_node_request_tunnel( auth.node.id ); } + + // A node with a tunnel is a node that can finally be tested end to end. + // Queued here rather than at approval, because at approval there was no + // tunnel and nothing to test — this is the first moment the gate can run, + // and a node whose hardware works should not wait for an admin to notice. + if let Err(e) = this + .work_sender + .send(WorkJob::HealthCheckNode { + node_id: auth.node.id, + }) + .await + { + log::error!( + "Allocated tunnel {} for node {} but could not queue its health check: {e}", + allocation.tunnel.id, + auth.node.id + ); + } ApiData::ok(allocation.into()) } diff --git a/lnvps_api/src/mocks.rs b/lnvps_api/src/mocks.rs index 8ddadbee..ddba94b8 100644 --- a/lnvps_api/src/mocks.rs +++ b/lnvps_api/src/mocks.rs @@ -39,7 +39,7 @@ use payments_rs::onchain::{ ChainPaymentUpdate, NewAddressRequest, NewAddressResponse, OnChainProvider, PaymentCursor, SendCoinsRequest, SendCoinsResponse, }; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::ops::Add; use std::pin::Pin; use std::sync::Arc; @@ -56,6 +56,11 @@ pub struct MockRouter { addresses: Arc>>>, /// Routes pointing down each tunnel interface routes: Arc>>>, + /// Whether this route server can reach anything at all. A caller cannot + /// know in advance which address the code under test will pick, so the + /// switch is all-or-nothing. + reach_all: Arc>, + probed: Arc>>, } impl Default for MockRouter { @@ -83,6 +88,8 @@ impl MockRouter { Arc::new(Mutex::new(HashMap::new())); static TL_ROUTES: Arc>>> = Arc::new(Mutex::new(HashMap::new())); + static TL_PROBED: Arc>> = Arc::new(Mutex::new(Vec::new())); + static TL_REACH_ALL: Arc> = Arc::new(Mutex::new(false)); static TL_DEFAULT_ROUTE: Arc>> = Arc::new(Mutex::new(Some(BgpRoute { prefix: "0.0.0.0/0".to_string(), @@ -97,6 +104,8 @@ impl MockRouter { default_route: TL_DEFAULT_ROUTE.with(|d| d.clone()), addresses: TL_ADDRESSES.with(|a| a.clone()), routes: TL_ROUTES.with(|r| r.clone()), + reach_all: TL_REACH_ALL.with(|r| r.clone()), + probed: TL_PROBED.with(|p| p.clone()), } } @@ -112,6 +121,8 @@ impl MockRouter { addresses.clear(); let mut routes = self.routes.lock().await; routes.clear(); + *self.reach_all.lock().await = false; + self.probed.lock().await.clear(); } /// Addresses configured on a tunnel interface @@ -121,6 +132,16 @@ impl MockRouter { } /// Routes pointing down a tunnel interface + /// Make this route server able to reach every address it is asked about. + pub async fn set_reach_all(&self, reach: bool) { + *self.reach_all.lock().await = reach; + } + + /// Every address this route server has been asked to probe. + pub async fn probed(&self) -> Vec { + self.probed.lock().await.clone() + } + pub async fn interface_routes(&self, interface: &str) -> Vec { let routes = self.routes.lock().await; routes.get(interface).cloned().unwrap_or_default() @@ -328,6 +349,13 @@ impl TunnelRouter for MockRouter { Ok(()) } + async fn probe_address(&self, address: &str) -> OpResult { + self.probed.lock().await.push(address.to_string()); + // A route server with nothing configured reaches nothing, which is the + // honest default and the one that makes a test say what it means. + Ok(*self.reach_all.lock().await) + } + 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()); diff --git a/lnvps_api/src/provisioner/tunnel.rs b/lnvps_api/src/provisioner/tunnel.rs index 0d37501d..09efd9ba 100644 --- a/lnvps_api/src/provisioner/tunnel.rs +++ b/lnvps_api/src/provisioner/tunnel.rs @@ -519,6 +519,36 @@ pub async fn node_guests( }); } } + // Any address a health gate is holding against this node. It is sent as a + // guest because that is the whole point: the gate proves the *guest* path — + // the route server's route and AllowedIPs, the node's bridge route, the + // packet filter's binding, proxy ARP for the range's gateway. A probe the + // node treated specially would prove a path no customer takes. + for ip in db.list_marketplace_probe_ips_for_node(node.id).await? { + let Some(probe) = db.get_marketplace_node_probe(node.id).await? else { + continue; + }; + let Some(address) = host_address(Some(&ip)) else { + continue; + }; + let Some(range_id) = probe.ip_range_id else { + continue; + }; + let range = db.get_ip_range(range_id).await?; + let gateway = lnvps_api_common::parse_gateway(&range.gateway) + .map(|g| g.ip().to_string()) + .unwrap_or(range.gateway); + out.push(GuestAddress { + address, + gateway, + // No MAC: nothing owns this address but the node itself, which + // holds it on the bridge rather than handing it to a guest. The + // filter admits it on address alone, which is the weaker check and + // visibly so. + mac: None, + }); + } + out.sort_by(|a, b| a.address.cmp(&b.address)); Ok(out) } @@ -553,6 +583,17 @@ async fn guest_addresses(db: &Arc, tunnel: &Tunnel) -> Result@[:]/` (port defaults to 22) /// - `token`: the SSH private key in PEM format +/// What a successful probe prints. +const PROBE_REACHED: &str = "lnvps-probe-reached"; +/// ...and what an unsuccessful one prints, so a silent command is neither. +const PROBE_UNREACHED: &str = "lnvps-probe-unreached"; + pub struct LinuxSshRouter { host: String, username: String, @@ -695,6 +700,33 @@ impl TunnelRouter for LinuxSshRouter { Ok(()) } + async fn probe_address(&self, address: &str) -> OpResult { + let parsed: std::net::IpAddr = match address.parse() { + Ok(ip) => ip, + // Not a transient failure: something upstream built a probe out of + // a value that is not an address, and retrying cannot fix it. + Err(e) => op_fatal!("{address} is not an address: {e}"), + }; + // `ping` for v4 and v6 alike on any current iproute2/iputils; the flag + // is what selects the family, so a v6 address is never sent to a v4 + // resolver. + let family = if parsed.is_ipv6() { "-6" } else { "-4" }; + // Three attempts, two seconds each: a node that answers at all answers + // in milliseconds — it is one hop away — and a gate that waited longer + // would only be waiting on a node that is not going to answer. + // The verdict is echoed rather than taken from ping's exit code: to + // this transport a non-zero exit means "the command failed", which is + // the right contract for every other call and the wrong one here. "No + // reply" is the answer the gate asked for, not a fault, and it must + // stay distinguishable from a route server that cannot be reached at + // all — those two results need different people. + let command = format!( + "ping {family} -c 3 -W 2 -q {} >/dev/null 2>&1 && echo {PROBE_REACHED} || echo {PROBE_UNREACHED}", + shq(address) + ); + Ok(self.exec_checked(&command).await?.contains(PROBE_REACHED)) + } + async fn sync_tunnel_routes(&self, interface: &str, prefixes: &[String]) -> OpResult<()> { // Both families have to be asked for separately: `ip route show` is // IPv4 only, and a v6 guest prefix would otherwise look like a route @@ -1410,6 +1442,61 @@ mod tests { assert!(ran[2].contains("wgln1"), "{}", ran[2]); } + /// The health gate asks the route server whether it can reach an address, + /// and both answers have to come back as answers. "No reply" is a verdict + /// about the node; a command that could not be run is a fault on the route + /// server, and the two need different people. + #[tokio::test] + async fn test_probe_address_reports_both_answers() { + // A shell that answers as the real one does: the command itself echoes + // the verdict, so the transport's "non-zero means failed" contract is + // left intact for every other call. + let log = std::sync::Arc::new(Mutex::new(Vec::new())); + let sink = log.clone(); + let r = LinuxSshRouter::with_exec(std::sync::Arc::new(move |cmd: &str| { + sink.lock().unwrap().push(cmd.to_string()); + Ok("lnvps-probe-reached\n".to_string()) + })); + + assert!(r.probe_address("203.0.113.9").await.unwrap()); + let ran = log.lock().unwrap().clone(); + assert!(ran[0].contains("ping -4 -c 3"), "{}", ran[0]); + assert!(ran[0].contains("203.0.113.9"), "{}", ran[0]); + + // A v6 address is asked about with the v6 flag, or the command resolves + // the wrong family and answers about nothing. + let log6 = std::sync::Arc::new(Mutex::new(Vec::new())); + let sink6 = log6.clone(); + let r6 = LinuxSshRouter::with_exec(std::sync::Arc::new(move |cmd: &str| { + sink6.lock().unwrap().push(cmd.to_string()); + Ok(String::new()) + })); + let _ = r6.probe_address("2001:db8::9").await; + assert!(log6.lock().unwrap()[0].contains("ping -6"), "{log6:?}"); + } + + /// A silent route server is not a reachable node. The verdict is echoed by + /// the command, so output that says nothing must not be read as success. + #[tokio::test] + async fn test_probe_address_does_not_assume_reachable() { + let r = LinuxSshRouter::with_exec(std::sync::Arc::new(|_cmd: &str| { + Ok("lnvps-probe-unreached".to_string()) + })); + assert!(!r.probe_address("203.0.113.9").await.unwrap()); + + let quiet = LinuxSshRouter::with_exec(std::sync::Arc::new(|_cmd: &str| Ok(String::new()))); + assert!(!quiet.probe_address("203.0.113.9").await.unwrap()); + } + + /// A value that is not an address is a fault upstream, not a node that + /// failed: retrying it forever would never make it an address. + #[tokio::test] + async fn test_probe_address_refuses_a_non_address() { + let r = LinuxSshRouter::with_exec(std::sync::Arc::new(|_cmd: &str| Ok(String::new()))); + let err = r.probe_address("not-an-address").await.unwrap_err(); + assert!(err.to_string().contains("not an address"), "{err}"); + } + /// The link route the kernel installs for the interface's own /31 is the /// link itself. Removing it to tidy a list would break the tunnel. #[tokio::test] diff --git a/lnvps_api/src/router/mod.rs b/lnvps_api/src/router/mod.rs index f30694e1..0f01f67b 100644 --- a/lnvps_api/src/router/mod.rs +++ b/lnvps_api/src/router/mod.rs @@ -212,6 +212,22 @@ pub trait TunnelRouter: Send + Sync { let _ = (interface, prefixes); op_fatal!("This router backend cannot manage tunnel routes") } + + /// Whether the route server can reach `address`. + /// + /// Asked *from the route server* rather than from LNVPS, because that is + /// where a customer's packet comes from: the address is statically routed + /// from the core network to this machine, and it is this machine's routing + /// table, this machine's `AllowedIPs` and this tunnel that decide whether + /// it arrives. A ping from anywhere else would prove a path nobody uses. + /// + /// `false` for unreachable, `Err` only when the question could not be + /// asked — the health gate has to tell "this node is broken" from "the + /// route server is broken", and they need different people. + async fn probe_address(&self, address: &str) -> OpResult { + let _ = address; + op_fatal!("This router backend cannot probe an address") + } } /// The kind of a tunnel interface @@ -466,6 +482,39 @@ mod tests { } } + /// A backend that cannot probe says so rather than answering "no". A + /// silent `false` would fail every node behind a Mikrotik route server and + /// blame the nodes for it. + #[tokio::test] + async fn a_backend_that_cannot_probe_refuses() { + struct NoProbe; + + #[async_trait::async_trait] + impl TunnelRouter for NoProbe { + async fn list_tunnels(&self) -> OpResult> { + Ok(vec![]) + } + async fn add_tunnel(&self, t: &Tunnel) -> OpResult { + Ok(t.clone()) + } + async fn remove_tunnel(&self, _id: &str) -> OpResult<()> { + Ok(()) + } + async fn update_tunnel(&self, t: &Tunnel) -> OpResult { + Ok(t.clone()) + } + async fn set_tunnel_enabled(&self, _id: &str, _enabled: bool) -> OpResult<()> { + Ok(()) + } + async fn tunnel_traffic(&self) -> OpResult> { + Ok(vec![]) + } + } + + let err = NoProbe.probe_address("203.0.113.9").await.unwrap_err(); + assert!(err.to_string().contains("cannot probe"), "{err}"); + } + #[tokio::test] async fn test_mock_tunnel_lifecycle() -> anyhow::Result<()> { let r = MockRouter::new(); diff --git a/lnvps_api/src/settings.rs b/lnvps_api/src/settings.rs index 0df610a1..aed3ad76 100644 --- a/lnvps_api/src/settings.rs +++ b/lnvps_api/src/settings.rs @@ -453,11 +453,9 @@ pub enum CaptchaConfig { Turnstile { secret_key: String }, } -#[derive(Debug, Clone, Deserialize, Serialize)] -pub struct NostrConfig { - pub relays: Vec, - pub nsec: String, -} +/// LNVPS's nostr identity, shared with the admin API and with the marketplace +/// control client — one account, not three keys that can disagree. +pub use lnvps_api_common::node_control::NostrConfig; #[derive(Debug, Clone, Deserialize, Serialize)] #[serde(rename_all = "kebab-case")] diff --git a/lnvps_api/src/worker.rs b/lnvps_api/src/worker.rs index 917a7b1b..45b4b2b2 100644 --- a/lnvps_api/src/worker.rs +++ b/lnvps_api/src/worker.rs @@ -7,6 +7,7 @@ use crate::subscription::SubscriptionHandler; use anyhow::{Context, Result, anyhow, bail}; use chrono::{DateTime, Days, TimeDelta, Utc}; use hickory_resolver::TokioResolver; +use lnvps_api_common::node_control::NodeControlApi; use lnvps_api_common::{ BlackholeWorkFeedback, ChannelWorkCommander, InMemoryKeyValueStore, JobFeedback, KeyValueStore, NetworkProvisioner, RedisConfig, RedisKeyValueStore, RedisWorkCommander, RedisWorkFeedback, @@ -16,9 +17,10 @@ use lnvps_api_common::{ retry::{OpError, Pipeline, RetryPolicy}, }; use lnvps_db::{ - CpuArch, CpuFeature, CpuMfg, IntervalType, LNVpsDb, PaymentMethod, RouterTunnelTraffic, - Subscription, SubscriptionLineItem, SubscriptionPayment, SubscriptionType, Vm, - VmHistoryActionType, VmHost, VmHostKind, VmIpAssignment, VmOsImage, + CpuArch, CpuFeature, CpuMfg, IntervalType, IpRange, LNVpsDb, MarketplaceNode, + MarketplaceProbeStatus, PaymentMethod, RouterTunnelTraffic, Subscription, SubscriptionLineItem, + SubscriptionPayment, SubscriptionType, Vm, VmHistoryActionType, VmHost, VmHostKind, + VmIpAssignment, VmOsImage, }; use log::{debug, error, info, warn}; use nostr_sdk::Client; @@ -199,6 +201,9 @@ pub struct Worker { http_client: reqwest::Client, referral_payouts: crate::referral::ReferralPayoutHandler, refunds: crate::refund::VmRefundHandler, + /// How marketplace nodes are called. `None` in a deployment with no nostr + /// identity, where the health gate says so rather than failing obscurely. + node_control: Option>, } #[derive(Clone)] @@ -226,11 +231,15 @@ pub struct WorkerSettings { pub referral_min_fiat_payout_sats: Option, /// Source of the on-chain fee-rate estimate for the cap above. pub referral_fee_estimator: crate::settings::FeeEstimatorConfig, + /// LNVPS's nostr identity, which is also how it calls marketplace nodes. + /// `None` in a deployment that runs no marketplace. + pub nostr: Option, } impl From<&Settings> for WorkerSettings { fn from(val: &Settings) -> Self { WorkerSettings { + nostr: val.nostr.clone(), delete_after: val.delete_after, smtp: val.smtp.clone(), telegram: val.telegram.clone(), @@ -284,6 +293,14 @@ impl Worker { Arc::new(InMemoryKeyValueStore::new()) }; + // The marketplace control client. A key that cannot be parsed is a + // startup failure, not a surprise on the first node this deployment + // tries to gate. + let node_control = match &settings.nostr { + Some(n) => Some(n.control()?), + None => None, + }; + let referral_payouts = crate::referral::ReferralPayoutHandler::new( db.clone(), node.clone(), @@ -328,6 +345,10 @@ impl Worker { http_client, referral_payouts, refunds, + // Built once, at startup, so a deployment configured with a key + // that cannot be parsed fails here rather than on the first node it + // tries to gate. + node_control: node_control.map(|c| Arc::new(c) as Arc), }) } @@ -1194,6 +1215,220 @@ impl Worker { Ok(()) } + /// Prove a node can carry a customer, and enable it if it can. + /// + /// The gate takes an address from the range customers are given, has the + /// node hold it as a guest, and pings it from the route server. That is the + /// production path exactly: a VM's address is statically routed from the + /// core network to the node's tunnel address, forwarded across the guest + /// bridge, and answered back out the node's default route. Anything less — + /// asking the node how it thinks it is doing, pinging its tunnel address — + /// tests a path no customer takes, and the failure worth catching is the + /// node that believes it is fine. + /// + /// Never returns `Err` for a node that fails: a failed gate is a recorded + /// verdict, not a job to retry forever. `Err` is reserved for not being able + /// to *ask* — no route server, no control key — which is LNVPS's problem + /// rather than the operator's. + pub async fn health_check_node(&self, node_id: u64) -> Result<()> { + let node = self.db.get_marketplace_node(node_id).await?; + let host = self + .db + .get_marketplace_node_host(node_id) + .await? + .ok_or_else(|| { + anyhow!( + "Node {node_id} has no host, so it has not been approved and cannot be gated" + ) + })?; + + // The run is recorded before it starts, so a node refused at the first + // step still gets a verdict an operator can read. A gate that only + // wrote a row once it had taken an address would tell the operator + // whose node never handshook precisely nothing. + self.db.start_marketplace_node_probe(node_id).await?; + let outcome = self.run_health_check(&node, &host).await; + + // The address goes back before anything else, including before the + // verdict is acted on. A gate that failed halfway and kept its address + // would leak one address per attempt out of a range customers are + // waiting for. + let (status, detail) = match &outcome { + Ok(()) => (MarketplaceProbeStatus::Passed, None), + Err(e) => (MarketplaceProbeStatus::Failed, Some(format!("{e:#}"))), + }; + self.db + .finish_marketplace_node_probe(node_id, status, detail.as_deref()) + .await?; + + // ...and the node is told to let go of it, so it stops answering for an + // address that is back in the pool. Best effort: the periodic refresh + // does the same thing, and a node that cannot be reached to release an + // address is a node that just failed its gate anyway. + if let Some(tunnel_id) = node.tunnel_id { + if let Err(e) = self.sync_node_tunnel(tunnel_id).await { + warn!("Could not un-route the probe address for node {node_id}: {e}"); + } + } + if let Some(control) = self.node_control() { + let _ = control.refresh_dataplane(&node, &host).await; + } + + match outcome { + Ok(()) => { + // Enabling is the whole point: an admin approved the hardware, + // and this is the machine check that was standing between that + // decision and the node taking customers. + let mut host = host; + if !host.enabled { + host.enabled = true; + self.db.update_host(&host).await?; + info!("Node {node_id} passed its health check and is now enabled"); + } + } + Err(e) => { + // Left approved but unusable, with the failing step named. That + // is the safe direction: a node that never carries a customer is + // a support conversation, and a node that carries one badly is + // an outage. + warn!("Node {node_id} failed its health check: {e:#}"); + } + } + Ok(()) + } + + /// The gate's steps, each failing with the sentence an operator needs. + async fn run_health_check(&self, node: &MarketplaceNode, host: &VmHost) -> Result<()> { + let tunnel_id = node.tunnel_id.ok_or_else(|| { + anyhow!("This node has no tunnel allocated, so nothing can reach it yet") + })?; + let tunnel = self.db.get_tunnel(tunnel_id).await?; + let pool_id = tunnel.pool_id.ok_or_else(|| { + anyhow!("This node's tunnel came from no pool, so it has no route server") + })?; + let pool = self.db.get_tunnel_pool(pool_id).await?; + + // The node's own report first. It cannot be trusted as a pass — a node + // that believes it is fine is the case this whole job exists for — but + // as a *failure* it is exact, and it names the step before an address is + // taken out of a customer range to prove the same thing slowly. + let control = self.node_control().ok_or_else(|| { + anyhow!("This deployment has no nostr identity configured, so no node can be called") + })?; + let status = control + .status(node, host) + .await + .map_err(|e| anyhow!("The node could not be reached: {e:#}"))?; + if !status.dataplane.tunnel_up || status.dataplane.last_handshake_secs.is_none() { + bail!( + "The node's tunnel has never handshaken with the route server, so nothing can reach its guests" + ); + } + if !status.dataplane.firewall.present { + bail!( + "The node is not filtering its guests, so any guest on it could claim another customer's address" + ); + } + + // A real address, from the range a customer would be given. The + // allocator excludes addresses held by other gates as well as assigned + // ones, so two nodes are never proved with the same address. + let (range, address) = self.pick_probe_address(host).await?; + self.db + .hold_marketplace_probe_address(node.id, range.id, &address.to_string()) + .await?; + + // The route server has to route it and admit it from this peer, and the + // node has to hold it on its bridge. Both are the ordinary guest path — + // the probe is in the node's guest list, so nothing here is special + // cased, which is what makes the result mean something. + self.sync_node_tunnel(tunnel_id).await.map_err(|e| { + anyhow!("The route server could not be configured for the probe: {e:#}") + })?; + control + .refresh_dataplane(node, host) + .await + .map_err(|e| anyhow!("The node would not apply its data plane: {e:#}"))?; + + let router = crate::router::get_router(&self.db, pool.router_id) + .await + .map_err(|e| anyhow!("Cannot reach the route server for this node: {e}"))?; + let tr = router + .tunnel() + .context("This node's route server cannot be asked to probe an address")?; + + if !tr + .probe_address(&address.to_string()) + .await + .map_err(|e| anyhow!("The route server could not run the probe: {e}"))? + { + bail!( + "The route server could not reach {address} on this node, so a customer's VM would not be reachable either" + ); + } + Ok(()) + } + + /// An address from a range in the node's region. + async fn pick_probe_address(&self, host: &VmHost) -> Result<(IpRange, std::net::IpAddr)> { + let network = NetworkProvisioner::new(self.db.clone()); + let mut ranges: Vec = self + .db + .list_ip_range_in_region(host.region_id) + .await? + .into_iter() + .filter(|r| r.enabled) + .collect(); + + // IPv4 first. A guest's IPv6 address is normally derived from its MAC + // (EUI64) and a probe has no guest, so the v4 path is the one a probe + // can stand in for exactly. A v6-only region is still gated, from the + // first free address in its range — a weaker stand-in, but the + // alternative is not gating those nodes at all. + ranges.sort_by_key(|r| { + r.cidr + .parse::() + .map(|c| c.is_ipv6()) + .unwrap_or(true) + }); + + let mut last_error = None; + for range in ranges { + match network.pick_ip_from_range(&range).await { + Ok(available) => return Ok((range, available.ip.ip())), + // A full range is not a reason to give up: a region usually has + // several, and the gate only needs one address from any of them. + Err(e) => last_error = Some(e), + } + } + match last_error { + Some(e) => Err(anyhow!( + "No address could be taken from any range in this region to test the node with: {e}" + )), + None => Err(anyhow!( + "This node's region has no enabled IP range, so there is no address a customer could be given here either" + )), + } + } + + /// The control client, when this deployment has an identity to call with. + fn node_control(&self) -> Option> { + self.node_control.clone() + } + + /// Point the health gate at a different node client. + /// + /// For tests: every case the gate has to get right is a node behaving + /// badly, and none of those are states a real node can be asked to be in. + #[cfg(test)] + pub(crate) fn with_node_control( + mut self, + control: Arc, + ) -> Self { + self.node_control = Some(control); + self + } + /// Remove a tunnel interface from a router, after its pool is deleted. /// /// Idempotent: an interface that is already gone is the desired state, not @@ -2794,6 +3029,9 @@ impl Worker { WorkJob::SyncNodeTunnel { tunnel_id } => { self.sync_node_tunnel(*tunnel_id).await?; } + WorkJob::HealthCheckNode { node_id } => { + self.health_check_node(*node_id).await?; + } WorkJob::DeleteVm { vm_id, reason, @@ -4312,6 +4550,82 @@ mod tests { } } + /// A node that answers everything as a working node would. + /// + /// Every interesting case in the gate is a node behaving badly, and none of + /// those are states a real node can be asked to be in on demand. + struct HealthyNode { + status: lnvps_api_common::node_control::NodeStatus, + refreshed: std::sync::Mutex, + } + + impl Default for HealthyNode { + fn default() -> Self { + use lnvps_api_common::node_control::*; + Self { + status: NodeStatus { + version: "test".to_string(), + dataplane: NodeDataPlaneState { + tunnel_up: true, + last_handshake_secs: Some(2), + bridge_up: true, + forwarding4: true, + firewall: NodeFirewallState { + available: true, + present: true, + isolated: true, + ..Default::default() + }, + ..Default::default() + }, + }, + refreshed: std::sync::Mutex::new(0), + } + } + } + + #[async_trait::async_trait] + impl lnvps_api_common::node_control::NodeControlApi for HealthyNode { + async fn status( + &self, + _node: &MarketplaceNode, + _host: &VmHost, + ) -> Result { + Ok(self.status.clone()) + } + + async fn refresh_dataplane( + &self, + _node: &MarketplaceNode, + _host: &VmHost, + ) -> Result> { + *self.refreshed.lock().unwrap() += 1; + Ok(vec![]) + } + } + + /// A node that is not there, or is not the node it claims to be. + struct SilentNode; + + #[async_trait::async_trait] + impl lnvps_api_common::node_control::NodeControlApi for SilentNode { + async fn status( + &self, + _node: &MarketplaceNode, + _host: &VmHost, + ) -> Result { + bail!("certificate does not match the pin") + } + + async fn refresh_dataplane( + &self, + _node: &MarketplaceNode, + _host: &VmHost, + ) -> Result> { + bail!("certificate does not match the pin") + } + } + async fn setup_worker(db: Arc) -> Result { setup_worker_with_delete_after(db, 0).await } @@ -5588,4 +5902,327 @@ mod tests { mr.clear().await; Ok(()) } + + /// A node that answers, filters and carries a packet is enabled. That is + /// the whole gate: an admin approved hardware they cannot see, and this is + /// the machine check standing between that decision and customers. + #[tokio::test] + async fn test_health_check_enables_a_node_that_carries_a_packet() -> Result<()> { + use crate::mocks::MockRouter; + + let db = Arc::new(MockDb::empty()); + let pool_id = setup_pool(&db, 51820).await?; + setup_node_tunnel(&db, pool_id).await?; + let mr = MockRouter::new(); + mr.clear().await; + + let dbt: Arc = db.clone(); + let node = dbt.get_marketplace_node(1).await?; + let worker = setup_worker(db.clone()) + .await? + .with_node_control(Arc::new(HealthyNode::default())); + // The interface has to exist before a peer can be pushed onto it, as it + // would on a route server LNVPS has already configured. + worker.sync_tunnel_pool(pool_id).await?; + + // The route server can reach whatever the gate asks it about, which is + // what a working node looks like from the only place that matters. + mr.set_reach_all(true).await; + worker.health_check_node(node.id).await?; + + let host = dbt.get_marketplace_node_host(node.id).await?.unwrap(); + assert!(host.enabled, "a node that carried a packet must be enabled"); + + let probe = dbt.get_marketplace_node_probe(node.id).await?.unwrap(); + assert_eq!(probe.status, MarketplaceProbeStatus::Passed); + assert_eq!(probe.detail, None); + // The address is given back. A gate that kept it would cost one address + // per run out of a range customers are waiting for. + assert_eq!(probe.ip, None); + assert!(probe.finished.is_some()); + Ok(()) + } + + /// The address the route server was asked about is a real one from the + /// region's range, and while the gate ran it was routed to the node like + /// any guest — that is what makes the result mean something. + #[tokio::test] + async fn test_health_check_uses_a_real_customer_address() -> Result<()> { + use crate::mocks::MockRouter; + + let db = Arc::new(MockDb::empty()); + let pool_id = setup_pool(&db, 51820).await?; + setup_node_tunnel(&db, pool_id).await?; + let mr = MockRouter::new(); + mr.clear().await; + + let dbt: Arc = db.clone(); + let node = dbt.get_marketplace_node(1).await?; + let range = dbt.get_ip_range(1).await?; + let cidr: ipnetwork::IpNetwork = range.cidr.parse()?; + + let control = Arc::new(HealthyNode::default()); + let worker = setup_worker(db.clone()) + .await? + .with_node_control(control.clone()); + worker.sync_tunnel_pool(pool_id).await?; + mr.set_reach_all(true).await; + worker.health_check_node(node.id).await?; + + let probed = mr.probed().await; + assert_eq!(probed.len(), 1, "{probed:?}"); + let address: std::net::IpAddr = probed[0].parse()?; + assert!( + cidr.contains(address), + "{probed:?} is not from {}", + range.cidr + ); + // Not an address a VM already has: the allocator sees held probe + // addresses and assigned ones alike. + assert_ne!(probed[0], "203.0.113.5"); + + // The node was told to apply the document rather than left to its + // heartbeat: an approval that took a minute to conclude would be a + // minute of an operator watching nothing happen. + assert!(*control.refreshed.lock().unwrap() >= 1); + Ok(()) + } + + /// A node the route server cannot reach is left approved and disabled, with + /// the failing step named. That is the safe direction: a node that never + /// carries a customer is a support conversation, and one that carries a + /// customer badly is an outage. + #[tokio::test] + async fn test_health_check_leaves_an_unreachable_node_disabled() -> Result<()> { + use crate::mocks::MockRouter; + + let db = Arc::new(MockDb::empty()); + let pool_id = setup_pool(&db, 51820).await?; + setup_node_tunnel(&db, pool_id).await?; + let mr = MockRouter::new(); + mr.clear().await; + + let dbt: Arc = db.clone(); + let node = dbt.get_marketplace_node(1).await?; + let worker = setup_worker(db.clone()) + .await? + .with_node_control(Arc::new(HealthyNode::default())); + worker.sync_tunnel_pool(pool_id).await?; + + // The route server reaches nothing, which is what a node with a broken + // bridge looks like from outside — while the node itself insists it is + // fine, the exact case the gate exists for. + worker.health_check_node(node.id).await?; + + let host = dbt.get_marketplace_node_host(node.id).await?.unwrap(); + assert!(!host.enabled); + let probe = dbt.get_marketplace_node_probe(node.id).await?.unwrap(); + assert_eq!(probe.status, MarketplaceProbeStatus::Failed); + assert!( + probe + .detail + .as_deref() + .unwrap() + .contains("would not be reachable"), + "{probe:?}" + ); + assert_eq!(probe.ip, None, "a failed gate gives its address back too"); + Ok(()) + } + + /// A node whose tunnel has never handshaken is refused before an address is + /// taken out of a customer range to prove the same thing slowly. + #[tokio::test] + async fn test_health_check_refuses_a_node_with_no_handshake() -> Result<()> { + use crate::mocks::MockRouter; + + let db = Arc::new(MockDb::empty()); + let pool_id = setup_pool(&db, 51820).await?; + setup_node_tunnel(&db, pool_id).await?; + MockRouter::new().clear().await; + + let dbt: Arc = db.clone(); + let node = dbt.get_marketplace_node(1).await?; + let mut control = HealthyNode::default(); + control.status.dataplane.last_handshake_secs = None; + let worker = setup_worker(db.clone()) + .await? + .with_node_control(Arc::new(control)); + + worker.health_check_node(node.id).await?; + + let probe = dbt.get_marketplace_node_probe(node.id).await?.unwrap(); + assert_eq!(probe.status, MarketplaceProbeStatus::Failed); + assert!( + probe.detail.as_deref().unwrap().contains("handshaken"), + "{probe:?}" + ); + assert!( + dbt.list_marketplace_probe_ips_in_range(1).await?.is_empty(), + "no address should have been taken" + ); + Ok(()) + } + + /// A node with no packet filter loaded is refused: on it, any guest could + /// claim another customer's address. + #[tokio::test] + async fn test_health_check_refuses_an_unfiltered_node() -> Result<()> { + use crate::mocks::MockRouter; + + let db = Arc::new(MockDb::empty()); + let pool_id = setup_pool(&db, 51820).await?; + setup_node_tunnel(&db, pool_id).await?; + MockRouter::new().clear().await; + + let dbt: Arc = db.clone(); + let node = dbt.get_marketplace_node(1).await?; + let mut control = HealthyNode::default(); + control.status.dataplane.firewall.present = false; + let worker = setup_worker(db.clone()) + .await? + .with_node_control(Arc::new(control)); + + worker.health_check_node(node.id).await?; + + let probe = dbt.get_marketplace_node_probe(node.id).await?.unwrap(); + assert_eq!(probe.status, MarketplaceProbeStatus::Failed); + assert!( + probe.detail.as_deref().unwrap().contains("filtering"), + "{probe:?}" + ); + Ok(()) + } + + /// A node that cannot be reached at all fails with the node's own error, + /// because "connection refused" and "certificate does not match" send an + /// operator to different places. + #[tokio::test] + async fn test_health_check_quotes_the_reason_a_node_is_unreachable() -> Result<()> { + use crate::mocks::MockRouter; + + let db = Arc::new(MockDb::empty()); + let pool_id = setup_pool(&db, 51820).await?; + setup_node_tunnel(&db, pool_id).await?; + MockRouter::new().clear().await; + + let dbt: Arc = db.clone(); + let node = dbt.get_marketplace_node(1).await?; + let worker = setup_worker(db.clone()) + .await? + .with_node_control(Arc::new(SilentNode)); + + worker.health_check_node(node.id).await?; + + let probe = dbt.get_marketplace_node_probe(node.id).await?.unwrap(); + assert_eq!(probe.status, MarketplaceProbeStatus::Failed); + assert!( + probe + .detail + .as_deref() + .unwrap() + .contains("certificate does not match"), + "{probe:?}" + ); + Ok(()) + } + + /// A held probe address is not handed to a VM. Two machines answering for + /// one address is the failure this gate is meant to prevent, not cause. + #[tokio::test] + async fn test_a_held_probe_address_is_not_allocated_to_a_vm() -> Result<()> { + let db = Arc::new(MockDb::empty()); + let pool_id = setup_pool(&db, 51820).await?; + setup_node_tunnel(&db, pool_id).await?; + + let dbt: Arc = db.clone(); + let range = dbt.get_ip_range(1).await?; + let network = NetworkProvisioner::new(dbt.clone()); + + let free = network + .pick_ip_from_range(&range) + .await? + .ip + .ip() + .to_string(); + dbt.start_marketplace_node_probe(1).await?; + dbt.hold_marketplace_probe_address(1, range.id, &free) + .await?; + + let next = network + .pick_ip_from_range(&range) + .await? + .ip + .ip() + .to_string(); + assert_ne!( + next, free, + "the allocator handed out an address a gate holds" + ); + + // ...and once the gate finishes, the address comes back into use. + dbt.finish_marketplace_node_probe(1, MarketplaceProbeStatus::Passed, None) + .await?; + assert!( + dbt.list_marketplace_probe_ips_in_range(range.id) + .await? + .is_empty() + ); + Ok(()) + } + + /// While a gate runs, the address is in the node's guest list — the route + /// server routes it, the peer is allowed it, and the node holds it on the + /// bridge. A probe the node treated specially would prove a path no + /// customer takes. + #[tokio::test] + async fn test_a_running_probe_is_routed_like_a_guest() -> Result<()> { + let db = Arc::new(MockDb::empty()); + let pool_id = setup_pool(&db, 51820).await?; + setup_node_tunnel(&db, pool_id).await?; + + let dbt: Arc = db.clone(); + let node = dbt.get_marketplace_node(1).await?; + dbt.start_marketplace_node_probe(node.id).await?; + dbt.hold_marketplace_probe_address(node.id, 1, "10.0.0.9") + .await?; + + // The VM the fixture put on this host has an assignment with no range, + // which the node's guest list cannot describe a gateway for. Pointing + // it at a real range is what a real assignment looks like. + { + let mut assignments = db.ip_assignments.lock().await; + for a in assignments.values_mut() { + a.ip_range_id = 1; + } + } + + let guests = crate::provisioner::node_guests(&dbt, &node).await?; + let probe = guests + .iter() + .find(|g| g.address.starts_with("10.0.0.9")) + .expect("the probe address is sent to the node as a guest"); + // The gateway is the range's, exactly as a customer's VM is given — + // the node answers for it on the bridge by proxy ARP. Normalised out of + // the range's stored `10.0.0.1/8`, because a guest is configured with an + // address, not a prefix. + assert_eq!(probe.gateway, "10.0.0.1"); + // No MAC: nothing owns it but the node, which holds it itself. + assert_eq!(probe.mac, None); + + // ...and the route server is told to route it and admit it from this + // peer. The node holding an address the route server does not route is + // a probe that fails for the wrong reason — and this list comes from a + // different function than the one above, so both have to be checked. + let pool = dbt.get_tunnel_pool(pool_id).await?; + let plan = crate::provisioner::plan_pool(&dbt, &pool).await?; + assert!(plan.routes.contains(&"10.0.0.9/32".to_string()), "{plan:?}"); + assert!( + plan.peers[0] + .allowed_ips + .contains(&"10.0.0.9/32".to_string()), + "{plan:?}" + ); + Ok(()) + } } diff --git a/lnvps_api_admin/src/admin/marketplace.rs b/lnvps_api_admin/src/admin/marketplace.rs index b5a7c636..0cc2cbd2 100644 --- a/lnvps_api_admin/src/admin/marketplace.rs +++ b/lnvps_api_admin/src/admin/marketplace.rs @@ -22,7 +22,7 @@ use std::sync::Arc; use lnvps_api_common::node_control::NodeStatus; use lnvps_api_common::{ - ApiData, ApiError, ApiPaginatedData, ApiPaginatedResult, ApiResult, PageQuery, + ApiData, ApiError, ApiPaginatedData, ApiPaginatedResult, ApiResult, PageQuery, WorkJob, deserialize_from_str_optional, }; use lnvps_db::{ @@ -50,6 +50,10 @@ pub fn router() -> Router { "/api/admin/v1/marketplace/nodes/{id}/status", get(admin_node_status), ) + .route( + "/api/admin/v1/marketplace/nodes/{id}/health_check", + post(admin_health_check_node), + ) .route( "/api/admin/v1/marketplace/operators", get(admin_list_operators), @@ -81,6 +85,13 @@ pub struct AdminMarketplaceNodeInfo { pub tunnel_id: Option, /// The backing host row, created by approval. `null` before then. pub host_id: Option, + /// Whether the host is taking placements. Approval alone does not enable a + /// node: the health gate does, and until it passes this stays `false` while + /// `status` reads `approved`. That pair is the answer to "why is my node + /// approved and empty?". + pub host_enabled: bool, + /// The last health-gate run, once a node has been gated. + pub probe: Option, /// Whether the one-off listing fee has been paid. `false` also covers "not /// started" — either way there is nothing to approve against. pub fee_paid: bool, @@ -90,6 +101,25 @@ pub struct AdminMarketplaceNodeInfo { pub created: DateTime, } +/// The last health-gate run. +#[derive(Serialize, Debug)] +pub struct AdminNodeProbeInfo { + /// `running`, `passed` or `failed`. + pub status: String, + /// The address held right now, `null` once the run is over. An address is + /// only held while a gate is running, so a value here on an old run means a + /// gate that never finished. + pub ip: Option, + /// The range the address came from — a failure reads as "this node cannot + /// carry addresses from that range". `null` for a run that was refused + /// before an address was worth taking out of a customer range. + pub ip_range_id: Option, + /// Which step failed, in the words an operator needs. + pub detail: Option, + pub created: DateTime, + pub finished: Option>, +} + /// An operator enrolment as an admin sees it. #[derive(Serialize, Debug)] pub struct AdminMarketplaceOperatorInfo { @@ -204,6 +234,7 @@ async fn node_info( let user = db.get_user(operator.user_id).await?; let host = db.get_marketplace_node_host(node.id).await?; let (fee_paid, fee_subscription_id) = fee_state(db, &node).await?; + let probe = db.get_marketplace_node_probe(node.id).await?; Ok(AdminMarketplaceNodeInfo { id: node.id, @@ -215,7 +246,16 @@ async fn node_info( trust_tier: node.trust_tier.to_string(), tls_fingerprint: node.tls_fingerprint.map(hex::encode), tunnel_id: node.tunnel_id, + host_enabled: host.as_ref().map(|h| h.enabled).unwrap_or(false), host_id: host.map(|h| h.id), + probe: probe.map(|p| AdminNodeProbeInfo { + status: p.status.to_string(), + ip: p.ip, + ip_range_id: p.ip_range_id, + detail: p.detail, + created: p.created, + finished: p.finished, + }), fee_paid, fee_subscription_id, last_seen: node.last_seen, @@ -341,6 +381,35 @@ async fn admin_node_status( ApiData::ok(status) } +/// Run the health gate against a node now. +/// +/// Queued rather than run inline: the gate takes an address, waits on a route +/// server and pings across a tunnel, which is not work to hold an HTTP request +/// open for. The verdict lands on the node's `probe`, which `GET +/// /marketplace/nodes/{id}` returns. +/// +/// Exists because the automatic run happens when a node's tunnel is allocated, +/// and the interesting cases are the ones after that: an operator who has +/// fixed their firewall, a node whose route server was down at the time. +async fn admin_health_check_node( + auth: AdminAuth, + State(this): State, + Path(id): Path, +) -> ApiResult<()> { + auth.require_permission(AdminResource::MarketplaceNode, AdminAction::Update)?; + let node = this.db.get_marketplace_node(id).await?; + if this.db.get_marketplace_node_host(node.id).await?.is_none() { + return Err(ApiError::bad_request( + "This node has not been approved, so it has no host to enable", + )); + } + this.work_commander + .send(WorkJob::HealthCheckNode { node_id: node.id }) + .await + .map_err(|e| ApiError::internal(format!("Cannot queue the health check: {e}")))?; + ApiData::ok(()) +} + /// Approve a node: create its backing host and make it placeable. async fn admin_approve_node( auth: AdminAuth, diff --git a/lnvps_api_common/src/mock.rs b/lnvps_api_common/src/mock.rs index 0e83db77..0a3a79f9 100644 --- a/lnvps_api_common/src/mock.rs +++ b/lnvps_api_common/src/mock.rs @@ -8,14 +8,15 @@ use lnvps_db::{ AppDeploymentVolumeUsage, AppTag, AsnSubscription, AsnSubscriptionStatus, AvailableIpSpace, Company, CpuArch, CpuMfg, DbError, DbResult, DiskInterface, DiskType, DnsServer, DnsServerKind, EncryptedString, IntervalType, IpRange, IpRangeAllocationMode, IpRangeSubscription, - IpSpacePricing, LNVpsDbBase, MarketplaceNode, MarketplaceNodeStatus, MarketplaceOperator, - NewAgentMessage, NostrDomain, NostrDomainHandle, OsDistribution, PaymentMethod, - PaymentMethodConfig, Referral, ReferralCostUsage, ReferralPayout, Region, Router, - RouterBgpRoute, RouterBgpSession, RouterTunnel, RouterTunnelTraffic, Subscription, - SubscriptionLineItem, SubscriptionPayment, SubscriptionPaymentWithCompany, Tunnel, TunnelPool, - User, UserPaymentMethod, UserSshKey, Vm, VmCostPlan, VmCustomPricing, VmCustomPricingDisk, - VmCustomTemplate, VmFirewallPolicy, VmFirewallRule, VmHistory, VmHost, VmHostDisk, VmHostKind, - VmIpAssignment, VmOsImage, VmTemplate, WebauthnCredential, + IpSpacePricing, LNVpsDbBase, MarketplaceNode, MarketplaceNodeProbe, MarketplaceNodeStatus, + MarketplaceOperator, MarketplaceProbeStatus, NewAgentMessage, NostrDomain, NostrDomainHandle, + OsDistribution, PaymentMethod, PaymentMethodConfig, Referral, ReferralCostUsage, + ReferralPayout, Region, Router, RouterBgpRoute, RouterBgpSession, RouterTunnel, + RouterTunnelTraffic, Subscription, SubscriptionLineItem, SubscriptionPayment, + SubscriptionPaymentWithCompany, Tunnel, TunnelPool, User, UserPaymentMethod, UserSshKey, Vm, + VmCostPlan, VmCustomPricing, VmCustomPricingDisk, VmCustomTemplate, VmFirewallPolicy, + VmFirewallRule, VmHistory, VmHost, VmHostDisk, VmHostKind, VmIpAssignment, VmOsImage, + VmTemplate, WebauthnCredential, }; use async_trait::async_trait; @@ -76,6 +77,9 @@ pub struct MockDb { pub referrals: Arc>>, pub marketplace_operators: Arc>>, pub marketplace_nodes: Arc>>, + /// Health-gate runs, keyed by node — one live gate per node, as the unique + /// key on the real table enforces. + pub marketplace_node_probes: Arc>>, pub tunnels: Arc>>, pub tunnel_pools: Arc>>, pub referral_payouts: Arc>>, @@ -463,6 +467,7 @@ impl Default for MockDb { referrals: Arc::new(Default::default()), marketplace_operators: Arc::new(Default::default()), marketplace_nodes: Arc::new(Default::default()), + marketplace_node_probes: Arc::new(Default::default()), tunnels: Arc::new(Default::default()), tunnel_pools: Arc::new(Default::default()), referral_payouts: Arc::new(Default::default()), @@ -3718,6 +3723,109 @@ impl LNVpsDbBase for MockDb { .cloned()) } + async fn get_marketplace_node_probe( + &self, + node_id: u64, + ) -> DbResult> { + Ok(self + .marketplace_node_probes + .lock() + .await + .get(&node_id) + .cloned()) + } + + async fn start_marketplace_node_probe(&self, node_id: u64) -> DbResult { + // FK marketplace_node_probe.node_id + if !self.marketplace_nodes.lock().await.contains_key(&node_id) { + return Err(anyhow!("Node {node_id} not found").into()); + } + let mut probes = self.marketplace_node_probes.lock().await; + let id = probes + .get(&node_id) + .map(|p| p.id) + .unwrap_or_else(|| probes.values().map(|p| p.id).max().unwrap_or(0) + 1); + probes.insert( + node_id, + MarketplaceNodeProbe { + id, + node_id, + ip_range_id: None, + ip: None, + status: MarketplaceProbeStatus::Running, + detail: None, + created: Utc::now(), + finished: None, + }, + ); + Ok(id) + } + + async fn hold_marketplace_probe_address( + &self, + node_id: u64, + ip_range_id: u64, + ip: &str, + ) -> DbResult<()> { + // FK marketplace_node_probe.ip_range_id + if !self.ip_range.lock().await.contains_key(&ip_range_id) { + return Err(anyhow!("IP range {ip_range_id} not found").into()); + } + let mut probes = self.marketplace_node_probes.lock().await; + // uk_marketplace_node_probe_ip: two nodes holding one address would + // both be routed it, and the route server would send it to whichever + // peer claimed it last. + if probes + .values() + .any(|p| p.node_id != node_id && p.ip.as_deref() == Some(ip)) + { + return Err(anyhow!("{ip} is already held by another probe").into()); + } + if let Some(probe) = probes.get_mut(&node_id) { + probe.ip_range_id = Some(ip_range_id); + probe.ip = Some(ip.to_string()); + } + Ok(()) + } + + async fn finish_marketplace_node_probe( + &self, + node_id: u64, + status: MarketplaceProbeStatus, + detail: Option<&str>, + ) -> DbResult<()> { + let mut probes = self.marketplace_node_probes.lock().await; + if let Some(probe) = probes.get_mut(&node_id) { + probe.status = status; + probe.detail = detail.map(str::to_string); + probe.ip = None; + probe.finished = Some(Utc::now()); + } + Ok(()) + } + + async fn list_marketplace_probe_ips_in_range(&self, range_id: u64) -> DbResult> { + Ok(self + .marketplace_node_probes + .lock() + .await + .values() + .filter(|p| p.ip_range_id == Some(range_id)) + .filter_map(|p| p.ip.clone()) + .collect()) + } + + async fn list_marketplace_probe_ips_for_node(&self, node_id: u64) -> DbResult> { + Ok(self + .marketplace_node_probes + .lock() + .await + .values() + .filter(|p| p.node_id == node_id) + .filter_map(|p| p.ip.clone()) + .collect()) + } + async fn insert_marketplace_node(&self, node: &MarketplaceNode) -> DbResult { let operators = self.marketplace_operators.lock().await; // FK marketplace_node.operator_id diff --git a/lnvps_api_common/src/network.rs b/lnvps_api_common/src/network.rs index ffbc6da9..0d4157ac 100644 --- a/lnvps_api_common/src/network.rs +++ b/lnvps_api_common/src/network.rs @@ -232,6 +232,18 @@ impl NetworkProvisioner { let mut ips: HashSet = ips.iter().filter_map(|i| i.ip.parse().ok()).collect(); ips.extend(exclude.iter().copied()); + // Addresses a marketplace health gate is currently holding. They are + // not `vm_ip_assignment` rows — no VM owns them — but they are being + // routed to a node and answered for right now, so handing one to a VM + // would put two machines on one address. + ips.extend( + self.db + .list_marketplace_probe_ips_in_range(range.id) + .await? + .iter() + .filter_map(|ip| ip.parse::().ok()), + ); + let gateway: IpNetwork = parse_gateway(&range.gateway)?; // Calculate the prefix to use: take the smallest prefix value (largest network) diff --git a/lnvps_api_common/src/node_control.rs b/lnvps_api_common/src/node_control.rs index 32b064c8..a5967cad 100644 --- a/lnvps_api_common/src/node_control.rs +++ b/lnvps_api_common/src/node_control.rs @@ -135,6 +135,35 @@ pub struct NodeFirewallState { pub spoofed_packets: u64, } +/// Calling a node, as the health gate needs it. +/// +/// A trait so the gate's decisions can be tested against a node that answers in +/// a chosen way. Every interesting case here is a node behaving badly — +/// claiming a tunnel it does not have, refusing to apply a document, answering +/// slowly — and none of those are states a real node can be asked to be in on +/// demand. +#[async_trait::async_trait] +pub trait NodeControlApi: Send + Sync { + async fn status(&self, node: &MarketplaceNode, host: &VmHost) -> Result; + async fn refresh_dataplane(&self, node: &MarketplaceNode, host: &VmHost) + -> Result>; +} + +#[async_trait::async_trait] +impl NodeControlApi for NodeControl { + async fn status(&self, node: &MarketplaceNode, host: &VmHost) -> Result { + NodeControl::status(self, node, host).await + } + + async fn refresh_dataplane( + &self, + node: &MarketplaceNode, + host: &VmHost, + ) -> Result> { + NodeControl::refresh_dataplane(self, node, host).await + } +} + /// Signs and sends control requests. /// /// Holds the key rather than taking it per call so that a deployment without a @@ -171,17 +200,55 @@ impl NodeControl { .with_context(|| format!("Node {} returned a status this LNVPS cannot read", node.id)) } + /// Ask a node to re-fetch and apply its data plane now. + /// + /// Nothing about the document is sent: the node fetches it itself, from + /// LNVPS, with its own credential. This only says *when*, so that a change + /// LNVPS has just made — a probe address, a new guest — does not have to + /// wait out the node's heartbeat before it can be tested. + pub async fn refresh_dataplane( + &self, + node: &MarketplaceNode, + host: &VmHost, + ) -> Result> { + let body = self + .send(node, host, "POST", "/api/v1/dataplane/refresh") + .await?; + #[derive(Deserialize, Default)] + struct RefreshResult { + #[serde(default)] + changed: Vec, + } + let result: RefreshResult = serde_json::from_str(&body).unwrap_or_default(); + Ok(result.changed) + } + /// `GET path` against a node, signed and pinned. async fn get(&self, node: &MarketplaceNode, host: &VmHost, path: &str) -> Result { + self.send(node, host, "GET", path).await + } + + /// One signed, pinned request. + async fn send( + &self, + node: &MarketplaceNode, + host: &VmHost, + method: &str, + path: &str, + ) -> Result { let url = endpoint(host, path)?; let fingerprint = node.tls_fingerprint.clone().context( "This node has no pinned certificate, so there is no way to tell its answers \ from anyone else's; it must re-register", )?; - let auth = self.authorization("GET", &url, &[])?; - let response = pinned_client(&fingerprint, self.timeout)? - .get(&url) + let auth = self.authorization(method, &url, &[])?; + let client = pinned_client(&fingerprint, self.timeout)?; + let request = match method { + "POST" => client.post(&url), + _ => client.get(&url), + }; + let response = request .header("Authorization", auth) .send() .await diff --git a/lnvps_api_common/src/work/mod.rs b/lnvps_api_common/src/work/mod.rs index f0fb6945..7e8f2575 100644 --- a/lnvps_api_common/src/work/mod.rs +++ b/lnvps_api_common/src/work/mod.rs @@ -254,6 +254,13 @@ pub enum WorkJob { /// the pool would work, but a node waiting on its first guest should not /// wait for every other node on the route server to be checked first. SyncNodeTunnel { tunnel_id: u64 }, + /// Prove a marketplace node can carry a customer, and enable it if it can. + /// + /// Everything between an admin approving hardware they cannot see and a + /// customer's VM working is machinery nobody has tested on that particular + /// machine. This tests it, with a real address from a real range, and + /// enables the host only if a packet arrives. + HealthCheckNode { node_id: u64 }, /// Re-apply forward + reverse DNS records for every IP assignment in a range. /// /// Used after changing a range's DNS server configuration (e.g. switching @@ -363,6 +370,7 @@ impl fmt::Display for WorkJob { WorkJob::RemoveTunnelInterface { .. } => write!(f, "RemoveTunnelInterface"), WorkJob::ReconcileTunnelPeers { .. } => write!(f, "ReconcileTunnelPeers"), WorkJob::SyncNodeTunnel { .. } => write!(f, "SyncNodeTunnel"), + WorkJob::HealthCheckNode { .. } => write!(f, "HealthCheckNode"), WorkJob::PatchIpRangeDns { .. } => write!(f, "PatchIpRangeDns"), } } @@ -423,5 +431,9 @@ mod tests { WorkJob::SyncNodeTunnel { tunnel_id: 5 }.to_string(), "SyncNodeTunnel" ); + assert_eq!( + WorkJob::HealthCheckNode { node_id: 9 }.to_string(), + "HealthCheckNode" + ); } } diff --git a/lnvps_db/migrations/20260810120000_marketplace_node_probe.sql b/lnvps_db/migrations/20260810120000_marketplace_node_probe.sql new file mode 100644 index 00000000..82bf28c6 --- /dev/null +++ b/lnvps_db/migrations/20260810120000_marketplace_node_probe.sql @@ -0,0 +1,75 @@ +-- The health gate's record: has this node ever carried a packet for a customer? +-- +-- A node is approved by a human looking at hardware they cannot see. Everything +-- between that decision and a customer's VM working — a tunnel that handshakes, +-- a route server that routes, a bridge, a packet filter, a forwarding knob — is +-- machinery nobody has tested on that particular machine. The gate tests it, by +-- taking an address from the range customers get, having the node hold it, and +-- pinging it from the route server. That is the customer's path exactly. +-- +-- Why a table rather than columns on `marketplace_node`: +-- +-- * The probe **holds an address** while it runs. That address must not be +-- handed to a VM at the same time, so the allocator has to be able to see it, +-- which means it has to be queryable by range — a column on the node would +-- make "which addresses in this range are taken?" a scan of every node. +-- * The result outlives the address. When the probe finishes the address goes +-- back (`ip` becomes NULL) but the verdict stays, so an operator can see why +-- their node was refused without the node still consuming an address. +-- +-- One row per node: this is the *last* gate run, not a history. A history would +-- be worth having and is not this — it belongs with SLA accounting, where the +-- retention and the questions are different. + +CREATE TABLE marketplace_node_probe ( + id INTEGER UNSIGNED NOT NULL AUTO_INCREMENT, + + node_id INTEGER UNSIGNED NOT NULL, + + -- The range the address came from. Recorded because the gateway a probe is + -- reachable through belongs to the range, not to the node, and a failure is + -- read as "this node cannot carry addresses from that range". + -- + -- NULL until an address is taken, and for a run that failed before it got + -- that far: a node whose tunnel never handshook is refused *before* an + -- address comes out of a customer range to prove the same thing slowly, and + -- that refusal still has to be recorded or the operator is told nothing. + ip_range_id INTEGER UNSIGNED NULL, + + -- The address currently held, or NULL once the run is over. + -- + -- Nullable rather than deleted with the row, because releasing the address + -- and keeping the verdict are two different things, and a gate that had to + -- destroy its own result to give an address back would leave no record of + -- why a node is disabled. + ip VARCHAR(255) NULL, + + -- 0 running, 1 passed, 2 failed. + status SMALLINT UNSIGNED NOT NULL DEFAULT 0, + + -- Which step failed, in the words an operator needs: "the node never + -- handshook", "the route server could not reach the probe address". A + -- verdict with no reason is a support conversation. + detail VARCHAR(255) NULL, + + created TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + finished TIMESTAMP NULL, + + CONSTRAINT PK_marketplace_node_probe PRIMARY KEY (id), + + -- One live gate per node. Two concurrent runs would hold two addresses and + -- disagree about the verdict. + CONSTRAINT uk_marketplace_node_probe_node UNIQUE KEY (node_id), + + -- An address is held by one probe. Two nodes holding the same address would + -- both be routed it, and the route server would send it to whichever peer + -- claimed it last. NULLs do not collide in MySQL, which is exactly the + -- behaviour wanted here: finished probes hold nothing. + CONSTRAINT uk_marketplace_node_probe_ip UNIQUE KEY (ip), + + CONSTRAINT fk_marketplace_node_probe_node FOREIGN KEY (node_id) + REFERENCES marketplace_node (id) ON DELETE CASCADE, + CONSTRAINT fk_marketplace_node_probe_range FOREIGN KEY (ip_range_id) + REFERENCES ip_range (id) +) ENGINE = InnoDB + DEFAULT CHARSET = utf8mb4; diff --git a/lnvps_db/src/lib.rs b/lnvps_db/src/lib.rs index 3ef2ed8b..ce83ba04 100644 --- a/lnvps_db/src/lib.rs +++ b/lnvps_db/src/lib.rs @@ -1177,6 +1177,49 @@ pub trait LNVpsDbBase: Send + Sync { /// Register a new node, returning the new id async fn insert_marketplace_node(&self, node: &MarketplaceNode) -> DbResult; + /// The last health-gate run for a node, if it has ever been gated. + async fn get_marketplace_node_probe( + &self, + node_id: u64, + ) -> DbResult>; + + /// Start a gate run, replacing any previous one for this node. + /// + /// Replacing rather than appending: this is the *last* run, not a history. + /// A history belongs with SLA accounting, where the retention and the + /// questions are different. + /// + /// No address yet — the early steps can refuse a node before one is worth + /// taking, and those refusals still have to be recorded. + async fn start_marketplace_node_probe(&self, node_id: u64) -> DbResult; + + /// Hold `ip` against a running gate until it finishes. + async fn hold_marketplace_probe_address( + &self, + node_id: u64, + ip_range_id: u64, + ip: &str, + ) -> DbResult<()>; + + /// Record the verdict and release the held address. + async fn finish_marketplace_node_probe( + &self, + node_id: u64, + status: MarketplaceProbeStatus, + detail: Option<&str>, + ) -> DbResult<()>; + + /// Addresses currently held by health gates in one range. + /// + /// Read by the IP allocator: a probe's address is not a `vm_ip_assignment`, + /// so without this a VM could be handed the address a gate is using, and + /// two machines would answer for it. + async fn list_marketplace_probe_ips_in_range(&self, range_id: u64) -> DbResult>; + + /// Addresses held by gates against one node, which the node must route and + /// admit exactly as it would a guest — that is the point of the exercise. + async fn list_marketplace_probe_ips_for_node(&self, node_id: u64) -> DbResult>; + /// Update a node's name, key, status, trust tier or assigned tunnel. /// `operator_id` and /// `created` are immutable and are not written: a node cannot change hands, diff --git a/lnvps_db/src/model.rs b/lnvps_db/src/model.rs index 72e8b3d2..efefbc46 100644 --- a/lnvps_db/src/model.rs +++ b/lnvps_db/src/model.rs @@ -1938,6 +1938,51 @@ impl TunnelPool { } } +/// How a health-gate run ended. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, sqlx::Type)] +#[repr(u16)] +pub enum MarketplaceProbeStatus { + /// The gate is running: an address is held and nothing is decided. + #[default] + Running = 0, + /// A packet reached the node's bridge from the route server. + Passed = 1, + /// It did not, and `detail` says at which step. + Failed = 2, +} + +impl Display for MarketplaceProbeStatus { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + f.write_str(match self { + MarketplaceProbeStatus::Running => "running", + MarketplaceProbeStatus::Passed => "passed", + MarketplaceProbeStatus::Failed => "failed", + }) + } +} + +/// The last health-gate run for a node. +/// +/// Holds an address from a real customer range while it runs, so a VM cannot be +/// given the same one, and keeps the verdict after releasing it — an operator +/// reading why their node is disabled should not be costing them an address to +/// do it. +#[derive(FromRow, Clone, Debug, Default)] +pub struct MarketplaceNodeProbe { + pub id: u64, + pub node_id: u64, + /// The range the held address came from, once one has been taken. `None` + /// for a run that failed before it got that far. + pub ip_range_id: Option, + /// The address held right now, or `None` once the run is over. + pub ip: Option, + pub status: MarketplaceProbeStatus, + /// Which step failed, in the words an operator needs. + pub detail: Option, + pub created: DateTime, + pub finished: Option>, +} + /// A single machine offered by an operator. /// /// There is deliberately no region here: an approved node's region lives on its diff --git a/lnvps_db/src/mysql.rs b/lnvps_db/src/mysql.rs index d3276f12..6ae7e20e 100644 --- a/lnvps_db/src/mysql.rs +++ b/lnvps_db/src/mysql.rs @@ -3,14 +3,14 @@ use crate::{ AppDeploymentFilter, AppDeploymentServiceUsage, AppDeploymentVolumeUsage, AppTag, AsnSubscription, AsnSubscriptionStatus, AvailableIpSpace, Company, DbError, DbResult, DnsServer, EncryptedString, IntervalType, IpRange, IpRangeSubscription, IpSpacePricing, - LNVpsDbBase, MarketplaceNode, MarketplaceNodeStatus, MarketplaceOperator, NewAgentMessage, - PaymentMethod, PaymentMethodConfig, PaymentType, Referral, ReferralCostUsage, ReferralPayout, - Region, RegionStats, Router, RouterBgpRoute, RouterBgpSession, RouterTunnel, - RouterTunnelTraffic, Subscription, SubscriptionLineItem, SubscriptionPayment, - SubscriptionPaymentWithCompany, Tunnel, TunnelPool, User, UserPaymentMethod, UserSshKey, Vm, - VmCostPlan, VmCustomPricing, VmCustomPricingDisk, VmCustomTemplate, VmFirewallPolicy, - VmFirewallRule, VmHistory, VmHost, VmHostDisk, VmIpAssignment, VmOsImage, VmTemplate, - WebauthnCredential, + LNVpsDbBase, MarketplaceNode, MarketplaceNodeProbe, MarketplaceNodeStatus, MarketplaceOperator, + MarketplaceProbeStatus, NewAgentMessage, PaymentMethod, PaymentMethodConfig, PaymentType, + Referral, ReferralCostUsage, ReferralPayout, Region, RegionStats, Router, RouterBgpRoute, + RouterBgpSession, RouterTunnel, RouterTunnelTraffic, Subscription, SubscriptionLineItem, + SubscriptionPayment, SubscriptionPaymentWithCompany, Tunnel, TunnelPool, User, + UserPaymentMethod, UserSshKey, Vm, VmCostPlan, VmCustomPricing, VmCustomPricingDisk, + VmCustomTemplate, VmFirewallPolicy, VmFirewallRule, VmHistory, VmHost, VmHostDisk, + VmIpAssignment, VmOsImage, VmTemplate, WebauthnCredential, }; #[cfg(feature = "admin")] use crate::{AdminDb, AdminRole, AdminRoleAssignment, AdminVmHost}; @@ -4216,6 +4216,94 @@ impl LNVpsDbBase for LNVpsDbMysql { Ok(res.try_get(0)?) } + async fn get_marketplace_node_probe( + &self, + node_id: u64, + ) -> DbResult> { + Ok( + sqlx::query_as("SELECT * FROM marketplace_node_probe WHERE node_id = ?") + .bind(node_id) + .fetch_optional(&self.db) + .await?, + ) + } + + async fn start_marketplace_node_probe(&self, node_id: u64) -> DbResult { + // One statement rather than delete-then-insert: the unique key on + // `node_id` is what stops two gates running against one node, and a + // gap between the two would be exactly the window for the second to + // start. `created` and `finished` are reset because this is a new run, + // not an amendment of the last one. + let res = sqlx::query( + "INSERT INTO marketplace_node_probe (node_id, ip_range_id, ip, status, detail, created, finished) \ + VALUES (?, NULL, NULL, ?, NULL, CURRENT_TIMESTAMP, NULL) \ + ON DUPLICATE KEY UPDATE ip_range_id = NULL, ip = NULL, \ + status = VALUES(status), detail = NULL, created = CURRENT_TIMESTAMP, finished = NULL, \ + id = LAST_INSERT_ID(id)", + ) + .bind(node_id) + .bind(MarketplaceProbeStatus::Running) + .execute(&self.db) + .await?; + Ok(res.last_insert_id()) + } + + async fn hold_marketplace_probe_address( + &self, + node_id: u64, + ip_range_id: u64, + ip: &str, + ) -> DbResult<()> { + sqlx::query("UPDATE marketplace_node_probe SET ip_range_id = ?, ip = ? WHERE node_id = ?") + .bind(ip_range_id) + .bind(ip) + .bind(node_id) + .execute(&self.db) + .await?; + Ok(()) + } + + async fn finish_marketplace_node_probe( + &self, + node_id: u64, + status: MarketplaceProbeStatus, + detail: Option<&str>, + ) -> DbResult<()> { + // The address is released in the same statement as the verdict: a gate + // that recorded its result and then failed to give the address back + // would leak one address per run, from a range customers are waiting + // for. + sqlx::query( + "UPDATE marketplace_node_probe \ + SET status = ?, detail = ?, ip = NULL, finished = CURRENT_TIMESTAMP \ + WHERE node_id = ?", + ) + .bind(status) + .bind(detail) + .bind(node_id) + .execute(&self.db) + .await?; + Ok(()) + } + + async fn list_marketplace_probe_ips_in_range(&self, range_id: u64) -> DbResult> { + Ok(sqlx::query_scalar( + "SELECT ip FROM marketplace_node_probe WHERE ip_range_id = ? AND ip IS NOT NULL", + ) + .bind(range_id) + .fetch_all(&self.db) + .await?) + } + + async fn list_marketplace_probe_ips_for_node(&self, node_id: u64) -> DbResult> { + Ok(sqlx::query_scalar( + "SELECT ip FROM marketplace_node_probe WHERE node_id = ? AND ip IS NOT NULL", + ) + .bind(node_id) + .fetch_all(&self.db) + .await?) + } + async fn update_marketplace_node(&self, node: &MarketplaceNode) -> DbResult<()> { sqlx::query( "UPDATE marketplace_node \ diff --git a/lnvps_node/src/control.rs b/lnvps_node/src/control.rs index 79fe0738..215b3352 100644 --- a/lnvps_node/src/control.rs +++ b/lnvps_node/src/control.rs @@ -22,10 +22,10 @@ use axum::extract::{DefaultBodyLimit, Request, State}; use axum::http::{StatusCode, header}; use axum::middleware::{self, Next}; use axum::response::{IntoResponse, Response}; -use axum::routing::get; +use axum::routing::{get, post}; use axum::{Json, Router}; use nostr::PublicKey; -use serde::Serialize; +use serde::{Deserialize, Serialize}; use crate::control_auth::{self, ReplayGuard}; use crate::tls::NodeTls; @@ -62,6 +62,14 @@ pub struct ControlState { pub net: Arc, /// How the node reads its packet filter back, for the same reason. pub fw: Arc, + /// How the node re-applies its data plane when LNVPS asks it to. + /// + /// Optional because the control API is constructed before the daemon has a + /// credential to fetch with, and because `dataplane observe` serves without + /// one at all. A node that cannot refresh says so rather than pretending it + /// did — LNVPS would otherwise conclude the node had applied a document it + /// never fetched. + pub refresh: Option>, } impl ControlState { @@ -78,9 +86,16 @@ impl ControlState { base_url: format!("https://{addr}"), net, fw, + refresh: None, } } + /// Let LNVPS ask this node to re-apply its data plane. + pub fn with_refresh(mut self, refresh: Arc) -> Self { + self.refresh = Some(refresh); + self + } + /// The data plane as this machine actually has it. pub async fn observe(&self) -> crate::net::DataPlaneState { crate::net::observe(self.net.as_ref(), self.fw.as_ref()) @@ -103,10 +118,23 @@ pub struct NodeStatus { pub dataplane: crate::net::DataPlaneState, } +/// Re-fetching and applying the data plane on demand. +/// +/// A trait rather than the daemon's own function, because the control API is +/// built long before the credential and the outbound client are — and because a +/// test needs to know that a refresh was asked for without one being performed. +#[async_trait::async_trait] +pub trait Refresh: Send + Sync { + /// Fetch the current document from LNVPS and apply it, returning what + /// changed. + async fn refresh(&self) -> Result>; +} + /// The control router, with authentication layered over everything. pub fn router(state: Arc) -> Router { Router::new() .route("/api/v1/status", get(get_status)) + .route("/api/v1/dataplane/refresh", post(refresh_dataplane)) // Order matters: the body limit is outermost so an oversized body is // rejected before authentication reads it into memory. .layer(middleware::from_fn_with_state(state.clone(), authenticate)) @@ -238,6 +266,40 @@ async fn get_status(State(state): State>) -> Json }) } +/// What a refresh changed. +#[derive(Debug, Serialize, Deserialize, Default, PartialEq)] +pub struct RefreshResult { + /// The applied changes, in the daemon's own words. Empty means the node was + /// already right, which is the normal answer on every refresh after the + /// first — and, for the health gate, the answer that says the probe address + /// was already in place. + pub changed: Vec, +} + +/// Re-fetch the data plane from LNVPS and apply it now. +/// +/// Exists so LNVPS does not have to wait out a heartbeat to see a change it +/// just made take effect. Nothing here is trusted from the request: the node +/// fetches the document itself, from LNVPS, with its own credential — this only +/// says *when*. +async fn refresh_dataplane( + State(state): State>, +) -> Result, (StatusCode, String)> { + let Some(refresh) = state.refresh.clone() else { + return Err(( + StatusCode::SERVICE_UNAVAILABLE, + "This node cannot refresh on demand; it has no credential to fetch with".to_string(), + )); + }; + match refresh.refresh().await { + Ok(changed) => Ok(Json(RefreshResult { changed })), + // The daemon's own message, because every cause needs a different + // person: an expired credential is the operator's, an unreachable LNVPS + // is ours, and a netlink error is neither. + Err(e) => Err((StatusCode::BAD_GATEWAY, format!("{e:#}"))), + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/lnvps_node/src/main.rs b/lnvps_node/src/main.rs index 7747c2fd..b636a3ba 100644 --- a/lnvps_node/src/main.rs +++ b/lnvps_node/src/main.rs @@ -239,19 +239,56 @@ async fn run(config_path: &Path) -> Result<()> { })?; control::serve_on( - Arc::new(ControlState::new(control_pubkey, addr, kernel, fw)), + Arc::new( + ControlState::new(control_pubkey, addr, kernel.clone(), fw.clone()).with_refresh( + Arc::new(OnDemandRefresh { + config: config.clone(), + kernel, + fw, + }), + ), + ), listener, tls, ) .await } +/// Re-applying the data plane because LNVPS asked, rather than because the +/// timer came round. +/// +/// The same code path as the periodic refresh, deliberately: a prompt path that +/// did something slightly different would be a second way for a node to be +/// configured, exercised only when LNVPS is in a hurry. +struct OnDemandRefresh { + config: NodeConfig, + kernel: Arc, + fw: Arc, +} + +#[async_trait::async_trait] +impl control::Refresh for OnDemandRefresh { + async fn refresh(&self) -> Result> { + apply_dataplane_changes(&self.config, self.kernel.as_ref(), self.fw.as_ref()).await + } +} + /// Fetch the data plane and apply it. async fn apply_dataplane( config: &NodeConfig, kernel: &dyn lnvps_node::net::NetOps, fw: &dyn lnvps_node::fw::FirewallOps, ) -> Result<()> { + apply_dataplane_changes(config, kernel, fw).await?; + Ok(()) +} + +/// The same, returning what changed, for the caller that has somebody waiting. +async fn apply_dataplane_changes( + config: &NodeConfig, + kernel: &dyn lnvps_node::net::NetOps, + fw: &dyn lnvps_node::fw::FirewallOps, +) -> Result> { let credential = Credential::load_checked(&config.credential)?; let api = lnvps_node::api::LnvpsApi::new(&config.api_url, &credential)?; @@ -269,5 +306,5 @@ async fn apply_dataplane( if !applied.is_empty() { log::debug!("Applied data plane: {}", applied.join("; ")); } - Ok(()) + Ok(applied) } diff --git a/work/marketplace.md b/work/marketplace.md index e0494a59..28a23604 100644 --- a/work/marketplace.md +++ b/work/marketplace.md @@ -1020,7 +1020,7 @@ for the guest, on the same address it would have had. competes for the port. An operator who changes it makes their own node unreachable, which the gate reports as unreachable — self-correcting, and cheaper than a column. -##### 4c3b — The gate itself (M/L) +##### 4c3b — The gate itself (M/L) ✅ - A probe address is taken from a real customer range in the node's region, sent to the node as a guest, and pinged from the route server. That is the production path exactly: a VM's address is statically routed from the core network to the node's `wgln0`, forwarded over `br-lnvps` @@ -1032,6 +1032,35 @@ for the guest, on the same address it would have had. direction: a node that never carries a customer is a support conversation, a node that carries one badly is an outage. +Built as `Worker::health_check_node`, with `marketplace_node_probe` holding the run. Decisions +worth recording: + +- **The run is recorded before it starts.** A gate that only wrote a row once it had taken an + address would tell the operator whose node never handshook precisely nothing, and that is the + most common failure. +- **The address is released in the same statement as the verdict.** A gate that recorded its + result and then failed to give the address back would leak one address per attempt out of a + range customers are waiting for. +- **The allocator sees held probe addresses.** They are not `vm_ip_assignment` rows, so without + that the gate could hand a VM the address it is currently proving a node with — two machines + answering for one address, which is the failure this whole increment exists to prevent. +- **IPv4 ranges are preferred.** A guest's IPv6 address is normally derived from its MAC + (EUI64) and a probe has no guest, so the v4 path is the one a probe stands in for exactly. A + v6-only region is still gated, from the first free address in its range. +- **The node is told to apply the document rather than left to its heartbeat**, via a new + `POST /api/v1/dataplane/refresh` on the node control API. Nothing about the document is sent + — the node fetches it itself, with its own credential; this only says *when*. An approval + that took a minute to conclude would be a minute of an operator watching nothing happen. +- **`probe_address` echoes its verdict rather than using ping's exit code.** To the SSH + transport a non-zero exit means "the command failed", which is right everywhere else and + wrong here: "no reply" is the answer the gate asked for, and it has to stay distinguishable + from a route server that cannot be reached at all. Those two results need different people. + +Caught while writing the tests: the node's guest list (`node_guests`) and the route server's +plan (`guest_addresses`) are **two separate functions** over the same idea. Adding the probe to +one and not the other would have produced a node holding an address the route server never +routed — a gate that failed for a reason that had nothing to do with the node. + ### Increment 5 — Confidential computing: attestation + encrypted disks (L) - Verify SEV-SNP attestation reports (`sev` crate) / TDX quotes (`dcap-qvl` crate) against AMD/Intel roots + measurement allow-list; store attestation state on `marketplace_node`;