sentinelpass_protocol/transport/
unix.rs

1//! Unix domain socket connection (client side; symmetric, also used by the
2//! core server's accept loop).
3
4use super::{TransportError, TransportResult, MAX_MESSAGE_SIZE};
5use std::path::PathBuf;
6use tokio::io::{AsyncReadExt, AsyncWriteExt};
7
8/// Unix socket connection
9pub struct UnixSocketConnection {
10    stream: tokio::net::UnixStream,
11}
12
13impl UnixSocketConnection {
14    /// Wrap an accepted (server-side) stream.
15    pub fn from_stream(stream: tokio::net::UnixStream) -> Self {
16        Self { stream }
17    }
18
19    /// Create a new connection as a client
20    pub async fn connect(path: PathBuf) -> TransportResult<Self> {
21        let stream = tokio::net::UnixStream::connect(&path).await.map_err(|e| {
22            TransportError::ConnectionFailed(format!(
23                "Failed to connect to {}: {}",
24                path.display(),
25                e
26            ))
27        })?;
28
29        Ok(Self { stream })
30    }
31
32    /// Read a message from the connection
33    pub async fn read_message(&mut self) -> TransportResult<Vec<u8>> {
34        // Read message length (4 bytes, big-endian)
35        let mut length_buf = [0u8; 4];
36        self.stream.read_exact(&mut length_buf).await?;
37
38        let length = u32::from_be_bytes(length_buf) as usize;
39
40        if length == 0 || length > MAX_MESSAGE_SIZE {
41            return Err(TransportError::MessageTooLarge {
42                size: length,
43                max: MAX_MESSAGE_SIZE,
44            });
45        }
46
47        // Read message payload
48        let mut buffer = vec![0u8; length];
49        self.stream.read_exact(&mut buffer).await?;
50
51        Ok(buffer)
52    }
53
54    /// Write a message to the connection
55    pub async fn write_message(&mut self, data: &[u8]) -> TransportResult<()> {
56        let length = data.len() as u32;
57
58        // Validate message size
59        if length as usize > MAX_MESSAGE_SIZE {
60            return Err(TransportError::MessageTooLarge {
61                size: data.len(),
62                max: MAX_MESSAGE_SIZE,
63            });
64        }
65
66        // Write length prefix
67        self.stream.write_all(&length.to_be_bytes()).await?;
68
69        // Write payload
70        self.stream.write_all(data).await?;
71
72        self.stream.flush().await?;
73
74        Ok(())
75    }
76
77    /// Close the connection
78    pub async fn close(&mut self) -> TransportResult<()> {
79        self.stream.shutdown().await?;
80        Ok(())
81    }
82
83    /// Check if the connection is still open
84    pub fn is_open(&self) -> bool {
85        // Try to get the peer address to check if still connected
86        self.stream.peer_addr().is_ok()
87    }
88}
89
90#[cfg(test)]
91mod tests {
92    use super::*;
93    use std::time::Duration;
94
95    #[tokio::test]
96    async fn test_unix_socket_connection_roundtrip() {
97        let temp_dir = std::env::temp_dir();
98        let socket_path = temp_dir.join(format!("test_ipc_{}.sock", uuid_v4()));
99
100        // Start server
101        let listener = tokio::net::UnixListener::bind(&socket_path).unwrap();
102        let server_handle = tokio::spawn(async move {
103            let (stream, _) = listener.accept().await.unwrap();
104            let mut conn = UnixSocketConnection::from_stream(stream);
105            let msg = conn.read_message().await.unwrap();
106            conn.write_message(&msg).await.unwrap();
107            conn.close().await.unwrap();
108        });
109
110        // Connect as client
111        tokio::time::sleep(Duration::from_millis(100)).await;
112        let mut client = UnixSocketConnection::connect(socket_path).await.unwrap();
113
114        // Send and receive
115        let test_data = b"Hello, IPC!";
116        client.write_message(test_data).await.unwrap();
117        let received = client.read_message().await.unwrap();
118
119        assert_eq!(received, test_data);
120
121        server_handle.await.unwrap();
122    }
123
124    fn uuid_v4() -> String {
125        // Simple unique suffix without pulling a uuid dependency
126        let nanos = std::time::SystemTime::now()
127            .duration_since(std::time::UNIX_EPOCH)
128            .unwrap()
129            .as_nanos();
130        format!("proto{n}", n = nanos)
131    }
132}
sentinelpass_protocol/transport/unix.rs

sentinelpass_protocol/transport/
unix.rs
