rwhod: react to audit pam events to broadcast updates

Listen to read-only audit packets from the kernel to react on user
logins/logouts. This way, we can immediately notify other machines on
the network whenever there is a change to the user session list.

src/bin/roowhod.rs

@@ -105,12 +105,20 @@ async fn main() -> anyhow::Result<()> {
             })
             .context("RWHOD server is enabled, but socket fd not provided by systemd")??;
 
+        let audit_socket: Option<OwnedFd> = fd_map
+            .get("audit_socket")
+            .map(|fd| fd.try_clone())
+            .transpose()
+            .context("Failed to clone audit socket fd")?;
+
         join_set.spawn(rwhod_server(
             socket,
             whod_status_store.clone(),
             rwhod_ignore_list.clone(),
             config.rwhod.interfaces.clone(),
             config.rwhod.send_interval(),
+            config.rwhod.realtime_updates_enabled(),
+            audit_socket,
         ));
     } else {
         tracing::debug!("RWHOD server is disabled in configuration");
@@ -148,13 +156,31 @@ async fn rwhod_server(
     ignore_list: Option<IgnoreList>,
     allowed_interfaces: Option<HashSet<String>>,
     send_interval: Duration,
+    realtime_updates_enabled: bool,
+    audit_socket: Option<OwnedFd>,
 ) -> anyhow::Result<()> {
     let socket = Arc::new(socket);
 
     let interfaces =
         roowho2_lib::server::rwhod::determine_relevant_interfaces(allowed_interfaces.as_ref())?;
+
+    let (tx, rx) = tokio::sync::mpsc::channel(1);
+    match (realtime_updates_enabled, audit_socket) {
+        (true, Some(fd)) => {
+            tokio::spawn(roowho2_lib::server::rwhod::audit_change_notifier(tx, fd));
+        }
+        (true, None) => {
+            tracing::warn!(
+                "rwhod.realtime_updates is enabled, but no audit socket fd was provided by \
+                 systemd (e.g. the kernel might not have been booted with `audit=1`); \
+                 falling back to interval-only updates"
+            );
+        }
+        (false, _) => {}
+    }
+
     let sender_task =
-        rwhod_packet_sender_task(socket.clone(), interfaces, ignore_list, send_interval);
+        rwhod_packet_sender_task(socket.clone(), interfaces, ignore_list, send_interval, rx);
     let receiver_task = rwhod_packet_receiver_task(socket.clone(), whod_status_store);
 
     tokio::select! {

src/server/config.rs

@@ -61,6 +61,13 @@ pub struct RwhodConfig {
     ///
     /// If left as `None`, defaults to 4096.
     pub max_status_entries: Option<usize>,
+
+    /// Whether to react to Linux audit log activity (e.g. logins/logouts)
+    /// and push a status update immediately, instead of only on the
+    /// regular `send_interval_seconds` interval.
+    ///
+    /// If left as `None`, defaults to `false`.
+    pub realtime_updates: Option<bool>,
 }
 
 impl RwhodConfig {
@@ -95,6 +102,11 @@ impl RwhodConfig {
             None => DEFAULT_MAX_STATUS_ENTRIES,
         }
     }
+
+    /// Resolves [`Self::realtime_updates`] into a concrete flag.
+    pub fn realtime_updates_enabled(&self) -> bool {
+        self.realtime_updates.unwrap_or(false)
+    }
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]

src/server/rwhod.rs

@@ -1,8 +1,10 @@
+mod audit_watcher;
 mod packet_receiver;
 mod packet_sender;
 mod rwhod_status;
 mod status_registry;
 
+pub use audit_watcher::audit_change_notifier;
 pub use packet_receiver::rwhod_packet_receiver_task;
 pub use packet_sender::{determine_relevant_interfaces, rwhod_packet_sender_task};
 pub use rwhod_status::{generate_rwhod_status_update, generate_rwhod_user_entries};

src/server/rwhod/audit_watcher.rs

@@ -0,0 +1,57 @@
+use std::os::fd::{FromRawFd, IntoRawFd, OwnedFd};
+
+use futures_util::stream::StreamExt;
+use netlink_packet_audit::AuditMessage;
+use netlink_proto::sys::TokioSocket;
+use tokio::sync::mpsc;
+
+/// `AUDIT_USER_START` is emitted when a PAM session is opened
+const AUDIT_USER_START: u16 = 1105;
+/// `AUDIT_USER_END` is emitted when a PAM session is closed
+const AUDIT_USER_END: u16 = 1106;
+
+/// Listens for audit messages on the systemd-provided netlink socket and sends a
+/// notification on the provided channel whenever a user session is opened or closed,
+/// indicating that there rwhod status may have changed.
+pub async fn audit_change_notifier(sender: mpsc::Sender<()>, socket_fd: OwnedFd) {
+    // SAFETY: `socket_fd` is a systemd-provided netlink socket, already
+    // bound and subscribed to the audit multicast group.
+    let socket = unsafe { TokioSocket::from_raw_fd(socket_fd.into_raw_fd()) };
+
+    let (connection, _handle, mut messages) = netlink_proto::from_socket_with_codec::<
+        AuditMessage,
+        TokioSocket,
+        netlink_packet_audit::NetlinkAuditCodec,
+    >(socket);
+
+    tokio::spawn(connection);
+
+    tracing::info!("Listening for realtime session updates via the Linux audit log");
+
+    loop {
+        match messages.next().await {
+            Some((msg, _addr))
+                if matches!(msg.header.message_type, AUDIT_USER_START | AUDIT_USER_END) =>
+            {
+                tracing::debug!("Received session-related audit message: {:?}", msg);
+                if sender.send(()).await.is_err() {
+                    tracing::debug!("Realtime update receiver dropped, stopping audit watcher");
+                    return;
+                }
+            }
+            Some((msg, _addr)) => {
+                tracing::trace!(
+                    "Ignoring audit message unrelated to sessions (type {})",
+                    msg.header.message_type
+                );
+            }
+            None => {
+                tracing::warn!(
+                    "Audit netlink connection closed unexpectedly; realtime updates disabled \
+                     for the rest of this run"
+                );
+                return;
+            }
+        }
+    }
+}

src/server/rwhod/packet_sender.rs

@@ -6,6 +6,7 @@ use std::{
 };
 use tokio::{
     net::UdpSocket,
+    sync::mpsc,
     time::{Duration as TokioDuration, interval},
 };
 
@@ -115,11 +116,31 @@ pub async fn rwhod_packet_sender_task(
     interfaces: Vec<RwhodSendTarget>,
     ignore_list: Option<IgnoreList>,
     send_interval: TokioDuration,
+    mut realtime_update_trigger: mpsc::Receiver<()>,
 ) -> anyhow::Result<()> {
     let mut interval = interval(send_interval);
+    let mut trigger_closed = false;
 
     loop {
-        interval.tick().await;
+        if trigger_closed {
+            interval.tick().await;
+        } else {
+            tokio::select! {
+                _ = interval.tick() => {}
+                triggered = realtime_update_trigger.recv() => {
+                    if triggered.is_some() {
+                        tracing::debug!("Sending an early rwhod update due to realtime trigger");
+                        interval.reset();
+                    } else {
+                        tracing::warn!(
+                            "Realtime update channel closed unexpectedly; falling back to interval-only updates"
+                        );
+                        trigger_closed = true;
+                        continue;
+                    }
+                }
+            }
+        }
 
         let status_update = generate_rwhod_status_update(ignore_list.as_ref())?;