From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from gate001.proxmox.com (gate001.proxmox.com [45.144.208.40]) by lore.proxmox.com (Postfix) with ESMTPS id AF9351FF0C2 for ; Thu, 22 Oct 2026 16:23:44 +0200 (CEST) Received: from gate001.proxmox.com (localhost.localdomain [127.0.0.1]) by gate001.proxmox.com (Proxmox) with ESMTP id 582DF21784; Thu, 22 Oct 2026 16:23:02 +0200 (CEST) From: Hannes Laimer To: pve-devel@lists.proxmox.com Subject: [PATCH pve-cluster 06/12] rust: ffi: add C ABI staticlib for pmxcfs Date: Thu, 22 Oct 2026 16:21:59 +0200 Message-ID: <20261022142205.930536-7-h.laimer@proxmox.com> X-Mailer: git-send-email 2.47.3 In-Reply-To: <20261022142205.930536-1-h.laimer@proxmox.com> References: <20261022142205.930536-1-h.laimer@proxmox.com> MIME-Version: 1.0 Content-Transfer-Encoding: 8bit X-Bm-Milter-Handled: 55990f41-d878-4baa-be0a-ee34c49e34d2 X-Bm-Transport-Timestamp: 1792678934087 X-SPAM-LEVEL: Spam detection results: 0 AWL 0.342 Adjusted score from AWL reputation of From: address DMARC_MISSING 0.1 Missing DMARC policy KAM_DMARC_STATUS 0.01 Test Rule for DKIM or SPF Failure with Strict Alignment (newer systems) RCVD_IN_DNSWL_MED -2.3 Sender listed at https://www.dnswl.org/, medium trust SPF_HELO_NONE 0.001 SPF: HELO does not publish an SPF Record SPF_PASS -0.001 SPF: sender matches SPF record Message-ID-Hash: QQKBB4MBJRQNGVMC3AM6ZMIDGL5J7UZM X-Message-ID-Hash: QQKBB4MBJRQNGVMC3AM6ZMIDGL5J7UZM X-MailFrom: h.laimer@proxmox.com X-Mailman-Rule-Misses: dmarc-mitigation; no-senders; approved; loop; banned-address; emergency; member-moderation; nonmember-moderation; administrivia; implicit-dest; max-recipients; max-size; news-moderation; no-subject; digests; suspicious-header X-Mailman-Version: 3.3.10 Precedence: list List-Id: Proxmox VE development discussion List-Help: List-Owner: List-Post: List-Subscribe: List-Unsubscribe: pmxcfs is a C daemon, so the notification socket gets a C interface in a static library the daemon can link. Three functions start the socket, emit an event and shut the socket down. Log messages go with a syslog priority to a callback from the caller, so they can end up in the daemon's own log. The event type crosses the boundary as the Rust enum itself, laid out as a C enum, so nothing translates between the two sides. Rust aborts the process when a panic reaches the C caller. In pmxcfs that would take /etc/pve down because of a bug in the notification socket. So every entry point catches a panic. A panic while starting or emitting disables notifications for the rest of the daemon's life and shuts the socket down. Clients then see EOF and the socket file gone, instead of waiting on a connection nobody serves. Signed-off-by: Hannes Laimer --- src/rust/Cargo.toml | 1 + src/rust/pmxcfs-ffi/Cargo.toml | 17 ++ src/rust/pmxcfs-ffi/include/pmxcfs-notify.h | 24 ++ src/rust/pmxcfs-ffi/src/lib.rs | 306 ++++++++++++++++++++ src/rust/pmxcfs-notify/src/protocol.rs | 6 +- 5 files changed, 353 insertions(+), 1 deletion(-) create mode 100644 src/rust/pmxcfs-ffi/Cargo.toml create mode 100644 src/rust/pmxcfs-ffi/include/pmxcfs-notify.h create mode 100644 src/rust/pmxcfs-ffi/src/lib.rs diff --git a/src/rust/Cargo.toml b/src/rust/Cargo.toml index 8cf928e..55e0565 100644 --- a/src/rust/Cargo.toml +++ b/src/rust/Cargo.toml @@ -1,5 +1,6 @@ [workspace] members = [ + "pmxcfs-ffi", "pmxcfs-notify", ] resolver = "3" diff --git a/src/rust/pmxcfs-ffi/Cargo.toml b/src/rust/pmxcfs-ffi/Cargo.toml new file mode 100644 index 0000000..8264cde --- /dev/null +++ b/src/rust/pmxcfs-ffi/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "pmxcfs-ffi" +version = "0.1.0" +description = "C ABI bridge exposing pmxcfs-notify to the pmxcfs daemon" +authors.workspace = true +edition.workspace = true +license.workspace = true +homepage.workspace = true +rust-version.workspace = true + +[lib] +crate-type = ["staticlib"] + +[dependencies] +libc.workspace = true +log.workspace = true +pmxcfs-notify.workspace = true diff --git a/src/rust/pmxcfs-ffi/include/pmxcfs-notify.h b/src/rust/pmxcfs-ffi/include/pmxcfs-notify.h new file mode 100644 index 0000000..8e0f0ce --- /dev/null +++ b/src/rust/pmxcfs-ffi/include/pmxcfs-notify.h @@ -0,0 +1,24 @@ +#ifndef PMXCFS_NOTIFY_H +#define PMXCFS_NOTIFY_H + +#include + +/* mirrored by EventKind on the Rust side, the values must stay the same */ +enum pmxcfs_notify_type { + PMXCFS_NOTIFY_CREATE = 0, + PMXCFS_NOTIFY_WRITE = 1, + PMXCFS_NOTIFY_MTIME = 2, + PMXCFS_NOTIFY_RENAME = 3, + PMXCFS_NOTIFY_DELETE = 4, + PMXCFS_NOTIFY_MKDIR = 5, + PMXCFS_NOTIFY_RESYNC = 6, +}; + +/* priority is a syslog(3) level */ +typedef void (*pmxcfs_notify_log_fn)(int priority, const char *msg); + +int pmxcfs_notify_init(const char *socket_path, gid_t gid, pmxcfs_notify_log_fn log_cb); +void pmxcfs_notify_emit(enum pmxcfs_notify_type type, const char *path, const char *to); +void pmxcfs_notify_shutdown(void); + +#endif /* PMXCFS_NOTIFY_H */ diff --git a/src/rust/pmxcfs-ffi/src/lib.rs b/src/rust/pmxcfs-ffi/src/lib.rs new file mode 100644 index 0000000..7291d7d --- /dev/null +++ b/src/rust/pmxcfs-ffi/src/lib.rs @@ -0,0 +1,306 @@ +//! C ABI for the pmxcfs daemon, declared in include/pmxcfs-notify.h. +//! +//! Rust aborts the process when a panic reaches an `extern "C"` frame, which must never happen to +//! pmxcfs. Every entry point therefore runs under `catch_unwind`, and a panic in one switches the +//! notifier off for the rest of the daemon's life instead of taking it down. + +use std::ffi::{CStr, CString, OsStr, c_char, c_int}; +use std::os::unix::ffi::OsStrExt; +use std::panic::{UnwindSafe, catch_unwind}; +use std::path::Path; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Mutex, MutexGuard, Once}; + +use log::{Level, LevelFilter, Log, Metadata, Record}; +use pmxcfs_notify::{Config, EventKind, Server}; + +pub type LogFn = unsafe extern "C" fn(c_int, *const c_char); + +const CAPACITY: usize = 8192; + +static SERVER: Mutex> = Mutex::new(None); +static DISABLED: AtomicBool = AtomicBool::new(false); +static LOG_CB: Mutex> = Mutex::new(None); +static LOGGER: CLogger = CLogger; +static LOGGER_INSTALLED: Once = Once::new(); + +struct CLogger; + +impl Log for CLogger { + fn enabled(&self, _: &Metadata) -> bool { + true + } + + fn log(&self, record: &Record) { + let Some(callback) = *lock(&LOG_CB) else { + return; + }; + let priority = match record.level() { + Level::Error => libc::LOG_ERR, + Level::Warn => libc::LOG_WARNING, + Level::Info => libc::LOG_INFO, + Level::Debug | Level::Trace => libc::LOG_DEBUG, + }; + let msg = CString::new(record.args().to_string().replace('\0', " ")) + .expect("no NUL left after the replace"); + unsafe { callback(priority, msg.as_ptr()) }; + } + + fn flush(&self) {} +} + +fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { + mutex + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +fn guard(what: &str, fallback: R, f: impl FnOnce() -> R + UnwindSafe) -> R { + if DISABLED.load(Ordering::Relaxed) { + return fallback; + } + match catch_unwind(f) { + Ok(result) => result, + Err(payload) => { + DISABLED.store(true, Ordering::Relaxed); + let reason = payload + .downcast_ref::<&str>() + .copied() + .or_else(|| payload.downcast_ref::().map(String::as_str)) + .unwrap_or(""); + let _ = catch_unwind(|| { + log::error!("{what} panicked, notifications disabled: {reason}"); + }); + let _ = catch_unwind(stop_server); + fallback + } + } +} + +fn stop_server() { + let server = lock(&SERVER).take(); + if let Some(server) = server { + server.shutdown(); + } +} + +unsafe fn c_str<'a>(ptr: *const c_char) -> Option<&'a CStr> { + if ptr.is_null() { + return None; + } + Some(unsafe { CStr::from_ptr(ptr) }) +} + +/// # Safety +/// +/// `socket_path` must be NULL or point to a NUL terminated string. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn pmxcfs_notify_init( + socket_path: *const c_char, + gid: libc::gid_t, + log_cb: Option, +) -> c_int { + guard("init", -1, || { + let Some(path) = (unsafe { c_str(socket_path) }) else { + return -1; + }; + let path = Path::new(OsStr::from_bytes(path.to_bytes())); + *lock(&LOG_CB) = log_cb; + LOGGER_INSTALLED.call_once(|| { + let _ = log::set_logger(&LOGGER); + log::set_max_level(LevelFilter::Debug); + }); + + let mut server = lock(&SERVER); + if server.is_some() { + log::warn!("notification socket is already running"); + return 0; + } + match Server::start(Config { + path: path.to_owned(), + gid, + capacity: CAPACITY, + }) { + Ok(started) => { + *server = Some(started); + log::info!("listening on {}", path.display()); + 0 + } + Err(err) => { + log::error!("failed to start notification socket: {err:#}"); + -1 + } + } + }) +} + +/// # Safety +/// +/// `kind` must be a value of enum pmxcfs_notify_type, and `path` and `to` must each be NULL or +/// point to a NUL terminated string. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn pmxcfs_notify_emit( + kind: EventKind, + path: *const c_char, + to: *const c_char, +) { + guard("emit", (), || { + let server = lock(&SERVER); + let Some(server) = server.as_ref() else { + return; + }; + match kind { + EventKind::Create + | EventKind::Write + | EventKind::Mtime + | EventKind::Rename + | EventKind::Delete + | EventKind::Mkdir + | EventKind::Resync => {} + // #[non_exhaustive] requires this arm. It cannot catch a bad value from C, which is + // undefined behavior before this point + _ => { + log::warn!("ignoring event with unknown type"); + return; + } + } + let path = unsafe { c_str(path) }.map(CStr::to_str).transpose(); + let to = unsafe { c_str(to) }.map(CStr::to_str).transpose(); + match (path, to) { + (Ok(path), Ok(to)) => server.emit(kind, path, to), + // a JSON string holds only Unicode, so a path that is not UTF-8 is dropped rather than + // sent mangled + _ => log::warn!("ignoring {kind:?} event for a path that is not UTF-8"), + } + }) +} + +#[unsafe(no_mangle)] +pub extern "C" fn pmxcfs_notify_shutdown() { + let _ = catch_unwind(|| { + stop_server(); + *lock(&LOG_CB) = None; + }); +} + +#[cfg(test)] +mod tests { + use std::io::{BufRead, BufReader, Read, Write}; + use std::os::unix::fs::MetadataExt; + use std::os::unix::net::UnixStream; + use std::time::Duration; + + use super::*; + + static LINES: Mutex> = Mutex::new(Vec::new()); + // the tests share the global notifier state, so they run one at a time + static SERIAL: Mutex<()> = Mutex::new(()); + + unsafe extern "C" fn collect(priority: c_int, msg: *const c_char) { + let msg = unsafe { CStr::from_ptr(msg) } + .to_string_lossy() + .into_owned(); + lock(&LINES).push((priority, msg)); + } + + fn warnings_containing(text: &str) -> usize { + lock(&LINES) + .iter() + .filter(|(prio, msg)| *prio == libc::LOG_WARNING && msg.contains(text)) + .count() + } + + #[test] + fn init_emit_shutdown_cycle() { + let _serial = lock(&SERIAL); + lock(&LINES).clear(); + let dir = std::env::temp_dir().join(format!("pmxcfs-ffi-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let gid = std::fs::metadata(&dir).unwrap().gid(); + let socket = CString::new(dir.join("sock").to_str().unwrap()).unwrap(); + + for _ in 0..2 { + let rc = unsafe { pmxcfs_notify_init(socket.as_ptr(), gid, Some(collect)) }; + assert_eq!(rc, 0); + let path = CString::new("nodes/n1/qemu-server/100.conf").unwrap(); + let odd = CString::new(b"nodes/n1/\xff.conf".to_vec()).unwrap(); + unsafe { + pmxcfs_notify_emit(EventKind::Write, path.as_ptr(), std::ptr::null()); + pmxcfs_notify_emit(EventKind::Resync, std::ptr::null(), std::ptr::null()); + pmxcfs_notify_emit(EventKind::Delete, std::ptr::null(), std::ptr::null()); + pmxcfs_notify_emit(EventKind::Rename, std::ptr::null(), path.as_ptr()); + pmxcfs_notify_emit(EventKind::Write, odd.as_ptr(), std::ptr::null()); + pmxcfs_notify_emit(EventKind::Rename, path.as_ptr(), odd.as_ptr()); + } + pmxcfs_notify_shutdown(); + } + + let rc = unsafe { pmxcfs_notify_init(std::ptr::null(), gid, Some(collect)) }; + assert_eq!(rc, -1); + let _ = std::fs::remove_dir_all(&dir); + + let listening = lock(&LINES) + .iter() + .filter(|(prio, msg)| *prio == libc::LOG_INFO && msg.starts_with("listening on ")) + .count(); + assert_eq!(listening, 2); + assert_eq!(warnings_containing("Delete event without a path"), 2); + assert_eq!(warnings_containing("Rename event without a path"), 2); + assert_eq!( + warnings_containing("Write event for a path that is not UTF-8"), + 2 + ); + assert_eq!( + warnings_containing("Rename event for a path that is not UTF-8"), + 2 + ); + } + + #[test] + fn panic_closes_the_connections() { + let _serial = lock(&SERIAL); + lock(&LINES).clear(); + let dir = std::env::temp_dir().join(format!("pmxcfs-ffi-panic-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let gid = std::fs::metadata(&dir).unwrap().gid(); + let path = dir.join("sock"); + let socket = CString::new(path.to_str().unwrap()).unwrap(); + assert_eq!( + unsafe { pmxcfs_notify_init(socket.as_ptr(), gid, Some(collect)) }, + 0 + ); + let mut client = UnixStream::connect(&path).unwrap(); + client + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + // a reply proves the server accepted the connection. Stopping the server after the panic + // below then closes an established socket, not a pending one the peer would see as a reset + client.write_all(b"{\"command\":\"hello\"}\n").unwrap(); + let mut reply = String::new(); + BufReader::new(client.try_clone().unwrap()) + .read_line(&mut reply) + .unwrap(); + assert!(reply.starts_with("{\"ok\""), "{reply}"); + + assert_eq!(guard("test", -1, || -> c_int { panic!("injected") }), -1); + + assert!(DISABLED.load(Ordering::Relaxed)); + assert!(lock(&SERVER).is_none()); + assert!(!path.exists()); + let mut sink = Vec::new(); + assert_eq!(client.read_to_end(&mut sink).unwrap(), 0, "client sees EOF"); + assert!( + lock(&LINES) + .iter() + .any(|(prio, msg)| *prio == libc::LOG_ERR && msg.contains("test panicked")) + ); + + // a disabled notifier ignores everything, including a new init + assert_eq!( + unsafe { pmxcfs_notify_init(socket.as_ptr(), gid, Some(collect)) }, + -1 + ); + DISABLED.store(false, Ordering::Relaxed); + let _ = std::fs::remove_dir_all(&dir); + } +} diff --git a/src/rust/pmxcfs-notify/src/protocol.rs b/src/rust/pmxcfs-notify/src/protocol.rs index d28c9dc..e706cda 100644 --- a/src/rust/pmxcfs-notify/src/protocol.rs +++ b/src/rust/pmxcfs-notify/src/protocol.rs @@ -10,6 +10,9 @@ pub(crate) const PROTOCOL_VERSION: u32 = 1; /// Keyed by subscription name, with one capture set per side of the event that matched. pub(crate) type Params<'a> = BTreeMap<&'a str, Vec>>; +// Crosses the C ABI by value, so the order has to match enum pmxcfs_notify_type in the pmxcfs-ffi +// header. +#[repr(C)] #[non_exhaustive] #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] #[serde(rename_all = "lowercase")] @@ -35,7 +38,8 @@ pub(crate) struct Event<'a> { pub params: Option>, } -/// Everything the server sends. The variant name is the single key of the JSON object on the wire. +/// Everything the server sends. The variant name is the single key of +/// the JSON object on the wire. #[derive(Serialize)] #[serde(rename_all = "lowercase")] pub(crate) enum Message<'a> { -- 2.47.3