diff --git a/Cargo.lock b/Cargo.lock index b32d7b3..aa0a5c9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -67,6 +67,18 @@ version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" +[[package]] +name = "bigfred-shared-daemon" +version = "0.1.0" +source = "git+https://github.com/dcc-bigfred/rust-commons.git?branch=main#5f5053ef9316139c7cde500e6e9e85a70dbbbd09" +dependencies = [ + "log", + "nix 0.29.0", + "serde", + "serde_json", + "thiserror 2.0.20", +] + [[package]] name = "bitflags" version = "2.13.1" @@ -1031,8 +1043,9 @@ dependencies = [ [[package]] name = "wireless-programmer" -version = "0.1.0" +version = "0.2.0" dependencies = [ + "bigfred-shared-daemon", "clap", "humantime", "libc", @@ -1073,8 +1086,9 @@ dependencies = [ [[package]] name = "wp-client" -version = "0.1.0" +version = "0.2.0" dependencies = [ + "bigfred-shared-daemon", "serde", "serde_json", "thiserror 2.0.20", @@ -1083,7 +1097,7 @@ dependencies = [ [[package]] name = "wp-core" -version = "0.1.0" +version = "0.2.0" dependencies = [ "serde", "serde_json", @@ -1093,7 +1107,7 @@ dependencies = [ [[package]] name = "wp-drivers" -version = "0.1.0" +version = "0.2.0" dependencies = [ "log", "quick-xml", @@ -1108,7 +1122,7 @@ dependencies = [ [[package]] name = "wp-fake" -version = "0.1.0" +version = "0.2.0" dependencies = [ "log", "parking_lot", @@ -1123,7 +1137,7 @@ dependencies = [ [[package]] name = "wp-link" -version = "0.1.0" +version = "0.2.0" dependencies = [ "futures", "libc", @@ -1140,11 +1154,11 @@ dependencies = [ [[package]] name = "wp-proto" -version = "0.1.0" +version = "0.2.0" dependencies = [ + "bigfred-shared-daemon", "serde", "serde_json", - "thiserror 2.0.20", ] [[package]] diff --git a/Makefile b/Makefile index 0fe4bf9..ba9ac0f 100644 --- a/Makefile +++ b/Makefile @@ -8,7 +8,7 @@ export RUSTUP_TOOLCHAIN # Optional wireless iface for `make dev` (e.g. INTERFACE=wlan0). INTERFACE ?= -.PHONY: all build release release-musl check test test-release-assertions clean fmt clippy dev +.PHONY: all build release release-musl check test test-release-assertions clean fmt clippy dev deps-update all: build @@ -54,6 +54,10 @@ fmt: clippy: $(CARGO) clippy --workspace --all-targets --locked -- -D warnings +# Refresh git crates (bigfred-shared-daemon) and rewrite Cargo.lock. Commit the lockfile afterwards. +deps-update: + $(CARGO) update -p bigfred-shared-daemon + clean: $(CARGO) clean rm -rf dist diff --git a/crates/wireless-programmer/Cargo.toml b/crates/wireless-programmer/Cargo.toml index 361fcb6..7e753e8 100644 --- a/crates/wireless-programmer/Cargo.toml +++ b/crates/wireless-programmer/Cargo.toml @@ -28,7 +28,8 @@ clap = { version = "4", features = ["derive"] } humantime = "2" thiserror = "2" tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "signal", "time", "sync", "io-util"] } -nix = { version = "0.29", features = ["user", "socket", "fs"] } +nix = { version = "0.29", features = ["user", "fs"] } +bigfred-shared-daemon = { git = "https://github.com/dcc-bigfred/rust-commons.git", branch = "main", default-features = false, features = ["ipc"] } libc = "0.2" parking_lot = "0.12" tracing = "0.1" diff --git a/crates/wireless-programmer/src/config.rs b/crates/wireless-programmer/src/config.rs index 0a32a74..bc47d69 100644 --- a/crates/wireless-programmer/src/config.rs +++ b/crates/wireless-programmer/src/config.rs @@ -196,13 +196,12 @@ fn resolve_fake_web_port_env() -> Option { /// Resolve the BigFred data directory. pub fn resolve_data_dir() -> PathBuf { - if let Ok(d) = std::env::var("BIGFRED_DATA_DIR") { - return PathBuf::from(d); - } - if let Ok(d) = std::env::var("DATA_DIR") { - return PathBuf::from(d); - } - PathBuf::from("/data") + bigfred_shared_daemon::DataDir::resolve( + bigfred_shared_daemon::EnvPolicy::BigfredThenDataDir, + bigfred_shared_daemon::PathRule::AcceptAny, + ) + .as_path() + .to_path_buf() } #[cfg(test)] diff --git a/crates/wireless-programmer/src/ipc.rs b/crates/wireless-programmer/src/ipc.rs index 30d757c..59cd4b4 100644 --- a/crates/wireless-programmer/src/ipc.rs +++ b/crates/wireless-programmer/src/ipc.rs @@ -7,16 +7,16 @@ //! checked against an allowlist via `SO_PEERCRED`. use std::io; -use std::os::unix::fs::PermissionsExt; -use std::os::unix::net::{UnixListener, UnixStream}; use std::path::Path; use std::sync::Arc; -use nix::sys::socket::{getsockopt, sockopt::PeerCredentials}; -use nix::unistd::{chown, Gid}; -use wp_proto::{ - read_frame, write_frame, ErrorBody, Params, Request, RequestKind, Response, ResultBody, +use bigfred_shared_daemon::ipc::{ + AcceptPolicy, Auth, BindOptions, Command, Connection, ErrorHandler, IpcError, RejectReason, + Router, SessionMode, }; +use nix::unistd::{chown, Gid}; +use serde_json::Value; +use wp_proto::{ErrorBody, Params, Request, RequestKind, Response, ResultBody, MAX_FRAME_BYTES}; use crate::config::Config; use crate::jobs::JobState; @@ -39,138 +39,189 @@ impl Server { /// /// Returns [`io::Error`] on bind/listen failure. pub fn run(self) -> io::Result<()> { - let socket = self.runtime.config().socket.clone(); - if let Some(parent) = socket.parent() { - std::fs::create_dir_all(parent)?; - } - if socket.exists() { - std::fs::remove_file(&socket)?; - } - let listener = UnixListener::bind(&socket)?; - let perms = std::fs::Permissions::from_mode(self.runtime.config().socket_mode); - std::fs::set_permissions(&socket, perms)?; - set_socket_group(&socket, self.runtime.config()); + let cfg = self.runtime.config(); + let socket = cfg.socket.clone(); + let listener = bigfred_shared_daemon::ipc::bind(&BindOptions { + path: socket.clone(), + mode: cfg.socket_mode, + chown: None, + process_name: "wireless-programmer", + }) + .map_err(|e| io::Error::other(e.to_string()))?; + set_socket_group(&socket, cfg); tracing::info!("listening on {}", socket.display()); - let inner = Arc::new(ServerInner { + let auth = if cfg.require_auth { + Auth::PeerUser { + allow_users: cfg.allow_users.clone(), + root_always: false, + fail_closed_if_empty: true, + } + } else { + Auth::None + }; + + let server = bigfred_shared_daemon::ipc::Server::from_listener( + listener, + socket, + AcceptPolicy { + auth, + session: SessionMode::Persistent, + max_clients: None, + max_frame: MAX_FRAME_BYTES, + }, + wp_router().map_err(io::Error::other)?, + WpHooks, + ); + server.serve(Arc::new(ServerInner { runtime: self.runtime, - }); - - for stream in listener.incoming() { - let stream = stream?; - let inner = Arc::clone(&inner); - std::thread::spawn(move || { - if let Err(e) = inner.handle_conn(stream) { - tracing::warn!("connection error: {e}"); - } - }); - } + })); Ok(()) } } -struct ServerInner { - runtime: Arc, -} - -impl ServerInner { - fn handle_conn(&self, mut stream: UnixStream) -> io::Result<()> { - if !self.peer_allowed(&stream) { - let _ = write_frame( - &mut stream, - &err_response(RequestKind::Hello, "forbidden", "peer not allowed"), - ); - return Ok(()); - } - loop { - let req: Request = match read_frame(&mut stream) { - Ok(r) => r, - Err(wp_proto::FrameError::UnexpectedEof { .. }) => return Ok(()), - Err(e) => { - tracing::warn!("frame read error: {e}"); - return Ok(()); - } - }; - // JobWatch streams many frames on one connection until terminal. - if req.kind == RequestKind::JobWatch { - if let Err(e) = self.stream_job_watch(&mut stream, req) { - tracing::warn!("job.watch stream error: {e}"); - } - return Ok(()); +macro_rules! wp_cmd { + ($ty:ident, $name:literal) => { + struct $ty; + impl Command for $ty { + fn name(&self) -> &'static str { + $name } - let resp = self.dispatch(req); - if let Err(e) = write_frame(&mut stream, &resp) { - tracing::warn!("frame write error: {e}"); - return Ok(()); + fn execute( + &self, + inner: &ServerInner, + body: Value, + conn: &mut Connection, + ) -> Result<(), IpcError> { + let req: Request = + serde_json::from_value(body).map_err(|e| IpcError::Other(e.to_string()))?; + let resp = inner.dispatch(req); + conn.reply(&resp).map_err(IpcError::from) } } + }; +} + +wp_cmd!(HelloCmd, "hello"); +wp_cmd!(ScanCmd, "scan"); +wp_cmd!(ProbeCmd, "probe"); +wp_cmd!(ProgramCmd, "program"); +wp_cmd!(JobGetCmd, "job.get"); +wp_cmd!(JobCancelCmd, "job.cancel"); +wp_cmd!(IdentifyCmd, "identify"); +wp_cmd!(LinkStatusCmd, "link.status"); +wp_cmd!(UpdateFirmwareCmd, "updateFirmware"); + +struct JobWatchCmd; + +impl Command for JobWatchCmd { + fn name(&self) -> &'static str { + "job.watch" + } + fn execute( + &self, + inner: &ServerInner, + body: Value, + conn: &mut Connection, + ) -> Result<(), IpcError> { + let req: Request = + serde_json::from_value(body).map_err(|e| IpcError::Other(e.to_string()))?; + inner.stream_job_watch(conn, req)?; + // Original handler closed the connection after the stream. + Err(IpcError::Hangup) } +} - fn peer_allowed(&self, stream: &UnixStream) -> bool { - let cfg = self.runtime.config(); - if !cfg.require_auth { - return true; - } - if cfg.allow_users.is_empty() { - // Auth on with an empty list should never happen after - // finalize_auth, but fail closed. - return false; - } - let creds = match getsockopt(stream, PeerCredentials) { - Ok(c) => c, - Err(_) => return false, - }; - let uid = creds.uid(); - let name = username_for_uid(uid); - match name { - Some(n) => cfg.allow_users.iter().any(|u| u == &n), - None => false, +struct WpHooks; + +impl ErrorHandler for WpHooks { + fn unknown(&self, _state: &ServerInner, type_name: &str, _body: &Value, conn: &mut Connection) { + tracing::warn!("unknown ipc type {type_name}"); + let _ = conn.reply(&err_response( + RequestKind::Hello, + "unknown_type", + &format!("unknown type `{type_name}`"), + )); + } + fn error(&self, _state: &ServerInner, err: &IpcError, conn: &mut Connection) { + tracing::warn!("connection error: {err}"); + let _ = conn.reply(&err_response( + RequestKind::Hello, + "internal", + &err.to_string(), + )); + } + fn reject(&self, _state: &ServerInner, reason: RejectReason, conn: &mut Connection) { + if reason == RejectReason::Auth { + let _ = conn.reply(&err_response( + RequestKind::Hello, + "forbidden", + "peer not allowed", + )); } } +} - fn stream_job_watch(&self, stream: &mut UnixStream, req: Request) -> io::Result<()> { - let write = |stream: &mut UnixStream, resp: &Response| { - write_frame(stream, resp).map_err(|e| io::Error::other(e.to_string())) - }; +fn wp_router() -> Result, bigfred_shared_daemon::ipc::DuplicateCommand> { + let mut router = Router::new(); + router.add(HelloCmd)?; + router.add(ScanCmd)?; + router.add(ProbeCmd)?; + router.add(ProgramCmd)?; + router.add(JobGetCmd)?; + router.add(JobWatchCmd)?; + router.add(JobCancelCmd)?; + router.add(IdentifyCmd)?; + router.add(LinkStatusCmd)?; + router.add(UpdateFirmwareCmd)?; + Ok(router) +} + +struct ServerInner { + runtime: Arc, +} + +impl ServerInner { + fn stream_job_watch(&self, conn: &mut Connection, req: Request) -> Result<(), IpcError> { let job_id = match req.params { Some(Params::Job(p)) => crate::jobs::JobId(p.job_id), _ => { - write( - stream, - &err_response(RequestKind::JobWatch, "bad_params", "missing params"), - )?; + conn.reply(&err_response( + RequestKind::JobWatch, + "bad_params", + "missing params", + ))?; return Ok(()); } }; if self.runtime.jobs().snapshot(&job_id).is_none() { - write( - stream, - &err_response(RequestKind::JobWatch, "not_found", "no such job"), - )?; + conn.reply(&err_response( + RequestKind::JobWatch, + "not_found", + "no such job", + ))?; return Ok(()); } let mut since = 0usize; let mut sent_snapshot = false; loop { let Some(frames) = self.runtime.jobs().frames_since(&job_id, since) else { - write( - stream, - &err_response(RequestKind::JobWatch, "not_found", "no such job"), - )?; + conn.reply(&err_response( + RequestKind::JobWatch, + "not_found", + "no such job", + ))?; return Ok(()); }; let mut terminal = false; for f in &frames { let wire = job_frame_to_wire(f); terminal = wire.state.is_terminal(); - write( - stream, - &Response { - kind: RequestKind::JobWatch, - result: Some(ResultBody::JobWatch(wire)), - error: None, - }, - )?; + conn.reply(&Response { + kind: RequestKind::JobWatch, + result: Some(ResultBody::JobWatch(wire)), + error: None, + })?; } since += frames.len(); if terminal { @@ -183,14 +234,11 @@ impl ServerInner { if let Some(s) = self.runtime.jobs().snapshot(&job_id) { let wire = snapshot_to_frame(s); let terminal = wire.state.is_terminal(); - write( - stream, - &Response { - kind: RequestKind::JobWatch, - result: Some(ResultBody::JobWatch(wire)), - error: None, - }, - )?; + conn.reply(&Response { + kind: RequestKind::JobWatch, + result: Some(ResultBody::JobWatch(wire)), + error: None, + })?; sent_snapshot = true; if terminal { return Ok(()); @@ -627,15 +675,6 @@ fn parse_passwd(content: &str) -> impl Iterator + '_ { }) } -/// Resolve a uid to a username via `/etc/passwd`. -fn username_for_uid(uid: u32) -> Option { - let content = std::fs::read_to_string("/etc/passwd").ok()?; - let found = parse_passwd(&content) - .find(|e| e.uid == uid) - .map(|e| e.name); - found -} - /// Resolve a login name's primary gid via `/etc/passwd`. fn primary_gid_for_user(name: &str) -> Option { let content = std::fs::read_to_string("/etc/passwd").ok()?; @@ -799,4 +838,18 @@ bigfred:x:1000:1001:BigFred loco-server:/home/bigfred:/bin/false }; assert_eq!(cfg.socket_group_owner(), None); } + + #[test] + fn error_frame_is_a_readable_response() { + let (mut a, mut b) = std::os::unix::net::UnixStream::pair().unwrap(); + wp_proto::write_frame( + &mut a, + &err_response(RequestKind::Hello, "unknown_type", "unknown type `nope`"), + ) + .unwrap(); + let resp: Response = wp_proto::read_frame(&mut b).unwrap(); + let err = resp.error.expect("error body"); + assert_eq!(err.code, "unknown_type"); + assert!(err.message.contains("nope")); + } } diff --git a/crates/wp-client/Cargo.toml b/crates/wp-client/Cargo.toml index 33bb772..d0fb229 100644 --- a/crates/wp-client/Cargo.toml +++ b/crates/wp-client/Cargo.toml @@ -12,6 +12,7 @@ wp-proto = { path = "../wp-proto" } serde = { version = "1", features = ["derive"] } serde_json = "1" thiserror = "2" +bigfred-shared-daemon = { git = "https://github.com/dcc-bigfred/rust-commons.git", branch = "main", default-features = false } [lints] workspace = true diff --git a/crates/wp-client/src/client.rs b/crates/wp-client/src/client.rs index 7e9d4aa..e947189 100644 --- a/crates/wp-client/src/client.rs +++ b/crates/wp-client/src/client.rs @@ -50,19 +50,11 @@ impl Client { /// Resolve the daemon socket path from the environment. pub fn resolve_socket() -> PathBuf { - if let Ok(d) = std::env::var("BIGFRED_DATA_DIR") { - return PathBuf::from(d) - .join("run") - .join("wireless-programmer") - .join("wireless-programmer.sock"); - } - if let Ok(d) = std::env::var("DATA_DIR") { - return PathBuf::from(d) - .join("run") - .join("wireless-programmer") - .join("wireless-programmer.sock"); - } - PathBuf::from(DEFAULT_SOCKET) + bigfred_shared_daemon::DataDir::resolve( + bigfred_shared_daemon::EnvPolicy::BigfredThenDataDir, + bigfred_shared_daemon::PathRule::AcceptAny, + ) + .run_nested_socket("wireless-programmer") } fn connect(&self) -> Result { diff --git a/crates/wp-proto/Cargo.toml b/crates/wp-proto/Cargo.toml index 5bc8c84..962112a 100644 --- a/crates/wp-proto/Cargo.toml +++ b/crates/wp-proto/Cargo.toml @@ -14,7 +14,7 @@ path = "src/lib.rs" [dependencies] serde = { version = "1", features = ["derive"] } serde_json = "1" -thiserror = "2" +bigfred-shared-daemon = { git = "https://github.com/dcc-bigfred/rust-commons.git", branch = "main", default-features = false, features = ["ipc"] } [lints] workspace = true diff --git a/crates/wp-proto/src/framing.rs b/crates/wp-proto/src/framing.rs index 9790e07..be69735 100644 --- a/crates/wp-proto/src/framing.rs +++ b/crates/wp-proto/src/framing.rs @@ -1,114 +1,14 @@ //! Length-prefixed JSON framing codec. -use std::io::{self, Read, Write}; - -use serde::de::DeserializeOwned; -use serde::Serialize; -use thiserror::Error; - -/// Absolute upper bound on a single frame payload: 1 MiB. -pub const MAX_FRAME_BYTES: usize = 1024 * 1024; - -const HEADER_LEN: usize = 4; - -/// Errors raised by the framing codec. -#[derive(Debug, Error)] -pub enum FrameError { - /// The payload length exceeded [`MAX_FRAME_BYTES`]. - #[error("frame payload of {len} bytes exceeds {max} bytes")] - TooLarge { - /// Declared payload length. - len: usize, - /// Configured maximum ([`MAX_FRAME_BYTES`]). - max: usize, - }, - /// The connection closed before a full header or payload was read. - #[error("connection closed: read {read} of {needed} bytes")] - UnexpectedEof { - /// Bytes actually read. - read: usize, - /// Bytes needed to complete the frame. - needed: usize, - }, - /// The JSON payload could not be (de)serialised. - #[error("json codec error: {0}")] - Json(#[from] serde_json::Error), - /// An underlying I/O failure. - #[error("io error: {0}")] - Io(#[from] io::Error), -} - -/// Serialise `msg` and write one framed message to `writer`. -/// -/// # Errors -/// -/// Returns [`FrameError::TooLarge`] when the serialised payload exceeds -/// [`MAX_FRAME_BYTES`], and propagates I/O and JSON codec errors. -pub fn write_frame(writer: &mut W, msg: &T) -> Result<(), FrameError> -where - W: Write, - T: Serialize, -{ - let payload = serde_json::to_vec(msg)?; - let len = payload.len(); - if len > MAX_FRAME_BYTES { - return Err(FrameError::TooLarge { - len, - max: MAX_FRAME_BYTES, - }); - } - let header = u32::try_from(len).unwrap_or(u32::MAX).to_le_bytes(); - writer.write_all(&header)?; - writer.write_all(&payload)?; - writer.flush()?; - Ok(()) -} - -/// Read one framed message from `reader` and decode it as `T`. -/// -/// # Errors -/// -/// Returns [`FrameError::TooLarge`] when the declared length exceeds -/// [`MAX_FRAME_BYTES`], [`FrameError::UnexpectedEof`] on a short read, and -/// propagates I/O and JSON codec errors. -pub fn read_frame(reader: &mut R) -> Result -where - R: Read, - T: DeserializeOwned, -{ - let mut header = [0u8; HEADER_LEN]; - read_exact(reader, &mut header)?; - let len = u32::from_le_bytes(header) as usize; - if len > MAX_FRAME_BYTES { - return Err(FrameError::TooLarge { - len, - max: MAX_FRAME_BYTES, - }); - } - let mut payload = vec![0u8; len]; - read_exact(reader, &mut payload)?; - Ok(serde_json::from_slice(&payload)?) -} - -fn read_exact(reader: &mut R, buf: &mut [u8]) -> Result<(), FrameError> { - let mut filled = 0usize; - while filled < buf.len() { - let n = reader.read(&mut buf[filled..])?; - if n == 0 { - return Err(FrameError::UnexpectedEof { - read: filled, - needed: buf.len(), - }); - } - filled += n; - } - Ok(()) -} +pub use bigfred_shared_daemon::ipc::{ + read_frame, write_frame, FrameError, DEFAULT_MAX_FRAME_BYTES as MAX_FRAME_BYTES, +}; #[cfg(test)] mod tests { use super::*; use serde::Deserialize; + use serde::Serialize; #[derive(Debug, PartialEq, Serialize, Deserialize)] struct Msg {