buntrail_services/connections/
authentication.rs

1//! Native capability boundary. Secrets travel only through the owned askpass socket/stdout.
2use crate::files::io_error;
3use buntrail_core::contracts::*;
4use std::{
5    collections::{HashSet, VecDeque},
6    fs,
7    io::{Read, Write},
8    os::unix::{
9        fs::PermissionsExt,
10        net::{UnixListener, UnixStream},
11    },
12    path::PathBuf,
13    process::Command,
14    sync::{
15        Arc, Mutex,
16        atomic::{AtomicBool, Ordering},
17    },
18    thread,
19    time::{Duration, Instant},
20};
21const LIMIT: usize = 4096;
22#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
23pub enum AuthenticationKind {
24    Secret,
25    Confirm,
26    Information,
27}
28pub struct AuthenticationRequest {
29    pub id: u64,
30    pub kind: AuthenticationKind,
31    pub prompt: String,
32}
33pub enum AuthenticationResponse {
34    Cancelled,
35    Confirm(bool),
36    Secret(Vec<u8>),
37}
38impl Drop for AuthenticationResponse {
39    fn drop(&mut self) {
40        if let Self::Secret(bytes) = self {
41            bytes.fill(0);
42        }
43    }
44}
45struct Pending {
46    id: u64,
47    response: Option<AuthenticationResponse>,
48}
49struct Shared {
50    sessions: Mutex<HashSet<String>>,
51    cancel: AtomicBool,
52    finished: AtomicBool,
53    waker: Mutex<Option<std::task::Waker>>,
54    queue: Mutex<VecDeque<AuthenticationRequest>>,
55    pending: Mutex<Option<Pending>>,
56}
57/// One visible prompt at a time. Create on a worker; polling, responses and Drop never join.
58pub struct AuthenticationBroker {
59    askpass: PathBuf,
60    socket: PathBuf,
61    shared: Arc<Shared>,
62    worker: Mutex<Option<thread::JoinHandle<()>>>,
63}
64#[derive(serde::Serialize, serde::Deserialize)]
65struct Prompt {
66    token: String,
67    kind: AuthenticationKind,
68    prompt: String,
69}
70impl AuthenticationBroker {
71    pub fn new(askpass: PathBuf) -> Result<Arc<Self>, CoreError> {
72        if !askpass.is_absolute() || !askpass.is_file() {
73            return Err(CoreError::new(
74                ErrorCode::InvalidArgument,
75                "Bundled SSH askpass executable unavailable",
76            ));
77        }
78        let directory = tempfile::Builder::new()
79            .prefix("buntrail-auth-")
80            .permissions(fs::Permissions::from_mode(0o700))
81            .tempdir()
82            .map_err(io_error)?;
83        let socket = directory.path().join("prompt");
84        let listener = UnixListener::bind(&socket).map_err(io_error)?;
85        fs::set_permissions(&socket, fs::Permissions::from_mode(0o600)).map_err(io_error)?;
86        listener.set_nonblocking(true).map_err(io_error)?;
87        let shared = Arc::new(Shared {
88            sessions: Mutex::new(HashSet::new()),
89            cancel: AtomicBool::new(false),
90            finished: AtomicBool::new(false),
91            waker: Mutex::new(None),
92            queue: Mutex::new(VecDeque::new()),
93            pending: Mutex::new(None),
94        });
95        let state = shared.clone();
96        let worker = thread::Builder::new()
97            .name("buntrail-auth".into())
98            .spawn(move || {
99                let mut id = 0;
100                while !state.cancel.load(Ordering::SeqCst) {
101                    match listener.accept() {
102                        Ok((mut stream, _)) => {
103                            let _ = stream.set_read_timeout(Some(Duration::from_secs(1)));
104                            let _ = stream.set_write_timeout(Some(Duration::from_secs(1)));
105                            let mut bytes = Vec::new();
106                            if (&mut stream)
107                                .take((LIMIT + 1024) as u64)
108                                .read_to_end(&mut bytes)
109                                .is_err()
110                                || bytes.len() >= LIMIT + 1024
111                            {
112                                continue;
113                            }
114                            let Ok(prompt) = serde_json::from_slice::<Prompt>(&bytes) else {
115                                continue;
116                            };
117                            if !state.sessions.lock().unwrap().contains(&prompt.token)
118                                || prompt.prompt.len() > LIMIT
119                            {
120                                continue;
121                            }
122                            id += 1;
123                            *state.pending.lock().unwrap() = Some(Pending { id, response: None });
124                            state
125                                .queue
126                                .lock()
127                                .unwrap()
128                                .push_back(AuthenticationRequest {
129                                    id,
130                                    kind: prompt.kind,
131                                    prompt: prompt.prompt,
132                                });
133                            let deadline = Instant::now() + Duration::from_secs(120);
134                            let response = loop {
135                                if state.cancel.load(Ordering::SeqCst)
136                                    || !state.sessions.lock().unwrap().contains(&prompt.token)
137                                    || Instant::now() >= deadline
138                                {
139                                    break AuthenticationResponse::Cancelled;
140                                }
141                                if let Some(response) = state
142                                    .pending
143                                    .lock()
144                                    .unwrap()
145                                    .as_mut()
146                                    .and_then(|p| p.response.take())
147                                {
148                                    break response;
149                                }
150                                thread::sleep(Duration::from_millis(10));
151                            };
152                            state.queue.lock().unwrap().retain(|r| r.id != id);
153                            *state.pending.lock().unwrap() = None;
154                            let value = match &response {
155                                AuthenticationResponse::Cancelled => None,
156                                AuthenticationResponse::Confirm(true) => Some(b"yes".as_slice()),
157                                AuthenticationResponse::Confirm(false) => Some(b"no".as_slice()),
158                                AuthenticationResponse::Secret(bytes) => Some(bytes.as_slice()),
159                            };
160                            if let Some(value) = value.filter(|v| {
161                                v.len() <= LIMIT && !v.contains(&0) && !v.contains(&b'\n')
162                            }) {
163                                let _ = stream.write_all(&[1]);
164                                let _ = stream.write_all(&(value.len() as u32).to_be_bytes());
165                                let _ = stream.write_all(value);
166                            } else {
167                                let _ = stream.write_all(&[0]);
168                            }
169                        }
170                        Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
171                            thread::sleep(Duration::from_millis(10))
172                        }
173                        Err(_) => break,
174                    }
175                }
176                drop(listener);
177                drop(directory);
178                state.finished.store(true, Ordering::SeqCst);
179                if let Some(waker) = state.waker.lock().unwrap().take() {
180                    waker.wake();
181                }
182            })
183            .map_err(io_error)?;
184        Ok(Arc::new(Self {
185            askpass,
186            socket,
187            shared,
188            worker: Mutex::new(Some(worker)),
189        }))
190    }
191    pub fn take_request(&self) -> Option<AuthenticationRequest> {
192        self.shared.queue.lock().unwrap().pop_front()
193    }
194    pub fn respond(&self, id: u64, response: AuthenticationResponse) -> bool {
195        let mut pending = self.shared.pending.lock().unwrap();
196        if let Some(p) = pending
197            .as_mut()
198            .filter(|p| p.id == id && p.response.is_none())
199        {
200            p.response = Some(response);
201            true
202        } else {
203            false
204        }
205    }
206    pub fn close(&self) -> AuthenticationShutdown {
207        self.cancel();
208        AuthenticationShutdown(self.shared.clone())
209    }
210    pub fn cancel(&self) {
211        self.shared.cancel.store(true, Ordering::SeqCst);
212    }
213    pub(crate) fn configure(
214        self: &Arc<Self>,
215        command: &mut Command,
216    ) -> Result<AuthenticationSession, CoreError> {
217        let token = super::bootstrap::hex(&super::local::new_id()?);
218        let mut sessions = self.shared.sessions.lock().unwrap();
219        if sessions.len() >= 8 {
220            return Err(CoreError::new(
221                ErrorCode::Busy,
222                "SSH authentication sessions are busy",
223            ));
224        }
225        sessions.insert(token.clone());
226        command
227            .args(["-oBatchMode=no", "-oNumberOfPasswordPrompts=1"])
228            .env("SSH_ASKPASS", &self.askpass)
229            .env("SSH_ASKPASS_REQUIRE", "force")
230            .env("DISPLAY", "buntrail-native")
231            .env("BUNTRAIL_ASKPASS_SOCKET", &self.socket)
232            .env("BUNTRAIL_ASKPASS_TOKEN", &token);
233        Ok(AuthenticationSession {
234            broker: self.clone(),
235            token,
236        })
237    }
238}
239impl Drop for AuthenticationBroker {
240    fn drop(&mut self) {
241        self.cancel();
242        self.worker.lock().unwrap().take();
243    }
244}
245/// Entry point of the bundled askpass executable; never writes diagnostics or secret arguments.
246pub fn askpass() -> Result<(), CoreError> {
247    let prompt = std::env::args().nth(1).ok_or_else(|| {
248        CoreError::new(ErrorCode::InvalidArgument, "Missing authentication prompt")
249    })?;
250    if prompt.len() > LIMIT {
251        return Err(CoreError::new(
252            ErrorCode::InvalidArgument,
253            "Authentication prompt too large",
254        ));
255    }
256    let kind = match std::env::var("SSH_ASKPASS_PROMPT").as_deref() {
257        Ok("confirm") => AuthenticationKind::Confirm,
258        Ok("none") => AuthenticationKind::Information,
259        _ => AuthenticationKind::Secret,
260    };
261    let mut stream =
262        UnixStream::connect(std::env::var_os("BUNTRAIL_ASKPASS_SOCKET").ok_or_else(|| {
263            CoreError::new(
264                ErrorCode::Unavailable,
265                "Authentication capability unavailable",
266            )
267        })?)
268        .map_err(io_error)?;
269    stream
270        .set_read_timeout(Some(Duration::from_secs(125)))
271        .map_err(io_error)?;
272    stream
273        .set_write_timeout(Some(Duration::from_secs(1)))
274        .map_err(io_error)?;
275    let request = Prompt {
276        token: std::env::var("BUNTRAIL_ASKPASS_TOKEN").map_err(io_error)?,
277        kind,
278        prompt,
279    };
280    stream
281        .write_all(&serde_json::to_vec(&request).map_err(io_error)?)
282        .map_err(io_error)?;
283    stream
284        .shutdown(std::net::Shutdown::Write)
285        .map_err(io_error)?;
286    let mut flag = [0];
287    stream.read_exact(&mut flag).map_err(io_error)?;
288    if flag[0] != 1 {
289        return Err(CoreError::new(
290            ErrorCode::Cancelled,
291            "Authentication cancelled",
292        ));
293    }
294    let mut length = [0; 4];
295    stream.read_exact(&mut length).map_err(io_error)?;
296    let length = u32::from_be_bytes(length) as usize;
297    if length > LIMIT {
298        return Err(CoreError::new(
299            ErrorCode::InvalidArgument,
300            "Authentication response too large",
301        ));
302    }
303    let mut secret = vec![0; length];
304    stream.read_exact(&mut secret).map_err(io_error)?;
305    let result = std::io::stdout().lock().write_all(&secret);
306    secret.fill(0);
307    result.map_err(io_error)
308}
309
310pub(crate) struct AuthenticationSession {
311    broker: Arc<AuthenticationBroker>,
312    token: String,
313}
314impl Drop for AuthenticationSession {
315    fn drop(&mut self) {
316        self.broker
317            .shared
318            .sessions
319            .lock()
320            .unwrap()
321            .remove(&self.token);
322    }
323}
324
325pub struct AuthenticationShutdown(Arc<Shared>);
326impl std::future::Future for AuthenticationShutdown {
327    type Output = ();
328    fn poll(
329        self: std::pin::Pin<&mut Self>,
330        cx: &mut std::task::Context<'_>,
331    ) -> std::task::Poll<()> {
332        let mut waker = self.0.waker.lock().unwrap();
333        if self.0.finished.load(Ordering::SeqCst) {
334            std::task::Poll::Ready(())
335        } else {
336            *waker = Some(cx.waker().clone());
337            std::task::Poll::Pending
338        }
339    }
340}
buntrail_services/connections/authentication.rs

buntrail_services/connections/
authentication.rs
