1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
//! Listener setup and accept loops for the Unix socket / Windows named pipe.
use super::connection::{ConnectionContext, handle_connection};
use crate::ipc::session::SessionType;
use crate::ipc::session_result::set_session_meta;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
/// Serve one accepted connection on its own task.
fn spawn_connection<S>(stream: S, ctx: ConnectionContext)
where
S: AsyncReadExt + AsyncWriteExt + Unpin + Send + 'static,
{
tokio::spawn(async move {
if let Err(e) = handle_connection(stream, ctx).await {
log::debug!("IPC connection error: {}", e);
}
});
}
/// Run the actual server loop.
#[cfg(unix)]
#[allow(clippy::too_many_arguments)]
pub(super) async fn run_server(
socket_path: &str,
started_at: &str,
r_home: Option<String>,
log_file: Option<String>,
history_session_id: Option<i64>,
session_type: SessionType,
ctx: ConnectionContext,
bind_tx: std::sync::mpsc::SyncSender<Result<(), std::io::Error>>,
) -> std::io::Result<()> {
let listener = match tokio::net::UnixListener::bind(socket_path) {
Ok(l) => {
// Restrict socket permissions so only the owner can connect.
// The default PID-based path lives under a 0700 sessions dir,
// but custom --ipc-bind paths inherit the parent dir's umask.
// Use fd-based fchmod to avoid TOCTOU symlink race.
//
// NOTE: There is a brief race window between bind() and fchmod()
// where the socket exists with umask-inherited permissions. For
// custom --ipc-bind paths in shared directories, operators should
// ensure the parent directory is restricted (e.g. 0700).
{
use std::os::unix::io::AsRawFd;
let ret = unsafe { libc::fchmod(l.as_raw_fd(), 0o600) };
if ret != 0 {
log::warn!(
"Could not set socket permissions on {}: {}",
socket_path,
std::io::Error::last_os_error()
);
}
}
// Cache session metadata BEFORE signalling bind success, so it
// is guaranteed to be available when the first request arrives.
set_session_meta(
socket_path.to_string(),
started_at.to_string(),
r_home,
log_file,
history_session_id,
session_type,
);
let _ = bind_tx.send(Ok(()));
l
}
Err(e) => {
let _ = bind_tx.send(Err(std::io::Error::new(e.kind(), e.to_string())));
return Err(e);
}
};
log::info!("IPC server listening on {}", socket_path);
loop {
tokio::select! {
result = listener.accept() => {
match result {
Ok((stream, _addr)) => {
spawn_connection(stream, ctx.clone());
}
Err(e) => {
if ctx.cancel.is_cancelled() {
break;
}
log::warn!("IPC accept error: {}", e);
}
}
}
_ = ctx.cancel.cancelled() => {
log::info!("IPC server shutting down");
break;
}
}
}
Ok(())
}
#[cfg(windows)]
#[allow(clippy::too_many_arguments)]
pub(super) async fn run_server(
socket_path: &str,
started_at: &str,
r_home: Option<String>,
log_file: Option<String>,
history_session_id: Option<i64>,
session_type: SessionType,
ctx: ConnectionContext,
bind_tx: std::sync::mpsc::SyncSender<Result<(), std::io::Error>>,
) -> std::io::Result<()> {
use tokio::net::windows::named_pipe::ServerOptions;
// Create the first pipe instance
let mut server = match ServerOptions::new()
.first_pipe_instance(true)
.create(socket_path)
{
Ok(s) => {
// Cache session metadata BEFORE signalling bind success, so it
// is guaranteed to be available when the first request arrives.
set_session_meta(
socket_path.to_string(),
started_at.to_string(),
r_home,
log_file,
history_session_id,
session_type,
);
let _ = bind_tx.send(Ok(()));
s
}
Err(e) => {
let _ = bind_tx.send(Err(std::io::Error::new(e.kind(), e.to_string())));
return Err(e);
}
};
log::info!("IPC server listening on {}", socket_path);
loop {
tokio::select! {
result = server.connect() => {
match result {
Ok(()) => {
let connected = server;
// Create a new pipe instance for the next connection
server = ServerOptions::new().create(socket_path)?;
spawn_connection(connected, ctx.clone());
}
Err(e) => {
if ctx.cancel.is_cancelled() {
break;
}
log::warn!("IPC accept error: {}", e);
}
}
}
_ = ctx.cancel.cancelled() => {
log::info!("IPC server shutting down");
break;
}
}
}
Ok(())
}