//! 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)?; // Make the socket and its parent directory group-accessible so // 'fips' group members can use fipsctl/fipstop without root. use std::os::unix::fs::PermissionsExt; std::fs::set_permissions(&socket_path, std::fs::Permissions::from_mode(0o770))?; Self::chown_to_fips_group(&socket_path); if let Some(parent) = socket_path.parent() { Self::chown_to_fips_group(parent); } 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(()) } } } /// Set group ownership of a path to the 'fips' group (best-effort). fn chown_to_fips_group(path: &Path) { use std::ffi::CString; use std::os::unix::ffi::OsStrExt; // Look up the 'fips' group let group_name = CString::new("fips").unwrap(); let grp = unsafe { libc::getgrnam(group_name.as_ptr()) }; if grp.is_null() { debug!("'fips' group not found, skipping chown for {}", path.display()); return; } let gid = unsafe { (*grp).gr_gid }; let c_path = match CString::new(path.as_os_str().as_bytes()) { Ok(p) => p, Err(_) => return, }; let ret = unsafe { libc::chown(c_path.as_ptr(), u32::MAX, gid) }; if ret != 0 { warn!( path = %path.display(), error = %std::io::Error::last_os_error(), "Failed to chown control socket to 'fips' group" ); } } /// 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(); } }