diff --git a/ADMIN_API_ENDPOINTS.md b/ADMIN_API_ENDPOINTS.md index 05f38120..9e7b57d7 100644 --- a/ADMIN_API_ENDPOINTS.md +++ b/ADMIN_API_ENDPOINTS.md @@ -16,7 +16,7 @@ Admin API request/response format reference for LLM consumption. `"redhatenterprise"` **IpRangeAllocationMode**: `"random"`, `"sequential"`, `"slaac_eui64"` **NetworkAccessPolicyKind**: `"static_arp"` -**RouterKind**: `"mikrotik"`, `"ovh_additional_ip"` +**RouterKind**: `"mikrotik"`, `"ovh_additional_ip"`, `"linux_ssh"` **AdminUserRole**: `"super_admin"`, `"admin"`, `"read_only"` **AdminUserStatus**: `"active"`, `"suspended"`, `"deleted"` **SubscriptionPaymentType**: `"purchase"`, `"renewal"`, `"upgrade"` @@ -2320,6 +2320,64 @@ Required Permission: `router::delete` Note: Routers that are used by access policies cannot be deleted. You must first remove the router from all access policies before deleting it. +#### List Router Tunnels + +``` +GET /api/admin/v1/routers/{router_id}/tunnels +``` + +Required Permission: `router::view` + +Returns the cached tunnel inventory discovered on the router (GRE/VXLAN/WireGuard). Each entry: `id`, `router_id`, +`name`, `kind` (`"gre"` | `"vxlan"` | `"wireguard"`), `local_addr`, `remote_addr`, `enabled`, `last_seen`. Refreshed +by a background sampler (~60s). + +#### Get Tunnel Traffic History + +``` +GET /api/admin/v1/routers/{router_id}/tunnels/{tunnel_name}/traffic +``` + +Required Permission: `router::view` + +Query parameters (optional): `from`, `to` (RFC3339 timestamps; default: last 24 hours). + +Returns per-tunnel traffic samples: `tunnel_name`, `rx_bytes`, `tx_bytes`, `sampled_at`. Tunnel interface counters +are the canonical per-session traffic source — BGP sessions expose no byte counters. + +#### List BGP Sessions + +``` +GET /api/admin/v1/routers/{router_id}/bgp/sessions +``` + +Required Permission: `router::view` + +Returns cached BGP sessions: `id`, `router_id`, `name`, `peer_ip`, `peer_asn`, `local_asn`, `state`, +`prefixes_received`, `prefixes_sent`, `enabled`, `direction` (`"upstream"` | `"downstream"` | `"peer"` | +`"unknown"`), `last_seen`. + +#### Toggle BGP Session + +``` +POST /api/admin/v1/routers/{router_id}/bgp/sessions/toggle +``` + +Required Permission: `router::update` + +Body: + +```json +{ + "session_id": "string", + // Backend session id from the sessions listing (protocol name on BIRD, .id on Mikrotik) + "enabled": boolean +} +``` + +Returns a `JobResponse` (`{ "job_id": "string" }`). The enable/disable is applied asynchronously by the worker and +the session cache is refreshed afterwards. + ### VM IP Assignment Management VM IP assignments bind specific IP addresses from IP ranges to virtual machines. These endpoints provide comprehensive diff --git a/API_CHANGELOG.md b/API_CHANGELOG.md index 0da7319a..54fc86d1 100644 --- a/API_CHANGELOG.md +++ b/API_CHANGELOG.md @@ -8,6 +8,14 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). ### Added +- **2026-06-21** - Route server management: BGP session and tunnel visibility/control (admin) + - New `RouterKind` value `linux_ssh` — a Linux router managed over SSH (BIRD/Pathvector routing, iproute2/WireGuard tunnels). Configure with `url = ssh://@[:]/` and `token` = the SSH private key (PEM). Selectable via `kind` on `POST/PATCH /api/admin/v1/routers`. + - `GET /api/admin/v1/routers/{id}/tunnels` — list cached tunnels discovered on the router. Each `AdminRouterTunnel`: `id`, `router_id`, `name`, `kind` (`gre`|`vxlan`|`wireguard`), `local_addr`, `remote_addr`, `enabled`, `last_seen`. Requires `router::view`. + - `GET /api/admin/v1/routers/{id}/tunnels/{name}/traffic` — per-tunnel traffic history (`AdminRouterTunnelTraffic`: `tunnel_name`, `rx_bytes`, `tx_bytes`, `sampled_at`). Optional `from`/`to` RFC3339 query params (default: last 24h). Tunnel interface counters are the canonical per-session traffic source — BGP sessions have no byte counters. Requires `router::view`. + - `GET /api/admin/v1/routers/{id}/bgp/sessions` — list cached BGP sessions (`AdminRouterBgpSession`: `id`, `router_id`, `name`, `peer_ip`, `peer_asn`, `local_asn`, `state`, `prefixes_received`, `prefixes_sent`, `enabled`, `direction` (`upstream`|`downstream`|`peer`|`unknown`), `last_seen`). Requires `router::view`. + - `POST /api/admin/v1/routers/{id}/bgp/sessions/toggle` — enable/disable a BGP session. Body: `{ "session_id": string, "enabled": boolean }` (`session_id` is the backend id from the sessions listing — protocol name on BIRD, `.id` on Mikrotik). Returns a `JobResponse` (`{ "job_id": string }`); the action is applied asynchronously and the session cache refreshed. Requires `router::update`. + - Tunnel inventory, per-tunnel traffic, and BGP session state are refreshed by a background sampler (~60s). All router queries are bounded and safe on routers carrying a full DFZ table. + - **2026-06-18** - Subscription line items now expose a typed `resource` reference - `SubscriptionLineItem` (public `GET /api/v1/subscriptions/{id}`) and `AdminSubscriptionLineItemInfo` (admin subscription + line-item endpoints) gain a `resource` field: a tagged union resolved server-side from the line item's subscription type. Shapes: `{ "type": "vps", "vm_id": number }`, `{ "type": "ip_range", "ip_range_subscription_id": number }`, or `null` when there is no linked resource. - `AdminSubscriptionLineItemInfo` now also includes the `subscription_type` discriminant. diff --git a/lnvps_api/Cargo.toml b/lnvps_api/Cargo.toml index 8f47c434..6340525f 100644 --- a/lnvps_api/Cargo.toml +++ b/lnvps_api/Cargo.toml @@ -15,6 +15,7 @@ default = [ "nostr-nwc", "nostr-domain", "proxmox", + "linux-ssh", "lnd", "cloudflare", "revolut", @@ -24,6 +25,8 @@ default = [ ] openapi = ["dep:openapi-build-gen"] mikrotik = [] +# Linux router managed over SSH (shares the ssh2-based SshClient with proxmox) +linux-ssh = ["dep:ssh2"] nostr-dm = ["dep:nostr-sdk", "nostr-sdk/nip44", "nostr-sdk/nip59"] nostr-dvm = ["dep:nostr-sdk"] nostr-domain = ["lnvps_db/nostr-domain", "dep:uuid"] diff --git a/lnvps_api/src/bin/api.rs b/lnvps_api/src/bin/api.rs index 45e7f9f1..ff240927 100644 --- a/lnvps_api/src/bin/api.rs +++ b/lnvps_api/src/bin/api.rs @@ -154,6 +154,9 @@ async fn main() -> Result<(), Error> { tasks.push(worker.spawn_job_interval(WorkJob::CheckVms, Duration::from_secs(30))); tasks.push(worker.spawn_job_interval(WorkJob::CheckSubscriptions, Duration::from_secs(30))); + // Sample router tunnel traffic + refresh BGP/tunnel state every 60s + tasks + .push(worker.spawn_job_interval(WorkJob::SampleRouterTraffic, Duration::from_secs(60))); tasks.push(worker.spawn_handler_loop()); // check all nostr domains every 10 minutes for CNAME entries (enable/disable as needed) diff --git a/lnvps_api/src/lib.rs b/lnvps_api/src/lib.rs index adcaf6d9..f29ac350 100644 --- a/lnvps_api/src/lib.rs +++ b/lnvps_api/src/lib.rs @@ -8,7 +8,7 @@ pub mod payments; pub mod provisioner; pub mod router; pub mod settings; -#[cfg(feature = "proxmox")] +#[cfg(any(feature = "proxmox", feature = "linux-ssh"))] pub mod ssh_client; pub mod subscription; pub mod worker; diff --git a/lnvps_api/src/mocks.rs b/lnvps_api/src/mocks.rs index 13cbed47..0bd41b6a 100644 --- a/lnvps_api/src/mocks.rs +++ b/lnvps_api/src/mocks.rs @@ -4,7 +4,9 @@ use crate::host::dummy_host::DummyVmHost; use crate::host::{ FullVmInfo, TerminalStream, TimeSeries, TimeSeriesData, VmHostClient, VmHostInfo, }; -use crate::router::{ArpEntry, Router}; +use crate::router::{ + ArpEntry, BgpPeer, BgpRoute, BgpRouter, BgpSession, Router, Tunnel, TunnelRouter, TunnelTraffic, +}; /// Type alias so tests can refer to the in-memory VM host as `MockVmHost`. pub type MockVmHost = DummyVmHost; @@ -44,6 +46,8 @@ use tokio::sync::Mutex; #[derive(Debug, Clone)] pub struct MockRouter { arp: Arc>>, + tunnels: Arc>>, + sessions: Arc>>, } impl Default for MockRouter { @@ -56,16 +60,32 @@ impl MockRouter { pub fn new() -> Self { static LAZY_ARP: LazyLock>>> = LazyLock::new(|| Arc::new(Mutex::new(HashMap::new()))); + static LAZY_TUNNELS: LazyLock>>> = + LazyLock::new(|| Arc::new(Mutex::new(HashMap::new()))); + static LAZY_SESSIONS: LazyLock>>> = + LazyLock::new(|| Arc::new(Mutex::new(HashMap::new()))); Self { arp: LAZY_ARP.clone(), + tunnels: LAZY_TUNNELS.clone(), + sessions: LAZY_SESSIONS.clone(), } } - /// Clear all ARP entries - useful for test isolation + /// Clear all ARP entries, tunnels and BGP sessions - useful for test isolation pub async fn clear(&self) { let mut arp = self.arp.lock().await; arp.clear(); + let mut tunnels = self.tunnels.lock().await; + tunnels.clear(); + let mut sessions = self.sessions.lock().await; + sessions.clear(); + } + + /// Seed a BGP session for tests + pub async fn add_session(&self, session: BgpSession) { + let mut sessions = self.sessions.lock().await; + sessions.insert(session.id.clone(), session); } } @@ -160,6 +180,117 @@ impl Router for MockRouter { } Ok(entry.clone()) } + + fn tunnel(&self) -> Option<&dyn TunnelRouter> { + Some(self) + } + + fn bgp(&self) -> Option<&dyn BgpRouter> { + Some(self) + } +} + +#[async_trait] +impl TunnelRouter for MockRouter { + async fn list_tunnels(&self) -> OpResult> { + let tunnels = self.tunnels.lock().await; + Ok(tunnels.values().cloned().collect()) + } + + async fn add_tunnel(&self, tunnel: &Tunnel) -> OpResult { + let mut tunnels = self.tunnels.lock().await; + if tunnels.contains_key(&tunnel.name) { + return Err(OpError::Fatal(anyhow::anyhow!( + "Tunnel already exists: {}", + tunnel.name + ))); + } + let stored = Tunnel { + id: Some(tunnel.name.clone()), + enabled: true, + ..tunnel.clone() + }; + tunnels.insert(tunnel.name.clone(), stored.clone()); + Ok(stored) + } + + async fn remove_tunnel(&self, id: &str) -> OpResult<()> { + let mut tunnels = self.tunnels.lock().await; + tunnels.remove(id); + Ok(()) + } + + async fn update_tunnel(&self, tunnel: &Tunnel) -> OpResult { + let mut tunnels = self.tunnels.lock().await; + let stored = Tunnel { + id: Some(tunnel.name.clone()), + ..tunnel.clone() + }; + tunnels.insert(tunnel.name.clone(), stored.clone()); + Ok(stored) + } + + async fn tunnel_traffic(&self) -> OpResult> { + let tunnels = self.tunnels.lock().await; + Ok(tunnels + .values() + .map(|t| TunnelTraffic { + name: t.name.clone(), + rx_bytes: 0, + tx_bytes: 0, + }) + .collect()) + } +} + +#[async_trait] +impl BgpRouter for MockRouter { + async fn list_sessions(&self) -> OpResult> { + let sessions = self.sessions.lock().await; + Ok(sessions.values().cloned().collect()) + } + + async fn originated_routes(&self, candidates: &[String]) -> OpResult> { + let all = vec![BgpRoute { + prefix: "203.0.113.0/24".to_string(), + next_hop: None, + }]; + if candidates.is_empty() { + Ok(all) + } else { + Ok(all + .into_iter() + .filter(|r| candidates.contains(&r.prefix)) + .collect()) + } + } + + async fn default_route(&self) -> OpResult> { + Ok(Some(BgpRoute { + prefix: "0.0.0.0/0".to_string(), + next_hop: Some("192.0.2.1".to_string()), + })) + } + + async fn discover_peers(&self) -> OpResult> { + let sessions = self.sessions.lock().await; + Ok(sessions + .values() + .map(|s| BgpPeer { + peer_ip: s.peer_ip.clone(), + asn: s.peer_asn, + direction: s.direction, + }) + .collect()) + } + + async fn set_session_enabled(&self, id: &str, enabled: bool) -> OpResult<()> { + let mut sessions = self.sessions.lock().await; + if let Some(s) = sessions.get_mut(id) { + s.enabled = enabled; + } + Ok(()) + } } #[derive(Clone, Debug, Default)] diff --git a/lnvps_api/src/router/linux_ssh.rs b/lnvps_api/src/router/linux_ssh.rs new file mode 100644 index 00000000..aac1cd62 --- /dev/null +++ b/lnvps_api/src/router/linux_ssh.rs @@ -0,0 +1,934 @@ +use anyhow::{Context, Result, bail}; +use async_trait::async_trait; +use reqwest::Url; +use serde::Deserialize; + +use lnvps_api_common::retry::OpResult; +use lnvps_api_common::{op_fatal, op_transient}; + +use crate::router::{ + ArpEntry, BgpPeer, BgpPeerDirection, BgpRoute, BgpRouter, BgpSession, GreConfig, Router, + Tunnel, TunnelConfig, TunnelRouter, TunnelTraffic, VxlanConfig, WireguardConfig, WireguardPeer, +}; +use crate::ssh_client::SshClient; + +/// A router backed by a generic Linux machine managed over SSH. +/// +/// ARP management is implemented with iproute2 (`ip neigh`). Static neighbour +/// entries are added as `PERMANENT` so they survive without ongoing ARP traffic, +/// mirroring the behaviour of the Mikrotik static-ARP router. +/// +/// Connection details are encoded in the router config: +/// - `url`: `ssh://@[:]/` (port defaults to 22) +/// - `token`: the SSH private key in PEM format +pub struct LinuxSshRouter { + host: String, + username: String, + /// Default network interface used for neighbour entries + interface: String, + /// SSH private key (PEM) + key: String, +} + +impl LinuxSshRouter { + /// Build a router from the stored config `url` and `token`. + pub fn new(url: &str, key: &str) -> Result { + let u = Url::parse(url).context("Invalid linux-ssh router url")?; + if u.scheme() != "ssh" { + bail!("linux-ssh router url must use the ssh:// scheme"); + } + let host = u + .host_str() + .context("Missing host in linux-ssh router url")?; + let port = u.port().unwrap_or(22); + let username = if u.username().is_empty() { + "root".to_string() + } else { + u.username().to_string() + }; + let interface = u.path().trim_matches('/').to_string(); + if interface.is_empty() { + bail!( + "linux-ssh router url must include an interface in the path, e.g. ssh://root@host/eth0" + ); + } + Ok(Self { + host: format!("{}:{}", host, port), + username, + interface, + key: key.to_string(), + }) + } + + /// Open a fresh SSH connection for a single operation. + /// + /// Connecting per-operation keeps the router `Send + Sync` without holding a + /// live (non-`Sync`) ssh2 session, and is naturally resilient to dropped + /// connections. + async fn connect(&self) -> Result { + let mut client = SshClient::new()?; + client + .connect_with_key(&self.host, &self.username, &self.key) + .await?; + Ok(client) + } + + /// Run a command, mapping connection failures to transient errors and + /// non-zero exits to fatal errors. Returns stdout on success. + async fn exec_checked(&self, cmd: &str) -> OpResult { + let mut client = match self.connect().await { + Ok(c) => c, + Err(e) => op_transient!(e), + }; + let (code, out) = match client.execute(cmd).await { + Ok(r) => r, + Err(e) => op_transient!(e), + }; + if code != 0 { + op_fatal!("command failed ({}): {} :: {}", code, cmd, out); + } + Ok(out) + } +} + +#[async_trait] +impl Router for LinuxSshRouter { + async fn generate_mac(&self, _ip: &str, _comment: &str) -> Result> { + // Linux doesn't require a specific MAC for the neighbour entry + Ok(None) + } + + async fn list_arp_entry(&self) -> OpResult> { + let mut client = match self.connect().await { + Ok(c) => c, + Err(e) => op_transient!(e), + }; + let (code, out) = match client.execute("ip -j neigh show").await { + Ok(r) => r, + Err(e) => op_transient!(e), + }; + if code != 0 { + op_fatal!("ip neigh show failed ({}): {}", code, out); + } + let entries: Vec = match serde_json::from_str(&out) { + Ok(e) => e, + Err(e) => op_fatal!("Failed to parse ip neigh output: {}", e), + }; + Ok(entries + .into_iter() + .filter(|e| e.dev == self.interface) + .filter_map(|e| e.into_arp_entry()) + .collect()) + } + + async fn add_arp_entry(&self, entry: &ArpEntry) -> OpResult { + let dev = entry.interface.as_deref().unwrap_or(&self.interface); + let cmd = format!( + "ip neigh replace {} lladdr {} dev {} nud permanent", + entry.address, entry.mac_address, dev + ); + let mut client = match self.connect().await { + Ok(c) => c, + Err(e) => op_transient!(e), + }; + let (code, out) = match client.execute(&cmd).await { + Ok(r) => r, + Err(e) => op_transient!(e), + }; + if code != 0 { + op_fatal!("ip neigh replace failed ({}): {}", code, out); + } + // Linux neighbour entries are keyed by (ip, dev); use the ip as the id + Ok(ArpEntry { + id: Some(entry.address.clone()), + interface: Some(dev.to_string()), + ..entry.clone() + }) + } + + async fn remove_arp_entry(&self, id: &str) -> OpResult<()> { + // `id` is the neighbour IP address (see add_arp_entry) + let cmd = format!("ip neigh del {} dev {}", id, self.interface); + let mut client = match self.connect().await { + Ok(c) => c, + Err(e) => op_transient!(e), + }; + let (code, out) = match client.execute(&cmd).await { + Ok(r) => r, + Err(e) => op_transient!(e), + }; + if code != 0 { + op_fatal!("ip neigh del failed ({}): {}", code, out); + } + Ok(()) + } + + async fn update_arp_entry(&self, entry: &ArpEntry) -> OpResult { + // `ip neigh replace` is idempotent and handles both add and update + self.add_arp_entry(entry).await + } + + fn tunnel(&self) -> Option<&dyn TunnelRouter> { + Some(self) + } + + fn bgp(&self) -> Option<&dyn BgpRouter> { + Some(self) + } +} + +#[async_trait] +impl BgpRouter for LinuxSshRouter { + async fn list_sessions(&self) -> OpResult> { + let out = self.exec_checked("birdc -r show protocols all").await?; + Ok(parse_bird_protocols(&out)) + } + + async fn originated_routes(&self, candidates: &[String]) -> OpResult> { + // Routes locally originated by static protocols. The `where` filter is a + // single in-memory pass in birdc and the *output* is bounded to our own + // originated prefixes, so this is safe even with a full table loaded. + let out = self + .exec_checked("birdc -r 'show route where source = RTS_STATIC'") + .await?; + let mut routes = parse_bird_routes(&out); + if !candidates.is_empty() { + let set: std::collections::HashSet<&str> = + candidates.iter().map(|s| s.as_str()).collect(); + routes.retain(|r| set.contains(r.prefix.as_str())); + } + Ok(routes) + } + + async fn default_route(&self) -> OpResult> { + let v4 = self + .exec_checked("birdc -r 'show route for 0.0.0.0/0'") + .await?; + let mut routes = parse_bird_routes(&v4); + if routes.is_empty() { + let v6 = self.exec_checked("birdc -r 'show route for ::/0'").await?; + routes = parse_bird_routes(&v6); + } + Ok(routes.into_iter().next()) + } + + async fn discover_peers(&self) -> OpResult> { + let sessions = self.list_sessions().await?; + Ok(sessions + .into_iter() + .map(|s| BgpPeer { + peer_ip: s.peer_ip, + asn: s.peer_asn, + direction: s.direction, + }) + .collect()) + } + + async fn set_session_enabled(&self, id: &str, enabled: bool) -> OpResult<()> { + let action = if enabled { "enable" } else { "disable" }; + self.exec_checked(&format!("birdc {} {}", action, shq(id))) + .await?; + Ok(()) + } +} + +/// Map a BIRD/RFC-9234 role string to a peer relationship (best-effort). +fn role_to_direction(role: &str) -> BgpPeerDirection { + match role.to_ascii_lowercase().as_str() { + "provider" | "rs-server" => BgpPeerDirection::Upstream, + "customer" | "rs-client" => BgpPeerDirection::Downstream, + "peer" => BgpPeerDirection::Peer, + _ => BgpPeerDirection::Unknown, + } +} + +/// Find the value after `key` on a line like ` Key: value` within a block. +fn bird_field<'a>(block: &'a str, key: &str) -> Option<&'a str> { + block.lines().find_map(|l| { + let t = l.trim(); + t.strip_prefix(key) + .map(|rest| rest.trim_start_matches(':').trim()) + }) +} + +/// Parse `birdc show protocols all` output into BGP sessions. +fn parse_bird_protocols(out: &str) -> Vec { + let mut sessions = Vec::new(); + let lines: Vec<&str> = out.lines().collect(); + let mut i = 0; + while i < lines.len() { + let line = lines[i]; + // Header lines start in column 0 + if line.is_empty() || line.starts_with(char::is_whitespace) { + i += 1; + continue; + } + if line.starts_with("BIRD") || line.starts_with("Name") { + i += 1; + continue; + } + let cols: Vec<&str> = line.split_whitespace().collect(); + let name = cols.first().copied().unwrap_or(""); + let proto = cols.get(1).copied().unwrap_or(""); + // Gather following indented detail lines into a block + let header = line; + let mut block = String::new(); + i += 1; + while i < lines.len() && (lines[i].is_empty() || lines[i].starts_with(char::is_whitespace)) + { + block.push_str(lines[i]); + block.push('\n'); + i += 1; + } + if proto != "BGP" { + continue; + } + let peer_ip = bird_field(&block, "Neighbor address").map(|s| s.to_string()); + let peer_asn = bird_field(&block, "Neighbor AS").and_then(|s| s.parse().ok()); + let local_asn = bird_field(&block, "Local AS").and_then(|s| s.parse().ok()); + let state = bird_field(&block, "BGP state") + .map(|s| s.to_string()) + .unwrap_or_else(|| "Unknown".to_string()); + let (prefixes_received, prefixes_sent) = parse_bird_routes_stats(&block); + let direction = bird_field(&block, "Role") + .map(role_to_direction) + .unwrap_or_default(); + // A disabled protocol is reported with "disabled" in the header info column + let enabled = !header.to_ascii_lowercase().contains("disabled"); + sessions.push(BgpSession { + id: name.to_string(), + name: name.to_string(), + peer_ip, + peer_asn, + local_asn, + state, + prefixes_received, + prefixes_sent, + enabled, + direction, + }); + } + sessions +} + +/// Extract `(imported, exported)` counts from a `Routes: N imported, M exported` +/// line within a protocol detail block. +fn parse_bird_routes_stats(block: &str) -> (Option, Option) { + for l in block.lines() { + let t = l.trim(); + if let Some(rest) = t.strip_prefix("Routes:") { + let mut imported = None; + let mut exported = None; + let tokens: Vec<&str> = rest.split_whitespace().collect(); + for w in tokens.windows(2) { + match w[1].trim_end_matches(',') { + "imported" => imported = w[0].parse().ok(), + "exported" => exported = w[0].parse().ok(), + _ => {} + } + } + return (imported, exported); + } + } + (None, None) +} + +/// Parse `birdc show route` output into routes. +fn parse_bird_routes(out: &str) -> Vec { + let mut routes = Vec::new(); + let mut cur: Option = None; + for line in out.lines() { + if line.starts_with("BIRD") || line.starts_with("Table") || line.trim().is_empty() { + continue; + } + if !line.starts_with(char::is_whitespace) { + if let Some(r) = cur.take() { + routes.push(r); + } + let prefix = line.split_whitespace().next().unwrap_or("").to_string(); + if prefix.contains('/') { + cur = Some(BgpRoute { + prefix, + next_hop: None, + }); + } + } else if let Some(rest) = line.trim().strip_prefix("via ") { + let nh = rest.split_whitespace().next().map(|s| s.to_string()); + if let Some(r) = cur.as_mut().filter(|r| r.next_hop.is_none()) { + r.next_hop = nh; + } + } + } + if let Some(r) = cur.take() { + routes.push(r); + } + routes +} + +/// Quote a value for safe use inside a single-quoted shell string. +fn shq(s: &str) -> String { + format!("'{}'", s.replace('\'', "'\\''")) +} + +/// Linux interface kinds we treat as tunnels +fn kind_from_info(info_kind: &str) -> Option<&'static str> { + match info_kind { + "gre" | "gretap" => Some("gre"), + "vxlan" => Some("vxlan"), + "wireguard" => Some("wireguard"), + _ => None, + } +} + +/// Parse a GRE key which `ip` may render either as a plain integer or as a +/// dotted-quad (e.g. `"0.0.0.10"`). +fn parse_gre_key(s: &str) -> Option { + if let Ok(v) = s.parse::() { + return Some(v); + } + let octets: Vec = s.split('.').filter_map(|o| o.parse().ok()).collect(); + if octets.len() == 4 { + Some((octets[0] << 24) | (octets[1] << 16) | (octets[2] << 8) | octets[3]) + } else { + None + } +} + +/// `ip` may render a GRE key as a JSON number or a dotted-quad string. +fn value_to_gre_key(v: &serde_json::Value) -> Option { + if let Some(n) = v.as_u64() { + return u32::try_from(n).ok(); + } + v.as_str().and_then(parse_gre_key) +} + +impl LinuxSshRouter { + /// Build a [`Tunnel`] from a parsed `ip` link entry, optionally augmented + /// with WireGuard details from `wg show all dump`. + fn link_to_tunnel( + link: &IpLink, + wg: &std::collections::HashMap, + ) -> Option { + let info = link.linkinfo.as_ref()?; + let mapped = kind_from_info(&info.info_kind)?; + let data = info.info_data.clone().unwrap_or_default(); + let enabled = link.flags.iter().any(|f| f == "UP"); + let config = match mapped { + "gre" => TunnelConfig::Gre(GreConfig { + key: data.ikey.as_ref().and_then(value_to_gre_key), + }), + "vxlan" => TunnelConfig::Vxlan(VxlanConfig { + vni: data.id.unwrap_or(0), + dst_port: data.port, + }), + "wireguard" => { + TunnelConfig::Wireguard(wg.get(&link.ifname).cloned().unwrap_or_default()) + } + _ => return None, + }; + Some(Tunnel { + id: Some(link.ifname.clone()), + name: link.ifname.clone(), + local_addr: data.local.clone(), + remote_addr: data.remote.clone(), + enabled, + config, + }) + } +} + +#[async_trait] +impl TunnelRouter for LinuxSshRouter { + async fn list_tunnels(&self) -> OpResult> { + let out = self.exec_checked("ip -s -d -j link show").await?; + let links: Vec = match serde_json::from_str(&out) { + Ok(l) => l, + Err(e) => op_fatal!("Failed to parse ip link output: {}", e), + }; + // Only query WireGuard if any wg interface exists + let has_wg = links.iter().any(|l| { + l.linkinfo + .as_ref() + .map(|i| i.info_kind == "wireguard") + .unwrap_or(false) + }); + let wg = if has_wg { + let dump = self.exec_checked("wg show all dump").await?; + parse_wg_dump(&dump) + } else { + std::collections::HashMap::new() + }; + Ok(links + .iter() + .filter_map(|l| Self::link_to_tunnel(l, &wg)) + .collect()) + } + + async fn add_tunnel(&self, tunnel: &Tunnel) -> OpResult { + let name = &tunnel.name; + let mut script = String::new(); + match &tunnel.config { + TunnelConfig::Gre(c) => { + script.push_str(&format!("ip link add {} type gre", shq(name))); + if let Some(l) = &tunnel.local_addr { + script.push_str(&format!(" local {}", shq(l))); + } + if let Some(r) = &tunnel.remote_addr { + script.push_str(&format!(" remote {}", shq(r))); + } + if let Some(k) = c.key { + script.push_str(&format!(" key {}", k)); + } + } + TunnelConfig::Vxlan(c) => { + script.push_str(&format!( + "ip link add {} type vxlan id {}", + shq(name), + c.vni + )); + if let Some(l) = &tunnel.local_addr { + script.push_str(&format!(" local {}", shq(l))); + } + if let Some(r) = &tunnel.remote_addr { + script.push_str(&format!(" remote {}", shq(r))); + } + if let Some(p) = c.dst_port { + script.push_str(&format!(" dstport {}", p)); + } + } + TunnelConfig::Wireguard(c) => { + script.push_str(&format!("ip link add {} type wireguard", shq(name))); + script.push_str(&format!(" && {}", wg_set_script(name, c))); + } + } + script.push_str(&format!(" && ip link set {} up", shq(name))); + self.exec_checked(&format!("sh -c {}", shq(&script))) + .await?; + Ok(Tunnel { + id: Some(name.clone()), + enabled: true, + ..tunnel.clone() + }) + } + + async fn remove_tunnel(&self, id: &str) -> OpResult<()> { + self.exec_checked(&format!("ip link del {}", shq(id))) + .await?; + Ok(()) + } + + async fn update_tunnel(&self, tunnel: &Tunnel) -> OpResult { + // Recreate the interface to apply config changes deterministically. + // `ip link del` is ignored if the interface does not yet exist. + let _ = self + .exec_checked(&format!( + "sh -c {}", + shq(&format!( + "ip link del {} 2>/dev/null; true", + shq(&tunnel.name) + )) + )) + .await?; + self.add_tunnel(tunnel).await + } + + async fn tunnel_traffic(&self) -> OpResult> { + let out = self.exec_checked("ip -s -d -j link show").await?; + let links: Vec = match serde_json::from_str(&out) { + Ok(l) => l, + Err(e) => op_fatal!("Failed to parse ip link output: {}", e), + }; + Ok(links + .iter() + .filter(|l| { + l.linkinfo + .as_ref() + .map(|i| kind_from_info(&i.info_kind).is_some()) + .unwrap_or(false) + }) + .filter_map(|l| { + l.stats64.as_ref().map(|s| TunnelTraffic { + name: l.ifname.clone(), + rx_bytes: s.rx.bytes, + tx_bytes: s.tx.bytes, + }) + }) + .collect()) + } +} + +/// Build a `wg set` command chain for a WireGuard interface configuration. +fn wg_set_script(name: &str, c: &WireguardConfig) -> String { + let mut parts = Vec::new(); + // Private key is written to a 0600 temp file and removed afterwards. + if let Some(pk) = &c.private_key { + let mut s = format!( + "umask 077; f=$(mktemp); printf '%s' {} > \"$f\"; wg set {}", + shq(pk), + shq(name) + ); + if let Some(port) = c.listen_port { + s.push_str(&format!(" listen-port {}", port)); + } + s.push_str(" private-key \"$f\"; rm -f \"$f\""); + parts.push(s); + } else if let Some(port) = c.listen_port { + parts.push(format!("wg set {} listen-port {}", shq(name), port)); + } + for p in &c.peers { + let mut s = format!("wg set {} peer {}", shq(name), shq(&p.public_key)); + if let Some(e) = &p.endpoint { + s.push_str(&format!(" endpoint {}", shq(e))); + } + if !p.allowed_ips.is_empty() { + s.push_str(&format!(" allowed-ips {}", shq(&p.allowed_ips.join(",")))); + } + if let Some(k) = p.persistent_keepalive { + s.push_str(&format!(" persistent-keepalive {}", k)); + } + parts.push(s); + } + if parts.is_empty() { + "true".to_string() + } else { + parts.join(" && ") + } +} + +/// Parse `wg show all dump` into per-interface WireGuard configs. +fn parse_wg_dump(dump: &str) -> std::collections::HashMap { + let mut map: std::collections::HashMap = + std::collections::HashMap::new(); + // Track which interfaces we've seen the header line for + let mut seen_header: std::collections::HashSet = std::collections::HashSet::new(); + for line in dump.lines() { + let f: Vec<&str> = line.split('\t').collect(); + if f.len() < 2 { + continue; + } + let iface = f[0].to_string(); + if !seen_header.contains(&iface) { + // Interface header: iface, private-key, public-key, listen-port, fwmark + seen_header.insert(iface.clone()); + let cfg = map.entry(iface).or_default(); + if f.len() >= 5 { + cfg.public_key = none_if_marker(f[2]).map(|s| s.to_string()); + cfg.listen_port = f[3].parse().ok(); + } + } else { + // Peer line: iface, public-key, psk, endpoint, allowed-ips, handshake, rx, tx, keepalive + let cfg = map.entry(iface).or_default(); + if f.len() >= 5 { + let allowed_ips = none_if_marker(f[4]) + .map(|s| s.split(',').map(|x| x.trim().to_string()).collect()) + .unwrap_or_default(); + cfg.peers.push(WireguardPeer { + public_key: f[1].to_string(), + endpoint: none_if_marker(f[3]).map(|s| s.to_string()), + allowed_ips, + persistent_keepalive: f.get(8).and_then(|v| v.parse().ok()), + }); + } + } + } + map +} + +/// WireGuard dump renders absent values as `(none)` or `off`. +fn none_if_marker(s: &str) -> Option<&str> { + match s { + "(none)" | "off" | "" => None, + other => Some(other), + } +} + +/// One entry from `ip -s -d -j link show` +#[derive(Debug, Clone, Deserialize)] +struct IpLink { + ifname: String, + #[serde(default)] + flags: Vec, + linkinfo: Option, + stats64: Option, +} + +#[derive(Debug, Clone, Deserialize)] +struct IpLinkInfo { + info_kind: String, + info_data: Option, +} + +#[derive(Debug, Clone, Default, Deserialize)] +struct IpLinkInfoData { + local: Option, + remote: Option, + /// VXLAN VNI + id: Option, + /// VXLAN UDP port + port: Option, + /// GRE input key (number or dotted-quad string) + ikey: Option, +} + +#[derive(Debug, Clone, Deserialize)] +struct IpStats64 { + rx: IpStatsDir, + tx: IpStatsDir, +} + +#[derive(Debug, Clone, Deserialize)] +struct IpStatsDir { + bytes: u64, +} + +/// One entry from `ip -j neigh show` +#[derive(Debug, Clone, Deserialize)] +struct IpNeighEntry { + dst: String, + dev: String, + lladdr: Option, +} + +impl IpNeighEntry { + /// Convert into an [`ArpEntry`], dropping entries without a resolved MAC + /// (e.g. `FAILED`/`INCOMPLETE` neighbours). + fn into_arp_entry(self) -> Option { + let mac = self.lladdr?; + Some(ArpEntry { + id: Some(self.dst.clone()), + address: self.dst, + mac_address: mac, + interface: Some(self.dev), + comment: None, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_new_parses_url() -> Result<()> { + let r = LinuxSshRouter::new("ssh://admin@10.0.0.1:2222/vmbr0", "KEY")?; + assert_eq!(r.host, "10.0.0.1:2222"); + assert_eq!(r.username, "admin"); + assert_eq!(r.interface, "vmbr0"); + assert_eq!(r.key, "KEY"); + Ok(()) + } + + #[test] + fn test_new_defaults() -> Result<()> { + let r = LinuxSshRouter::new("ssh://10.0.0.1/eth0", "K")?; + assert_eq!(r.host, "10.0.0.1:22"); + assert_eq!(r.username, "root"); + assert_eq!(r.interface, "eth0"); + Ok(()) + } + + #[test] + fn test_new_rejects_bad_input() { + assert!(LinuxSshRouter::new("http://10.0.0.1/eth0", "K").is_err()); + assert!(LinuxSshRouter::new("ssh://10.0.0.1", "K").is_err()); + assert!(LinuxSshRouter::new("ssh://10.0.0.1/", "K").is_err()); + } + + #[test] + fn test_parse_gre_key() { + assert_eq!(parse_gre_key("10"), Some(10)); + assert_eq!(parse_gre_key("0.0.0.10"), Some(10)); + assert_eq!(parse_gre_key("0.0.1.0"), Some(256)); + assert_eq!(parse_gre_key("garbage"), None); + assert_eq!(value_to_gre_key(&serde_json::json!(42)), Some(42)); + assert_eq!(value_to_gre_key(&serde_json::json!("0.0.0.5")), Some(5)); + } + + #[test] + fn test_shq_escaping() { + assert_eq!(shq("eth0"), "'eth0'"); + assert_eq!(shq("a'b"), "'a'\\''b'"); + } + + #[test] + fn test_link_to_tunnel_gre_vxlan() { + let json = r#"[ + {"ifname":"gre1","flags":["POINTOPOINT","UP"],"linkinfo":{"info_kind":"gre","info_data":{"local":"10.0.0.1","remote":"10.0.0.2","ikey":"0.0.0.7"}},"stats64":{"rx":{"bytes":100},"tx":{"bytes":200}}}, + {"ifname":"vx1","flags":["UP"],"linkinfo":{"info_kind":"vxlan","info_data":{"id":42,"local":"10.0.0.1","remote":"10.0.0.3","port":4789}}}, + {"ifname":"eth0","flags":["UP"]} + ]"#; + let links: Vec = serde_json::from_str(json).unwrap(); + let wg = std::collections::HashMap::new(); + let tuns: Vec = links + .iter() + .filter_map(|l| LinuxSshRouter::link_to_tunnel(l, &wg)) + .collect(); + assert_eq!(tuns.len(), 2); + let gre = &tuns[0]; + assert_eq!(gre.name, "gre1"); + assert!(gre.enabled); + assert_eq!(gre.local_addr.as_deref(), Some("10.0.0.1")); + match &gre.config { + TunnelConfig::Gre(c) => assert_eq!(c.key, Some(7)), + _ => panic!("expected gre"), + } + match &tuns[1].config { + TunnelConfig::Vxlan(c) => { + assert_eq!(c.vni, 42); + assert_eq!(c.dst_port, Some(4789)); + } + _ => panic!("expected vxlan"), + } + } + + #[test] + fn test_parse_wg_dump() { + let dump = "wg0\tPRIVKEY\tPUBKEY\t51820\toff\n\ + wg0\tPEERPUB\t(none)\t1.2.3.4:51820\t10.0.0.0/24,10.0.1.0/24\t1700000000\t1024\t2048\t25\n" + .replace(" wg0", "wg0"); + let map = parse_wg_dump(&dump); + let cfg = map.get("wg0").unwrap(); + assert_eq!(cfg.public_key.as_deref(), Some("PUBKEY")); + assert_eq!(cfg.listen_port, Some(51820)); + assert_eq!(cfg.peers.len(), 1); + let p = &cfg.peers[0]; + assert_eq!(p.public_key, "PEERPUB"); + assert_eq!(p.endpoint.as_deref(), Some("1.2.3.4:51820")); + assert_eq!(p.allowed_ips, vec!["10.0.0.0/24", "10.0.1.0/24"]); + assert_eq!(p.persistent_keepalive, Some(25)); + } + + #[test] + fn test_wg_set_script() { + let c = WireguardConfig { + listen_port: Some(51820), + private_key: Some("KEY".to_string()), + public_key: None, + peers: vec![WireguardPeer { + public_key: "PUB".to_string(), + endpoint: Some("1.2.3.4:51820".to_string()), + allowed_ips: vec!["10.0.0.0/24".to_string()], + persistent_keepalive: Some(25), + }], + }; + let s = wg_set_script("wg0", &c); + assert!(s.contains("listen-port 51820")); + assert!(s.contains("private-key")); + assert!(s.contains("peer 'PUB'")); + assert!(s.contains("allowed-ips '10.0.0.0/24'")); + assert!(s.contains("persistent-keepalive 25")); + } + + #[test] + fn test_traffic_filters_non_tunnels() { + let json = r#"[ + {"ifname":"gre1","flags":["UP"],"linkinfo":{"info_kind":"gre"},"stats64":{"rx":{"bytes":5},"tx":{"bytes":6}}}, + {"ifname":"eth0","flags":["UP"],"stats64":{"rx":{"bytes":1},"tx":{"bytes":2}}} + ]"#; + let links: Vec = serde_json::from_str(json).unwrap(); + let traffic: Vec = links + .iter() + .filter(|l| { + l.linkinfo + .as_ref() + .map(|i| kind_from_info(&i.info_kind).is_some()) + .unwrap_or(false) + }) + .filter_map(|l| { + l.stats64.as_ref().map(|s| TunnelTraffic { + name: l.ifname.clone(), + rx_bytes: s.rx.bytes, + tx_bytes: s.tx.bytes, + }) + }) + .collect(); + assert_eq!(traffic.len(), 1); + assert_eq!(traffic[0].name, "gre1"); + assert_eq!(traffic[0].rx_bytes, 5); + } + + #[test] + fn test_parse_bird_protocols() { + let out = [ + "BIRD 2.0.7 ready.", + "Name Proto Table State Since Info", + "device1 Device --- up 2024-06-01 ", + "bgp1 BGP --- up 2024-06-01 Established", + " BGP state: Established", + " Neighbor address: 192.0.2.1", + " Neighbor AS: 64512", + " Local AS: 64500", + " Role: provider", + " Channel ipv4", + " Routes: 5 imported, 2 exported, 3 preferred", + "bgp2 BGP --- down 2024-06-01 disabled", + " BGP state: Idle", + " Neighbor address: 192.0.2.5", + " Neighbor AS: 64600", + ] + .join("\n"); + let sessions = parse_bird_protocols(&out); + assert_eq!(sessions.len(), 2); + let s = &sessions[0]; + assert_eq!(s.name, "bgp1"); + assert_eq!(s.peer_ip.as_deref(), Some("192.0.2.1")); + assert_eq!(s.peer_asn, Some(64512)); + assert_eq!(s.local_asn, Some(64500)); + assert_eq!(s.state, "Established"); + assert_eq!(s.prefixes_received, Some(5)); + assert_eq!(s.prefixes_sent, Some(2)); + assert_eq!(s.direction, BgpPeerDirection::Upstream); + assert!(s.enabled); + + let s2 = &sessions[1]; + assert_eq!(s2.name, "bgp2"); + assert_eq!(s2.state, "Idle"); + assert!(!s2.enabled); + assert_eq!(s2.direction, BgpPeerDirection::Unknown); + } + + #[test] + fn test_parse_bird_routes() { + let out = [ + "BIRD 2.0.7 ready.", + "Table master4:", + "198.51.100.0/24 unicast [static1 2024-06-01] * (200)", + "\tvia 192.0.2.1 on eth0", + "203.0.113.0/24 unicast [static1 2024-06-01] * (200)", + "\tdev eth0", + ] + .join("\n"); + let routes = parse_bird_routes(&out); + assert_eq!(routes.len(), 2); + assert_eq!(routes[0].prefix, "198.51.100.0/24"); + assert_eq!(routes[0].next_hop.as_deref(), Some("192.0.2.1")); + assert_eq!(routes[1].prefix, "203.0.113.0/24"); + assert_eq!(routes[1].next_hop, None); + } + + #[test] + fn test_role_to_direction() { + assert_eq!(role_to_direction("provider"), BgpPeerDirection::Upstream); + assert_eq!(role_to_direction("customer"), BgpPeerDirection::Downstream); + assert_eq!(role_to_direction("peer"), BgpPeerDirection::Peer); + assert_eq!(role_to_direction("rs-server"), BgpPeerDirection::Upstream); + assert_eq!(role_to_direction("weird"), BgpPeerDirection::Unknown); + } + + #[test] + fn test_neigh_parse_filters_no_mac() { + let json = r#"[ + {"dst":"10.0.0.5","dev":"vmbr0","lladdr":"aa:bb:cc:dd:ee:ff","state":["REACHABLE"]}, + {"dst":"10.0.0.6","dev":"vmbr0","state":["FAILED"]} + ]"#; + let entries: Vec = serde_json::from_str(json).unwrap(); + let arp: Vec = entries + .into_iter() + .filter_map(|e| e.into_arp_entry()) + .collect(); + assert_eq!(arp.len(), 1); + assert_eq!(arp[0].address, "10.0.0.5"); + assert_eq!(arp[0].mac_address, "aa:bb:cc:dd:ee:ff"); + assert_eq!(arp[0].id.as_deref(), Some("10.0.0.5")); + } +} diff --git a/lnvps_api/src/router/mikrotik.rs b/lnvps_api/src/router/mikrotik.rs index fffe6e51..957120af 100644 --- a/lnvps_api/src/router/mikrotik.rs +++ b/lnvps_api/src/router/mikrotik.rs @@ -1,5 +1,9 @@ use crate::json_api::JsonApi; -use crate::router::{ArpEntry, Router}; +use crate::router::{ + ArpEntry, BgpPeer, BgpPeerDirection, BgpRoute, BgpRouter, BgpSession, GreConfig, Router, + Tunnel, TunnelConfig, TunnelKind, TunnelRouter, TunnelTraffic, VxlanConfig, WireguardConfig, + WireguardPeer, +}; use anyhow::{Context, Result}; use async_trait::async_trait; use base64::Engine; @@ -8,6 +12,7 @@ use lnvps_api_common::op_fatal; use lnvps_api_common::retry::{OpError, OpResult}; use reqwest::Method; use serde::{Deserialize, Serialize}; +use serde_json::json; pub struct MikrotikRouter { api: JsonApi, @@ -69,6 +74,487 @@ impl Router for MikrotikRouter { .await?; rsp.try_into().map_err(OpError::Fatal) } + + fn tunnel(&self) -> Option<&dyn TunnelRouter> { + Some(self) + } + + fn bgp(&self) -> Option<&dyn BgpRouter> { + Some(self) + } +} + +#[async_trait] +impl BgpRouter for MikrotikRouter { + async fn list_sessions(&self) -> OpResult> { + let connections: Vec = self + .api + .req::<_, ()>(Method::GET, "/rest/routing/bgp/connection", None) + .await?; + let sessions: Vec = self + .api + .req::<_, ()>(Method::GET, "/rest/routing/bgp/session", None) + .await?; + Ok(connections + .into_iter() + .map(|c| { + let live = sessions.iter().find(|s| s.name.as_deref() == Some(&c.name)); + let established = live + .map(|s| matches!(s.established.as_deref(), Some("true") | Some("yes"))) + .unwrap_or(false); + BgpSession { + id: c.id.clone().unwrap_or_default(), + name: c.name.clone(), + peer_ip: live + .and_then(|s| s.remote_address.clone()) + .or(c.remote_address), + peer_asn: live + .and_then(|s| s.remote_as.as_deref()) + .or(c.remote_as.as_deref()) + .and_then(|s| s.parse().ok()), + local_asn: c.local_as.and_then(|s| s.parse().ok()), + state: if established { + "Established".to_string() + } else { + "Idle".to_string() + }, + prefixes_received: live + .and_then(|s| s.prefix_count.as_deref()) + .and_then(|s| s.parse().ok()), + prefixes_sent: None, + enabled: mt_enabled(&c.disabled), + direction: BgpPeerDirection::Unknown, + } + }) + .collect()) + } + + async fn originated_routes(&self, candidates: &[String]) -> OpResult> { + // IMPORTANT: never `GET /rest/ip/route` unfiltered — on a full-table router + // that returns hundreds of MB of JSON. Query each candidate prefix with a + // server-side `dst-address` filter and a trimmed proplist instead. + let mut out = Vec::new(); + for prefix in candidates { + let path = format!( + "/rest/ip/route?dst-address={}&.proplist=dst-address,gateway,bgp", + prefix + ); + let routes: Vec = self + .api + .req::<_, ()>(Method::GET, &path, None) + .await + .unwrap_or_default(); + for r in routes { + if let Some(dst) = r.dst_address { + out.push(BgpRoute { + prefix: dst, + next_hop: r.gateway, + }); + } + } + } + Ok(out) + } + + async fn default_route(&self) -> OpResult> { + // Filter server-side; do not fetch the whole table. + for dst in ["0.0.0.0/0", "::/0"] { + let path = format!( + "/rest/ip/route?dst-address={}&.proplist=dst-address,gateway", + dst + ); + let routes: Vec = self.api.req::<_, ()>(Method::GET, &path, None).await?; + if let Some(r) = routes.into_iter().next() { + return Ok(Some(BgpRoute { + prefix: r.dst_address.unwrap_or_else(|| dst.to_string()), + next_hop: r.gateway, + })); + } + } + Ok(None) + } + + async fn discover_peers(&self) -> OpResult> { + let sessions = self.list_sessions().await?; + Ok(sessions + .into_iter() + .map(|s| BgpPeer { + peer_ip: s.peer_ip, + asn: s.peer_asn, + direction: s.direction, + }) + .collect()) + } + + async fn set_session_enabled(&self, id: &str, enabled: bool) -> OpResult<()> { + let body = json!({ "disabled": (!enabled).to_string() }); + let _: serde_json::Value = self + .api + .req( + Method::PATCH, + &format!("/rest/routing/bgp/connection/{}", id), + Some(body), + ) + .await?; + Ok(()) + } +} + +#[derive(Debug, Deserialize)] +struct MtBgpConnection { + #[serde(rename = ".id")] + id: Option, + name: String, + #[serde(rename = "remote.address")] + remote_address: Option, + #[serde(rename = "remote.as")] + remote_as: Option, + #[serde(rename = "local.as")] + local_as: Option, + disabled: Option, +} + +#[derive(Debug, Deserialize)] +struct MtBgpSession { + name: Option, + #[serde(rename = "remote.address")] + remote_address: Option, + #[serde(rename = "remote.as")] + remote_as: Option, + established: Option, + #[serde(rename = "prefix-count")] + prefix_count: Option, +} + +#[derive(Debug, Deserialize)] +struct MtRoute { + #[serde(rename = "dst-address")] + dst_address: Option, + gateway: Option, +} + +/// RouterOS REST menu path for a tunnel kind +fn tunnel_endpoint(kind: TunnelKind) -> &'static str { + match kind { + TunnelKind::Gre => "gre", + TunnelKind::Vxlan => "vxlan", + TunnelKind::Wireguard => "wireguard", + } +} + +/// RouterOS renders booleans as the strings "true"/"false"; an interface is +/// enabled when it is not disabled. +fn mt_enabled(disabled: &Option) -> bool { + !matches!(disabled.as_deref(), Some("true") | Some("yes")) +} + +#[async_trait] +impl TunnelRouter for MikrotikRouter { + async fn list_tunnels(&self) -> OpResult> { + let mut out = Vec::new(); + + let gres: Vec = self + .api + .req::<_, ()>(Method::GET, "/rest/interface/gre", None) + .await?; + for g in gres { + out.push(Tunnel { + id: Some(format!("gre:{}", g.id.clone().unwrap_or_default())), + name: g.name, + local_addr: g.local_address, + remote_addr: g.remote_address, + enabled: mt_enabled(&g.disabled), + config: TunnelConfig::Gre(GreConfig::default()), + }); + } + + let vxlans: Vec = self + .api + .req::<_, ()>(Method::GET, "/rest/interface/vxlan", None) + .await?; + for v in vxlans { + out.push(Tunnel { + id: Some(format!("vxlan:{}", v.id.clone().unwrap_or_default())), + name: v.name, + local_addr: v.local_address, + remote_addr: None, + enabled: mt_enabled(&v.disabled), + config: TunnelConfig::Vxlan(VxlanConfig { + vni: v.vni.and_then(|s| s.parse().ok()).unwrap_or(0), + dst_port: v.port.and_then(|s| s.parse().ok()), + }), + }); + } + + let wgs: Vec = self + .api + .req::<_, ()>(Method::GET, "/rest/interface/wireguard", None) + .await?; + if !wgs.is_empty() { + let peers: Vec = self + .api + .req::<_, ()>(Method::GET, "/rest/interface/wireguard/peers", None) + .await?; + for w in wgs { + let cfg = WireguardConfig { + listen_port: w.listen_port.and_then(|s| s.parse().ok()), + private_key: None, // never return secrets when listing + public_key: w.public_key, + peers: peers + .iter() + .filter(|p| p.interface.as_deref() == Some(&w.name)) + .map(|p| WireguardPeer { + public_key: p.public_key.clone().unwrap_or_default(), + endpoint: p.endpoint(), + allowed_ips: p + .allowed_address + .as_deref() + .map(|s| s.split(',').map(|x| x.trim().to_string()).collect()) + .unwrap_or_default(), + persistent_keepalive: p.keepalive_secs(), + }) + .collect(), + }; + out.push(Tunnel { + id: Some(format!("wireguard:{}", w.id.clone().unwrap_or_default())), + name: w.name, + local_addr: None, + remote_addr: None, + enabled: mt_enabled(&w.disabled), + config: TunnelConfig::Wireguard(cfg), + }); + } + } + + Ok(out) + } + + async fn add_tunnel(&self, tunnel: &Tunnel) -> OpResult { + let endpoint = tunnel_endpoint(tunnel.kind()); + let path = format!("/rest/interface/{}", endpoint); + let created: MtCreated = match &tunnel.config { + TunnelConfig::Gre(_) => { + let body = json!({ + "name": tunnel.name, + "local-address": tunnel.local_addr, + "remote-address": tunnel.remote_addr, + }); + self.api.req(Method::PUT, &path, Some(body)).await? + } + TunnelConfig::Vxlan(c) => { + let body = json!({ + "name": tunnel.name, + "vni": c.vni.to_string(), + "port": c.dst_port.map(|p| p.to_string()), + "local-address": tunnel.local_addr, + }); + self.api.req(Method::PUT, &path, Some(body)).await? + } + TunnelConfig::Wireguard(c) => { + let body = json!({ + "name": tunnel.name, + "listen-port": c.listen_port.map(|p| p.to_string()), + "private-key": c.private_key, + }); + let created: MtCreated = self.api.req(Method::PUT, &path, Some(body)).await?; + for p in &c.peers { + let (ep_addr, ep_port) = split_endpoint(p.endpoint.as_deref()); + let pbody = json!({ + "interface": tunnel.name, + "public-key": p.public_key, + "endpoint-address": ep_addr, + "endpoint-port": ep_port, + "allowed-address": if p.allowed_ips.is_empty() { None } else { Some(p.allowed_ips.join(",")) }, + "persistent-keepalive": p.persistent_keepalive.map(|k| k.to_string()), + }); + let _: MtCreated = self + .api + .req(Method::PUT, "/rest/interface/wireguard/peers", Some(pbody)) + .await?; + } + created + } + }; + Ok(Tunnel { + id: Some(format!("{}:{}", endpoint, created.id.unwrap_or_default())), + enabled: true, + ..tunnel.clone() + }) + } + + async fn remove_tunnel(&self, id: &str) -> OpResult<()> { + let (endpoint, ros_id) = split_tunnel_id(id)?; + let _: serde_json::Value = self + .api + .req::<_, ()>( + Method::DELETE, + &format!("/rest/interface/{}/{}", endpoint, ros_id), + None, + ) + .await + .or_else(|e| match e { + // DELETE returns an empty body which fails JSON parsing; treat as success + OpError::Fatal(_) => Ok(serde_json::Value::Null), + other => Err(other), + })?; + Ok(()) + } + + async fn update_tunnel(&self, tunnel: &Tunnel) -> OpResult { + let id = tunnel + .id + .as_deref() + .ok_or_else(|| OpError::Fatal(anyhow::anyhow!("update_tunnel requires an id")))?; + let (endpoint, ros_id) = split_tunnel_id(id)?; + let path = format!("/rest/interface/{}/{}", endpoint, ros_id); + let body = match &tunnel.config { + TunnelConfig::Gre(_) => json!({ + "local-address": tunnel.local_addr, + "remote-address": tunnel.remote_addr, + "disabled": (!tunnel.enabled).to_string(), + }), + TunnelConfig::Vxlan(c) => json!({ + "vni": c.vni.to_string(), + "port": c.dst_port.map(|p| p.to_string()), + "local-address": tunnel.local_addr, + "disabled": (!tunnel.enabled).to_string(), + }), + TunnelConfig::Wireguard(c) => json!({ + "listen-port": c.listen_port.map(|p| p.to_string()), + "disabled": (!tunnel.enabled).to_string(), + }), + }; + let _: serde_json::Value = self.api.req(Method::PATCH, &path, Some(body)).await?; + Ok(tunnel.clone()) + } + + async fn tunnel_traffic(&self) -> OpResult> { + let ifaces: Vec = self + .api + .req::<_, ()>(Method::GET, "/rest/interface", None) + .await?; + Ok(ifaces + .into_iter() + .filter(|i| matches!(i.kind.as_deref(), Some("gre") | Some("vxlan") | Some("wg"))) + .map(|i| TunnelTraffic { + name: i.name, + rx_bytes: i.rx_byte.and_then(|s| s.parse().ok()).unwrap_or(0), + tx_bytes: i.tx_byte.and_then(|s| s.parse().ok()).unwrap_or(0), + }) + .collect()) + } +} + +/// Split a tunnel id of the form `":"`. +fn split_tunnel_id(id: &str) -> OpResult<(&str, &str)> { + match id.split_once(':') { + Some((endpoint @ ("gre" | "vxlan" | "wireguard"), ros_id)) => Ok((endpoint, ros_id)), + _ => Err(OpError::Fatal(anyhow::anyhow!( + "Invalid mikrotik tunnel id: {}", + id + ))), + } +} + +/// Split a `host:port` endpoint into separate address/port components. +fn split_endpoint(ep: Option<&str>) -> (Option, Option) { + match ep.and_then(|e| e.rsplit_once(':')) { + Some((addr, port)) => (Some(addr.to_string()), Some(port.to_string())), + None => (ep.map(|s| s.to_string()), None), + } +} + +#[derive(Debug, Deserialize)] +struct MtCreated { + #[serde(rename = ".id")] + id: Option, +} + +#[derive(Debug, Deserialize)] +struct MtGre { + #[serde(rename = ".id")] + id: Option, + name: String, + #[serde(rename = "local-address")] + local_address: Option, + #[serde(rename = "remote-address")] + remote_address: Option, + disabled: Option, +} + +#[derive(Debug, Deserialize)] +struct MtVxlan { + #[serde(rename = ".id")] + id: Option, + name: String, + vni: Option, + port: Option, + #[serde(rename = "local-address")] + local_address: Option, + disabled: Option, +} + +#[derive(Debug, Deserialize)] +struct MtWireguard { + #[serde(rename = ".id")] + id: Option, + name: String, + #[serde(rename = "listen-port")] + listen_port: Option, + #[serde(rename = "public-key")] + public_key: Option, + disabled: Option, +} + +#[derive(Debug, Deserialize)] +struct MtWgPeer { + interface: Option, + #[serde(rename = "public-key")] + public_key: Option, + #[serde(rename = "endpoint-address")] + endpoint_address: Option, + #[serde(rename = "endpoint-port")] + endpoint_port: Option, + #[serde(rename = "allowed-address")] + allowed_address: Option, + #[serde(rename = "persistent-keepalive")] + persistent_keepalive: Option, +} + +impl MtWgPeer { + /// Combine RouterOS endpoint-address/endpoint-port into a `host:port` string. + fn endpoint(&self) -> Option { + match (&self.endpoint_address, &self.endpoint_port) { + (Some(a), Some(p)) if !a.is_empty() => Some(format!("{}:{}", a, p)), + (Some(a), None) if !a.is_empty() => Some(a.clone()), + _ => None, + } + } + + /// RouterOS renders persistent-keepalive as a duration string (e.g. "25s" or + /// "00:00:25"); extract the seconds component best-effort. + fn keepalive_secs(&self) -> Option { + let v = self.persistent_keepalive.as_deref()?; + if let Some(stripped) = v.strip_suffix('s') { + return stripped.parse().ok(); + } + // hh:mm:ss form + if let Some(sec) = v.rsplit(':').next() { + return sec.parse().ok(); + } + v.parse().ok() + } +} + +#[derive(Debug, Deserialize)] +struct MtInterface { + name: String, + #[serde(rename = "type")] + kind: Option, + #[serde(rename = "rx-byte")] + rx_byte: Option, + #[serde(rename = "tx-byte")] + tx_byte: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -110,3 +596,70 @@ impl From for MikrotikArpEntry { } } } + +#[cfg(test)] +mod tunnel_tests { + use super::*; + + #[test] + fn test_mt_enabled() { + assert!(mt_enabled(&None)); + assert!(mt_enabled(&Some("false".to_string()))); + assert!(!mt_enabled(&Some("true".to_string()))); + assert!(!mt_enabled(&Some("yes".to_string()))); + } + + #[test] + fn test_tunnel_endpoint() { + assert_eq!(tunnel_endpoint(TunnelKind::Gre), "gre"); + assert_eq!(tunnel_endpoint(TunnelKind::Vxlan), "vxlan"); + assert_eq!(tunnel_endpoint(TunnelKind::Wireguard), "wireguard"); + } + + #[test] + fn test_split_tunnel_id() { + assert_eq!(split_tunnel_id("gre:*1").unwrap(), ("gre", "*1")); + assert_eq!( + split_tunnel_id("wireguard:*A").unwrap(), + ("wireguard", "*A") + ); + assert!(split_tunnel_id("bogus:*1").is_err()); + assert!(split_tunnel_id("noseparator").is_err()); + } + + #[test] + fn test_split_endpoint() { + assert_eq!( + split_endpoint(Some("1.2.3.4:51820")), + (Some("1.2.3.4".to_string()), Some("51820".to_string())) + ); + assert_eq!( + split_endpoint(Some("host")), + (Some("host".to_string()), None) + ); + assert_eq!(split_endpoint(None), (None, None)); + } + + #[test] + fn test_wg_peer_endpoint_and_keepalive() { + let p = MtWgPeer { + interface: Some("wg0".to_string()), + public_key: Some("PUB".to_string()), + endpoint_address: Some("1.2.3.4".to_string()), + endpoint_port: Some("51820".to_string()), + allowed_address: Some("10.0.0.0/24,10.0.1.0/24".to_string()), + persistent_keepalive: Some("25s".to_string()), + }; + assert_eq!(p.endpoint().as_deref(), Some("1.2.3.4:51820")); + assert_eq!(p.keepalive_secs(), Some(25)); + + let p2 = MtWgPeer { + persistent_keepalive: Some("00:00:30".to_string()), + endpoint_address: None, + endpoint_port: None, + ..p + }; + assert_eq!(p2.keepalive_secs(), Some(30)); + assert_eq!(p2.endpoint(), None); + } +} diff --git a/lnvps_api/src/router/mod.rs b/lnvps_api/src/router/mod.rs index b67fdd7c..163523b6 100644 --- a/lnvps_api/src/router/mod.rs +++ b/lnvps_api/src/router/mod.rs @@ -19,6 +19,272 @@ pub trait Router: Send + Sync { async fn add_arp_entry(&self, entry: &ArpEntry) -> OpResult; async fn remove_arp_entry(&self, id: &str) -> OpResult<()>; async fn update_arp_entry(&self, entry: &ArpEntry) -> OpResult; + + /// Tunnel-management capability, if this router supports it. + /// + /// Returns `None` for routers that cannot manage GRE/VXLAN/WireGuard tunnels. + fn tunnel(&self) -> Option<&dyn TunnelRouter> { + None + } + + /// BGP capability, if this router supports it. + /// + /// Returns `None` for routers that do not run BGP. + fn bgp(&self) -> Option<&dyn BgpRouter> { + None + } +} + +/// Optional capability for routers that run BGP (route servers / edge routers). +/// +/// Note: BGP itself exposes no per-session byte counters — traffic accounting is +/// done at the tunnel/interface level (see [`TunnelRouter`]). +#[async_trait] +pub trait BgpRouter: Send + Sync { + /// Detect configured BGP sessions and their state (issue task 2) + async fn list_sessions(&self) -> OpResult>; + /// Detect which of the `candidates` prefixes (e.g. VM IP ranges) the router + /// actually originates/announces (issue task 3). + /// + /// Scoped to a candidate set rather than enumerating the table so it stays + /// bounded on routers carrying a full DFZ table (~1M+ routes). Passing an + /// empty slice returns all locally-originated prefixes (which is inherently + /// small — a router only originates its own ranges). + async fn originated_routes(&self, candidates: &[String]) -> OpResult>; + /// Detect a default route, if present (issue task 4 — route-server detection) + async fn default_route(&self) -> OpResult>; + /// Discover BGP peers and classify upstream/downstream (issue task 5) + async fn discover_peers(&self) -> OpResult>; + /// Enable or disable a BGP session by its backend id (issue task 6) + async fn set_session_enabled(&self, id: &str, enabled: bool) -> OpResult<()>; +} + +/// Relationship of a BGP peer relative to this router +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum BgpPeerDirection { + /// The peer is a transit provider (we are its customer) + Upstream, + /// The peer is our customer (we provide transit) + Downstream, + /// Settlement-free / lateral peer + Peer, + /// Relationship could not be determined + #[default] + Unknown, +} + +/// A detected BGP session +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BgpSession { + /// Backend identifier used for toggling (protocol name on BIRD, `.id` on Mikrotik) + pub id: String, + /// Human-readable session/protocol name + pub name: String, + /// Neighbour address + pub peer_ip: Option, + /// Neighbour AS number + pub peer_asn: Option, + /// Local AS number + pub local_asn: Option, + /// Session state (e.g. `Established`, `Active`, `Idle`) + pub state: String, + /// Number of prefixes received from the peer + pub prefixes_received: Option, + /// Number of prefixes sent to the peer + pub prefixes_sent: Option, + /// Whether the session is administratively enabled + pub enabled: bool, + /// Inferred peer relationship + pub direction: BgpPeerDirection, +} + +/// A discovered BGP peer +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BgpPeer { + /// Peer address + pub peer_ip: Option, + /// Peer AS number + pub asn: Option, + /// Inferred relationship + pub direction: BgpPeerDirection, +} + +/// A route in the routing table +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BgpRoute { + /// Destination prefix (CIDR) + pub prefix: String, + /// Next hop / gateway, if any + pub next_hop: Option, +} + +/// Optional capability for routers that can manage point-to-point/overlay tunnels +/// (GRE, VXLAN, WireGuard) and report per-tunnel traffic counters. +/// +/// Per-tunnel byte counters are the canonical source of "per session" traffic for +/// route servers — BGP itself exposes no byte counters. +#[async_trait] +pub trait TunnelRouter: Send + Sync { + /// List all tunnels currently configured on the router + async fn list_tunnels(&self) -> OpResult>; + /// Create a new tunnel + async fn add_tunnel(&self, tunnel: &Tunnel) -> OpResult; + /// Remove a tunnel by its backend id (interface name on Linux) + async fn remove_tunnel(&self, id: &str) -> OpResult<()>; + /// Update an existing tunnel + async fn update_tunnel(&self, tunnel: &Tunnel) -> OpResult; + /// Report per-tunnel rx/tx byte counters + async fn tunnel_traffic(&self) -> OpResult>; +} + +/// The kind of a tunnel interface +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TunnelKind { + Gre, + Vxlan, + Wireguard, +} + +/// A tunnel interface (GRE, VXLAN or WireGuard) +#[derive(Debug, Clone, PartialEq)] +pub struct Tunnel { + /// Backend identifier (interface name on Linux, `.id` on Mikrotik) + pub id: Option, + /// Interface name + pub name: String, + /// Local tunnel endpoint address + pub local_addr: Option, + /// Remote tunnel endpoint address + pub remote_addr: Option, + /// Whether the interface is administratively up + pub enabled: bool, + /// Kind-specific configuration + pub config: TunnelConfig, +} + +impl Tunnel { + /// The kind of this tunnel, derived from its config + pub fn kind(&self) -> TunnelKind { + match self.config { + TunnelConfig::Gre(_) => TunnelKind::Gre, + TunnelConfig::Vxlan(_) => TunnelKind::Vxlan, + TunnelConfig::Wireguard(_) => TunnelKind::Wireguard, + } + } +} + +/// Kind-specific tunnel configuration +#[derive(Debug, Clone, PartialEq)] +pub enum TunnelConfig { + Gre(GreConfig), + Vxlan(VxlanConfig), + Wireguard(WireguardConfig), +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct GreConfig { + /// GRE key (shared between local/remote tunnel ends) + pub key: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct VxlanConfig { + /// VXLAN network identifier (VNI) + pub vni: u32, + /// UDP destination port (default 4789) + pub dst_port: Option, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct WireguardConfig { + /// UDP listen port + pub listen_port: Option, + /// Interface private key (PEM/base64); never returned when listing + pub private_key: Option, + /// Interface public key + pub public_key: Option, + /// Configured peers + pub peers: Vec, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct WireguardPeer { + /// Peer public key + pub public_key: String, + /// Peer endpoint (host:port) + pub endpoint: Option, + /// Allowed IP ranges (CIDR) + pub allowed_ips: Vec, + /// Persistent keepalive interval in seconds + pub persistent_keepalive: Option, +} + +/// Per-tunnel traffic counters +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TunnelTraffic { + /// Tunnel interface name + pub name: String, + /// Bytes received + pub rx_bytes: u64, + /// Bytes transmitted + pub tx_bytes: u64, +} + +impl From for lnvps_db::RouterTunnelKind { + fn from(k: TunnelKind) -> Self { + match k { + TunnelKind::Gre => lnvps_db::RouterTunnelKind::Gre, + TunnelKind::Vxlan => lnvps_db::RouterTunnelKind::Vxlan, + TunnelKind::Wireguard => lnvps_db::RouterTunnelKind::Wireguard, + } + } +} + +impl Tunnel { + /// Build a cacheable DB row for this tunnel under the given router. + pub fn to_db(&self, router_id: u64) -> lnvps_db::RouterTunnel { + lnvps_db::RouterTunnel { + id: 0, + router_id, + name: self.name.clone(), + kind: self.kind().into(), + local_addr: self.local_addr.clone(), + remote_addr: self.remote_addr.clone(), + enabled: self.enabled, + last_seen: chrono::Utc::now(), + } + } +} + +impl From for lnvps_db::RouterBgpDirection { + fn from(d: BgpPeerDirection) -> Self { + match d { + BgpPeerDirection::Upstream => lnvps_db::RouterBgpDirection::Upstream, + BgpPeerDirection::Downstream => lnvps_db::RouterBgpDirection::Downstream, + BgpPeerDirection::Peer => lnvps_db::RouterBgpDirection::Peer, + BgpPeerDirection::Unknown => lnvps_db::RouterBgpDirection::Unknown, + } + } +} + +impl BgpSession { + /// Build a cacheable DB row for this session under the given router. + pub fn to_db(&self, router_id: u64) -> lnvps_db::RouterBgpSession { + lnvps_db::RouterBgpSession { + id: 0, + router_id, + name: self.name.clone(), + peer_ip: self.peer_ip.clone(), + peer_asn: self.peer_asn, + local_asn: self.local_asn, + state: self.state.clone(), + prefixes_received: self.prefixes_received, + prefixes_sent: self.prefixes_sent, + enabled: self.enabled, + direction: self.direction.into(), + last_seen: chrono::Utc::now(), + } + } } #[derive(Debug, Clone)] @@ -46,10 +312,14 @@ impl ArpEntry { } } +#[cfg(feature = "linux-ssh")] +mod linux_ssh; #[cfg(feature = "mikrotik")] mod mikrotik; mod ovh; +#[cfg(feature = "linux-ssh")] +pub use linux_ssh::LinuxSshRouter; #[cfg(feature = "mikrotik")] pub use mikrotik::MikrotikRouter; pub use ovh::OvhDedicatedServerVMacRouter; @@ -68,6 +338,12 @@ pub async fn get_router(db: &Arc, router_id: u64) -> OpResult Ok(Arc::new( OvhDedicatedServerVMacRouter::new(&cfg.url, &cfg.name, cfg.token.as_str()).await?, )), + #[cfg(feature = "linux-ssh")] + RouterKind::LinuxSsh => Ok(Arc::new(LinuxSshRouter::new(&cfg.url, cfg.token.as_str())?)), + #[cfg(not(feature = "linux-ssh"))] + RouterKind::LinuxSsh => Err(lnvps_api_common::retry::OpError::Fatal(anyhow::anyhow!( + "LinuxSsh router support is not enabled in this build" + ))), RouterKind::MockRouter => { #[cfg(test)] return Ok(Arc::new(crate::mocks::MockRouter::new())); @@ -79,3 +355,103 @@ pub async fn get_router(db: &Arc, router_id: u64) -> OpResult Tunnel { + Tunnel { + id: None, + name: name.to_string(), + local_addr: Some("10.0.0.1".to_string()), + remote_addr: Some("10.0.0.2".to_string()), + enabled: false, + config: TunnelConfig::Gre(GreConfig { key: Some(7) }), + } + } + + #[tokio::test] + async fn test_mock_tunnel_lifecycle() -> anyhow::Result<()> { + let r = MockRouter::new(); + r.clear().await; + let tr = r.tunnel().expect("mock router supports tunnels"); + + assert!(tr.list_tunnels().await.unwrap().is_empty()); + + let added = tr.add_tunnel(&sample_tunnel("gre1")).await.unwrap(); + assert_eq!(added.id.as_deref(), Some("gre1")); + assert!(added.enabled); + assert_eq!(added.kind(), TunnelKind::Gre); + + // duplicate add fails + assert!(tr.add_tunnel(&sample_tunnel("gre1")).await.is_err()); + + let list = tr.list_tunnels().await.unwrap(); + assert_eq!(list.len(), 1); + + let traffic = tr.tunnel_traffic().await.unwrap(); + assert_eq!(traffic.len(), 1); + assert_eq!(traffic[0].name, "gre1"); + + let mut upd = sample_tunnel("gre1"); + upd.remote_addr = Some("10.0.0.9".to_string()); + let updated = tr.update_tunnel(&upd).await.unwrap(); + assert_eq!(updated.remote_addr.as_deref(), Some("10.0.0.9")); + + tr.remove_tunnel("gre1").await.unwrap(); + assert!(tr.list_tunnels().await.unwrap().is_empty()); + Ok(()) + } + + #[tokio::test] + async fn test_mock_bgp() -> anyhow::Result<()> { + let r = MockRouter::new(); + r.clear().await; + r.add_session(BgpSession { + id: "s1".to_string(), + name: "upstream1".to_string(), + peer_ip: Some("192.0.2.1".to_string()), + peer_asn: Some(64512), + local_asn: Some(64500), + state: "Established".to_string(), + prefixes_received: Some(10), + prefixes_sent: Some(2), + enabled: true, + direction: BgpPeerDirection::Upstream, + }) + .await; + let bgp = r.bgp().expect("mock router supports bgp"); + + let sessions = bgp.list_sessions().await.unwrap(); + assert_eq!(sessions.len(), 1); + assert_eq!(sessions[0].peer_asn, Some(64512)); + + let peers = bgp.discover_peers().await.unwrap(); + assert_eq!(peers.len(), 1); + assert_eq!(peers[0].direction, BgpPeerDirection::Upstream); + + // empty candidates => all originated; scoped candidates => filtered subset + assert!(!bgp.originated_routes(&[]).await.unwrap().is_empty()); + assert_eq!( + bgp.originated_routes(&["203.0.113.0/24".to_string()]) + .await + .unwrap() + .len(), + 1 + ); + assert!( + bgp.originated_routes(&["10.0.0.0/8".to_string()]) + .await + .unwrap() + .is_empty() + ); + assert!(bgp.default_route().await.unwrap().is_some()); + + bgp.set_session_enabled("s1", false).await.unwrap(); + let sessions = bgp.list_sessions().await.unwrap(); + assert!(!sessions[0].enabled); + Ok(()) + } +} diff --git a/lnvps_api/src/worker.rs b/lnvps_api/src/worker.rs index 9a0232d7..a1620f37 100644 --- a/lnvps_api/src/worker.rs +++ b/lnvps_api/src/worker.rs @@ -18,8 +18,8 @@ use lnvps_api_common::{ retry::{OpError, Pipeline, RetryPolicy}, }; use lnvps_db::{ - CpuArch, CpuFeature, CpuMfg, IntervalType, LNVpsDb, Subscription, SubscriptionLineItem, - SubscriptionType, Vm, VmHost, VmHostKind, VmIpAssignment, VmOsImage, + CpuArch, CpuFeature, CpuMfg, IntervalType, LNVpsDb, RouterTunnelTraffic, Subscription, + SubscriptionLineItem, SubscriptionType, Vm, VmHost, VmHostKind, VmIpAssignment, VmOsImage, }; use log::{debug, error, info, warn}; use nostr_sdk::{Client, EventBuilder, PublicKey, ToBech32}; @@ -441,6 +441,111 @@ impl Worker { } /// Check all active subscriptions for expiry, auto-renewal, and deactivation. + /// Poll every enabled router for tunnel/BGP state and record per-tunnel + /// traffic samples. + /// + /// Only tunnel traffic counters are sampled into the time-series table; BGP + /// sessions are refreshed as cached state (no byte counters exist for BGP). + /// All route/tunnel queries used here are bounded and full-table safe. + pub async fn sample_router_traffic(&self) -> Result<()> { + let routers = self.db.list_routers().await?; + for router in routers.iter().filter(|r| r.enabled) { + if let Err(e) = self.sample_one_router(router.id).await { + error!("Failed to sample router {}: {}", router.id, e); + } + } + Ok(()) + } + + async fn sample_one_router(&self, router_id: u64) -> Result<()> { + let router = crate::router::get_router(&self.db, router_id) + .await + .map_err(|e| anyhow!("failed to load router {}: {}", router_id, e))?; + + // Tunnels: refresh cached inventory and record traffic samples + if let Some(tr) = router.tunnel() { + match tr.list_tunnels().await { + Ok(tunnels) => { + for t in &tunnels { + if let Err(e) = self.db.upsert_router_tunnel(&t.to_db(router_id)).await { + warn!( + "Failed to cache tunnel {} on router {}: {}", + t.name, router_id, e + ); + } + } + } + Err(e) => warn!("Failed to list tunnels on router {}: {}", router_id, e), + } + match tr.tunnel_traffic().await { + Ok(traffic) => { + for tt in traffic { + let sample = RouterTunnelTraffic { + id: 0, + router_id, + tunnel_name: tt.name, + rx_bytes: tt.rx_bytes, + tx_bytes: tt.tx_bytes, + sampled_at: Utc::now(), + }; + if let Err(e) = self.db.insert_router_tunnel_traffic(&sample).await { + warn!("Failed to record traffic on router {}: {}", router_id, e); + } + } + } + Err(e) => warn!( + "Failed to read tunnel traffic on router {}: {}", + router_id, e + ), + } + } + + // BGP: refresh cached session state (no traffic counters) + if let Some(bgp) = router.bgp() { + match bgp.list_sessions().await { + Ok(sessions) => { + for s in &sessions { + if let Err(e) = self.db.upsert_router_bgp_session(&s.to_db(router_id)).await + { + warn!( + "Failed to cache BGP session {} on router {}: {}", + s.name, router_id, e + ); + } + } + } + Err(e) => warn!("Failed to list BGP sessions on router {}: {}", router_id, e), + } + } + + Ok(()) + } + + /// Enable/disable a BGP session on a router and refresh its cached state. + pub async fn toggle_bgp_session( + &self, + router_id: u64, + session_id: &str, + enabled: bool, + ) -> Result<()> { + let router = crate::router::get_router(&self.db, router_id) + .await + .map_err(|e| anyhow!("failed to load router {}: {}", router_id, e))?; + let bgp = router.bgp().context("router does not support BGP")?; + bgp.set_session_enabled(session_id, enabled) + .await + .map_err(|e| anyhow!("failed to toggle BGP session: {}", e))?; + // Refresh cached session state so the admin API reflects the change + if let Ok(sessions) = bgp.list_sessions().await { + for s in &sessions { + if let Err(e) = self.db.upsert_router_bgp_session(&s.to_db(router_id)).await { + warn!("Failed to refresh BGP session cache: {}", e); + } + } + } + Ok(()) + } + pub async fn check_subscriptions(&self) -> Result<()> { let last_check = self.get_last_check_subscriptions().await?; let time_since = Utc::now().signed_duration_since(last_check); @@ -1719,6 +1824,17 @@ impl Worker { WorkJob::CheckSubscriptions => { self.check_subscriptions().await?; } + WorkJob::SampleRouterTraffic => { + self.sample_router_traffic().await?; + } + WorkJob::ToggleBgpSession { + router_id, + session_id, + enabled, + } => { + self.toggle_bgp_session(*router_id, session_id, *enabled) + .await?; + } WorkJob::DeleteVm { vm_id, reason, @@ -2911,4 +3027,132 @@ mod tests { ); Ok(()) } + + #[tokio::test] + async fn test_sample_router_traffic() -> Result<()> { + use crate::mocks::MockRouter; + use crate::router::{ + BgpPeerDirection, BgpSession, GreConfig, Router as _, Tunnel, TunnelConfig, + }; + use lnvps_db::{Router, RouterKind}; + + let db = Arc::new(MockDb::empty()); + { + let mut routers = db.router.lock().await; + routers.insert( + 1, + Router { + id: 1, + name: "r1".to_string(), + enabled: true, + kind: RouterKind::MockRouter, + url: "mock://".to_string(), + token: "".into(), + }, + ); + } + + // Seed the shared mock-router state with a tunnel and a BGP session + let mr = MockRouter::new(); + mr.clear().await; + mr.tunnel() + .unwrap() + .add_tunnel(&Tunnel { + id: None, + name: "gre1".to_string(), + local_addr: None, + remote_addr: None, + enabled: true, + config: TunnelConfig::Gre(GreConfig { key: None }), + }) + .await + .unwrap(); + mr.add_session(BgpSession { + id: "s1".to_string(), + name: "peer1".to_string(), + peer_ip: Some("192.0.2.1".to_string()), + peer_asn: Some(64512), + local_asn: Some(64500), + state: "Established".to_string(), + prefixes_received: Some(5), + prefixes_sent: Some(1), + enabled: true, + direction: BgpPeerDirection::Upstream, + }) + .await; + + let worker = setup_worker(db.clone()).await?; + worker.sample_router_traffic().await?; + + let tunnels = db.list_router_tunnels(1).await?; + assert_eq!(tunnels.len(), 1); + assert_eq!(tunnels[0].name, "gre1"); + + let traffic = db + .list_router_tunnel_traffic( + 1, + "gre1", + Utc::now() - TimeDelta::hours(1), + Utc::now() + TimeDelta::hours(1), + ) + .await?; + assert_eq!(traffic.len(), 1); + + let sessions = db.list_router_bgp_sessions(1).await?; + assert_eq!(sessions.len(), 1); + assert_eq!(sessions[0].peer_asn, Some(64512)); + + // Clean up shared state for other tests + mr.clear().await; + Ok(()) + } + + #[tokio::test] + async fn test_toggle_bgp_session() -> Result<()> { + use crate::mocks::MockRouter; + use crate::router::{BgpPeerDirection, BgpSession}; + use lnvps_db::{Router, RouterKind}; + + let db = Arc::new(MockDb::empty()); + { + let mut routers = db.router.lock().await; + routers.insert( + 1, + Router { + id: 1, + name: "r1".to_string(), + enabled: true, + kind: RouterKind::MockRouter, + url: "mock://".to_string(), + token: "".into(), + }, + ); + } + let mr = MockRouter::new(); + mr.clear().await; + mr.add_session(BgpSession { + id: "s1".to_string(), + name: "peer1".to_string(), + peer_ip: Some("192.0.2.1".to_string()), + peer_asn: Some(64512), + local_asn: Some(64500), + state: "Established".to_string(), + prefixes_received: None, + prefixes_sent: None, + enabled: true, + direction: BgpPeerDirection::Upstream, + }) + .await; + + let worker = setup_worker(db.clone()).await?; + worker.toggle_bgp_session(1, "s1", false).await?; + + // The cached session should reflect the disabled state after refresh + let sessions = db.list_router_bgp_sessions(1).await?; + assert_eq!(sessions.len(), 1); + assert!(!sessions[0].enabled); + + mr.clear().await; + Ok(()) + } } diff --git a/lnvps_api_admin/src/admin/model.rs b/lnvps_api_admin/src/admin/model.rs index 50de88b3..73359958 100644 --- a/lnvps_api_admin/src/admin/model.rs +++ b/lnvps_api_admin/src/admin/model.rs @@ -100,6 +100,7 @@ impl From for NetworkAccessPolicy { pub enum AdminRouterKind { Mikrotik, OvhAdditionalIp, + LinuxSsh, } impl From for AdminRouterKind { @@ -107,6 +108,7 @@ impl From for AdminRouterKind { match router_kind { RouterKind::Mikrotik => AdminRouterKind::Mikrotik, RouterKind::OvhAdditionalIp => AdminRouterKind::OvhAdditionalIp, + RouterKind::LinuxSsh => AdminRouterKind::LinuxSsh, // MockRouter is a test-only variant and should never appear in production. // Map it to Mikrotik as a safe fallback rather than panicking. RouterKind::MockRouter => AdminRouterKind::Mikrotik, @@ -119,6 +121,7 @@ impl From for RouterKind { match admin_router_kind { AdminRouterKind::Mikrotik => RouterKind::Mikrotik, AdminRouterKind::OvhAdditionalIp => RouterKind::OvhAdditionalIp, + AdminRouterKind::LinuxSsh => RouterKind::LinuxSsh, } } } @@ -1868,6 +1871,112 @@ impl From for AdminRouterDetail { } } +/// A cached tunnel discovered on a router +#[derive(Serialize)] +pub struct AdminRouterTunnel { + pub id: u64, + pub router_id: u64, + pub name: String, + pub kind: String, + pub local_addr: Option, + pub remote_addr: Option, + pub enabled: bool, + pub last_seen: DateTime, +} + +impl From for AdminRouterTunnel { + fn from(t: lnvps_db::RouterTunnel) -> Self { + let kind = match t.kind { + lnvps_db::RouterTunnelKind::Gre => "gre", + lnvps_db::RouterTunnelKind::Vxlan => "vxlan", + lnvps_db::RouterTunnelKind::Wireguard => "wireguard", + } + .to_string(); + Self { + id: t.id, + router_id: t.router_id, + name: t.name, + kind, + local_addr: t.local_addr, + remote_addr: t.remote_addr, + enabled: t.enabled, + last_seen: t.last_seen, + } + } +} + +/// A single per-tunnel traffic sample +#[derive(Serialize)] +pub struct AdminRouterTunnelTraffic { + pub tunnel_name: String, + pub rx_bytes: u64, + pub tx_bytes: u64, + pub sampled_at: DateTime, +} + +impl From for AdminRouterTunnelTraffic { + fn from(t: lnvps_db::RouterTunnelTraffic) -> Self { + Self { + tunnel_name: t.tunnel_name, + rx_bytes: t.rx_bytes, + tx_bytes: t.tx_bytes, + sampled_at: t.sampled_at, + } + } +} + +/// A cached BGP session on a router +#[derive(Serialize)] +pub struct AdminRouterBgpSession { + pub id: u64, + pub router_id: u64, + /// Backend session id used for toggling (protocol name / RouterOS .id) + pub name: String, + pub peer_ip: Option, + pub peer_asn: Option, + pub local_asn: Option, + pub state: String, + pub prefixes_received: Option, + pub prefixes_sent: Option, + pub enabled: bool, + pub direction: String, + pub last_seen: DateTime, +} + +impl From for AdminRouterBgpSession { + fn from(s: lnvps_db::RouterBgpSession) -> Self { + let direction = match s.direction { + lnvps_db::RouterBgpDirection::Upstream => "upstream", + lnvps_db::RouterBgpDirection::Downstream => "downstream", + lnvps_db::RouterBgpDirection::Peer => "peer", + lnvps_db::RouterBgpDirection::Unknown => "unknown", + } + .to_string(); + Self { + id: s.id, + router_id: s.router_id, + name: s.name, + peer_ip: s.peer_ip, + peer_asn: s.peer_asn, + local_asn: s.local_asn, + state: s.state, + prefixes_received: s.prefixes_received, + prefixes_sent: s.prefixes_sent, + enabled: s.enabled, + direction, + last_seen: s.last_seen, + } + } +} + +/// Toggle a BGP session on/off +#[derive(Deserialize)] +pub struct ToggleBgpSessionRequest { + /// Backend session id (protocol name on BIRD, `.id` on Mikrotik) + pub session_id: String, + pub enabled: bool, +} + impl CreateRouterRequest { pub fn to_router(&self) -> anyhow::Result { let db_kind = RouterKind::from(self.kind); @@ -3347,4 +3456,72 @@ mod tests { let req: UpdateCustomPricingRequest = serde_json::from_str(json).unwrap(); assert_eq!(req.cpu_mfg, Some(Some("amd".to_string()))); } + + #[test] + fn test_admin_router_tunnel_from() { + use chrono::Utc; + let t = lnvps_db::RouterTunnel { + id: 7, + router_id: 1, + name: "wg0".to_string(), + kind: lnvps_db::RouterTunnelKind::Wireguard, + local_addr: Some("10.0.0.1".to_string()), + remote_addr: None, + enabled: true, + last_seen: Utc::now(), + }; + let a = AdminRouterTunnel::from(t); + assert_eq!(a.id, 7); + assert_eq!(a.kind, "wireguard"); + assert_eq!(a.local_addr.as_deref(), Some("10.0.0.1")); + } + + #[test] + fn test_admin_router_tunnel_traffic_from() { + use chrono::Utc; + let t = lnvps_db::RouterTunnelTraffic { + id: 1, + router_id: 1, + tunnel_name: "gre1".to_string(), + rx_bytes: 100, + tx_bytes: 200, + sampled_at: Utc::now(), + }; + let a = AdminRouterTunnelTraffic::from(t); + assert_eq!(a.tunnel_name, "gre1"); + assert_eq!(a.rx_bytes, 100); + assert_eq!(a.tx_bytes, 200); + } + + #[test] + fn test_admin_router_bgp_session_from() { + use chrono::Utc; + let s = lnvps_db::RouterBgpSession { + id: 3, + router_id: 1, + name: "peer1".to_string(), + peer_ip: Some("192.0.2.1".to_string()), + peer_asn: Some(64512), + local_asn: Some(64500), + state: "Established".to_string(), + prefixes_received: Some(5), + prefixes_sent: Some(1), + enabled: true, + direction: lnvps_db::RouterBgpDirection::Upstream, + last_seen: Utc::now(), + }; + let a = AdminRouterBgpSession::from(s); + assert_eq!(a.id, 3); + assert_eq!(a.peer_asn, Some(64512)); + assert_eq!(a.direction, "upstream"); + assert_eq!(a.state, "Established"); + } + + #[test] + fn test_toggle_bgp_session_request_deserialize() { + let json = r#"{"session_id": "bgp1", "enabled": false}"#; + let req: ToggleBgpSessionRequest = serde_json::from_str(json).unwrap(); + assert_eq!(req.session_id, "bgp1"); + assert!(!req.enabled); + } } diff --git a/lnvps_api_admin/src/admin/routers.rs b/lnvps_api_admin/src/admin/routers.rs index b8c3a359..04186a69 100644 --- a/lnvps_api_admin/src/admin/routers.rs +++ b/lnvps_api_admin/src/admin/routers.rs @@ -1,13 +1,19 @@ use crate::admin::RouterState; use crate::admin::auth::AdminAuth; use crate::admin::model::{ - AdminRouterDetail, AdminRouterKind, CreateRouterRequest, UpdateRouterRequest, + AdminRouterBgpSession, AdminRouterDetail, AdminRouterTunnel, AdminRouterTunnelTraffic, + CreateRouterRequest, JobResponse, ToggleBgpSessionRequest, UpdateRouterRequest, }; use axum::extract::{Path, Query, State}; -use axum::routing::get; +use axum::routing::{get, post}; use axum::{Json, Router}; -use lnvps_api_common::{ApiData, ApiPaginatedData, ApiPaginatedResult, ApiResult, PageQuery}; +use chrono::{DateTime, TimeDelta, Utc}; +use lnvps_api_common::{ + ApiData, ApiPaginatedData, ApiPaginatedResult, ApiResult, PageQuery, WorkJob, +}; use lnvps_db::{AdminAction, AdminResource}; +use log::{error, info}; +use serde::Deserialize; pub fn router() -> Router { Router::new() @@ -21,6 +27,99 @@ pub fn router() -> Router { .patch(admin_update_router) .delete(admin_delete_router), ) + .route( + "/api/admin/v1/routers/{id}/tunnels", + get(admin_list_router_tunnels), + ) + .route( + "/api/admin/v1/routers/{id}/tunnels/{name}/traffic", + get(admin_get_tunnel_traffic), + ) + .route( + "/api/admin/v1/routers/{id}/bgp/sessions", + get(admin_list_bgp_sessions), + ) + .route( + "/api/admin/v1/routers/{id}/bgp/sessions/toggle", + post(admin_toggle_bgp_session), + ) +} + +/// Time-range filter for traffic history (defaults to the last 24 hours) +#[derive(Deserialize)] +struct TrafficQuery { + from: Option>, + to: Option>, +} + +async fn admin_list_router_tunnels( + auth: AdminAuth, + State(this): State, + Path(router_id): Path, +) -> ApiResult> { + auth.require_permission(AdminResource::Router, AdminAction::View)?; + let tunnels = this.db.list_router_tunnels(router_id).await?; + ApiData::ok(tunnels.into_iter().map(AdminRouterTunnel::from).collect()) +} + +async fn admin_get_tunnel_traffic( + auth: AdminAuth, + State(this): State, + Path((router_id, name)): Path<(u64, String)>, + Query(q): Query, +) -> ApiResult> { + auth.require_permission(AdminResource::Router, AdminAction::View)?; + let to = q.to.unwrap_or_else(Utc::now); + let from = q.from.unwrap_or_else(|| to - TimeDelta::hours(24)); + let samples = this + .db + .list_router_tunnel_traffic(router_id, &name, from, to) + .await?; + ApiData::ok( + samples + .into_iter() + .map(AdminRouterTunnelTraffic::from) + .collect(), + ) +} + +async fn admin_list_bgp_sessions( + auth: AdminAuth, + State(this): State, + Path(router_id): Path, +) -> ApiResult> { + auth.require_permission(AdminResource::Router, AdminAction::View)?; + let sessions = this.db.list_router_bgp_sessions(router_id).await?; + ApiData::ok( + sessions + .into_iter() + .map(AdminRouterBgpSession::from) + .collect(), + ) +} + +async fn admin_toggle_bgp_session( + auth: AdminAuth, + State(this): State, + Path(router_id): Path, + Json(request): Json, +) -> ApiResult { + auth.require_permission(AdminResource::Router, AdminAction::Update)?; + let job = WorkJob::ToggleBgpSession { + router_id, + session_id: request.session_id, + enabled: request.enabled, + }; + match this.work_commander.send(job).await { + Ok(stream_id) => { + info!("BGP toggle job queued with stream ID: {}", stream_id); + ApiData::ok(JobResponse { job_id: stream_id }) + } + Err(e) => { + error!("Failed to queue BGP toggle job: {}", e); + ApiData::err("Failed to queue BGP toggle job") + } + } } async fn admin_list_routers( @@ -110,10 +209,7 @@ async fn admin_update_router( router.enabled = enabled; } if let Some(kind) = &request.kind { - router.kind = match kind { - AdminRouterKind::Mikrotik => lnvps_db::RouterKind::Mikrotik, - AdminRouterKind::OvhAdditionalIp => lnvps_db::RouterKind::OvhAdditionalIp, - }; + router.kind = lnvps_db::RouterKind::from(*kind); } if let Some(url) = &request.url { router.url = url.trim().to_string(); diff --git a/lnvps_api_common/src/mock.rs b/lnvps_api_common/src/mock.rs index ff72e02b..869c962f 100644 --- a/lnvps_api_common/src/mock.rs +++ b/lnvps_api_common/src/mock.rs @@ -6,10 +6,11 @@ use lnvps_db::{ AccessPolicy, AvailableIpSpace, Company, CpuArch, CpuMfg, DbError, DbResult, DiskInterface, DiskType, IntervalType, IpRange, IpRangeAllocationMode, IpRangeSubscription, IpSpacePricing, LNVpsDbBase, NostrDomain, NostrDomainHandle, OsDistribution, PaymentMethod, - PaymentMethodConfig, Referral, ReferralCostUsage, ReferralPayout, Router, Subscription, - SubscriptionLineItem, SubscriptionPayment, SubscriptionPaymentWithCompany, User, UserSshKey, - Vm, VmCostPlan, VmCustomPricing, VmCustomPricingDisk, VmCustomTemplate, VmHistory, VmHost, - VmHostDisk, VmHostKind, VmHostRegion, VmIpAssignment, VmOsImage, VmPayment, VmTemplate, + PaymentMethodConfig, Referral, ReferralCostUsage, ReferralPayout, Router, RouterBgpSession, + RouterTunnel, RouterTunnelTraffic, Subscription, SubscriptionLineItem, SubscriptionPayment, + SubscriptionPaymentWithCompany, User, UserSshKey, Vm, VmCostPlan, VmCustomPricing, + VmCustomPricingDisk, VmCustomTemplate, VmHistory, VmHost, VmHostDisk, VmHostKind, VmHostRegion, + VmIpAssignment, VmOsImage, VmPayment, VmTemplate, }; use async_trait::async_trait; @@ -48,6 +49,9 @@ pub struct MockDb { pub payment_method_configs: Arc>>, pub referrals: Arc>>, pub referral_payouts: Arc>>, + pub router_tunnels: Arc>>, + pub router_tunnel_traffic: Arc>>, + pub router_bgp_sessions: Arc>>, } impl MockDb { @@ -297,6 +301,9 @@ impl Default for MockDb { payment_method_configs: Arc::new(Default::default()), referrals: Arc::new(Default::default()), referral_payouts: Arc::new(Default::default()), + router_tunnels: Arc::new(Default::default()), + router_tunnel_traffic: Arc::new(Default::default()), + router_bgp_sessions: Arc::new(Default::default()), } } } @@ -1097,6 +1104,119 @@ impl LNVpsDbBase for MockDb { Ok(routers.values().cloned().collect()) } + async fn list_router_tunnels(&self, router_id: u64) -> DbResult> { + let t = self.router_tunnels.lock().await; + Ok(t.values() + .filter(|x| x.router_id == router_id) + .cloned() + .collect()) + } + + async fn upsert_router_tunnel(&self, tunnel: &RouterTunnel) -> DbResult { + let mut t = self.router_tunnels.lock().await; + if let Some(existing) = t + .values_mut() + .find(|x| x.router_id == tunnel.router_id && x.name == tunnel.name) + { + let id = existing.id; + *existing = RouterTunnel { + id, + last_seen: Utc::now(), + ..tunnel.clone() + }; + return Ok(id); + } + let id = t.keys().max().copied().unwrap_or(0) + 1; + t.insert( + id, + RouterTunnel { + id, + last_seen: Utc::now(), + ..tunnel.clone() + }, + ); + Ok(id) + } + + async fn delete_router_tunnel(&self, id: u64) -> DbResult<()> { + let mut t = self.router_tunnels.lock().await; + t.remove(&id); + Ok(()) + } + + async fn insert_router_tunnel_traffic(&self, sample: &RouterTunnelTraffic) -> DbResult { + let mut t = self.router_tunnel_traffic.lock().await; + let id = t.len() as u64 + 1; + t.push(RouterTunnelTraffic { + id, + sampled_at: Utc::now(), + ..sample.clone() + }); + Ok(id) + } + + async fn list_router_tunnel_traffic( + &self, + router_id: u64, + tunnel_name: &str, + from: chrono::DateTime, + to: chrono::DateTime, + ) -> DbResult> { + let t = self.router_tunnel_traffic.lock().await; + let mut out: Vec = t + .iter() + .filter(|x| { + x.router_id == router_id + && x.tunnel_name == tunnel_name + && x.sampled_at >= from + && x.sampled_at <= to + }) + .cloned() + .collect(); + out.sort_by_key(|x| x.sampled_at); + Ok(out) + } + + async fn list_router_bgp_sessions(&self, router_id: u64) -> DbResult> { + let s = self.router_bgp_sessions.lock().await; + Ok(s.values() + .filter(|x| x.router_id == router_id) + .cloned() + .collect()) + } + + async fn upsert_router_bgp_session(&self, session: &RouterBgpSession) -> DbResult { + let mut s = self.router_bgp_sessions.lock().await; + if let Some(existing) = s + .values_mut() + .find(|x| x.router_id == session.router_id && x.name == session.name) + { + let id = existing.id; + *existing = RouterBgpSession { + id, + last_seen: Utc::now(), + ..session.clone() + }; + return Ok(id); + } + let id = s.keys().max().copied().unwrap_or(0) + 1; + s.insert( + id, + RouterBgpSession { + id, + last_seen: Utc::now(), + ..session.clone() + }, + ); + Ok(id) + } + + async fn delete_router_bgp_session(&self, id: u64) -> DbResult<()> { + let mut s = self.router_bgp_sessions.lock().await; + s.remove(&id); + Ok(()) + } + async fn get_vm_ip_assignment(&self, id: u64) -> DbResult { let assignments = self.ip_assignments.lock().await; Ok(assignments @@ -3374,4 +3494,105 @@ mod tests { assert!(!updated.is_active); assert!(updated.ended_at.is_some()); } + + #[tokio::test] + async fn test_router_tunnel_crud() { + use lnvps_db::{RouterTunnel, RouterTunnelKind}; + let db = MockDb::empty(); + + let t = RouterTunnel { + id: 0, + router_id: 1, + name: "gre1".to_string(), + kind: RouterTunnelKind::Gre, + local_addr: Some("10.0.0.1".to_string()), + remote_addr: Some("10.0.0.2".to_string()), + enabled: true, + last_seen: Utc::now(), + }; + let id = db.upsert_router_tunnel(&t).await.unwrap(); + assert_eq!(db.list_router_tunnels(1).await.unwrap().len(), 1); + + // upsert by (router_id, name) updates in place + let mut t2 = t.clone(); + t2.enabled = false; + let id2 = db.upsert_router_tunnel(&t2).await.unwrap(); + assert_eq!(id, id2); + let tunnels = db.list_router_tunnels(1).await.unwrap(); + assert_eq!(tunnels.len(), 1); + assert!(!tunnels[0].enabled); + + db.delete_router_tunnel(id).await.unwrap(); + assert!(db.list_router_tunnels(1).await.unwrap().is_empty()); + } + + #[tokio::test] + async fn test_router_tunnel_traffic_window() { + use lnvps_db::RouterTunnelTraffic; + let db = MockDb::empty(); + db.insert_router_tunnel_traffic(&RouterTunnelTraffic { + id: 0, + router_id: 1, + tunnel_name: "gre1".to_string(), + rx_bytes: 100, + tx_bytes: 200, + sampled_at: Utc::now(), + }) + .await + .unwrap(); + + let in_window = db + .list_router_tunnel_traffic( + 1, + "gre1", + Utc::now() - chrono::Duration::hours(1), + Utc::now() + chrono::Duration::hours(1), + ) + .await + .unwrap(); + assert_eq!(in_window.len(), 1); + + let out_window = db + .list_router_tunnel_traffic( + 1, + "gre1", + Utc::now() + chrono::Duration::hours(1), + Utc::now() + chrono::Duration::hours(2), + ) + .await + .unwrap(); + assert!(out_window.is_empty()); + } + + #[tokio::test] + async fn test_router_bgp_session_crud() { + use lnvps_db::{RouterBgpDirection, RouterBgpSession}; + let db = MockDb::empty(); + let s = RouterBgpSession { + id: 0, + router_id: 1, + name: "peer1".to_string(), + peer_ip: Some("192.0.2.1".to_string()), + peer_asn: Some(64512), + local_asn: Some(64500), + state: "Established".to_string(), + prefixes_received: Some(5), + prefixes_sent: Some(1), + enabled: true, + direction: RouterBgpDirection::Upstream, + last_seen: Utc::now(), + }; + let id = db.upsert_router_bgp_session(&s).await.unwrap(); + assert_eq!(db.list_router_bgp_sessions(1).await.unwrap().len(), 1); + + let mut s2 = s.clone(); + s2.state = "Idle".to_string(); + let id2 = db.upsert_router_bgp_session(&s2).await.unwrap(); + assert_eq!(id, id2); + let sessions = db.list_router_bgp_sessions(1).await.unwrap(); + assert_eq!(sessions[0].state, "Idle"); + + db.delete_router_bgp_session(id).await.unwrap(); + assert!(db.list_router_bgp_sessions(1).await.unwrap().is_empty()); + } } diff --git a/lnvps_api_common/src/work/mod.rs b/lnvps_api_common/src/work/mod.rs index 262f06a3..c035343d 100644 --- a/lnvps_api_common/src/work/mod.rs +++ b/lnvps_api_common/src/work/mod.rs @@ -126,6 +126,15 @@ pub enum WorkJob { DownloadOsImages { image_id: Option }, /// Check all active subscriptions for expiry, auto-renewal, and deactivation. CheckSubscriptions, + /// Poll routers for tunnel/BGP state and record per-tunnel traffic samples. + SampleRouterTraffic, + /// Enable or disable a BGP session on a router (admin action). + ToggleBgpSession { + router_id: u64, + /// Backend session id (protocol name on BIRD, `.id` on Mikrotik) + session_id: String, + enabled: bool, + }, } impl WorkJob { @@ -167,6 +176,8 @@ impl fmt::Display for WorkJob { WorkJob::DownloadOsImages { .. } => write!(f, "DownloadOsImages"), WorkJob::CheckSubscriptions => write!(f, "CheckSubscriptions"), WorkJob::SpawnVm { .. } => write!(f, "SpawnVm"), + WorkJob::SampleRouterTraffic => write!(f, "SampleRouterTraffic"), + WorkJob::ToggleBgpSession { .. } => write!(f, "ToggleBgpSession"), } } } diff --git a/lnvps_db/migrations/20260621220706_router_tunnels_bgp.sql b/lnvps_db/migrations/20260621220706_router_tunnels_bgp.sql new file mode 100644 index 00000000..b42219da --- /dev/null +++ b/lnvps_db/migrations/20260621220706_router_tunnels_bgp.sql @@ -0,0 +1,47 @@ +-- Cached tunnel inventory discovered on routers (GRE/VXLAN/WireGuard) +create table router_tunnel +( + id integer unsigned not null auto_increment primary key, + router_id integer unsigned not null, + name varchar(100) not null, + kind smallint unsigned not null, + local_addr varchar(255), + remote_addr varchar(255), + enabled bit(1) not null default 1, + last_seen timestamp not null default current_timestamp, + constraint fk_router_tunnel_router foreign key (router_id) references router (id), + constraint uq_router_tunnel_name unique (router_id, name) +); + +-- Per-tunnel traffic samples (the canonical "per session" traffic for route servers). +-- BGP sessions have no byte counters; counters come from the tunnel interfaces. +create table router_tunnel_traffic +( + id bigint unsigned not null auto_increment primary key, + router_id integer unsigned not null, + tunnel_name varchar(100) not null, + rx_bytes bigint unsigned not null default 0, + tx_bytes bigint unsigned not null default 0, + sampled_at timestamp not null default current_timestamp, + constraint fk_router_tunnel_traffic_router foreign key (router_id) references router (id) +); +create index ix_router_tunnel_traffic_lookup on router_tunnel_traffic (router_id, tunnel_name, sampled_at); + +-- Cached BGP session discovery state (no byte counters) +create table router_bgp_session +( + id integer unsigned not null auto_increment primary key, + router_id integer unsigned not null, + name varchar(100) not null, + peer_ip varchar(64), + peer_asn integer unsigned, + local_asn integer unsigned, + state varchar(32) not null, + prefixes_received bigint unsigned, + prefixes_sent bigint unsigned, + enabled bit(1) not null default 1, + direction smallint unsigned not null default 0, + last_seen timestamp not null default current_timestamp, + constraint fk_router_bgp_session_router foreign key (router_id) references router (id), + constraint uq_router_bgp_session_name unique (router_id, name) +); diff --git a/lnvps_db/src/lib.rs b/lnvps_db/src/lib.rs index 36d3b07e..9d0426b3 100644 --- a/lnvps_db/src/lib.rs +++ b/lnvps_db/src/lib.rs @@ -1,5 +1,6 @@ use anyhow::{Error, Result, anyhow}; use async_trait::async_trait; +use chrono::{DateTime, Utc}; use sqlx::migrate::MigrateError; use thiserror::Error; @@ -374,6 +375,36 @@ pub trait LNVpsDbBase: Send + Sync { /// List all routers async fn list_routers(&self) -> DbResult>; + /// List cached tunnels for a router + async fn list_router_tunnels(&self, router_id: u64) -> DbResult>; + + /// Insert or update (by router_id+name) a cached tunnel, returning its id + async fn upsert_router_tunnel(&self, tunnel: &RouterTunnel) -> DbResult; + + /// Delete a cached tunnel by id + async fn delete_router_tunnel(&self, id: u64) -> DbResult<()>; + + /// Insert a per-tunnel traffic sample, returning its id + async fn insert_router_tunnel_traffic(&self, sample: &RouterTunnelTraffic) -> DbResult; + + /// List traffic samples for a router/tunnel within a time window + async fn list_router_tunnel_traffic( + &self, + router_id: u64, + tunnel_name: &str, + from: DateTime, + to: DateTime, + ) -> DbResult>; + + /// List cached BGP sessions for a router + async fn list_router_bgp_sessions(&self, router_id: u64) -> DbResult>; + + /// Insert or update (by router_id+name) a cached BGP session, returning its id + async fn upsert_router_bgp_session(&self, session: &RouterBgpSession) -> DbResult; + + /// Delete a cached BGP session by id + async fn delete_router_bgp_session(&self, id: u64) -> DbResult<()>; + /// Get VM IP assignment async fn get_vm_ip_assignment(&self, id: u64) -> DbResult; diff --git a/lnvps_db/src/model.rs b/lnvps_db/src/model.rs index ed8eff25..9b589906 100644 --- a/lnvps_db/src/model.rs +++ b/lnvps_db/src/model.rs @@ -683,10 +683,73 @@ pub enum RouterKind { Mikrotik = 0, /// A pseudo-router which allows adding virtual mac addresses to a dedicated server OvhAdditionalIp = 1, + /// A Linux machine managed over SSH (BIRD/Pathvector routing, iproute2 tunnels) + LinuxSsh = 2, /// Mock router access in tests MockRouter = u16::MAX, } +/// Cached tunnel inventory discovered on a router (GRE/VXLAN/WireGuard) +#[derive(FromRow, Clone, Debug)] +pub struct RouterTunnel { + pub id: u64, + pub router_id: u64, + /// Tunnel interface name + pub name: String, + pub kind: RouterTunnelKind, + pub local_addr: Option, + pub remote_addr: Option, + pub enabled: bool, + pub last_seen: DateTime, +} + +#[derive(Debug, Clone, Copy, sqlx::Type, PartialEq, Eq)] +#[repr(u16)] +pub enum RouterTunnelKind { + Gre = 0, + Vxlan = 1, + Wireguard = 2, +} + +/// A single per-tunnel traffic sample. Tunnel interface counters are the +/// canonical source of per-session traffic for route servers (BGP has none). +#[derive(FromRow, Clone, Debug)] +pub struct RouterTunnelTraffic { + pub id: u64, + pub router_id: u64, + pub tunnel_name: String, + pub rx_bytes: u64, + pub tx_bytes: u64, + pub sampled_at: DateTime, +} + +/// Cached BGP session discovery state (no byte counters) +#[derive(FromRow, Clone, Debug)] +pub struct RouterBgpSession { + pub id: u64, + pub router_id: u64, + pub name: String, + pub peer_ip: Option, + pub peer_asn: Option, + pub local_asn: Option, + pub state: String, + pub prefixes_received: Option, + pub prefixes_sent: Option, + pub enabled: bool, + pub direction: RouterBgpDirection, + pub last_seen: DateTime, +} + +#[derive(Debug, Clone, Copy, sqlx::Type, PartialEq, Eq, Default)] +#[repr(u16)] +pub enum RouterBgpDirection { + #[default] + Unknown = 0, + Upstream = 1, + Downstream = 2, + Peer = 3, +} + #[derive(FromRow, Clone, Debug, Default)] pub struct IpRange { pub id: u64, diff --git a/lnvps_db/src/mysql.rs b/lnvps_db/src/mysql.rs index 307e19cf..d8256557 100644 --- a/lnvps_db/src/mysql.rs +++ b/lnvps_db/src/mysql.rs @@ -1,11 +1,11 @@ use crate::{ AccessPolicy, AvailableIpSpace, Company, DbError, DbResult, IntervalType, IpRange, IpRangeSubscription, IpSpacePricing, LNVpsDbBase, PaymentMethod, PaymentMethodConfig, - PaymentType, Referral, ReferralCostUsage, ReferralPayout, RegionStats, Router, Subscription, - SubscriptionLineItem, SubscriptionPayment, SubscriptionPaymentWithCompany, User, UserSshKey, - Vm, VmCostPlan, VmCustomPricing, VmCustomPricingDisk, VmCustomTemplate, VmForMigration, - VmHistory, VmHost, VmHostDisk, VmHostRegion, VmIpAssignment, VmOsImage, VmPayment, - VmPaymentRaw, VmTemplate, + PaymentType, Referral, ReferralCostUsage, ReferralPayout, RegionStats, Router, + RouterBgpSession, RouterTunnel, RouterTunnelTraffic, Subscription, SubscriptionLineItem, + SubscriptionPayment, SubscriptionPaymentWithCompany, User, UserSshKey, Vm, VmCostPlan, + VmCustomPricing, VmCustomPricingDisk, VmCustomTemplate, VmForMigration, VmHistory, VmHost, + VmHostDisk, VmHostRegion, VmIpAssignment, VmOsImage, VmPayment, VmPaymentRaw, VmTemplate, }; #[cfg(feature = "admin")] use crate::{AdminDb, AdminRole, AdminRoleAssignment, AdminVmHost}; @@ -13,6 +13,7 @@ use crate::{AdminDb, AdminRole, AdminRoleAssignment, AdminVmHost}; use crate::{LNVPSNostrDb, NostrDomain, NostrDomainHandle}; use anyhow::{Result, anyhow}; use async_trait::async_trait; +use chrono::{DateTime, Utc}; use sqlx::{Executor, MySqlPool, QueryBuilder, Row}; #[derive(Clone)] @@ -1236,6 +1237,127 @@ impl LNVpsDbBase for LNVpsDbMysql { .await?) } + async fn list_router_tunnels(&self, router_id: u64) -> DbResult> { + Ok( + sqlx::query_as("select * from router_tunnel where router_id=?") + .bind(router_id) + .fetch_all(&self.db) + .await?, + ) + } + + async fn upsert_router_tunnel(&self, tunnel: &RouterTunnel) -> DbResult { + Ok(sqlx::query( + r#"insert into router_tunnel (router_id, name, kind, local_addr, remote_addr, enabled, last_seen) + values (?, ?, ?, ?, ?, ?, current_timestamp) + on duplicate key update + kind = values(kind), + local_addr = values(local_addr), + remote_addr = values(remote_addr), + enabled = values(enabled), + last_seen = current_timestamp"#, + ) + .bind(tunnel.router_id) + .bind(&tunnel.name) + .bind(tunnel.kind) + .bind(&tunnel.local_addr) + .bind(&tunnel.remote_addr) + .bind(tunnel.enabled) + .execute(&self.db) + .await? + .last_insert_id()) + } + + async fn delete_router_tunnel(&self, id: u64) -> DbResult<()> { + sqlx::query("delete from router_tunnel where id=?") + .bind(id) + .execute(&self.db) + .await?; + Ok(()) + } + + async fn insert_router_tunnel_traffic(&self, sample: &RouterTunnelTraffic) -> DbResult { + Ok(sqlx::query( + r#"insert into router_tunnel_traffic (router_id, tunnel_name, rx_bytes, tx_bytes, sampled_at) + values (?, ?, ?, ?, current_timestamp)"#, + ) + .bind(sample.router_id) + .bind(&sample.tunnel_name) + .bind(sample.rx_bytes) + .bind(sample.tx_bytes) + .execute(&self.db) + .await? + .last_insert_id()) + } + + async fn list_router_tunnel_traffic( + &self, + router_id: u64, + tunnel_name: &str, + from: DateTime, + to: DateTime, + ) -> DbResult> { + Ok(sqlx::query_as( + r#"select * from router_tunnel_traffic + where router_id=? and tunnel_name=? and sampled_at >= ? and sampled_at <= ? + order by sampled_at asc"#, + ) + .bind(router_id) + .bind(tunnel_name) + .bind(from) + .bind(to) + .fetch_all(&self.db) + .await?) + } + + async fn list_router_bgp_sessions(&self, router_id: u64) -> DbResult> { + Ok( + sqlx::query_as("select * from router_bgp_session where router_id=?") + .bind(router_id) + .fetch_all(&self.db) + .await?, + ) + } + + async fn upsert_router_bgp_session(&self, session: &RouterBgpSession) -> DbResult { + Ok(sqlx::query( + r#"insert into router_bgp_session + (router_id, name, peer_ip, peer_asn, local_asn, state, prefixes_received, prefixes_sent, enabled, direction, last_seen) + values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, current_timestamp) + on duplicate key update + peer_ip = values(peer_ip), + peer_asn = values(peer_asn), + local_asn = values(local_asn), + state = values(state), + prefixes_received = values(prefixes_received), + prefixes_sent = values(prefixes_sent), + enabled = values(enabled), + direction = values(direction), + last_seen = current_timestamp"#, + ) + .bind(session.router_id) + .bind(&session.name) + .bind(&session.peer_ip) + .bind(session.peer_asn) + .bind(session.local_asn) + .bind(&session.state) + .bind(session.prefixes_received) + .bind(session.prefixes_sent) + .bind(session.enabled) + .bind(session.direction) + .execute(&self.db) + .await? + .last_insert_id()) + } + + async fn delete_router_bgp_session(&self, id: u64) -> DbResult<()> { + sqlx::query("delete from router_bgp_session where id=?") + .bind(id) + .execute(&self.db) + .await?; + Ok(()) + } + async fn get_vm_ip_assignment(&self, id: u64) -> DbResult { Ok(sqlx::query_as("select * from vm_ip_assignment where id=?") .bind(id) diff --git a/work/route-server-management.md b/work/route-server-management.md new file mode 100644 index 00000000..c7fd122c --- /dev/null +++ b/work/route-server-management.md @@ -0,0 +1,116 @@ +# Route Server Management (#138) + +**Status:** in-progress +**Started:** 2026-06-21 +**Last updated:** 2026-06-21 (increments 1-5 complete) + +## Goal + +Extend the router subsystem to support: SSH-based Linux (BIRD/Pathvector) routers, +BGP session detection/toggle, originated-route + default-route detection, peer +discovery, and tunnel (GRE/VXLAN/WireGuard) detection/management with per-tunnel +traffic counters. Mikrotik must implement the same new capabilities. Traffic +counters come from tunnel interfaces (not BGP sessions). + +## Findings + +- Router trait + factory: `lnvps_api/src/router/mod.rs` (ARP-only today). +- Backends: `lnvps_api/src/router/{mikrotik.rs,ovh.rs}`. Mikrotik uses `JsonApi`. +- DB model `Router` + `RouterKind` enum: `lnvps_db/src/model.rs` (~L670). `RouterKind` + is `#[repr(u16)] sqlx::Type`. DB methods in `lnvps_db/src/mysql.rs` (`get_router`, + `list_routers`, `admin_*_router`). +- Admin CRUD: `lnvps_api_admin/src/admin/routers.rs` + models in + `lnvps_api_admin/src/admin/model.rs` (`AdminRouterKind`, `AdminRouterDetail`, + `Create/UpdateRouterRequest`). +- Mock router: `lnvps_api/src/mocks.rs` (`MockRouter`, uses LazyLock shared state). +- Reusable SSH: `lnvps_api/src/ssh_client.rs` (`SshClient`, ssh2, key-file + + in-memory PEM). `ssh2` is gated behind `proxmox`/`libvirt` features currently. +- Feature flags in `lnvps_api/Cargo.toml`: `mikrotik` (default on), `proxmox` (pulls `ssh2`). +- Decision: abstraction-first. Capability traits `BgpRouter` + `TunnelRouter` + separate from `Router`. Ship SSH/CLI backend first; netlink agent later behind + same traits (`RouterKind::LinuxAgent` future). + +## Tasks + +### Increment 1 — Linux SSH backend skeleton + RouterKind ✅ DONE +- [x] Add `RouterKind::LinuxSsh = 2` (db model) + `AdminRouterKind::LinuxSsh` mappings +- [x] New `lnvps_api/src/router/linux_ssh.rs` implementing `Router` (ARP via `ip neigh`) +- [x] Wire into `get_router()` factory + feature flag (`linux-ssh` pulling `ssh2`) +- [x] Unit tests for url parsing / neigh parsing +- Notes: url=`ssh://user@host:port/interface`, token=PEM key. ssh_client now gated + on `any(proxmox, linux-ssh)`. Connect-per-operation (Send/Sync-safe). ARP via + `ip -j neigh show` / `ip neigh replace ... nud permanent` / `ip neigh del`. + +### Increment 2 — TunnelRouter trait + Linux impl ✅ DONE +- [x] Define `TunnelRouter` trait + `Tunnel`/`TunnelKind`/`TunnelConfig`/`TunnelTraffic` + + `Gre/Vxlan/WireguardConfig` + `WireguardPeer` (in router/mod.rs) +- [x] Capability accessor: `Router::tunnel() -> Option<&dyn TunnelRouter>` (default None) +- [x] Linux detect via `ip -s -d -j link show` (+ `wg show all dump` for WG) + manage + (`ip link add/del`, `wg set` with 0600 temp key file) + traffic from stats64 +- [x] MockRouter `TunnelRouter` impl + `tunnel()` override +- [x] Unit tests (gre/vxlan/wg parsing, gre-key, shq escaping, wg-set script, traffic + filter, mock lifecycle) +- Notes: `update_tunnel` recreates iface (del+add) for deterministic config apply. + GRE key accepts int or dotted-quad. WG private key never returned on list. + +### Increment 3 — Mikrotik TunnelRouter impl ✅ DONE +- [x] `/rest/interface/{gre,vxlan,wireguard}` (+ `/wireguard/peers`) list/add/remove/update +- [x] Traffic via `/rest/interface` rx-byte/tx-byte filtered by type {gre,vxlan,wg} +- [x] `Router::tunnel()` override; tunnel id encoded as `":"` +- [x] Unit tests for helpers (mt_enabled, endpoint map, id split, endpoint split, wg peer) +- Notes: RouterOS booleans/numbers are strings. GRE has no key field (ignored). + VXLAN remote handled via vteps (not yet supported — remote_addr=None). WG private + key never returned on list. DELETE returns empty body → treated as success. + +### Increment 4 — BgpRouter trait + Linux birdc + Mikrotik BGP ✅ DONE +- [x] `BgpRouter` trait + `BgpSession`/`BgpPeer`/`BgpRoute`/`BgpPeerDirection` + + `Router::bgp()` accessor (router/mod.rs) +- [x] Linux birdc: `show protocols all` parse (sessions), `show route` parse + (originated/default), `birdc enable/disable` toggle; role→direction mapping +- [x] Mikrotik `/rest/routing/bgp/{connection,session,advertisements}` + `/rest/ip/route`; + toggle PATCHes connection.disabled +- [x] MockRouter `BgpRouter` impl (+ `add_session` seed helper) + `bgp()` override +- [x] Unit tests: bird protocols/routes parsers, role mapping, mock bgp lifecycle +- Notes: BGP has NO byte counters (prefixes only). Mikrotik direction=Unknown + (no customer/provider signal). birdc text parsing is inherently fragile — + parsers are table-tested against captured sample output. +- **Full-table (DFZ) safety** — routers carry a full internet table (~1M+ routes): + - `originated_routes(candidates)` is SCOPED to candidate prefixes (VM ranges), + never enumerates the table. Empty slice => only locally-originated (small). + - BIRD uses `show route for ` (LPM lookup) + `show route where source = + RTS_STATIC` (bounded output). Never a bare `show route` dump. + - Mikrotik NEVER does unfiltered `GET /rest/ip/route`. Uses server-side + `?dst-address=&.proplist=...` filters per candidate / for default route. + - Sampler (incr 5) samples ONLY tunnel traffic (bounded), never routes. + +### Increment 5 — Persistence + background sampler ✅ DONE +- [x] Migration `20260621220706_router_tunnels_bgp.sql`: `router_tunnel`, + `router_tunnel_traffic`, `router_bgp_session` (FKs + unique(router_id,name)) +- [x] DB models `RouterTunnel`/`RouterTunnelKind`/`RouterTunnelTraffic`/ + `RouterBgpSession`/`RouterBgpDirection` (model.rs) +- [x] DB trait methods (list/upsert/delete tunnels, insert/list traffic, list/upsert/delete + bgp sessions) + MySQL impl (upsert via ON DUPLICATE KEY) + MockDb impl +- [x] `Tunnel::to_db`/`BgpSession::to_db` conversions (router/mod.rs) +- [x] `WorkJob::SampleRouterTraffic` + `Worker::sample_router_traffic`/`sample_one_router`; + registered 60s interval in bin/api.rs +- [x] Tests: MockDb CRUD (tunnel/traffic/bgp), worker sampler integration test +- Notes: sampler only writes tunnel traffic to time-series; BGP refreshed as cached + state. All queries bounded/full-table safe. 100% function coverage maintained. + +### Increment 6 — Admin API + docs ✅ DONE +- [x] Admin endpoints (routers.rs): GET tunnels, GET tunnel traffic (from/to, def 24h), + GET bgp/sessions, POST bgp/sessions/toggle (dispatches WorkJob) +- [x] Admin models (model.rs): `AdminRouterTunnel`/`AdminRouterTunnelTraffic`/ + `AdminRouterBgpSession` From impls + `ToggleBgpSessionRequest` +- [x] `WorkJob::ToggleBgpSession` + `Worker::toggle_bgp_session` handler (refreshes cache) +- [x] API_CHANGELOG.md + ADMIN_API_ENDPOINTS.md updated (incl. linux_ssh RouterKind) +- [x] Tests: admin model conversions + request deserialize; worker toggle test +- Notes: admin crate has NO dep on lnvps_api, so reads cached DB state and dispatches + live actions via WorkCommander. Originated/default-route admin exposure deferred + (needs caching or RPC; capability exists in BgpRouter trait). + +## Notes + +- Open questions (asked on issue): Pathvector toggle persistence, WG key storage, + tunnel→customer/subscription linkage, router count × sampling cadence. +- Per-session BGP byte counters do NOT exist — traffic is per-tunnel only.