Merge remote-tracking branch 'refs/remotes/origin/main'

iota-daemon-lib/Cargo.toml

@@ -16,6 +16,7 @@ mtp = { git = "https://git.methanium.net/Methanium/mtp.git" }
 libc = "0.2"
 sysinfo = "0.38.0"
 serde_yaml = "0.9"
+serde_json = "1"
 tokio = { version = "1.50.0", features = ["full"] }
 tokio-util = { version = "0.7", features = ["rt"] }
 uuid = { version = "*", features = ["v4"] }

iota-daemon-lib/src/command_router.rs

@@ -1,10 +1,10 @@
 use crate::log_buffer::LogBuffer;
 use crate::{DaemonRuntime, DaemonServices};
 use iota_ipc::{
-    CommunitySummary, ComponentStatusResponse, ConfigResponse, ExitIntent, IpcErrorCode,
-    LocalRequest, LogEntriesResponse, OmikronStatusResponse, ResponseEnvelope, ResponsePayload,
-    ResponseResult, StatusResponse, TaskSummary, UpdateStatusResponse, UserDetailResponse,
-    UserSummary,
+    CommunitySummary, ComponentStatusResponse, ConfigResponse, DaemonMessage, ExitIntent,
+    IpcErrorCode, LocalRequest, LogEntriesResponse, LogEntry, MAX_MESSAGE_SIZE,
+    OmikronStatusResponse, ResponseEnvelope, ResponsePayload, ResponseResult, StatusResponse,
+    TaskSummary, UpdateStatusResponse, UserDetailResponse, UserSummary,
 };
 use iota_logger::{log, log_command};
 use iota_storage::users::user_manager;
@@ -15,6 +15,37 @@ use std::time::Duration;
 
 use crate::daemon_state::{ShutdownReason, StartupPhase};
 
+pub use iota_ipc::IpcRole;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct PeerContext {
+    pub pid: i32,
+    pub uid: u32,
+    pub role: IpcRole,
+}
+
+const MAX_LOG_ENTRIES_PER_RESPONSE: usize = 512;
+
+fn bounded_log_entries(mut entries: Vec<LogEntry>) -> Vec<LogEntry> {
+    entries.truncate(MAX_LOG_ENTRIES_PER_RESPONSE);
+    while !entries.is_empty() {
+        let response = DaemonMessage::Response(ResponseEnvelope {
+            request_id: u64::MAX,
+            result: ResponseResult::Ok(ResponsePayload::LogEntries(LogEntriesResponse {
+                entries: entries.clone(),
+            })),
+        });
+        let fits = serde_json::to_vec(&response)
+            .map(|encoded| encoded.len() <= MAX_MESSAGE_SIZE)
+            .unwrap_or(false);
+        if fits {
+            return entries;
+        }
+        entries.remove(0);
+    }
+    entries
+}
+
 #[derive(Clone)]
 pub struct CommandRouter {
     runtime: Arc<DaemonRuntime>,
@@ -35,8 +66,33 @@ impl CommandRouter {
         }
     }
 
-    pub async fn route(&self, request_id: u64, request: LocalRequest) -> ResponseEnvelope {
-        log_command!("{:?}", request);
+    pub async fn route(
+        &self,
+        peer: &PeerContext,
+        request_id: u64,
+        request: LocalRequest,
+    ) -> ResponseEnvelope {
+        if !peer.role.allows(request.required_role()) {
+            log!(
+                "IPC authorization denied: pid={}, uid={}, role={:?}, request={:?}",
+                peer.pid,
+                peer.uid,
+                peer.role,
+                request
+            );
+            return ResponseEnvelope {
+                request_id,
+                result: ResponseResult::Error(IpcErrorCode::Unauthorized),
+            };
+        }
+
+        log_command!(
+            "pid={} uid={} role={:?} request={:?}",
+            peer.pid,
+            peer.uid,
+            peer.role,
+            request
+        );
         let result = self.execute(request).await;
         ResponseEnvelope { request_id, result }
     }
@@ -284,7 +340,7 @@ impl CommandRouter {
                 {
                     return ResponseResult::Error(IpcErrorCode::Conflict);
                 }
-                self.runtime.shutdown(match intent {
+                self.runtime.request_shutdown(match intent {
                     ExitIntent::Stop => ShutdownReason::Stop,
                     ExitIntent::Restart => ShutdownReason::Restart,
                 });
@@ -298,13 +354,13 @@ impl CommandRouter {
                 },
             )),
             LocalRequest::RestartDaemon => {
-                self.runtime.shutdown(ShutdownReason::Restart);
+                self.runtime.request_shutdown(ShutdownReason::Restart);
                 ResponseResult::Ok(ResponsePayload::Acknowledged {
                     message: "Daemon restart requested".into(),
                 })
             }
             LocalRequest::StopDaemon => {
-                self.runtime.shutdown(ShutdownReason::Stop);
+                self.runtime.request_shutdown(ShutdownReason::Stop);
                 ResponseResult::Ok(ResponsePayload::Acknowledged {
                     message: "Daemon shutdown requested".into(),
                 })
@@ -378,7 +434,7 @@ impl CommandRouter {
             LocalRequest::ImportUser { .. } => ResponseResult::Error(IpcErrorCode::InvalidRequest),
             LocalRequest::GetLogs { limit } => {
                 let entries = if let Ok(buf) = self.log_buffer.lock() {
-                    buf.recent(limit)
+                    bounded_log_entries(buf.recent(limit.min(MAX_LOG_ENTRIES_PER_RESPONSE)))
                 } else {
                     Vec::new()
                 };
@@ -393,11 +449,10 @@ impl CommandRouter {
                 Err(_e) => ResponseResult::Error(IpcErrorCode::InternalFailure),
             },
             LocalRequest::ListCommunities => {
-                let iota_id = config_util::CONFIG
-                    .load()
-                    .iota_id
-                    .map(|id| id as i64)
-                    .unwrap_or(0);
+                let iota_id = config_util::CONFIG.load().iota_id;
+                let Ok(iota_id) = iota_id.map(i64::try_from).unwrap_or(Ok(0)) else {
+                    return ResponseResult::Ok(ResponsePayload::Communities(Vec::new()));
+                };
                 let stored =
                     iota_storage::util::communities_util::CommunitiesUtil::get_communities(iota_id);
                 let summaries: Vec<CommunitySummary> = stored
@@ -412,3 +467,77 @@ impl CommandRouter {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{IpcRole, LocalRequest, bounded_log_entries};
+    use iota_ipc::{ExitIntent, LogEntry, SecretString};
+
+    #[test]
+    fn every_request_has_an_explicit_role_policy() {
+        let requests = [
+            LocalRequest::GetStatus,
+            LocalRequest::ListTasks,
+            LocalRequest::ListUsers,
+            LocalRequest::CreateUser {
+                username: "alice".into(),
+            },
+            LocalRequest::AttachUserFromTu {
+                credential: SecretString("credential".into()),
+            },
+            LocalRequest::PurgeUserData { user_id: 1 },
+            LocalRequest::ReleaseUser { user_id: 1 },
+            LocalRequest::CompleteDeleteUser {
+                user_id: 1,
+                credential: None,
+            },
+            LocalRequest::RemoveUser { user_id: 1 },
+            LocalRequest::ReconnectOmikron,
+            LocalRequest::RotateIotaIdentity,
+            LocalRequest::RequestProcessExit {
+                intent: ExitIntent::Stop,
+            },
+            LocalRequest::GetDaemonStatus,
+            LocalRequest::RestartDaemon,
+            LocalRequest::StopDaemon,
+            LocalRequest::GetConfig,
+            LocalRequest::SetConfig {
+                key: "port".into(),
+                value: "1984".into(),
+            },
+            LocalRequest::ReloadConfig,
+            LocalRequest::GetOmikronStatus,
+            LocalRequest::ListComponents,
+            LocalRequest::GetUser { user_id: 1 },
+            LocalRequest::ImportUser {
+                username: "alice".into(),
+            },
+            LocalRequest::GetLogs { limit: 10 },
+            LocalRequest::CheckUpdate,
+            LocalRequest::ListCommunities,
+        ];
+
+        assert_eq!(requests.len(), 25);
+        for request in requests {
+            let required = request.required_role();
+            assert!(IpcRole::Admin.allows(required));
+            assert_eq!(
+                IpcRole::Operate.allows(required),
+                required != IpcRole::Admin
+            );
+            assert_eq!(IpcRole::Read.allows(required), required == IpcRole::Read);
+        }
+    }
+
+    #[test]
+    fn log_responses_drop_entries_that_cannot_fit_one_ipc_frame() {
+        let entries = vec![LogEntry {
+            timestamp_ms: 0,
+            sender: "test".into(),
+            message: "x".repeat(2 * 1024 * 1024),
+            is_error: false,
+        }];
+
+        assert!(bounded_log_entries(entries).is_empty());
+    }
+}

iota-daemon-lib/src/daemon_state.rs

@@ -55,7 +55,7 @@ impl From<StartupPhase> for iota_ipc::StartupPhase {
 
 /* This wrapper exposes daemon state as IPC-safe snapshots while preserving a
  * single owned state instance for all daemon subsystems. The cancellation token
- * is the single lifecycle signal — all subsystems check it instead of a
+ * is the single lifecycle signal, and all subsystems check it instead of a
  * separate boolean. */
 pub struct DaemonRuntime {
     pub state: Arc<DaemonState>,
@@ -131,12 +131,20 @@ impl DaemonRuntime {
     }
 
     pub fn shutdown(&self, reason: ShutdownReason) {
+        self.request_shutdown(reason);
+        self.begin_shutdown();
+    }
+
+    pub fn request_shutdown(&self, reason: ShutdownReason) {
         if self.shutdown_tx.borrow().is_none() {
             let _ = self.shutdown_tx.send(Some(reason));
-            self.cancellation.cancel();
         }
     }
 
+    pub fn begin_shutdown(&self) {
+        self.cancellation.cancel();
+    }
+
     pub fn shutdown_reason(&self) -> Option<ShutdownReason> {
         self.shutdown_tx.borrow().clone()
     }

iota-daemon-lib/src/ipc_server.rs

@@ -1,23 +1,26 @@
 use crate::deployment::from_environment;
 use crate::log_buffer::LogBuffer;
-use crate::{CommandRouter, DaemonRuntime, DaemonServices};
+use crate::{CommandRouter, DaemonRuntime, DaemonServices, IpcRole, PeerContext};
 use iota_ipc::{
     ClientMessage, DaemonMessage, HelloAck, MIN_PROTOCOL_VERSION, PROTOCOL_VERSION, read_msg,
     write_msg,
 };
 use iota_logger::log;
+use iota_storage::util::config_util;
 use std::io::Result;
-use std::os::unix::fs::{FileTypeExt, MetadataExt, OpenOptionsExt};
+use std::os::unix::fs::{FileTypeExt, MetadataExt, OpenOptionsExt, PermissionsExt};
 use std::path::{Path, PathBuf};
 use std::sync::{Arc, Mutex};
 use std::{env, fs::File, os::fd::FromRawFd, os::unix::net::UnixListener as StdUnixListener};
+use tokio::io::AsyncWriteExt;
 use tokio::net::{UnixListener, UnixStream};
-use tokio::sync::{broadcast, mpsc, watch};
+use tokio::sync::{Semaphore, broadcast, mpsc, watch};
 use tokio::time::timeout;
 use uuid::Uuid;
 
 /// Per-client outbound queue capacity.
 const CLIENT_CHANNEL_SIZE: usize = 256;
+const MAX_CONFIGURED_IPC_CLIENTS: usize = 4096;
 
 /// Maximum handshake retries before giving up.
 const MAX_HANDSHAKE_RETRIES: u32 = 1;
@@ -36,6 +39,20 @@ struct ClientSubscription {
     metric_interval_ms: u64,
 }
 
+enum WriterCommand {
+    Message(DaemonMessage),
+    Flush {
+        complete: tokio::sync::oneshot::Sender<()>,
+    },
+}
+
+fn configured_client_limit() -> usize {
+    config_util::CONFIG
+        .load()
+        .max_ipc_clients
+        .clamp(1, MAX_CONFIGURED_IPC_CLIENTS)
+}
+
 pub struct IpcServer {
     listener: UnixListener,
     runtime: Arc<DaemonRuntime>,
@@ -45,6 +62,7 @@ pub struct IpcServer {
     state_rx: watch::Receiver<iota_ipc::StateSnapshot>,
     instance_id: String,
     _instance_lock: File,
+    client_limit: Arc<Semaphore>,
 }
 
 impl IpcServer {
@@ -96,11 +114,18 @@ impl IpcServer {
                 }
                 remove_stale_socket(&path).await?;
                 let listener = UnixListener::bind(&path)?;
-                let _ = tokio::fs::set_permissions(
-                    &path,
-                    std::os::unix::fs::PermissionsExt::from_mode(0o600),
-                )
-                .await;
+                if let Err(error) =
+                    tokio::fs::set_permissions(&path, PermissionsExt::from_mode(0o600)).await
+                {
+                    drop(listener);
+                    let _ = tokio::fs::remove_file(&path).await;
+                    return Err(error);
+                }
+                if let Err(error) = validate_manual_socket(&path).await {
+                    drop(listener);
+                    let _ = tokio::fs::remove_file(&path).await;
+                    return Err(error);
+                }
                 return Ok(Self {
                     listener,
                     runtime,
@@ -110,6 +135,7 @@ impl IpcServer {
                     state_rx,
                     instance_id: Uuid::new_v4().to_string(),
                     _instance_lock: lock,
+                    client_limit: Arc::new(Semaphore::new(configured_client_limit())),
                 });
             }
         };
@@ -122,12 +148,21 @@ impl IpcServer {
             state_rx,
             instance_id: Uuid::new_v4().to_string(),
             _instance_lock: File::options().read(true).open("/dev/null")?,
+            client_limit: Arc::new(Semaphore::new(configured_client_limit())),
         })
     }
 
     pub async fn serve(self) -> Result<()> {
         loop {
             let (stream, _addr) = self.listener.accept().await?;
+            let permit = match self.client_limit.clone().try_acquire_owned() {
+                Ok(permit) => permit,
+                Err(_) => {
+                    eprintln!("IPC connection rejected: active client limit reached");
+                    drop(stream);
+                    continue;
+                }
+            };
             eprintln!("IPC client accepted");
             let runtime = self.runtime.clone();
             let services = self.services.clone();
@@ -136,6 +171,7 @@ impl IpcServer {
             let state_rx = self.state_rx.clone();
             let instance_id = self.instance_id.clone();
             tokio::spawn(async move {
+                let _permit = permit;
                 if let Err(error) = handle_client(
                     stream,
                     runtime,
@@ -232,6 +268,33 @@ async fn remove_stale_socket(path: &Path) -> Result<()> {
     }
 }
 
+async fn validate_manual_socket(path: &Path) -> Result<()> {
+    let metadata = tokio::fs::symlink_metadata(path).await?;
+    if metadata.file_type().is_symlink() || !metadata.file_type().is_socket() {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::InvalidData,
+            "bound IPC path is no longer a Unix socket",
+        ));
+    }
+
+    let mode = metadata.permissions().mode() & 0o777;
+    if mode != 0o600 {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::PermissionDenied,
+            format!("IPC socket has unexpected mode {mode:o}"),
+        ));
+    }
+
+    let expected_uid = unsafe { libc::geteuid() } as u32;
+    if metadata.uid() != expected_uid {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::PermissionDenied,
+            "IPC socket ownership changed after bind",
+        ));
+    }
+    Ok(())
+}
+
 #[derive(Clone, Debug)]
 struct PeerIdentity {
     pid: i32,
@@ -274,6 +337,14 @@ fn peer_credentials(stream: &UnixStream) -> Result<PeerIdentity> {
     }
 }
 
+fn role_for_peer(_peer: &PeerIdentity) -> IpcRole {
+    // This deployment has one IPC listener. Its Unix socket permissions are
+    // the admission boundary: systemd grants access to root, the daemon, and
+    // members of iota-operators. Once a peer has passed that boundary, it is
+    // an administrator for the operator console protocol.
+    IpcRole::Admin
+}
+
 async fn handle_client(
     stream: UnixStream,
     runtime: Arc<DaemonRuntime>,
@@ -283,16 +354,17 @@ async fn handle_client(
     mut state_rx: watch::Receiver<iota_ipc::StateSnapshot>,
     instance_id: String,
 ) -> Result<()> {
-    let peer = peer_credentials(&stream)?;
-    // Access control belongs to the Unix socket.  The systemd socket grants
-    // iota-operators group access (0660); rejecting every UID other than the
-    // service account here would make that authorization ineffective.  Manual
-    // sockets remain owner-only (0600) at bind time.
+    let peer_identity = peer_credentials(&stream)?;
+    let peer = PeerContext {
+        pid: peer_identity.pid,
+        uid: peer_identity.uid,
+        role: role_for_peer(&peer_identity),
+    };
     let (mut reader, mut writer) = stream.into_split();
     // A failed writer must stop the reader and any subsequent command work
     // for this client; otherwise the reader can remain parked forever.
     let session_cancellation = runtime.cancellation.child_token();
-    let (directed_tx, directed_rx) = mpsc::channel::<DaemonMessage>(CLIENT_CHANNEL_SIZE);
+    let (directed_tx, directed_rx) = mpsc::channel::<WriterCommand>(CLIENT_CHANNEL_SIZE);
     eprintln!("IPC handshake started (pid={}, uid={})", peer.pid, peer.uid);
 
     // --- Handshake ---
@@ -343,7 +415,7 @@ async fn handle_client(
                     break;
                 }
                 Ok(_) => {
-                    // Unexpected first message — send error and close.
+                    // Unexpected first message, send an error and close.
                     return Err(std::io::Error::new(
                         std::io::ErrorKind::InvalidData,
                         "Expected Hello as first message",
@@ -353,7 +425,7 @@ async fn handle_client(
             },
         }
     }
-    let _version = negotiated_version.ok_or_else(|| {
+    let negotiated_version = negotiated_version.ok_or_else(|| {
         std::io::Error::new(std::io::ErrorKind::Other, "Handshake failed after retries")
     })?;
 
@@ -361,7 +433,7 @@ async fn handle_client(
 
     // --- Send initial state snapshot ---
     let initial = DaemonMessage::StateUpdate(runtime.snapshot());
-    let _ = directed_tx.send(initial).await;
+    let _ = directed_tx.send(WriterCommand::Message(initial)).await;
 
     // --- Writer task: merge directed responses + shared log events ---
     let mut log_rx = log_tx.subscribe();
@@ -375,19 +447,29 @@ async fn handle_client(
         tokio::spawn(async move {
             let mut directed_rx = directed_rx;
             let mut last_metric_sent = tokio::time::Instant::now();
+            let mut state_updates_open = true;
             loop {
                 let metric_interval = sub_rx.borrow().metric_interval_ms;
                 tokio::select! {
+                    _ = session_cancellation.cancelled() => break,
                     // Directed messages (responses to this client's requests)
-                    msg = directed_rx.recv() => {
-                        match msg {
-                            Some(message) => {
+                    command = directed_rx.recv() => {
+                        match command {
+                            Some(WriterCommand::Message(message)) => {
                                 if let Err(error) = write_client_message(&mut writer, &message).await {
                                     eprintln!("IPC client writer stopped while sending directed message: {error}");
                                     session_cancellation.cancel();
                                     break;
                                 }
                             }
+                            Some(WriterCommand::Flush { complete }) => {
+                                if let Err(error) = writer.flush().await {
+                                    eprintln!("IPC client writer stopped while flushing: {error}");
+                                    session_cancellation.cancel();
+                                    break;
+                                }
+                                let _ = complete.send(());
+                            }
                             None => break,
                         }
                     }
@@ -435,12 +517,16 @@ async fn handle_client(
                                     break;
                                 }
                             }
-                            Err(broadcast::error::RecvError::Closed) => break,
+                            Err(broadcast::error::RecvError::Closed) => {
+                                session_cancellation.cancel();
+                                break;
+                            }
                         }
                     }
-                    changed = state_rx.changed() => {
+                    changed = state_rx.changed(), if state_updates_open => {
                         if changed.is_err() {
-                            break;
+                            state_updates_open = false;
+                            continue;
                         }
                         let snapshot = state_rx.borrow().clone();
                         if let Err(error) = write_client_message(&mut writer, &DaemonMessage::StateUpdate(snapshot)).await {
@@ -475,9 +561,14 @@ async fn handle_client(
                     iota_ipc::LocalRequest::StopDaemon => Some("shutdown requested"),
                     _ => None,
                 };
-                let response = if envelope.protocol_version < MIN_PROTOCOL_VERSION
-                    || envelope.protocol_version > PROTOCOL_VERSION
-                {
+                let response = if envelope.protocol_version != negotiated_version {
+                    log!(
+                        "IPC protocol mismatch: pid={}, uid={}, negotiated={}, request={}",
+                        peer.pid,
+                        peer.uid,
+                        negotiated_version,
+                        envelope.protocol_version
+                    );
                     iota_ipc::ResponseEnvelope {
                         request_id: envelope.request_id,
                         result: iota_ipc::ResponseResult::Error(
@@ -485,21 +576,42 @@ async fn handle_client(
                         ),
                     }
                 } else {
-                    router.route(envelope.request_id, envelope.request).await
+                    router
+                        .route(&peer, envelope.request_id, envelope.request)
+                        .await
                 };
-                let _ = directed_tx.send(DaemonMessage::Response(response)).await;
-                if let Some(reason) = shutdown_reason {
+                let should_shutdown = shutdown_reason.is_some()
+                    && matches!(&response.result, iota_ipc::ResponseResult::Ok(_));
+                let _ = directed_tx
+                    .send(WriterCommand::Message(DaemonMessage::Response(response)))
+                    .await;
+                if let Some(reason) = shutdown_reason.filter(|_| should_shutdown) {
                     let _ = directed_tx
-                        .send(DaemonMessage::LifecycleEvent(
+                        .send(WriterCommand::Message(DaemonMessage::LifecycleEvent(
                             iota_ipc::LifecycleEvent::Shutdown {
                                 reason: reason.into(),
                             },
-                        ))
+                        )))
                         .await;
-                    // The request itself initiates daemon cancellation. Give
-                    // the dedicated writer a chance to flush the response
-                    // and lifecycle event before this session is torn down.
-                    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+                    let (flush_tx, flush_rx) = tokio::sync::oneshot::channel();
+                    let _ = directed_tx
+                        .send(WriterCommand::Flush { complete: flush_tx })
+                        .await;
+                    timeout(CLIENT_IO_TIMEOUT, flush_rx)
+                        .await
+                        .map_err(|_| {
+                            std::io::Error::new(
+                                std::io::ErrorKind::TimedOut,
+                                "IPC shutdown response flush timed out",
+                            )
+                        })?
+                        .map_err(|_| {
+                            std::io::Error::new(
+                                std::io::ErrorKind::BrokenPipe,
+                                "IPC writer stopped before shutdown flush",
+                            )
+                        })?;
+                    runtime.begin_shutdown();
                     break;
                 }
             }
@@ -515,26 +627,51 @@ async fn handle_client(
                     metric_interval_ms: interval,
                 });
                 let snapshot = DaemonMessage::StateUpdate(runtime.snapshot());
-                let _ = directed_tx.send(snapshot).await;
-                let _ = directed_tx.send(DaemonMessage::Subscribed).await;
+                let _ = directed_tx.send(WriterCommand::Message(snapshot)).await;
+                let _ = directed_tx
+                    .send(WriterCommand::Message(DaemonMessage::Subscribed))
+                    .await;
             }
             Ok(ClientMessage::Ping { seq }) => {
-                let _ = directed_tx.send(DaemonMessage::Pong { seq }).await;
+                let _ = directed_tx
+                    .send(WriterCommand::Message(DaemonMessage::Pong { seq }))
+                    .await;
             }
             Ok(ClientMessage::Hello { .. }) => {
                 // Re-handshake on existing connection: treat as resubscribe
                 let snapshot = DaemonMessage::StateUpdate(runtime.snapshot());
-                let _ = directed_tx.send(snapshot).await;
+                let _ = directed_tx.send(WriterCommand::Message(snapshot)).await;
             }
             Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => break,
             Err(error) => {
-                writer_task.abort();
+                session_cancellation.cancel();
+                drop(directed_tx);
+                let mut writer_task = writer_task;
+                match timeout(CLIENT_IO_TIMEOUT, &mut writer_task).await {
+                    Ok(_) => {}
+                    Err(_) => {
+                        writer_task.abort();
+                        let _ = writer_task.await;
+                    }
+                }
                 return Err(error);
             }
         }
     }
+    drop(directed_tx);
     session_cancellation.cancel();
-    writer_task.abort();
+    let mut writer_task = writer_task;
+    match timeout(CLIENT_IO_TIMEOUT, &mut writer_task).await {
+        Ok(Ok(())) => {}
+        Ok(Err(error)) => {
+            eprintln!("IPC client writer task failed: {error}");
+        }
+        Err(_) => {
+            eprintln!("IPC client writer did not stop before timeout");
+            writer_task.abort();
+            let _ = writer_task.await;
+        }
+    }
     log!(
         "IPC client disconnected (pid={}, uid={})",
         peer.pid,
@@ -563,4 +700,46 @@ mod tests {
                 .is_err()
         );
     }
+
+    #[tokio::test]
+    async fn manual_socket_validation_requires_owner_only_mode() {
+        let directory = tempfile::tempdir().unwrap();
+        let path = directory.path().join("ipc.sock");
+        let listener = StdUnixListener::bind(&path).expect("test socket binds");
+        tokio::fs::set_permissions(&path, PermissionsExt::from_mode(0o600))
+            .await
+            .expect("test socket permissions apply");
+
+        validate_manual_socket(&path)
+            .await
+            .expect("manual socket validation succeeds");
+        drop(listener);
+    }
+
+    #[tokio::test]
+    async fn manual_socket_validation_rejects_unexpected_mode() {
+        let directory = tempfile::tempdir().unwrap();
+        let path = directory.path().join("ipc.sock");
+        let listener = StdUnixListener::bind(&path).expect("test socket binds");
+        tokio::fs::set_permissions(&path, PermissionsExt::from_mode(0o660))
+            .await
+            .expect("test socket permissions apply");
+
+        let error = validate_manual_socket(&path)
+            .await
+            .expect_err("group-accessible manual socket must be rejected");
+        assert_eq!(error.kind(), std::io::ErrorKind::PermissionDenied);
+        drop(listener);
+    }
+
+    #[test]
+    fn an_admitted_operator_peer_receives_administrator_role() {
+        let peer = PeerIdentity {
+            pid: 123,
+            uid: 1000,
+            _gid: 1000,
+        };
+
+        assert_eq!(role_for_peer(&peer), IpcRole::Admin);
+    }
 }

iota-daemon-lib/src/lib.rs

@@ -7,7 +7,7 @@ pub mod log_buffer;
 pub mod services;
 pub mod task_registry;
 
-pub use command_router::CommandRouter;
+pub use command_router::{CommandRouter, IpcRole, PeerContext};
 pub use daemon_state::{DaemonRuntime, ShutdownReason, StartupPhase};
 pub use ipc_server::IpcServer;
 pub use services::DaemonServices;

iota-daemon-lib/tests/command_router.rs

@@ -1,7 +1,7 @@
 use async_trait::async_trait;
 use iota_daemon_lib::log_buffer::LogBuffer;
-use iota_daemon_lib::{CommandRouter, DaemonRuntime, DaemonServices};
-use iota_ipc::{LocalRequest, ResponseResult};
+use iota_daemon_lib::{CommandRouter, DaemonRuntime, DaemonServices, IpcRole, PeerContext};
+use iota_ipc::{IpcErrorCode, LocalRequest, ResponseResult};
 use mtp::codec::CommunicationValue;
 use omikron_connector::{OmikronClient, OmikronError};
 use std::sync::{
@@ -13,6 +13,22 @@ use std::time::Duration;
 struct FakeOmikron {
     reconnects: AtomicUsize,
 }
+
+fn admin_peer() -> PeerContext {
+    PeerContext {
+        pid: 1,
+        uid: 0,
+        role: IpcRole::Admin,
+    }
+}
+
+fn read_peer() -> PeerContext {
+    PeerContext {
+        pid: 2,
+        uid: 1000,
+        role: IpcRole::Read,
+    }
+}
 #[async_trait]
 impl OmikronClient for FakeOmikron {
     async fn send_message(&self, _: &CommunicationValue) -> Result<(), OmikronError> {
@@ -55,7 +71,10 @@ async fn reconnect_uses_the_injected_client() {
         Arc::new(Mutex::new(LogBuffer::new(100))),
     );
     assert!(matches!(
-        router.route(1, LocalRequest::ReconnectOmikron).await.result,
+        router
+            .route(&admin_peer(), 1, LocalRequest::ReconnectOmikron)
+            .await
+            .result,
         ResponseResult::Ok(_)
     ));
     assert_eq!(fake.reconnects.load(Ordering::SeqCst), 1);
@@ -79,10 +98,44 @@ async fn identity_rotation_is_available_while_omikron_is_offline() {
     );
     assert!(matches!(
         router
-            .route(1, LocalRequest::RotateIotaIdentity)
+            .route(&admin_peer(), 1, LocalRequest::RotateIotaIdentity)
             .await
             .result,
         ResponseResult::Ok(_)
     ));
     assert_eq!(fake.reconnects.load(Ordering::SeqCst), 1);
 }
+
+#[tokio::test]
+async fn read_role_cannot_execute_an_administrative_request() {
+    let fake = Arc::new(FakeOmikron {
+        reconnects: AtomicUsize::new(0),
+    });
+    let services = Arc::new(DaemonServices {
+        omikron: fake.clone(),
+        users: Default::default(),
+        config: Default::default(),
+        active: true,
+    });
+    let router = CommandRouter::new(
+        Arc::new(DaemonRuntime::new()),
+        services,
+        Arc::new(Mutex::new(LogBuffer::new(100))),
+    );
+
+    assert!(matches!(
+        router
+            .route(
+                &read_peer(),
+                9,
+                LocalRequest::SetConfig {
+                    key: "port".into(),
+                    value: "1984".into(),
+                },
+            )
+            .await
+            .result,
+        ResponseResult::Error(IpcErrorCode::Unauthorized)
+    ));
+    assert_eq!(fake.reconnects.load(Ordering::SeqCst), 0);
+}

iota-daemon-lib/tests/ipc_server.rs

@@ -0,0 +1,252 @@
+use async_trait::async_trait;
+use iota_daemon_lib::{DaemonRuntime, DaemonServices, IpcServer};
+use iota_ipc::{
+    ClientMessage, DaemonMessage, ExitIntent, IpcErrorCode, LocalRequest, PROTOCOL_VERSION,
+    RequestEnvelope, ResponseResult, read_msg, write_msg,
+};
+use iota_storage::util::config_util::{self, IotaConfig};
+use mtp::codec::CommunicationValue;
+use omikron_connector::{OmikronClient, OmikronError};
+use std::path::Path;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::net::UnixStream;
+use tokio::sync::{broadcast, watch};
+
+struct ConfigRestore(Arc<IotaConfig>);
+
+impl Drop for ConfigRestore {
+    fn drop(&mut self) {
+        config_util::CONFIG.store(self.0.clone());
+    }
+}
+
+fn set_client_limit(limit: usize) -> ConfigRestore {
+    let previous = config_util::CONFIG.load_full();
+    let mut config = (*previous).clone();
+    config.max_ipc_clients = limit;
+    config_util::CONFIG.store(Arc::new(config));
+    ConfigRestore(previous)
+}
+
+async fn start_server(
+    path: &Path,
+    services: Arc<DaemonServices>,
+) -> (Arc<DaemonRuntime>, tokio::task::JoinHandle<()>) {
+    let runtime = Arc::new(DaemonRuntime::new());
+    let (log_tx, _) = broadcast::channel(32);
+    let log_buffer = Arc::new(std::sync::Mutex::new(
+        iota_daemon_lib::log_buffer::LogBuffer::new(32),
+    ));
+    let (_, state_rx) = watch::channel(runtime.snapshot());
+    let server = IpcServer::bind(
+        path.to_owned(),
+        runtime.clone(),
+        services,
+        log_tx,
+        log_buffer,
+        state_rx,
+    )
+    .await
+    .expect("IPC server binds");
+    let task = tokio::spawn(async move {
+        let _ = server.serve().await;
+    });
+    (runtime, task)
+}
+
+async fn try_connect_and_await_hello(path: &Path) -> std::io::Result<UnixStream> {
+    let mut stream = UnixStream::connect(path).await?;
+    write_msg(
+        &mut stream,
+        &ClientMessage::Hello {
+            supported_versions: vec![PROTOCOL_VERSION],
+        },
+    )
+    .await?;
+    let message: DaemonMessage = read_msg(&mut stream).await?;
+    if !matches!(message, DaemonMessage::HelloAck(_)) {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::InvalidData,
+            "expected HelloAck",
+        ));
+    }
+    Ok(stream)
+}
+
+async fn connect_and_await_hello(path: &Path) -> UnixStream {
+    try_connect_and_await_hello(path)
+        .await
+        .expect("IPC connection completes the Hello exchange")
+}
+
+#[tokio::test]
+async fn active_client_limit_rejects_excess_clients_and_releases_permits() {
+    let _config = set_client_limit(1);
+    let directory = tempfile::tempdir().unwrap();
+    let socket = directory.path().join("ipc.sock");
+    let (_runtime, server_task) = start_server(&socket, DaemonServices::inactive()).await;
+
+    let first = connect_and_await_hello(&socket).await;
+    let mut rejected = UnixStream::connect(&socket)
+        .await
+        .expect("second connection reaches the Unix listener");
+    let rejected_result = tokio::time::timeout(
+        Duration::from_secs(2),
+        read_msg::<_, DaemonMessage>(&mut rejected),
+    )
+    .await
+    .expect("rejected client is closed promptly");
+    assert!(rejected_result.is_err());
+
+    drop(first);
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
+    let _released = loop {
+        match try_connect_and_await_hello(&socket).await {
+            Ok(stream) => break stream,
+            Err(_error) if tokio::time::Instant::now() < deadline => {
+                tokio::time::sleep(Duration::from_millis(10)).await;
+            }
+            Err(error) => panic!("client permit was not released: {error}"),
+        }
+    };
+    server_task.abort();
+    let _ = server_task.await;
+}
+
+#[tokio::test]
+async fn request_with_version_different_from_hello_is_rejected() {
+    let directory = tempfile::tempdir().unwrap();
+    let socket = directory.path().join("ipc.sock");
+    let (_runtime, server_task) = start_server(&socket, DaemonServices::inactive()).await;
+    let mut stream = connect_and_await_hello(&socket).await;
+
+    write_msg(
+        &mut stream,
+        &ClientMessage::Request(RequestEnvelope {
+            request_id: 7,
+            protocol_version: PROTOCOL_VERSION + 1,
+            request: LocalRequest::GetStatus,
+        }),
+    )
+    .await
+    .expect("request sends");
+
+    let response = loop {
+        match read_msg::<_, DaemonMessage>(&mut stream)
+            .await
+            .expect("daemon response arrives")
+        {
+            DaemonMessage::Response(response) => break response,
+            _ => continue,
+        }
+    };
+    assert_eq!(response.request_id, 7);
+    assert!(matches!(
+        response.result,
+        ResponseResult::Error(IpcErrorCode::UnsupportedVersion)
+    ));
+
+    drop(stream);
+    server_task.abort();
+    let _ = server_task.await;
+}
+
+struct TestOmikron;
+
+#[async_trait]
+impl OmikronClient for TestOmikron {
+    async fn send_message(&self, _: &CommunicationValue) -> Result<(), OmikronError> {
+        Ok(())
+    }
+
+    async fn await_response(
+        &self,
+        _: &CommunicationValue,
+        _: Duration,
+    ) -> Result<CommunicationValue, OmikronError> {
+        Err(OmikronError::Disconnected("test client".into()))
+    }
+
+    async fn reconnect(&self) -> Result<(), OmikronError> {
+        Ok(())
+    }
+
+    async fn rotate_identity(&self) -> Result<(), OmikronError> {
+        Ok(())
+    }
+
+    async fn is_connected(&self) -> bool {
+        true
+    }
+}
+
+fn active_services() -> Arc<DaemonServices> {
+    Arc::new(DaemonServices {
+        omikron: Arc::new(TestOmikron),
+        users: Default::default(),
+        config: Default::default(),
+        active: true,
+    })
+}
+
+#[tokio::test]
+async fn shutdown_delivers_response_and_lifecycle_event_before_eof() {
+    let directory = tempfile::tempdir().unwrap();
+    let socket = directory.path().join("ipc.sock");
+    let (runtime, server_task) = start_server(&socket, active_services()).await;
+    let mut stream = connect_and_await_hello(&socket).await;
+
+    write_msg(
+        &mut stream,
+        &ClientMessage::Request(RequestEnvelope {
+            request_id: 8,
+            protocol_version: PROTOCOL_VERSION,
+            request: LocalRequest::RequestProcessExit {
+                intent: ExitIntent::Stop,
+            },
+        }),
+    )
+    .await
+    .expect("shutdown request sends");
+
+    let mut response_seen = false;
+    let mut lifecycle_seen = false;
+    for _ in 0..4 {
+        match tokio::time::timeout(
+            Duration::from_secs(2),
+            read_msg::<_, DaemonMessage>(&mut stream),
+        )
+        .await
+        .expect("shutdown message arrives")
+        .expect("shutdown stream remains readable")
+        {
+            DaemonMessage::Response(response) => {
+                assert_eq!(response.request_id, 8);
+                assert!(matches!(response.result, ResponseResult::Ok(_)));
+                response_seen = true;
+            }
+            DaemonMessage::LifecycleEvent(iota_ipc::LifecycleEvent::Shutdown { .. }) => {
+                lifecycle_seen = true;
+            }
+            _ => {}
+        }
+        if response_seen && lifecycle_seen {
+            break;
+        }
+    }
+
+    assert!(response_seen);
+    assert!(lifecycle_seen);
+    let eof = tokio::time::timeout(
+        Duration::from_secs(2),
+        read_msg::<_, DaemonMessage>(&mut stream),
+    )
+    .await
+    .expect("shutdown connection closes after flush");
+    assert!(eof.is_err());
+    assert!(runtime.is_shutting_down());
+
+    server_task.abort();
+    let _ = server_task.await;
+}