diff --git a/Cargo.lock b/Cargo.lock index 6250de0..7f6a1ff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -543,6 +543,7 @@ dependencies = [ "rtnetlink", "secp256k1", "serde", + "serde_json", "serde_yaml", "sha2", "simple-dns", diff --git a/Cargo.toml b/Cargo.toml index 68de178..06b4817 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,6 +17,7 @@ rand = "0.8" thiserror = "2.0" bech32 = "0.11" serde = { version = "1.0", features = ["derive"] } +serde_json = "1.0" serde_yaml = "0.9" dirs = "6.0" hex = "0.4" @@ -35,6 +36,10 @@ socket2 = { version = "0.5", features = ["all"] } tempfile = "3.15" criterion = { version = "0.5", features = ["html_reports"] } +[[bin]] +name = "fipsctl" +path = "src/bin/fipsctl.rs" + [[bench]] name = "bloom" harness = false diff --git a/src/bin/fipsctl.rs b/src/bin/fipsctl.rs new file mode 100644 index 0000000..afadf63 --- /dev/null +++ b/src/bin/fipsctl.rs @@ -0,0 +1,165 @@ +//! fipsctl — FIPS control client +//! +//! Connects to the FIPS daemon's Unix domain control socket, sends a +//! query command, and pretty-prints the JSON response. + +use clap::{Parser, Subcommand}; +use std::io::{BufRead, BufReader, Write}; +use std::os::unix::net::UnixStream; +use std::path::PathBuf; +use std::time::Duration; + +/// FIPS control client +#[derive(Parser, Debug)] +#[command(name = "fipsctl", version, about = "Query a running FIPS daemon")] +struct Cli { + /// Control socket path override + #[arg(short = 's', long)] + socket: Option, + + #[command(subcommand)] + command: Commands, +} + +#[derive(Subcommand, Debug)] +enum Commands { + /// Show node information + Show { + #[command(subcommand)] + what: ShowCommands, + }, +} + +#[derive(Subcommand, Debug)] +enum ShowCommands { + /// Node status overview + Status, + /// Authenticated peers + Peers, + /// Active links + Links, + /// Spanning tree state + Tree, + /// End-to-end sessions + Sessions, + /// Bloom filter state + Bloom, + /// MMP metrics summary + Mmp, + /// Coordinate cache stats + Cache, + /// Pending handshake connections + Connections, + /// Transport instances + Transports, + /// Routing table summary + Routing, +} + +impl ShowCommands { + fn command_name(&self) -> &'static str { + match self { + ShowCommands::Status => "show_status", + ShowCommands::Peers => "show_peers", + ShowCommands::Links => "show_links", + ShowCommands::Tree => "show_tree", + ShowCommands::Sessions => "show_sessions", + ShowCommands::Bloom => "show_bloom", + ShowCommands::Mmp => "show_mmp", + ShowCommands::Cache => "show_cache", + ShowCommands::Connections => "show_connections", + ShowCommands::Transports => "show_transports", + ShowCommands::Routing => "show_routing", + } + } +} + +/// Determine the default socket path. +fn default_socket_path() -> PathBuf { + if let Ok(runtime_dir) = std::env::var("XDG_RUNTIME_DIR") { + PathBuf::from(format!("{}/fips/control.sock", runtime_dir)) + } else { + PathBuf::from("/tmp/fips-control.sock") + } +} + +fn main() { + let cli = Cli::parse(); + + let socket_path = cli.socket.unwrap_or_else(default_socket_path); + let command_name = match &cli.command { + Commands::Show { what } => what.command_name(), + }; + + // Connect to the control socket + let mut stream = match UnixStream::connect(&socket_path) { + Ok(s) => s, + Err(e) => { + eprintln!( + "error: cannot connect to {}: {}", + socket_path.display(), + e + ); + eprintln!("Is the FIPS daemon running?"); + std::process::exit(1); + } + }; + + // Set timeouts + let timeout = Duration::from_secs(2); + let _ = stream.set_read_timeout(Some(timeout)); + let _ = stream.set_write_timeout(Some(timeout)); + + // Send request + let request = format!("{{\"command\":\"{}\"}}\n", command_name); + if let Err(e) = stream.write_all(request.as_bytes()) { + eprintln!("error: failed to send request: {}", e); + std::process::exit(1); + } + + // Shutdown write half to signal end of request + let _ = stream.shutdown(std::net::Shutdown::Write); + + // Read response + let reader = BufReader::new(&stream); + let response_line = match reader.lines().next() { + Some(Ok(line)) => line, + Some(Err(e)) => { + eprintln!("error: failed to read response: {}", e); + std::process::exit(1); + } + None => { + eprintln!("error: no response from daemon"); + std::process::exit(1); + } + }; + + // Parse and pretty-print + let Ok(value) = serde_json::from_str::(&response_line) else { + // Not JSON, print raw + println!("{}", response_line); + return; + }; + + let status = value + .get("status") + .and_then(|v| v.as_str()) + .unwrap_or("unknown"); + + if status == "error" { + let msg = value + .get("message") + .and_then(|v| v.as_str()) + .unwrap_or("unknown error"); + eprintln!("error: {}", msg); + std::process::exit(1); + } + + // Pretty-print the data field (or whole response if no data field) + let output = if let Some(data) = value.get("data") { + serde_json::to_string_pretty(data).unwrap_or(response_line) + } else { + serde_json::to_string_pretty(&value).unwrap_or(response_line) + }; + println!("{}", output); +} diff --git a/src/config/mod.rs b/src/config/mod.rs index 5e51955..bbe37e5 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -29,8 +29,8 @@ use std::path::{Path, PathBuf}; use thiserror::Error; pub use node::{ - BloomConfig, BuffersConfig, CacheConfig, DiscoveryConfig, LimitsConfig, NodeConfig, - RateLimitConfig, RetryConfig, SessionConfig, SessionMmpConfig, TreeConfig, + BloomConfig, BuffersConfig, CacheConfig, ControlConfig, DiscoveryConfig, LimitsConfig, + NodeConfig, RateLimitConfig, RetryConfig, SessionConfig, SessionMmpConfig, TreeConfig, }; pub use peer::{ConnectPolicy, PeerAddress, PeerConfig}; pub use transport::{TransportInstances, TransportsConfig, UdpConfig}; diff --git a/src/config/node.rs b/src/config/node.rs index 62e8add..42e4da0 100644 --- a/src/config/node.rs +++ b/src/config/node.rs @@ -314,6 +314,38 @@ impl SessionMmpConfig { fn default_owd_window_size() -> usize { DEFAULT_OWD_WINDOW_SIZE } } +/// Control socket configuration (`node.control.*`). +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ControlConfig { + /// Enable the control socket (`node.control.enabled`). + #[serde(default = "ControlConfig::default_enabled")] + pub enabled: bool, + /// Unix socket path (`node.control.socket_path`). + #[serde(default = "ControlConfig::default_socket_path")] + pub socket_path: String, +} + +impl Default for ControlConfig { + fn default() -> Self { + Self { + enabled: true, + socket_path: Self::default_socket_path(), + } + } +} + +impl ControlConfig { + fn default_enabled() -> bool { true } + + fn default_socket_path() -> String { + if let Ok(runtime_dir) = std::env::var("XDG_RUNTIME_DIR") { + format!("{}/fips/control.sock", runtime_dir) + } else { + "/tmp/fips-control.sock".to_string() + } + } +} + /// Internal buffers (`node.buffers.*`). #[derive(Debug, Clone, Serialize, Deserialize)] pub struct BuffersConfig { @@ -412,6 +444,10 @@ pub struct NodeConfig { #[serde(default)] pub buffers: BuffersConfig, + /// Control socket (`node.control.*`). + #[serde(default)] + pub control: ControlConfig, + /// Metrics Measurement Protocol — link layer (`node.mmp.*`). #[serde(default)] pub mmp: MmpConfig, @@ -439,6 +475,7 @@ impl Default for NodeConfig { bloom: BloomConfig::default(), session: SessionConfig::default(), buffers: BuffersConfig::default(), + control: ControlConfig::default(), mmp: MmpConfig::default(), session_mmp: SessionMmpConfig::default(), } diff --git a/src/control/mod.rs b/src/control/mod.rs new file mode 100644 index 0000000..b09c11d --- /dev/null +++ b/src/control/mod.rs @@ -0,0 +1,216 @@ +//! Control socket for runtime observability. +//! +//! Provides a Unix domain socket that accepts query commands and returns +//! structured JSON data about the node's current state. Read-only queries +//! only — no state mutation through this channel. + +pub mod protocol; +pub mod queries; + +use crate::config::ControlConfig; +use protocol::{Request, Response}; +use std::path::{Path, PathBuf}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::UnixListener; +use tokio::sync::{mpsc, oneshot}; +use tracing::{debug, info, warn}; + +/// Maximum request size in bytes (4 KB). +const MAX_REQUEST_SIZE: usize = 4096; + +/// I/O timeout for client connections. +const IO_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); + +/// A message sent from the accept loop to the main event loop. +pub type ControlMessage = (Request, oneshot::Sender); + +/// Control socket listener. +/// +/// Manages the Unix domain socket lifecycle: bind, accept, cleanup. +pub struct ControlSocket { + listener: UnixListener, + socket_path: PathBuf, +} + +impl ControlSocket { + /// Bind a new control socket. + /// + /// Creates parent directories if needed, removes stale socket files, + /// and binds the Unix listener. + pub fn bind(config: &ControlConfig) -> Result { + let socket_path = PathBuf::from(&config.socket_path); + + // Create parent directory if it doesn't exist + if let Some(parent) = socket_path.parent() + && !parent.exists() + { + std::fs::create_dir_all(parent)?; + debug!(path = %parent.display(), "Created control socket directory"); + } + + // Remove stale socket if it exists + if socket_path.exists() { + Self::remove_stale_socket(&socket_path)?; + } + + let listener = UnixListener::bind(&socket_path)?; + info!(path = %socket_path.display(), "Control socket listening"); + + Ok(Self { + listener, + socket_path, + }) + } + + /// Remove a stale socket file. + /// + /// If the file exists but no one is listening, remove it so we can + /// bind. This handles unclean daemon exits. + fn remove_stale_socket(path: &Path) -> Result<(), std::io::Error> { + // Try connecting to see if someone is listening + match std::os::unix::net::UnixStream::connect(path) { + Ok(_) => { + // Someone is listening — don't remove it + Err(std::io::Error::new( + std::io::ErrorKind::AddrInUse, + format!("control socket already in use: {}", path.display()), + )) + } + Err(_) => { + // No one listening — remove the stale socket + debug!(path = %path.display(), "Removing stale control socket"); + std::fs::remove_file(path)?; + Ok(()) + } + } + } + + /// Run the accept loop, forwarding requests to the main event loop via mpsc. + /// + /// Each accepted connection is handled in a spawned task: + /// 1. Read one line of JSON (the request) + /// 2. Send (Request, oneshot::Sender) to the main loop + /// 3. Wait for the response via oneshot + /// 4. Write the response as one line of JSON + /// 5. Close the connection + pub async fn accept_loop(self, control_tx: mpsc::Sender) { + loop { + let (stream, _addr) = match self.listener.accept().await { + Ok(conn) => conn, + Err(e) => { + warn!(error = %e, "Control socket accept failed"); + continue; + } + }; + + let tx = control_tx.clone(); + tokio::spawn(async move { + if let Err(e) = Self::handle_connection(stream, tx).await { + debug!(error = %e, "Control connection error"); + } + }); + } + } + + /// Handle a single client connection. + async fn handle_connection( + stream: tokio::net::UnixStream, + control_tx: mpsc::Sender, + ) -> Result<(), Box> { + let (reader, mut writer) = stream.into_split(); + let mut buf_reader = BufReader::new(reader); + let mut line = String::new(); + + // Read one line with timeout and size limit + let read_result = tokio::time::timeout(IO_TIMEOUT, async { + let mut total = 0usize; + loop { + let n = buf_reader.read_line(&mut line).await?; + if n == 0 { + break; // EOF + } + total += n; + if total > MAX_REQUEST_SIZE { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "request too large", + )); + } + if line.ends_with('\n') { + break; + } + } + Ok(()) + }) + .await; + + let response = match read_result { + Ok(Ok(())) if line.is_empty() => { + Response::error("empty request") + } + Ok(Ok(())) => { + // Parse the request + match serde_json::from_str::(line.trim()) { + Ok(request) => { + // Send to main loop and wait for response + let (resp_tx, resp_rx) = oneshot::channel(); + if control_tx.send((request, resp_tx)).await.is_err() { + Response::error("node shutting down") + } else { + match tokio::time::timeout(IO_TIMEOUT, resp_rx).await { + Ok(Ok(resp)) => resp, + Ok(Err(_)) => Response::error("response channel closed"), + Err(_) => Response::error("query timeout"), + } + } + } + Err(e) => Response::error(format!("invalid request: {}", e)), + } + } + Ok(Err(e)) => Response::error(format!("read error: {}", e)), + Err(_) => Response::error("read timeout"), + }; + + // Write response with timeout + let json = serde_json::to_string(&response)?; + let write_result = tokio::time::timeout(IO_TIMEOUT, async { + writer.write_all(json.as_bytes()).await?; + writer.write_all(b"\n").await?; + writer.shutdown().await?; + Ok::<_, std::io::Error>(()) + }) + .await; + + if let Err(_) | Ok(Err(_)) = write_result { + debug!("Control socket write failed or timed out"); + } + + Ok(()) + } + + /// Get the socket path. + pub fn socket_path(&self) -> &Path { + &self.socket_path + } + + /// Clean up the socket file. + fn cleanup(&self) { + if self.socket_path.exists() { + if let Err(e) = std::fs::remove_file(&self.socket_path) { + warn!( + path = %self.socket_path.display(), + error = %e, + "Failed to remove control socket" + ); + } else { + debug!(path = %self.socket_path.display(), "Control socket removed"); + } + } + } +} + +impl Drop for ControlSocket { + fn drop(&mut self) { + self.cleanup(); + } +} diff --git a/src/control/protocol.rs b/src/control/protocol.rs new file mode 100644 index 0000000..081eae7 --- /dev/null +++ b/src/control/protocol.rs @@ -0,0 +1,90 @@ +//! Control socket protocol types. +//! +//! Line-delimited JSON protocol for the Unix domain socket. +//! Each request is one JSON line, each response is one JSON line. + +use serde::{Deserialize, Serialize}; + +/// A control request from a client. +#[derive(Debug, Deserialize)] +pub struct Request { + /// The command to execute (e.g., "show_status", "show_peers"). + pub command: String, +} + +/// A control response to a client. +#[derive(Debug, Serialize)] +pub struct Response { + /// "ok" or "error". + pub status: String, + /// Response data (present on success). + #[serde(skip_serializing_if = "Option::is_none")] + pub data: Option, + /// Error message (present on failure). + #[serde(skip_serializing_if = "Option::is_none")] + pub message: Option, +} + +impl Response { + /// Create a success response with data. + pub fn ok(data: serde_json::Value) -> Self { + Self { + status: "ok".to_string(), + data: Some(data), + message: None, + } + } + + /// Create an error response with a message. + pub fn error(message: impl Into) -> Self { + Self { + status: "error".to_string(), + data: None, + message: Some(message.into()), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_deserialize_request() { + let json = r#"{"command": "show_status"}"#; + let req: Request = serde_json::from_str(json).unwrap(); + assert_eq!(req.command, "show_status"); + } + + #[test] + fn test_serialize_ok_response() { + let resp = Response::ok(serde_json::json!({"state": "running"})); + let json = serde_json::to_string(&resp).unwrap(); + assert!(json.contains("\"status\":\"ok\"")); + assert!(json.contains("\"state\":\"running\"")); + assert!(!json.contains("\"message\"")); + } + + #[test] + fn test_serialize_error_response() { + let resp = Response::error("unknown command: foo"); + let json = serde_json::to_string(&resp).unwrap(); + assert!(json.contains("\"status\":\"error\"")); + assert!(json.contains("unknown command: foo")); + assert!(!json.contains("\"data\"")); + } + + #[test] + fn test_deserialize_unknown_fields_ignored() { + let json = r#"{"command": "show_peers", "extra": true}"#; + let req: Request = serde_json::from_str(json).unwrap(); + assert_eq!(req.command, "show_peers"); + } + + #[test] + fn test_deserialize_malformed_request() { + let json = r#"{"not_command": "foo"}"#; + let result: Result = serde_json::from_str(json); + assert!(result.is_err()); + } +} diff --git a/src/control/queries.rs b/src/control/queries.rs new file mode 100644 index 0000000..edcf760 --- /dev/null +++ b/src/control/queries.rs @@ -0,0 +1,414 @@ +//! Control query implementations. +//! +//! Each function takes `&Node` and returns a `serde_json::Value`. +//! Query logic is kept separate from socket handling. + +use crate::node::Node; +use serde_json::{json, Value}; + +/// Helper: get current Unix time in milliseconds. +fn now_ms() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +/// Classify a DualEwma trend as "rising", "falling", or "stable". +fn trend_label(short: f64, long: f64) -> &'static str { + if !short.is_finite() || !long.is_finite() || long == 0.0 { + return "stable"; + } + let ratio = short / long; + if ratio > 1.05 { + "rising" + } else if ratio < 0.95 { + "falling" + } else { + "stable" + } +} + +/// `show_status` — Node overview. +pub fn show_status(node: &Node) -> Value { + json!({ + "npub": node.npub(), + "node_addr": hex::encode(node.node_addr().as_bytes()), + "fips_address": format!("{}", node.identity().address()), + "state": format!("{}", node.state()), + "is_leaf_only": node.is_leaf_only(), + "peer_count": node.peer_count(), + "session_count": node.session_count(), + "link_count": node.link_count(), + "transport_count": node.transport_count(), + "connection_count": node.connection_count(), + "tun_state": format!("{}", node.tun_state()), + "effective_ipv6_mtu": node.effective_ipv6_mtu(), + }) +} + +/// `show_peers` — Authenticated peers. +pub fn show_peers(node: &Node) -> Value { + let peers: Vec = node.peers().map(|peer| { + let node_addr = *peer.node_addr(); + let addr_hex = hex::encode(node_addr.as_bytes()); + + let mut peer_json = json!({ + "node_addr": addr_hex, + "npub": peer.npub(), + "display_name": node.peer_display_name(&node_addr), + "fips_address": format!("{}", peer.address()), + "connectivity": format!("{}", peer.connectivity()), + "link_id": peer.link_id().as_u64(), + "authenticated_at_ms": peer.authenticated_at(), + "last_seen_ms": peer.last_seen(), + "has_tree_position": peer.has_tree_position(), + "has_bloom_filter": peer.filter_sequence() > 0, + "filter_sequence": peer.filter_sequence(), + }); + + // Add transport address if available + if let Some(addr) = peer.current_addr() { + peer_json["transport_addr"] = json!(format!("{}", addr)); + } + + // Add tree depth if available + if let Some(coords) = peer.coords() { + peer_json["tree_depth"] = json!(coords.depth()); + } + + // Add link stats + let stats = peer.link_stats(); + peer_json["stats"] = json!({ + "packets_sent": stats.packets_sent, + "packets_recv": stats.packets_recv, + "bytes_sent": stats.bytes_sent, + "bytes_recv": stats.bytes_recv, + }); + + // Add MMP metrics if available + if let Some(mmp) = peer.mmp() { + let mut mmp_json = json!({ + "mode": format!("{}", mmp.mode()), + }); + if let Some(srtt) = mmp.metrics.srtt_ms() { + mmp_json["srtt_ms"] = json!(srtt); + } + mmp_json["loss_rate"] = json!(mmp.metrics.loss_rate()); + mmp_json["etx"] = json!(mmp.metrics.etx); + mmp_json["goodput_bps"] = json!(mmp.metrics.goodput_bps); + mmp_json["delivery_ratio_forward"] = json!(mmp.metrics.delivery_ratio_forward); + mmp_json["delivery_ratio_reverse"] = json!(mmp.metrics.delivery_ratio_reverse); + peer_json["mmp"] = mmp_json; + } + + peer_json + }).collect(); + + json!({ "peers": peers }) +} + +/// `show_links` — Active links. +pub fn show_links(node: &Node) -> Value { + let links: Vec = node.links().map(|link| { + let stats = link.stats(); + json!({ + "link_id": link.link_id().as_u64(), + "transport_id": link.transport_id().as_u32(), + "remote_addr": format!("{}", link.remote_addr()), + "direction": format!("{}", link.direction()), + "state": format!("{}", link.state()), + "created_at_ms": link.created_at(), + "stats": { + "packets_sent": stats.packets_sent, + "packets_recv": stats.packets_recv, + "bytes_sent": stats.bytes_sent, + "bytes_recv": stats.bytes_recv, + "last_recv_ms": stats.last_recv_ms, + }, + }) + }).collect(); + + json!({ "links": links }) +} + +/// `show_tree` — Spanning tree state. +pub fn show_tree(node: &Node) -> Value { + let tree = node.tree_state(); + let my_coords = tree.my_coords(); + let decl = tree.my_declaration(); + + // Build coords array as hex strings + let coords: Vec = my_coords.entries() + .iter() + .map(|e| hex::encode(e.node_addr.as_bytes())) + .collect(); + + // Build peer tree data + let peers: Vec = tree.peer_ids().map(|peer_id| { + let mut peer_json = json!({ + "node_addr": hex::encode(peer_id.as_bytes()), + "display_name": node.peer_display_name(peer_id), + }); + if let Some(coords) = tree.peer_coords(peer_id) { + peer_json["depth"] = json!(coords.depth()); + peer_json["root"] = json!(hex::encode(coords.root_id().as_bytes())); + peer_json["distance_to_us"] = json!(my_coords.distance_to(coords)); + } + peer_json + }).collect(); + + // Determine parent display name + let parent_addr = my_coords.parent_id(); + let parent_hex = hex::encode(parent_addr.as_bytes()); + let parent_display = node.peer_display_name(parent_addr); + + json!({ + "my_node_addr": hex::encode(tree.my_node_addr().as_bytes()), + "root": hex::encode(tree.root().as_bytes()), + "is_root": tree.is_root(), + "depth": my_coords.depth(), + "my_coords": coords, + "parent": parent_hex, + "parent_display_name": parent_display, + "declaration_sequence": decl.sequence(), + "declaration_signed": decl.is_signed(), + "peer_tree_count": tree.peer_count(), + "peers": peers, + }) +} + +/// `show_sessions` — End-to-end sessions. +pub fn show_sessions(node: &Node) -> Value { + let sessions: Vec = node.session_entries().map(|(addr, entry)| { + let state_str = if entry.is_established() { + "established" + } else if entry.is_initiating() { + "initiating" + } else if entry.is_responding() { + "responding" + } else { + "unknown" + }; + + let mut session_json = json!({ + "remote_addr": hex::encode(addr.as_bytes()), + "display_name": node.peer_display_name(addr), + "state": state_str, + "is_initiator": entry.is_initiator(), + "last_activity_ms": entry.last_activity(), + }); + + // Add session MMP if available + if let Some(mmp) = entry.mmp() { + let mut mmp_json = json!({ + "mode": format!("{}", mmp.mode()), + }); + if let Some(srtt) = mmp.metrics.srtt_ms() { + mmp_json["srtt_ms"] = json!(srtt); + } + mmp_json["path_mtu"] = json!(mmp.path_mtu.current_mtu()); + session_json["mmp"] = mmp_json; + } + + session_json + }).collect(); + + json!({ "sessions": sessions }) +} + +/// `show_bloom` — Bloom filter state. +pub fn show_bloom(node: &Node) -> Value { + let bloom = node.bloom_state(); + + let leaf_deps: Vec = bloom.leaf_dependents() + .iter() + .map(|addr| hex::encode(addr.as_bytes())) + .collect(); + + // Build per-peer filter info + let peer_filters: Vec = node.peers().map(|peer| { + let addr = *peer.node_addr(); + json!({ + "peer": hex::encode(addr.as_bytes()), + "display_name": node.peer_display_name(&addr), + "has_filter": peer.filter_sequence() > 0, + "filter_sequence": peer.filter_sequence(), + }) + }).collect(); + + json!({ + "own_node_addr": hex::encode(node.node_addr().as_bytes()), + "is_leaf_only": node.is_leaf_only(), + "sequence": bloom.sequence(), + "leaf_dependent_count": bloom.leaf_dependents().len(), + "leaf_dependents": leaf_deps, + "peer_filters": peer_filters, + }) +} + +/// `show_mmp` — MMP metrics summary. +pub fn show_mmp(node: &Node) -> Value { + // Link-layer MMP per peer + let peers: Vec = node.peers().filter_map(|peer| { + let mmp = peer.mmp()?; + let addr = *peer.node_addr(); + let metrics = &mmp.metrics; + + let mut link_layer = json!({ + "loss_rate": metrics.loss_rate(), + "etx": metrics.etx, + "goodput_bps": metrics.goodput_bps, + "spin_bit_role": if mmp.spin_bit.is_initiator() { "initiator" } else { "responder" }, + }); + + if let Some(srtt) = metrics.srtt_ms() { + link_layer["srtt_ms"] = json!(srtt); + } + + // Trend indicators + if metrics.rtt_trend.initialized() { + link_layer["rtt_trend"] = json!(trend_label(metrics.rtt_trend.short(), metrics.rtt_trend.long())); + } + if metrics.loss_trend.initialized() { + link_layer["loss_trend"] = json!(trend_label(metrics.loss_trend.short(), metrics.loss_trend.long())); + } + if metrics.goodput_trend.initialized() { + link_layer["goodput_trend"] = json!(trend_label(metrics.goodput_trend.short(), metrics.goodput_trend.long())); + } + if metrics.jitter_trend.initialized() { + link_layer["jitter_trend"] = json!(trend_label(metrics.jitter_trend.short(), metrics.jitter_trend.long())); + } + + link_layer["delivery_ratio_forward"] = json!(metrics.delivery_ratio_forward); + link_layer["delivery_ratio_reverse"] = json!(metrics.delivery_ratio_reverse); + + Some(json!({ + "peer": hex::encode(addr.as_bytes()), + "display_name": node.peer_display_name(&addr), + "mode": format!("{}", mmp.mode()), + "link_layer": link_layer, + })) + }).collect(); + + // Session-layer MMP + let sessions: Vec = node.session_entries().filter_map(|(addr, entry)| { + let mmp = entry.mmp()?; + let metrics = &mmp.metrics; + + let mut session_layer = json!({ + "loss_rate": metrics.loss_rate(), + "etx": metrics.etx, + "path_mtu": mmp.path_mtu.current_mtu(), + }); + + if let Some(srtt) = metrics.srtt_ms() { + session_layer["srtt_ms"] = json!(srtt); + } + + Some(json!({ + "remote": hex::encode(addr.as_bytes()), + "display_name": node.peer_display_name(addr), + "mode": format!("{}", mmp.mode()), + "session_layer": session_layer, + })) + }).collect(); + + json!({ + "peers": peers, + "sessions": sessions, + }) +} + +/// `show_cache` — Coordinate cache stats. +pub fn show_cache(node: &Node) -> Value { + let cache = node.coord_cache(); + let stats = cache.stats(now_ms()); + + json!({ + "entries": stats.entries, + "max_entries": stats.max_entries, + "fill_ratio": stats.fill_ratio(), + "default_ttl_ms": cache.default_ttl_ms(), + "expired": stats.expired, + "avg_age_ms": stats.avg_age_ms, + }) +} + +/// `show_connections` — Pending handshakes. +pub fn show_connections(node: &Node) -> Value { + let now = now_ms(); + let connections: Vec = node.connections().map(|conn| { + let mut conn_json = json!({ + "link_id": conn.link_id().as_u64(), + "direction": format!("{}", conn.direction()), + "handshake_state": format!("{}", conn.handshake_state()), + "started_at_ms": conn.started_at(), + "idle_ms": now.saturating_sub(conn.last_activity()), + "resend_count": conn.resend_count(), + }); + + if let Some(identity) = conn.expected_identity() { + conn_json["expected_peer"] = json!(identity.npub()); + } + + conn_json + }).collect(); + + json!({ "connections": connections }) +} + +/// `show_transports` — Transport instances. +pub fn show_transports(node: &Node) -> Value { + let transports: Vec = node.transport_ids().map(|id| { + let handle = node.get_transport(id).unwrap(); + let mut t_json = json!({ + "transport_id": id.as_u32(), + "type": handle.transport_type().name, + "state": format!("{}", handle.state()), + "mtu": handle.mtu(), + }); + + if let Some(name) = handle.name() { + t_json["name"] = json!(name); + } + if let Some(addr) = handle.local_addr() { + t_json["local_addr"] = json!(format!("{}", addr)); + } + + t_json + }).collect(); + + json!({ "transports": transports }) +} + +/// `show_routing` — Routing table summary. +pub fn show_routing(node: &Node) -> Value { + let cache = node.coord_cache(); + let stats = cache.stats(now_ms()); + + json!({ + "coord_cache_entries": stats.entries, + "identity_cache_entries": node.identity_cache_len(), + "pending_lookups": node.pending_lookup_count(), + "recent_requests": node.recent_request_count(), + }) +} + +/// Dispatch a command string to the appropriate query function. +pub fn dispatch(node: &Node, command: &str) -> super::protocol::Response { + match command { + "show_status" => super::protocol::Response::ok(show_status(node)), + "show_peers" => super::protocol::Response::ok(show_peers(node)), + "show_links" => super::protocol::Response::ok(show_links(node)), + "show_tree" => super::protocol::Response::ok(show_tree(node)), + "show_sessions" => super::protocol::Response::ok(show_sessions(node)), + "show_bloom" => super::protocol::Response::ok(show_bloom(node)), + "show_mmp" => super::protocol::Response::ok(show_mmp(node)), + "show_cache" => super::protocol::Response::ok(show_cache(node)), + "show_connections" => super::protocol::Response::ok(show_connections(node)), + "show_transports" => super::protocol::Response::ok(show_transports(node)), + "show_routing" => super::protocol::Response::ok(show_routing(node)), + _ => super::protocol::Response::error(format!("unknown command: {}", command)), + } +} diff --git a/src/lib.rs b/src/lib.rs index 5a80327..3fda2c3 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -6,6 +6,7 @@ pub mod bloom; pub mod cache; pub mod config; +pub mod control; pub mod identity; pub mod mmp; pub mod noise; diff --git a/src/mmp/algorithms.rs b/src/mmp/algorithms.rs index 521e766..75d4ce8 100644 --- a/src/mmp/algorithms.rs +++ b/src/mmp/algorithms.rs @@ -299,6 +299,11 @@ impl SpinBitState { } } + /// Check if this is the spin bit initiator. + pub fn is_initiator(&self) -> bool { + self.is_initiator + } + /// Get the spin bit value to set on an outgoing frame. pub fn tx_bit(&self) -> bool { self.current_value diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index b9ba1ce..a52ddaa 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -1,10 +1,12 @@ //! RX event loop and packet dispatch. +use crate::control::ControlSocket; +use crate::control::queries; use crate::node::{Node, NodeError}; use crate::transport::ReceivedPacket; use crate::node::wire::{CommonPrefix, PHASE_ESTABLISHED, PHASE_MSG1, PHASE_MSG2, FMP_VERSION, COMMON_PREFIX_SIZE}; use std::time::Duration; -use tracing::{debug, info}; +use tracing::{debug, info, warn}; impl Node { /// Run the receive event loop. @@ -53,6 +55,28 @@ impl Node { let mut tick = tokio::time::interval(Duration::from_secs(self.config.node.tick_interval_secs)); + // Set up control socket channel + let (control_tx, mut control_rx) = tokio::sync::mpsc::channel::< + crate::control::ControlMessage, + >(32); + + if self.config.node.control.enabled { + let config = self.config.node.control.clone(); + let tx = control_tx.clone(); + tokio::spawn(async move { + match ControlSocket::bind(&config) { + Ok(socket) => { + socket.accept_loop(tx).await; + } + Err(e) => { + warn!(error = %e, "Failed to bind control socket"); + } + } + }); + } + // Drop unused sender to avoid keeping channel open if control is disabled + drop(control_tx); + info!("RX event loop started"); loop { @@ -73,6 +97,10 @@ impl Node { ); self.register_identity(identity.node_addr, identity.pubkey); } + Some((request, response_tx)) = control_rx.recv() => { + let response = queries::dispatch(self, &request.command); + let _ = response_tx.send(response); + } _ = tick.tick() => { self.check_timeouts(); let now_ms = std::time::SystemTime::now() diff --git a/src/node/mod.rs b/src/node/mod.rs index 438238c..24e65ca 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -919,6 +919,11 @@ impl Node { self.sessions.len() } + /// Iterate over all session entries (for control queries). + pub(crate) fn session_entries(&self) -> impl Iterator { + self.sessions.iter() + } + // === Identity Cache === /// Register a node in the identity cache for FipsAddress → NodeAddr lookup. @@ -952,6 +957,16 @@ impl Node { self.identity_cache.len() } + /// Number of pending discovery lookups. + pub fn pending_lookup_count(&self) -> usize { + self.pending_lookups.len() + } + + /// Number of recent discovery requests tracked. + pub fn recent_request_count(&self) -> usize { + self.recent_requests.len() + } + // === Routing === /// Find next hop for a destination node address.