diff --git a/Cargo.lock b/Cargo.lock index b37e619..08e74ca 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -458,6 +458,10 @@ dependencies = [ name = "ferese-layout" version = "0.1.0" +[[package]] +name = "ferese-malformed-client" +version = "0.1.0" + [[package]] name = "feresectl" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 3e1b0eb..5e7d86c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,6 +6,7 @@ members = [ "crates/ferese-core", "crates/ferese-ipc", "crates/ferese-layout", + "tools/ferese-malformed-client", ] resolver = "3" diff --git a/README.md b/README.md index c1e2f0b..f19537d 100644 --- a/README.md +++ b/README.md @@ -201,7 +201,8 @@ The v0 IPC slice also supports `move`, `resize`, `move-to-workspace`, `center-column`, `consume`, `expel`, `close`, `get-focused-window`, and `get-outputs`. Requests use versioned, 1 MiB-limited length-prefixed JSON. The runtime directory and socket are restricted to modes `0700` and `0600`, and the -server rejects peers whose effective UID differs from the compositor's. +server rejects peers whose effective UID differs from the compositor's. IPC has +a bounded request queue and accepts at most 64 concurrent client connections. `get-outputs` includes disabled-but-connected monitors, connector and persistent identity matchers, current and available modes with refresh rates, fractional scale, transform, logical position and size, and physical dimensions. diff --git a/compositor/src/ipc.rs b/compositor/src/ipc.rs index 30d38ea..bbfda9f 100644 --- a/compositor/src/ipc.rs +++ b/compositor/src/ipc.rs @@ -5,7 +5,11 @@ use std::os::fd::AsRawFd; use std::os::unix::fs::{FileTypeExt, MetadataExt, PermissionsExt}; use std::os::unix::net::{UnixListener, UnixStream}; use std::path::{Path, PathBuf}; -use std::sync::mpsc::{SyncSender, sync_channel}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{ + Arc, + mpsc::{SyncSender, sync_channel}, +}; use std::thread; use ferese_core::LayoutMode; @@ -18,6 +22,7 @@ use smithay::utils::Transform; use crate::{Ferese, config::OutputTransform}; const REQUEST_QUEUE_CAPACITY: usize = 128; +const MAX_CONNECTIONS: usize = 64; #[derive(Debug)] struct IpcCall { @@ -25,6 +30,16 @@ struct IpcCall { response: SyncSender, } +struct ConnectionPermit { + active: Arc, +} + +impl Drop for ConnectionPermit { + fn drop(&mut self) { + self.active.fetch_sub(1, Ordering::AcqRel); + } +} + #[derive(Debug)] pub(crate) struct IpcSocketGuard { path: PathBuf, @@ -78,6 +93,8 @@ pub(crate) fn init( } fn accept_connections(listener: UnixListener, sender: channel::SyncSender) { + let active_connections = Arc::new(AtomicUsize::new(0)); + loop { let (stream, _) = match listener.accept() { Ok(connection) => connection, @@ -100,16 +117,43 @@ fn accept_connections(listener: UnixListener, sender: channel::SyncSender) -> Option { + let mut count = active.load(Ordering::Acquire); + loop { + if count >= MAX_CONNECTIONS { + return None; + } + match active.compare_exchange_weak(count, count + 1, Ordering::AcqRel, Ordering::Acquire) { + Ok(_) => break, + Err(actual) => count = actual, + } + } + + Some(ConnectionPermit { + active: active.clone(), + }) +} + fn serve_connection(mut stream: UnixStream, sender: channel::SyncSender) { loop { let request = match read_frame(&mut stream) { @@ -147,15 +191,8 @@ fn serve_connection(mut stream: UnixStream, sender: channel::SyncSender impl Ferese { fn handle_ipc_request(&mut self, request: Request) -> Response { - if request.version != VERSION { - return Response::error( - request.id, - "unsupported_version", - format!("supported version is {VERSION}"), - ); - } - if request.kind != "command" { - return Response::error(request.id, "invalid_type", "expected type \"command\""); + if let Err(error) = validate_request(&request) { + return Response::error(request.id, error.code, error.message); } match self.dispatch_ipc_command(&request.command, &request.args) { @@ -345,6 +382,23 @@ impl Ferese { } } +fn validate_request(request: &Request) -> Result<(), CommandError> { + if request.version != VERSION { + return Err(CommandError::new( + "unsupported_version", + format!("supported version is {VERSION}"), + )); + } + if request.kind != "command" { + return Err(CommandError::new( + "invalid_type", + "expected type \"command\"", + )); + } + + Ok(()) +} + fn transform_name(transform: Transform) -> &'static str { match transform { Transform::Normal => "normal", @@ -500,8 +554,22 @@ fn effective_uid() -> u32 { #[cfg(test)] mod tests { + use std::io::{Read, Write}; + use std::net::Shutdown; + use std::os::unix::fs::PermissionsExt; + use super::*; + fn request(version: u32, kind: &str) -> Request { + Request { + version, + id: 7, + kind: kind.to_owned(), + command: "get-outputs".to_owned(), + args: json!({}), + } + } + #[test] fn parses_direction_arguments() { assert_eq!( @@ -517,4 +585,112 @@ mod tests { assert!(workspace_arg(&json!({ "index": 0 })).is_err()); assert!(workspace_arg(&json!({ "index": -1 })).is_err()); } + + #[test] + fn rejects_unsupported_versions_and_request_types_with_stable_codes() { + let version = validate_request(&request(VERSION + 1, "command")).unwrap_err(); + let kind = validate_request(&request(VERSION, "event")).unwrap_err(); + + assert_eq!(version.code, "unsupported_version"); + assert_eq!(kind.code, "invalid_type"); + assert!(validate_request(&request(VERSION, "command")).is_ok()); + } + + #[test] + fn reads_same_user_peer_credentials() { + let (left, right) = UnixStream::pair().unwrap(); + + assert_eq!(peer_uid(&left).unwrap(), effective_uid()); + assert_eq!(peer_uid(&right).unwrap(), effective_uid()); + } + + #[test] + fn malformed_ipc_connection_is_closed_without_reaching_the_request_queue() { + let (mut client, server) = UnixStream::pair().unwrap(); + let (sender, _receiver) = channel::sync_channel(1); + let worker = thread::spawn(move || serve_connection(server, sender)); + let invalid = b"not-json"; + + client + .write_all(&(invalid.len() as u32).to_be_bytes()) + .unwrap(); + client.write_all(invalid).unwrap(); + client.shutdown(Shutdown::Write).unwrap(); + client + .set_read_timeout(Some(std::time::Duration::from_secs(1))) + .unwrap(); + let mut byte = [0_u8; 1]; + + assert_eq!(client.read(&mut byte).unwrap(), 0); + worker.join().unwrap(); + } + + #[test] + fn runtime_socket_directory_is_private() { + let root = unique_test_directory("private-parent"); + let parent = root.join("ferese"); + let socket = parent.join("control.sock"); + + fs::create_dir(&root).unwrap(); + prepare_parent(&socket).unwrap(); + let metadata = fs::metadata(&parent).unwrap(); + + assert_eq!(metadata.permissions().mode() & 0o777, 0o700); + fs::remove_dir_all(root).unwrap(); + } + + #[test] + fn listener_reclaims_only_a_stale_owned_socket() { + let root = unique_test_directory("stale-socket"); + fs::create_dir(&root).unwrap(); + let socket = root.join("control.sock"); + let first = bind_listener(&socket).unwrap(); + + assert_eq!( + bind_listener(&socket).unwrap_err().kind(), + io::ErrorKind::AddrInUse + ); + drop(first); + let replacement = bind_listener(&socket).unwrap(); + + drop(replacement); + fs::remove_file(socket).unwrap(); + fs::remove_dir(root).unwrap(); + } + + #[test] + fn listener_never_replaces_a_non_socket_path() { + let root = unique_test_directory("non-socket"); + fs::create_dir(&root).unwrap(); + let socket = root.join("control.sock"); + fs::write(&socket, b"keep").unwrap(); + + assert_eq!( + bind_listener(&socket).unwrap_err().kind(), + io::ErrorKind::AddrInUse + ); + assert_eq!(fs::read(&socket).unwrap(), b"keep"); + fs::remove_dir_all(root).unwrap(); + } + + #[test] + fn ipc_worker_count_is_bounded_and_permits_are_reusable() { + let active = Arc::new(AtomicUsize::new(0)); + let mut permits = (0..MAX_CONNECTIONS) + .map(|_| try_acquire_connection(&active).unwrap()) + .collect::>(); + + assert!(try_acquire_connection(&active).is_none()); + permits.pop(); + assert_eq!(active.load(Ordering::Acquire), MAX_CONNECTIONS - 1); + assert!(try_acquire_connection(&active).is_some()); + } + + fn unique_test_directory(label: &str) -> PathBuf { + let nonce = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + env::temp_dir().join(format!("ferese-ipc-{label}-{}-{nonce}", std::process::id())) + } } diff --git a/docs/ferese-window-manager-spec.md b/docs/ferese-window-manager-spec.md index a10f2a3..9348aff 100644 --- a/docs/ferese-window-manager-spec.md +++ b/docs/ferese-window-manager-spec.md @@ -1347,6 +1347,10 @@ Events include `window-opened`, `window-closed`, `window-focused`, Subscribers have bounded queues. The server MUST disconnect a subscriber whose queue overflows rather than block the compositor. +The IPC server MUST bound concurrent connection workers as well as queued +requests. Reaching either limit rejects the excess client without blocking the +compositor event loop or allocating another worker thread. + ## 14. Input Ferese MUST support click-to-focus, configurable focus-follows-pointer, floating diff --git a/docs/native-soak-testing.md b/docs/native-soak-testing.md index 8bb9aae..eced6d5 100644 --- a/docs/native-soak-testing.md +++ b/docs/native-soak-testing.md @@ -2,9 +2,11 @@ The M6 soak runner repeatedly creates and destroys native Wayland clients while exercising focus, resize, column width, fullscreen, floating, and workspace -paths. Every tenth iteration also terminates a client abruptly. The runner -checks that the compositor and authenticated IPC socket remain responsive and -records compositor RSS growth. +paths. Every tenth iteration also terminates a client abruptly and opens three +isolated connections that send malformed Wayland messages. The runner requires +Ferese to disconnect each malformed client and then checks that the compositor +and authenticated IPC socket remain responsive. It also records compositor RSS +growth. Build Ferese and `feresectl`, start the compositor, and run: @@ -30,12 +32,24 @@ FERESE_SOAK_SECONDS=300 \ ./scripts/soak-native.sh ``` +The malformed-client check can also be run independently against a disposable +Ferese session: + +```bash +WAYLAND_DISPLAY=wayland-1 cargo run -p ferese-malformed-client +target/debug/feresectl get-outputs +``` + +Do not point the malformed-client probe at the host compositor socket. + Environment controls: - `FERESE_SOAK_SECONDS`: run duration, default `86400`; - `FERESE_SOAK_DELAY`: delay between operations, default `0.15` seconds; - `FERESE_SOAK_CLIENT`: native client executable, default `foot`; - `FERESECTL`: path to `feresectl`, default `target/debug/feresectl`; +- `FERESE_MALFORMED_CLIENT`: path to the malformed protocol probe, default + `target/debug/ferese-malformed-client`; - `FERESE_PID`: compositor PID; automatic lookup is only a fallback; - `FERESE_SOAK_LOG`: result log, default `/tmp/ferese-soak-.log`. diff --git a/scripts/soak-native.sh b/scripts/soak-native.sh index 7547822..15ea55c 100755 --- a/scripts/soak-native.sh +++ b/scripts/soak-native.sh @@ -6,6 +6,7 @@ soak_seconds="${FERESE_SOAK_SECONDS:-86400}" iteration_delay="${FERESE_SOAK_DELAY:-0.15}" client_program="${FERESE_SOAK_CLIENT:-foot}" feresectl_program="${FERESECTL:-target/debug/feresectl}" +malformed_client_program="${FERESE_MALFORMED_CLIENT:-target/debug/ferese-malformed-client}" compositor_pid="${FERESE_PID:-}" log_path="${FERESE_SOAK_LOG:-/tmp/ferese-soak-$$.log}" @@ -23,6 +24,10 @@ if [[ ! -x "$feresectl_program" ]]; then cargo build -p feresectl fi +if [[ ! -x "$malformed_client_program" ]]; then + cargo build -p ferese-malformed-client +fi + if ! command -v "$client_program" >/dev/null 2>&1; then echo "soak client not found: $client_program" >&2 exit 2 @@ -135,6 +140,8 @@ while ((SECONDS < deadline)); do launch_client wait_for_focus kill -TERM "${client_pids[-1]}" 2>/dev/null || true + "$malformed_client_program" >>"$log_path" 2>&1 + control get-outputs fi reap_clients diff --git a/tools/ferese-malformed-client/Cargo.toml b/tools/ferese-malformed-client/Cargo.toml new file mode 100644 index 0000000..ee23b85 --- /dev/null +++ b/tools/ferese-malformed-client/Cargo.toml @@ -0,0 +1,8 @@ +[package] +name = "ferese-malformed-client" +version.workspace = true +edition.workspace = true +license.workspace = true +rust-version.workspace = true + +[dependencies] diff --git a/tools/ferese-malformed-client/src/main.rs b/tools/ferese-malformed-client/src/main.rs new file mode 100644 index 0000000..0c4ab47 --- /dev/null +++ b/tools/ferese-malformed-client/src/main.rs @@ -0,0 +1,150 @@ +use std::env; +use std::error::Error; +use std::io::{self, Read, Write}; +use std::net::Shutdown; +use std::os::unix::net::UnixStream; +use std::path::PathBuf; +use std::time::Duration; + +const DISCONNECT_TIMEOUT: Duration = Duration::from_secs(2); +const UNKNOWN_OBJECT_FRAME: [u8; 8] = frame_header(u32::MAX, 0, 8); +const UNDERSIZED_FRAME: [u8; 8] = frame_header(1, 0, 4); +const UNALIGNED_FRAME: [u8; 9] = { + let header = frame_header(1, 0, 9); + [ + header[0], header[1], header[2], header[3], header[4], header[5], header[6], header[7], 0, + ] +}; + +struct MalformedCase { + name: &'static str, + bytes: &'static [u8], +} + +fn main() -> Result<(), Box> { + let socket = display_socket()?; + let cases = malformed_cases(); + + for case in cases { + exercise_case(&socket, &case)?; + println!("PASS {}", case.name); + } + + Ok(()) +} + +fn display_socket() -> Result { + let display = env::var_os("WAYLAND_DISPLAY") + .filter(|value| !value.is_empty()) + .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "WAYLAND_DISPLAY is not set"))?; + let display = PathBuf::from(display); + if display.is_absolute() { + return Ok(display); + } + + env::var_os("XDG_RUNTIME_DIR") + .filter(|value| !value.is_empty()) + .map(PathBuf::from) + .map(|directory| directory.join(display)) + .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "XDG_RUNTIME_DIR is not set")) +} + +fn malformed_cases() -> [MalformedCase; 3] { + [ + MalformedCase { + name: "unknown-object", + bytes: &UNKNOWN_OBJECT_FRAME, + }, + MalformedCase { + name: "undersized-frame", + bytes: &UNDERSIZED_FRAME, + }, + MalformedCase { + name: "unaligned-frame", + bytes: &UNALIGNED_FRAME, + }, + ] +} + +const fn frame_header(object: u32, opcode: u16, size: u16) -> [u8; 8] { + let object = object.to_ne_bytes(); + let size_and_opcode = ((size as u32) << 16 | opcode as u32).to_ne_bytes(); + [ + object[0], + object[1], + object[2], + object[3], + size_and_opcode[0], + size_and_opcode[1], + size_and_opcode[2], + size_and_opcode[3], + ] +} + +fn exercise_case(socket: &PathBuf, case: &MalformedCase) -> Result<(), Box> { + let mut stream = UnixStream::connect(socket)?; + stream.set_read_timeout(Some(DISCONNECT_TIMEOUT))?; + stream.write_all(case.bytes)?; + stream.shutdown(Shutdown::Write)?; + + let mut response = [0_u8; 256]; + loop { + match stream.read(&mut response) { + Ok(0) => return Ok(()), + Ok(_) => continue, + Err(error) + if matches!( + error.kind(), + io::ErrorKind::ConnectionReset | io::ErrorKind::BrokenPipe + ) => + { + return Ok(()); + } + Err(error) + if matches!( + error.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + ) => + { + return Err(format!( + "compositor did not disconnect malformed case {:?} within {:?}", + case.name, DISCONNECT_TIMEOUT + ) + .into()); + } + Err(error) => return Err(error.into()), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn encodes_native_wayland_headers() { + let header = UNKNOWN_OBJECT_FRAME; + + assert_eq!( + u32::from_ne_bytes(header[..4].try_into().unwrap()), + u32::MAX + ); + assert_eq!(u32::from_ne_bytes(header[4..].try_into().unwrap()), 8 << 16); + } + + #[test] + fn malformed_sizes_match_the_transmitted_byte_counts() { + let undersized = UNDERSIZED_FRAME; + let unaligned = UNALIGNED_FRAME; + + assert_eq!( + u32::from_ne_bytes(undersized[4..].try_into().unwrap()) >> 16, + 4 + ); + assert_eq!( + u32::from_ne_bytes(unaligned[4..8].try_into().unwrap()) >> 16, + 9 + ); + assert_eq!(unaligned.len(), 9); + } +}