From 5053cf673d0ccd652d6bf1ace6a0ee4f9b206b6a Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Fri, 20 Mar 2026 05:44:31 +0000 Subject: [PATCH] Add connect/disconnect control commands and maelstrom chaos scenario Add runtime peer management to the FIPS daemon via control socket commands, and a new chaos simulation scenario that exercises dynamic topology mutation with ephemeral node identities. Daemon (connect/disconnect commands): - Extend control socket Request with optional params field - Add commands.rs module for mutating command dispatch, separate from read-only queries - Add api_connect() on Node: builds ephemeral PeerConfig (no auto- reconnect), pre-seeds identity cache, reuses initiate_peer_connection - Add api_disconnect() on Node: calls remove_active_peer(), clears retry_pending to suppress reconnection - Route non-show_* commands to async command dispatch in rx_loop fipsctl CLI: - Add Connect and Disconnect subcommands accepting npub or hostname - Resolve hostnames from /etc/fips/hosts before sending to daemon - Refactor socket I/O into reusable send_request helper Chaos simulator (maelstrom scenario): - Add PeerChurnManager: periodically disconnects a random active link and connects a random unconnected node pair via control socket - Add send_command() to control.py using base64-encoded JSON payloads to avoid shell quoting issues in docker exec - Add PeerChurnConfig to scenario with interval and ephemeral_fraction - Ephemeral identity support: nodes configured without nsec generate fresh keypairs on restart; simulator queries show_status for new npub and updates its cache via on_node_restart callback - Add maelstrom.yaml: all chaos dimensions (netem, link flaps, node churn, peer topology churn, traffic) with 50% ephemeral identity --- src/bin/fipsctl.rs | 282 +++++++++++++++---------- src/control/commands.rs | 68 ++++++ src/control/mod.rs | 9 +- src/control/protocol.rs | 23 +- src/node/handlers/rx_loop.rs | 12 +- src/node/lifecycle.rs | 72 +++++++ testing/chaos/scenarios/maelstrom.yaml | 76 +++++++ testing/chaos/sim/config_gen.py | 19 +- testing/chaos/sim/control.py | 46 ++++ testing/chaos/sim/nodes.py | 6 + testing/chaos/sim/peer_churn.py | 178 ++++++++++++++++ testing/chaos/sim/runner.py | 64 +++++- testing/chaos/sim/scenario.py | 26 +++ 13 files changed, 757 insertions(+), 124 deletions(-) create mode 100644 src/control/commands.rs create mode 100644 testing/chaos/scenarios/maelstrom.yaml create mode 100644 testing/chaos/sim/peer_churn.py diff --git a/src/bin/fipsctl.rs b/src/bin/fipsctl.rs index e30f44f..c2ca2aa 100644 --- a/src/bin/fipsctl.rs +++ b/src/bin/fipsctl.rs @@ -1,10 +1,11 @@ //! fipsctl — FIPS control client //! -//! Connects to the FIPS daemon's Unix domain control socket, sends a -//! query command, and pretty-prints the JSON response. +//! Connects to the FIPS daemon's Unix domain control socket, sends +//! commands, and pretty-prints the JSON response. use clap::{Parser, Subcommand}; use fips::config::{write_key_file, write_pub_file}; +use fips::upper::hosts::HostMap; use fips::version; use fips::{encode_nsec, Identity}; use std::io::{BufRead, BufReader, Write}; @@ -18,7 +19,7 @@ use std::time::Duration; name = "fipsctl", version = version::short_version(), long_version = version::long_version(), - about = "Query a running FIPS daemon" + about = "Control a running FIPS daemon" )] struct Cli { /// Control socket path override @@ -48,6 +49,20 @@ enum Commands { #[arg(short = 's', long = "stdout")] stdout: bool, }, + /// Connect to a peer + Connect { + /// Peer identifier: npub (bech32) or hostname from /etc/fips/hosts + peer: String, + /// Transport address (e.g., "192.168.1.1:2121") + address: String, + /// Transport type: udp, tcp, tor, ethernet + transport: String, + }, + /// Disconnect a peer + Disconnect { + /// Peer identifier: npub (bech32) or hostname from /etc/fips/hosts + peer: String, + }, } #[derive(Subcommand, Debug)] @@ -108,123 +123,58 @@ fn default_socket_path() -> PathBuf { } } -fn main() { - let cli = Cli::parse(); - - // Commands that don't require a running daemon - match &cli.command { - Commands::Keygen { - dir, - force, - stdout, - } => { - let identity = Identity::generate(); - let nsec = encode_nsec(&identity.keypair().secret_key()); - let npub = identity.npub(); - - if *stdout { - println!("{}", nsec); - println!("{}", npub); - return; - } - - let key_path = dir.join("fips.key"); - let pub_path = dir.join("fips.pub"); - - if key_path.exists() && !force { - eprintln!("error: key file already exists: {}", key_path.display()); - eprintln!("Use --force to overwrite."); - std::process::exit(1); - } - - if let Err(e) = std::fs::create_dir_all(dir) { - eprintln!("error: cannot create directory {}: {}", dir.display(), e); - std::process::exit(1); - } - - if let Err(e) = write_key_file(&key_path, &nsec) { - eprintln!("error: failed to write key file: {}", e); - std::process::exit(1); - } - - if let Err(e) = write_pub_file(&pub_path, &npub) { - eprintln!("error: failed to write pub file: {}", e); - std::process::exit(1); - } - - eprintln!("{}", npub); - eprintln!("Key files written to: {}/", dir.display()); - eprintln!(); - eprintln!("NOTE: Set 'node.identity.persistent: true' in fips.yaml"); - eprintln!(" or these keys will be overwritten on next daemon start."); - return; - } - Commands::Show { .. } => {} - } - - let socket_path = cli.socket.unwrap_or_else(default_socket_path); - let command_name = match &cli.command { - Commands::Show { what } => what.command_name(), - Commands::Keygen { .. } => unreachable!(), - }; - - // Connect to the control socket - let mut stream = match UnixStream::connect(&socket_path) { - Ok(s) => s, - Err(e) => { - eprintln!( - "error: cannot connect to {}: {}", +/// Send a JSON request to the control socket and return the response. +fn send_request(socket_path: &Path, request_json: &str) -> Result { + let mut stream = UnixStream::connect(socket_path).map_err(|e| { + if e.kind() == std::io::ErrorKind::PermissionDenied { + format!( + "cannot connect to {}: {}\n\ + Hint: add your user to the 'fips' group: sudo usermod -aG fips $USER\n\ + Then log out and back in for the change to take effect.", socket_path.display(), e - ); - if e.kind() == std::io::ErrorKind::PermissionDenied { - eprintln!( - "Hint: add your user to the 'fips' group: sudo usermod -aG fips $USER" - ); - eprintln!("Then log out and back in for the change to take effect."); - } else { - eprintln!("Is the FIPS daemon running?"); - } - std::process::exit(1); + ) + } else { + format!( + "cannot connect to {}: {}\nIs the FIPS daemon running?", + socket_path.display(), + e + ) } - }; + })?; - // Set timeouts - let timeout = Duration::from_secs(2); + let timeout = Duration::from_secs(5); 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 + stream + .write_all(request_json.as_bytes()) + .map_err(|e| format!("failed to send request: {e}"))?; 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); - } - }; + let line = reader + .lines() + .next() + .ok_or("no response from daemon")? + .map_err(|e| format!("failed to read response: {e}"))?; - // Parse and pretty-print - let Ok(value) = serde_json::from_str::(&response_line) else { - // Not JSON, print raw - println!("{}", response_line); - return; - }; + serde_json::from_str(&line).map_err(|e| format!("invalid response JSON: {e}")) +} +/// Build a request JSON string for a simple command (no params). +fn build_query(command: &str) -> String { + format!("{{\"command\":\"{command}\"}}\n") +} + +/// Build a request JSON string for a command with params. +fn build_command(command: &str, params: serde_json::Value) -> String { + let req = serde_json::json!({"command": command, "params": params}); + format!("{}\n", serde_json::to_string(&req).unwrap()) +} + +/// Print a control socket response, handling error status. +fn print_response(value: &serde_json::Value) { let status = value .get("status") .and_then(|v| v.as_str()) @@ -235,15 +185,121 @@ fn main() { .get("message") .and_then(|v| v.as_str()) .unwrap_or("unknown error"); - eprintln!("error: {}", msg); + 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) + serde_json::to_string_pretty(data) } else { - serde_json::to_string_pretty(&value).unwrap_or(response_line) + serde_json::to_string_pretty(value) }; - println!("{}", output); + println!("{}", output.unwrap_or_else(|_| format!("{value}"))); +} + +/// Resolve a peer identifier to an npub. +/// +/// If the identifier starts with "npub1", it's returned as-is. +/// Otherwise, it's looked up as a hostname in /etc/fips/hosts. +fn resolve_peer(peer: &str) -> String { + if peer.starts_with("npub1") { + return peer.to_string(); + } + + let hosts = HostMap::load_hosts_file(Path::new(fips::upper::hosts::DEFAULT_HOSTS_PATH)); + match hosts.lookup_npub(peer) { + Some(npub) => npub.to_string(), + None => { + eprintln!("error: unknown host '{peer}'"); + eprintln!("Not found in /etc/fips/hosts and not an npub."); + std::process::exit(1); + } + } +} + +fn main() { + let cli = Cli::parse(); + + // Commands that don't require a running daemon + if let Commands::Keygen { + dir, + force, + stdout, + } = &cli.command + { + let identity = Identity::generate(); + let nsec = encode_nsec(&identity.keypair().secret_key()); + let npub = identity.npub(); + + if *stdout { + println!("{nsec}"); + println!("{npub}"); + return; + } + + let key_path = dir.join("fips.key"); + let pub_path = dir.join("fips.pub"); + + if key_path.exists() && !force { + eprintln!("error: key file already exists: {}", key_path.display()); + eprintln!("Use --force to overwrite."); + std::process::exit(1); + } + + if let Err(e) = std::fs::create_dir_all(dir) { + eprintln!("error: cannot create directory {}: {e}", dir.display()); + std::process::exit(1); + } + + if let Err(e) = write_key_file(&key_path, &nsec) { + eprintln!("error: failed to write key file: {e}"); + std::process::exit(1); + } + + if let Err(e) = write_pub_file(&pub_path, &npub) { + eprintln!("error: failed to write pub file: {e}"); + std::process::exit(1); + } + + eprintln!("{npub}"); + eprintln!("Key files written to: {}/", dir.display()); + eprintln!(); + eprintln!("NOTE: Set 'node.identity.persistent: true' in fips.yaml"); + eprintln!(" or these keys will be overwritten on next daemon start."); + return; + } + + let socket_path = cli.socket.unwrap_or_else(default_socket_path); + + let request = match &cli.command { + Commands::Show { what } => build_query(what.command_name()), + Commands::Connect { + peer, + address, + transport, + } => { + let npub = resolve_peer(peer); + build_command( + "connect", + serde_json::json!({ + "npub": npub, + "address": address, + "transport": transport, + }), + ) + } + Commands::Disconnect { peer } => { + let npub = resolve_peer(peer); + build_command("disconnect", serde_json::json!({"npub": npub})) + } + Commands::Keygen { .. } => unreachable!(), + }; + + match send_request(&socket_path, &request) { + Ok(value) => print_response(&value), + Err(e) => { + eprintln!("error: {e}"); + std::process::exit(1); + } + } } diff --git a/src/control/commands.rs b/src/control/commands.rs new file mode 100644 index 0000000..01a9acb --- /dev/null +++ b/src/control/commands.rs @@ -0,0 +1,68 @@ +//! Mutating control socket commands. +//! +//! Commands that modify node state (connect, disconnect) are handled here, +//! separate from read-only queries in `queries.rs`. + +use super::protocol::Response; +use crate::node::Node; +use serde_json::Value; +use tracing::info; + +/// Dispatch a mutating command to the appropriate handler. +pub async fn dispatch(node: &mut Node, command: &str, params: Option<&Value>) -> Response { + match command { + "connect" => connect(node, params).await, + "disconnect" => disconnect(node, params), + _ => Response::error(format!("unknown command: {command}")), + } +} + +/// Connect to a peer. +/// +/// Params: `{"npub": "npub1...", "address": "host:port", "transport": "udp"}` +async fn connect(node: &mut Node, params: Option<&Value>) -> Response { + let Some(params) = params else { + return Response::error("missing params for connect"); + }; + + let npub = match params.get("npub").and_then(|v| v.as_str()) { + Some(v) => v, + None => return Response::error("missing 'npub' parameter"), + }; + let address = match params.get("address").and_then(|v| v.as_str()) { + Some(v) => v, + None => return Response::error("missing 'address' parameter"), + }; + let transport = match params.get("transport").and_then(|v| v.as_str()) { + Some(v) => v, + None => return Response::error("missing 'transport' parameter"), + }; + + info!(npub = %npub, address = %address, transport = %transport, "API connect requested"); + + match node.api_connect(npub, address, transport).await { + Ok(data) => Response::ok(data), + Err(msg) => Response::error(msg), + } +} + +/// Disconnect a peer. +/// +/// Params: `{"npub": "npub1..."}` +fn disconnect(node: &mut Node, params: Option<&Value>) -> Response { + let Some(params) = params else { + return Response::error("missing params for disconnect"); + }; + + let npub = match params.get("npub").and_then(|v| v.as_str()) { + Some(v) => v, + None => return Response::error("missing 'npub' parameter"), + }; + + info!(npub = %npub, "API disconnect requested"); + + match node.api_disconnect(npub) { + Ok(data) => Response::ok(data), + Err(msg) => Response::error(msg), + } +} diff --git a/src/control/mod.rs b/src/control/mod.rs index 3112529..7590eb3 100644 --- a/src/control/mod.rs +++ b/src/control/mod.rs @@ -1,9 +1,10 @@ -//! Control socket for runtime observability. +//! Control socket for runtime management and 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. +//! Provides a Unix domain socket that accepts commands and returns +//! structured JSON responses. Supports both read-only queries (show_*) +//! and mutating commands (connect, disconnect). +pub mod commands; pub mod protocol; pub mod queries; diff --git a/src/control/protocol.rs b/src/control/protocol.rs index 081eae7..e913093 100644 --- a/src/control/protocol.rs +++ b/src/control/protocol.rs @@ -8,8 +8,11 @@ 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"). + /// The command to execute (e.g., "show_status", "connect"). pub command: String, + /// Optional parameters for mutating commands. + #[serde(default)] + pub params: Option, } /// A control response to a client. @@ -81,6 +84,24 @@ mod tests { assert_eq!(req.command, "show_peers"); } + #[test] + fn test_deserialize_request_with_params() { + let json = r#"{"command": "connect", "params": {"npub": "npub1abc", "address": "1.2.3.4:2121", "transport": "udp"}}"#; + let req: Request = serde_json::from_str(json).unwrap(); + assert_eq!(req.command, "connect"); + let params = req.params.unwrap(); + assert_eq!(params["npub"], "npub1abc"); + assert_eq!(params["transport"], "udp"); + } + + #[test] + fn test_deserialize_request_without_params() { + let json = r#"{"command": "show_status"}"#; + let req: Request = serde_json::from_str(json).unwrap(); + assert_eq!(req.command, "show_status"); + assert!(req.params.is_none()); + } + #[test] fn test_deserialize_malformed_request() { let json = r#"{"not_command": "foo"}"#; diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index 67ec513..ce653df 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -1,6 +1,6 @@ //! RX event loop and packet dispatch. -use crate::control::ControlSocket; +use crate::control::{commands, ControlSocket}; use crate::control::queries; use crate::node::{Node, NodeError}; use crate::transport::ReceivedPacket; @@ -98,7 +98,15 @@ 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 = if request.command.starts_with("show_") { + queries::dispatch(self, &request.command) + } else { + commands::dispatch( + self, + &request.command, + request.params.as_ref(), + ).await + }; let _ = response_tx.send(response); } _ = tick.tick() => { diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index b88746b..db81922 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -739,4 +739,76 @@ impl Node { info!(sent, total = peer_addrs.len(), reason = %reason, "Sent disconnect notifications"); } + + // === Control API methods === + + /// Connect to a peer via the control API. + /// + /// Creates an ephemeral peer connection (not persisted to config, no + /// auto-reconnect). Reuses the same connection path as auto-connect + /// peers. Returns JSON data on success or an error message. + pub(crate) async fn api_connect( + &mut self, + npub: &str, + address: &str, + transport: &str, + ) -> Result { + let peer_config = crate::config::PeerConfig { + npub: npub.to_string(), + alias: None, + addresses: vec![crate::config::PeerAddress::new(transport, address)], + connect_policy: crate::config::ConnectPolicy::Manual, + auto_reconnect: false, + }; + + // Pre-seed identity cache (same as initiate_peer_connections does) + if let Ok(identity) = PeerIdentity::from_npub(npub) { + self.peer_aliases + .insert(*identity.node_addr(), identity.short_npub()); + self.register_identity(*identity.node_addr(), identity.pubkey_full()); + } + + self.initiate_peer_connection(&peer_config) + .await + .map(|()| { + info!( + npub = %npub, + address = %address, + transport = %transport, + "API connect initiated" + ); + serde_json::json!({ + "npub": npub, + "address": address, + "transport": transport, + }) + }) + .map_err(|e| e.to_string()) + } + + /// Disconnect a peer via the control API. + /// + /// Removes the peer and suppresses auto-reconnect. + pub(crate) fn api_disconnect(&mut self, npub: &str) -> Result { + let peer_identity = PeerIdentity::from_npub(npub) + .map_err(|e| format!("invalid npub '{npub}': {e}"))?; + let node_addr = *peer_identity.node_addr(); + + if !self.peers.contains_key(&node_addr) { + return Err(format!("peer not found: {npub}")); + } + + // Remove the peer (full cleanup: sessions, indices, links, tree, bloom) + self.remove_active_peer(&node_addr); + + // Suppress any pending auto-reconnect + self.retry_pending.remove(&node_addr); + + info!(npub = %npub, "API disconnect completed"); + + Ok(serde_json::json!({ + "npub": npub, + "disconnected": true, + })) + } } diff --git a/testing/chaos/scenarios/maelstrom.yaml b/testing/chaos/scenarios/maelstrom.yaml new file mode 100644 index 0000000..1f5b1ca --- /dev/null +++ b/testing/chaos/scenarios/maelstrom.yaml @@ -0,0 +1,76 @@ +# Maelstrom: topology mutation + node churn + ephemeral identity +# +# Combines all chaos dimensions: netem impairments, link flaps, node +# churn, AND peer-level topology mutation (connect/disconnect). +# Half the nodes have ephemeral identities that change on restart. +# +# Default: 20 nodes. Override with --nodes flag. + +scenario: + name: "maelstrom" + seed: 42 + duration_secs: 600 + +topology: + num_nodes: 20 + algorithm: erdos_renyi + params: + p: 0.3 + ensure_connected: true + subnet: "172.20.0.0/16" + ip_start: 10 + transport_mix: + udp: 0.8 + tcp: 0.2 + +netem: + enabled: true + default_policy: + delay_ms: { min: 5, max: 50 } + jitter_ms: { min: 1, max: 10 } + loss_pct: { min: 0, max: 2 } + mutation: + interval_secs: { min: 20, max: 45 } + fraction: 0.3 + policies: + normal: + delay_ms: [5, 20] + loss_pct: [0, 1] + degraded: + delay_ms: [50, 100] + jitter_ms: [10, 30] + loss_pct: [3, 8] + +link_flaps: + enabled: true + interval_secs: { min: 30, max: 60 } + max_down_links: 3 + down_duration_secs: { min: 10, max: 30 } + protect_connectivity: true + +traffic: + enabled: true + max_concurrent: 5 + interval_secs: { min: 5, max: 30 } + duration_secs: { min: 5, max: 60 } + parallel_streams: 4 + +node_churn: + enabled: true + interval_secs: { min: 60, max: 120 } + max_down_nodes: 3 + down_duration_secs: { min: 30, max: 90 } + protect_connectivity: false + +peer_churn: + enabled: true + interval_secs: { min: 8, max: 12 } + ephemeral_fraction: 0.5 + +bandwidth: + enabled: true + tiers_mbps: [1, 10, 100, 1000] + +logging: + rust_log: "debug" + output_dir: "./sim-results" diff --git a/testing/chaos/sim/config_gen.py b/testing/chaos/sim/config_gen.py index e2ade0e..bd519fb 100644 --- a/testing/chaos/sim/config_gen.py +++ b/testing/chaos/sim/config_gen.py @@ -107,8 +107,13 @@ def generate_node_config( node_id: str, outbound_peers: list[str], fips_overrides: dict | None = None, + ephemeral: bool = False, ) -> str: - """Generate a complete FIPS config YAML for one node.""" + """Generate a complete FIPS config YAML for one node. + + If ephemeral is True, the nsec is omitted from the config so the + daemon generates a fresh keypair on each restart. + """ template = _load_template() node = topology.nodes[node_id] peers_yaml = generate_peers_block(topology, node_id, outbound_peers) @@ -120,6 +125,13 @@ def generate_node_config( config = config.replace("{{NSEC}}", node.nsec) config = config.replace("{{PEERS}}", peers_yaml) + # Ephemeral nodes: remove nsec so daemon generates fresh keys on restart + if ephemeral: + parsed = yaml.safe_load(config) + identity = parsed.get("node", {}).get("identity", {}) + identity.pop("nsec", None) + config = yaml.dump(parsed, default_flow_style=False, sort_keys=False) + # Determine which transports this node participates in eth_ifaces = topology.ethernet_interfaces(node_id) has_tcp = bool(topology.tcp_peers(node_id)) @@ -168,14 +180,17 @@ def write_configs( topology: SimTopology, output_dir: str, fips_overrides: dict | None = None, + ephemeral_nodes: set[str] | None = None, ): """Write all node configs and npubs.env to the output directory.""" os.makedirs(output_dir, exist_ok=True) + ephemeral_nodes = ephemeral_nodes or set() outbound = topology.directed_outbound() for node_id in topology.nodes: config = generate_node_config( - topology, node_id, outbound[node_id], fips_overrides + topology, node_id, outbound[node_id], fips_overrides, + ephemeral=(node_id in ephemeral_nodes), ) path = os.path.join(output_dir, f"{node_id}.yaml") with open(path, "w") as f: diff --git a/testing/chaos/sim/control.py b/testing/chaos/sim/control.py index 9f9af40..cb2fc01 100644 --- a/testing/chaos/sim/control.py +++ b/testing/chaos/sim/control.py @@ -56,6 +56,52 @@ def query_node(container: str, command: str, timeout: int = 10) -> dict | None: return response.get("data", {}) +def send_command( + container: str, command: str, params: dict, timeout: int = 10 +) -> dict | None: + """Send a mutating command with params to a node's control socket. + + Returns the response data dict, or None on failure. + Uses base64-encoded JSON to avoid shell quoting issues with + embedded quotes in the payload. + """ + import base64 + + payload = json.dumps({"command": command, "params": params}) + b64 = base64.b64encode(payload.encode()).decode() + script = ( + "import socket,json,sys,base64; " + "s=socket.socket(socket.AF_UNIX,socket.SOCK_STREAM); " + f"s.connect('{CONTROL_SOCKET}'); " + f"s.sendall(base64.b64decode('{b64}')+b'\\n'); " + "s.shutdown(socket.SHUT_WR); " + "chunks=[]; " + "[chunks.append(d) for d in iter(lambda:s.recv(65536),b'')]; " + "print(b''.join(chunks).decode())" + ) + stdout = docker_exec_quiet(container, f"python3 -c \"{script}\"", timeout=timeout) + if stdout is None: + return None + + try: + response = json.loads(stdout.strip()) + except json.JSONDecodeError as e: + log.warning("Invalid JSON from %s: %s", container, e) + return None + + if response.get("status") != "ok": + msg = response.get("message", "unknown error") + log.debug("Command %s on %s: %s", command, container, msg) + return None + + return response.get("data", {}) + + +def query_status(container: str) -> dict | None: + """Query a node's status (npub, ipv6, uptime, etc.).""" + return query_node(container, "show_status") + + def query_tree(container: str) -> dict | None: """Query a node's spanning tree state.""" return query_node(container, "show_tree") diff --git a/testing/chaos/sim/nodes.py b/testing/chaos/sim/nodes.py index 8648107..bff2d6a 100644 --- a/testing/chaos/sim/nodes.py +++ b/testing/chaos/sim/nodes.py @@ -43,6 +43,7 @@ class NodeManager: netem_mgr=None, down_nodes: set[str] | None = None, veth_mgr=None, + on_node_restart=None, ): self.topology = topology self.config = config @@ -50,6 +51,7 @@ class NodeManager: self.netem_mgr = netem_mgr self.veth_mgr = veth_mgr self.down_nodes = down_nodes or set() + self.on_node_restart = on_node_restart self.node_states: dict[str, NodeState] = { nid: NodeState(node_id=nid) for nid in topology.nodes } @@ -162,6 +164,10 @@ class NodeManager: time.sleep(1) self.netem_mgr.setup_node(node_id) + # Notify callback (e.g., refresh npub for ephemeral identity nodes) + if self.on_node_restart: + self.on_node_restart(node_id) + def _would_disconnect(self, node_id: str) -> bool: """Check if removing this node (plus currently-down nodes) disconnects the graph. diff --git a/testing/chaos/sim/peer_churn.py b/testing/chaos/sim/peer_churn.py new file mode 100644 index 0000000..88cfd3f --- /dev/null +++ b/testing/chaos/sim/peer_churn.py @@ -0,0 +1,178 @@ +"""Peer-level topology churn via connect/disconnect commands. + +Unlike NodeManager (which stops/starts containers), this uses the +fipsctl connect/disconnect API to dynamically add and remove individual +peer connections while nodes stay running. The topology graph evolves +over time: random links are disconnected and new random pairs connected. + +Supports ephemeral identity nodes — half the nodes (configurable) get +new keypairs on each container restart, requiring the simulator to +track current npubs via show_status queries. +""" + +from __future__ import annotations + +import logging +import random +import time + +from .control import query_status, send_command +from .scenario import PeerChurnConfig +from .topology import SimTopology + +log = logging.getLogger(__name__) + + +class PeerChurnManager: + """Manages peer-level topology churn using connect/disconnect commands.""" + + def __init__( + self, + topology: SimTopology, + config: PeerChurnConfig, + rng: random.Random, + down_nodes: set[str], + ephemeral_nodes: set[str] | None = None, + ): + self.topology = topology + self.config = config + self.rng = rng + self.down_nodes = down_nodes + self.ephemeral_nodes = ephemeral_nodes or set() + + # Current npub for each node (populated during init) + self.npub_cache: dict[str, str] = {} + + # Currently active edges (start with topology edges) + self.active_edges: set[tuple[str, str]] = set() + for a, b in topology.edges: + self.active_edges.add(self._canonical(a, b)) + + # Edges disconnected by peer churn (not yet reconnected) + self.churned_count = 0 + + @staticmethod + def _canonical(a: str, b: str) -> tuple[str, str]: + """Canonical edge ordering (sorted).""" + return (min(a, b), max(a, b)) + + @property + def churn_count(self) -> int: + return self.churned_count + + def refresh_npub(self, node_id: str) -> str | None: + """Query a node's current npub and update the cache. + + Returns the npub or None if the query failed. + """ + container = self.topology.container_name(node_id) + status = query_status(container) + if status and "npub" in status: + npub = status["npub"] + self.npub_cache[node_id] = npub + return npub + return None + + def refresh_all_npubs(self): + """Populate npub cache for all nodes.""" + for node_id in self.topology.nodes: + if node_id not in self.down_nodes: + self.refresh_npub(node_id) + log.info( + "Cached npubs for %d/%d nodes", + len(self.npub_cache), + len(self.topology.nodes), + ) + + def maybe_churn(self): + """Disconnect a random active link, then connect a random new pair.""" + # Skip if too many nodes are down + up_nodes = [n for n in self.topology.nodes if n not in self.down_nodes] + if len(up_nodes) < 3: + return + + # Phase 1: Disconnect a random active link between up nodes + candidates = [ + (a, b) + for a, b in self.active_edges + if a not in self.down_nodes and b not in self.down_nodes + ] + if candidates: + edge = self.rng.choice(candidates) + if self._disconnect_edge(edge[0], edge[1]): + self.active_edges.discard(self._canonical(edge[0], edge[1])) + self.churned_count += 1 + + # Phase 2: Connect a random pair that isn't currently connected + non_edges = [] + for i, a in enumerate(up_nodes): + for b in up_nodes[i + 1 :]: + if self._canonical(a, b) not in self.active_edges: + non_edges.append((a, b)) + + if non_edges: + a, b = self.rng.choice(non_edges) + if self._connect_edge(a, b): + self.active_edges.add(self._canonical(a, b)) + + def _disconnect_edge(self, a: str, b: str) -> bool: + """Disconnect both sides of a link.""" + npub_a = self.npub_cache.get(a) + npub_b = self.npub_cache.get(b) + if not npub_a or not npub_b: + log.debug("Missing npub for %s or %s, skipping disconnect", a, b) + return False + + container_a = self.topology.container_name(a) + container_b = self.topology.container_name(b) + + ok_a = send_command(container_a, "disconnect", {"npub": npub_b}) + ok_b = send_command(container_b, "disconnect", {"npub": npub_a}) + + if ok_a is not None or ok_b is not None: + log.info("Peer DISCONNECT: %s -- %s", a, b) + return True + + log.debug("Disconnect failed for %s -- %s", a, b) + return False + + def _connect_edge(self, a: str, b: str) -> bool: + """Connect both sides of a new link (mutual outbound).""" + npub_a = self.npub_cache.get(a) + npub_b = self.npub_cache.get(b) + if not npub_a or not npub_b: + log.debug("Missing npub for %s or %s, skipping connect", a, b) + return False + + # Use UDP transport with the node's Docker IP + ip_a = self.topology.nodes[a].docker_ip + ip_b = self.topology.nodes[b].docker_ip + port = 2121 # Default UDP port + + container_a = self.topology.container_name(a) + container_b = self.topology.container_name(b) + + # Node A connects to B + ok_a = send_command( + container_a, + "connect", + {"npub": npub_b, "address": f"{ip_b}:{port}", "transport": "udp"}, + ) + + # Node B connects to A + ok_b = send_command( + container_b, + "connect", + {"npub": npub_a, "address": f"{ip_a}:{port}", "transport": "udp"}, + ) + + if ok_a is not None or ok_b is not None: + log.info("Peer CONNECT: %s -- %s (udp)", a, b) + return True + + log.debug("Connect failed for %s -- %s", a, b) + return False + + def restore_all(self): + """No-op for teardown — peer connections are ephemeral.""" + pass diff --git a/testing/chaos/sim/runner.py b/testing/chaos/sim/runner.py index b6dbd16..1a07bb7 100644 --- a/testing/chaos/sim/runner.py +++ b/testing/chaos/sim/runner.py @@ -19,6 +19,7 @@ from .links import LinkManager from .logs import AnalysisResult, analyze_logs, collect_logs, write_sim_metadata from .netem import NetemManager from .nodes import NodeManager +from .peer_churn import PeerChurnManager from .scenario import Scenario from .topology import SimTopology, generate_topology from .traffic import TrafficManager @@ -46,6 +47,7 @@ class SimRunner: self.link_mgr: LinkManager | None = None self.traffic_mgr: TrafficManager | None = None self.node_mgr: NodeManager | None = None + self.peer_churn_mgr: PeerChurnManager | None = None def run(self) -> AnalysisResult | None: """Run the full simulation lifecycle.""" @@ -113,7 +115,23 @@ class SimRunner: config_dir = os.path.normpath( os.path.join(docker_network_dir, "generated-configs", "sim") ) - write_configs(self.topology, config_dir, self.scenario.fips_overrides) + # Select ephemeral identity nodes (if peer churn enabled) + self._ephemeral_nodes: set[str] = set() + if s.peer_churn.enabled and s.peer_churn.ephemeral_fraction > 0: + all_nodes = sorted(self.topology.nodes.keys()) + count = int(len(all_nodes) * s.peer_churn.ephemeral_fraction) + self._ephemeral_nodes = set(self.rng.sample(all_nodes, count)) + log.info( + "Ephemeral identity nodes (%d/%d): %s", + len(self._ephemeral_nodes), + len(all_nodes), + ", ".join(sorted(self._ephemeral_nodes)), + ) + + write_configs( + self.topology, config_dir, self.scenario.fips_overrides, + ephemeral_nodes=self._ephemeral_nodes, + ) log.info("Wrote node configs to %s", config_dir) # 3. Generate docker-compose.yml @@ -167,6 +185,14 @@ class SimRunner: self.topology, s.node_churn, self.rng, netem_mgr=self.netem_mgr, down_nodes=self._down_nodes, veth_mgr=self.veth_mgr, + on_node_restart=self._handle_node_restart, + ) + + if s.peer_churn.enabled: + self.peer_churn_mgr = PeerChurnManager( + self.topology, s.peer_churn, self.rng, + down_nodes=self._down_nodes, + ephemeral_nodes=self._ephemeral_nodes, ) def _warmup(self): @@ -177,6 +203,30 @@ class SimRunner: self._sleep(wait) self._take_snapshot("warmup") + # Populate npub cache after convergence (nodes must be running) + if self.peer_churn_mgr: + self.peer_churn_mgr.refresh_all_npubs() + + def _handle_node_restart(self, node_id: str): + """Called after a node container is restarted. + + For ephemeral identity nodes, waits briefly for the daemon to + start, then queries its new npub and updates the peer churn + manager's cache. + """ + if not self.peer_churn_mgr: + return + if node_id not in self.peer_churn_mgr.ephemeral_nodes: + return + + # Brief delay for daemon startup before querying control socket + time.sleep(2) + new_npub = self.peer_churn_mgr.refresh_npub(node_id) + if new_npub: + log.info("Ephemeral node %s new identity: %s...%s", node_id, new_npub[:12], new_npub[-6:]) + else: + log.warning("Failed to refresh npub for ephemeral node %s", node_id) + def _simulation_loop(self): """Main event loop driving stochastic behavior.""" start = time.time() @@ -189,6 +239,7 @@ class SimRunner: next_flap = self._schedule_next(start, s.link_flaps.interval_secs) if self.link_mgr else float("inf") next_traffic = self._schedule_next(start, s.traffic.interval_secs) if self.traffic_mgr else float("inf") next_churn = self._schedule_next(start, s.node_churn.interval_secs) if self.node_mgr else float("inf") + next_peer_churn = self._schedule_next(start, s.peer_churn.interval_secs) if self.peer_churn_mgr else float("inf") while not self._interrupted: now = time.time() @@ -222,17 +273,26 @@ class SimRunner: next_churn = self._schedule_next(now, s.node_churn.interval_secs) self.node_mgr.restore_expired() + # Peer churn (topology mutation) + if self.peer_churn_mgr: + if now >= next_peer_churn: + self.peer_churn_mgr.maybe_churn() + next_peer_churn = self._schedule_next(now, s.peer_churn.interval_secs) + # Status line down_links = self.link_mgr.down_count if self.link_mgr else 0 down_nodes = self.node_mgr.down_count if self.node_mgr else 0 active = self.traffic_mgr.active_count if self.traffic_mgr else 0 + peer_churns = self.peer_churn_mgr.churn_count if self.peer_churn_mgr else 0 + status_extra = f" peer_churns={peer_churns}" if self.peer_churn_mgr else "" print( f"\r [{elapsed:.0f}s/{duration}s] " f"nodes={len(self.topology.nodes)} " f"edges={len(self.topology.edges)} " f"links_down={down_links} " f"nodes_down={down_nodes} " - f"traffic={active} ", + f"traffic={active}" + f"{status_extra} ", end="", flush=True, ) diff --git a/testing/chaos/sim/scenario.py b/testing/chaos/sim/scenario.py index f1c5efe..6374ac5 100644 --- a/testing/chaos/sim/scenario.py +++ b/testing/chaos/sim/scenario.py @@ -106,6 +106,20 @@ class NodeChurnConfig: protect_connectivity: bool = True +@dataclass +class PeerChurnConfig: + """Peer-level topology churn via connect/disconnect commands. + + When enabled, periodically disconnects a random active link and + connects a random unconnected node pair, causing the mesh topology + to evolve over time. + """ + + enabled: bool = False + interval_secs: Range = field(default_factory=lambda: Range(8, 12)) + ephemeral_fraction: float = 0.0 + + @dataclass class BandwidthConfig: """Per-link bandwidth pacing via HTB rate limiting. @@ -152,6 +166,7 @@ class Scenario: link_flaps: LinkFlapsConfig = field(default_factory=LinkFlapsConfig) traffic: TrafficConfig = field(default_factory=TrafficConfig) node_churn: NodeChurnConfig = field(default_factory=NodeChurnConfig) + peer_churn: PeerChurnConfig = field(default_factory=PeerChurnConfig) bandwidth: BandwidthConfig = field(default_factory=BandwidthConfig) ingress: IngressConfig = field(default_factory=IngressConfig) logging: LoggingConfig = field(default_factory=LoggingConfig) @@ -279,6 +294,13 @@ def load_scenario(path: str) -> Scenario: ) s.node_churn.protect_connectivity = nc2.get("protect_connectivity", True) + # Peer churn section + pc = raw.get("peer_churn", {}) + s.peer_churn.enabled = pc.get("enabled", False) + if "interval_secs" in pc: + s.peer_churn.interval_secs = _parse_range(pc["interval_secs"], "peer_churn.interval_secs") + s.peer_churn.ephemeral_fraction = float(pc.get("ephemeral_fraction", 0.0)) + # Bandwidth section bw = raw.get("bandwidth", {}) s.bandwidth.enabled = bw.get("enabled", False) @@ -399,6 +421,10 @@ def _validate(s: Scenario): s.node_churn.down_duration_secs.validate("node_churn.down_duration_secs") if s.node_churn.max_down_nodes >= s.topology.num_nodes: raise ValueError("node_churn.max_down_nodes must be < topology.num_nodes") + if s.peer_churn.enabled: + s.peer_churn.interval_secs.validate("peer_churn.interval_secs") + if not 0.0 <= s.peer_churn.ephemeral_fraction <= 1.0: + raise ValueError("peer_churn.ephemeral_fraction must be between 0.0 and 1.0") if s.bandwidth.enabled: for tier in s.bandwidth.tiers_mbps: if tier <= 0: