1use std::sync::Arc;
22use std::sync::atomic::{AtomicUsize, Ordering};
23use std::time::{Duration, Instant};
24
25use serde_json::json;
26use tungstenite::protocol::{CloseFrame, Role, WebSocketConfig, frame::coding::CloseCode};
27use tungstenite::{Message, WebSocket};
28
29use super::http::{Duplex, Request, Response};
30use super::mcp::Caller;
31
32pub const MAX_SESSIONS: usize = 16;
33pub const IDLE: Duration = Duration::from_secs(30 * 60);
34pub const MAX_AGE: Duration = Duration::from_secs(8 * 3600);
35pub const MAX_MESSAGE: usize = 64 * 1024;
36const POLL: Duration = Duration::from_millis(15);
39
40#[derive(Debug, Clone, PartialEq, Eq)]
42pub struct TermRequest {
43 pub app: String,
45 pub slot: Option<u32>,
47 pub instance: Option<String>,
49 pub session: Option<String>,
52 pub cols: u16,
53 pub rows: u16,
54}
55
56impl TermRequest {
57 pub fn query(&self) -> String {
59 let mut q = match &self.instance {
60 Some(i) => format!("instance={i}"),
61 None => format!("app={}", self.app),
62 };
63 q.push_str(&format!("&cols={}&rows={}", self.cols, self.rows));
64 if let Some(n) = self.slot {
65 q.push_str(&format!("&slot={n}"));
66 }
67 if let Some(s) = &self.session {
68 q.push_str(&format!("&session={}", pct_encode(s)));
69 }
70 q
71 }
72
73 pub fn target(&self) -> &str {
75 self.instance.as_deref().unwrap_or(&self.app)
76 }
77}
78
79#[derive(Debug)]
81pub struct Limits {
82 pub active: &'static AtomicUsize,
83 pub max_sessions: usize,
84 pub idle: Duration,
85 pub max_age: Duration,
86 pub busy: &'static str,
88}
89
90pub static TERMINALS: Limits = Limits {
92 active: &ACTIVE,
93 max_sessions: MAX_SESSIONS,
94 idle: IDLE,
95 max_age: MAX_AGE,
96 busy: "too many terminals are open on this server; close one and try again",
97};
98
99#[derive(Debug, PartialEq, Eq)]
101pub enum PtyOutput {
102 Data(Vec<u8>),
103 Idle,
105 Exit(Option<i32>),
107 Failed(String),
109 Note(serde_json::Value),
113}
114
115pub trait Pty: Send {
117 fn input(&mut self, data: &[u8]) -> crate::Result<()>;
118 fn resize(&mut self, cols: u16, rows: u16);
119 fn output(&mut self, wait: Duration) -> PtyOutput;
120 fn close(&mut self);
122 fn target(&self) -> Option<String> {
124 None
125 }
126 fn details(&self) -> Option<serde_json::Map<String, serde_json::Value>> {
129 None
130 }
131}
132
133pub type Terminal = Arc<
135 dyn Fn(&Caller, &crate::org::OrgId, &TermRequest) -> crate::Result<Box<dyn Pty>> + Send + Sync,
136>;
137
138static ACTIVE: AtomicUsize = AtomicUsize::new(0);
139
140struct Slot(&'static AtomicUsize);
141
142impl Drop for Slot {
143 fn drop(&mut self) {
144 self.0.fetch_sub(1, Ordering::SeqCst);
145 }
146}
147
148fn param(req: &Request, key: &str) -> Option<String> {
149 req.query.as_deref()?.split('&').find_map(|kv| {
150 let (k, v) = kv.split_once('=').unwrap_or((kv, ""));
151 (k == key).then(|| v.to_string())
152 })
153}
154
155pub fn plain_name(s: &str) -> bool {
158 !s.is_empty()
159 && s.len() <= 64
160 && s.bytes()
161 .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-')
162}
163
164pub fn query_param(req: &Request, key: &str) -> Option<String> {
166 param(req, key)
167}
168
169pub fn pct_encode(s: &str) -> String {
171 s.bytes()
172 .map(|b| match b {
173 b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => {
174 (b as char).to_string()
175 }
176 b => format!("%{b:02X}"),
177 })
178 .collect()
179}
180
181pub fn pct_decode(s: &str) -> Option<String> {
184 let b = s.as_bytes();
185 let mut out = Vec::with_capacity(b.len());
186 let mut i = 0;
187 while i < b.len() {
188 match b[i] {
189 b'%' => {
190 let hex = std::str::from_utf8(b.get(i + 1..i + 3)?).ok()?;
191 out.push(u8::from_str_radix(hex, 16).ok()?);
192 i += 3;
193 }
194 b'+' => {
195 out.push(b' ');
196 i += 1;
197 }
198 c => {
199 out.push(c);
200 i += 1;
201 }
202 }
203 }
204 String::from_utf8(out).ok()
205}
206
207pub fn term_request(req: &Request) -> std::result::Result<TermRequest, String> {
209 let (app, instance) = match (param(req, "app"), param(req, "instance")) {
210 (Some(a), None) if plain_name(&a) => (a, None),
211 (Some(_), None) => return Err("app= is not an app name".into()),
212 (None, Some(i)) if plain_name(&i) => (String::new(), Some(i)),
213 (None, Some(_)) => return Err("instance= is not an instance name".into()),
214 (Some(_), Some(_)) => return Err("give app= or instance=, not both".into()),
215 (None, None) => return Err("app= (or instance=) is required".into()),
216 };
217 let num = |k: &str, lo: u32, hi: u32, def: u32| -> std::result::Result<u32, String> {
218 match param(req, k) {
219 None => Ok(def),
220 Some(v) => v
221 .parse::<u32>()
222 .ok()
223 .filter(|n| (lo..=hi).contains(n))
224 .ok_or_else(|| format!("{k}= must be {lo}-{hi}")),
225 }
226 };
227 let slot = match param(req, "slot") {
228 None => None,
229 Some(_) if instance.is_some() => return Err("slot= is for apps".into()),
230 Some(_) => Some(num("slot", 1, 1000, 1)?),
231 };
232 let session = match param(req, "session") {
233 None => None,
234 Some(_) if instance.is_none() => return Err("session= is for instances".into()),
235 Some(v) => match pct_decode(&v) {
236 Some(s)
237 if !s.is_empty() && s.chars().count() <= 40 && !s.chars().any(char::is_control) =>
238 {
239 Some(s)
240 }
241 _ => return Err("session= is not a session name".into()),
242 },
243 };
244 Ok(TermRequest {
245 app,
246 slot,
247 instance,
248 session,
249 cols: num("cols", 2, 1000, 80)? as u16,
250 rows: num("rows", 2, 1000, 24)? as u16,
251 })
252}
253
254fn origin_authority(origin: &str) -> Option<&str> {
256 let rest = origin
257 .strip_prefix("https://")
258 .or_else(|| origin.strip_prefix("http://"))?;
259 (!rest.is_empty() && !rest.contains('/')).then_some(rest)
260}
261
262pub fn origin_allowed(req: &Request) -> bool {
265 let bearer = req.header("authorization").is_some();
266 match req.header("origin") {
267 None => bearer,
268 Some(o) => match (origin_authority(o), req.header("host")) {
269 (Some(a), Some(h)) => a.eq_ignore_ascii_case(h.trim()),
270 _ => false,
271 },
272 }
273}
274
275pub fn websocket_key(req: &Request) -> Option<String> {
277 let has = |name: &str, token: &str| {
278 req.header(name)
279 .is_some_and(|v| v.split(',').any(|t| t.trim().eq_ignore_ascii_case(token)))
280 };
281 if req.method != "GET" || !has("upgrade", "websocket") || !has("connection", "upgrade") {
282 return None;
283 }
284 if req.header("sec-websocket-version").map(str::trim) != Some("13") {
285 return None;
286 }
287 let key = req.header("sec-websocket-key")?.trim();
288 (key.len() == 24).then(|| key.to_string())
289}
290
291pub fn upgrade<F>(key: &str, open: F) -> Response
295where
296 F: FnOnce() -> crate::Result<Box<dyn Pty>> + Send + 'static,
297{
298 upgrade_with(key, &TERMINALS, open)
299}
300
301pub fn upgrade_with<F>(key: &str, limits: &'static Limits, open: F) -> Response
303where
304 F: FnOnce() -> crate::Result<Box<dyn Pty>> + Send + 'static,
305{
306 let accept = tungstenite::handshake::derive_accept_key(key.as_bytes());
307 Response::upgrade(
308 "websocket",
309 Box::new(move |s: &mut dyn Duplex| {
310 let mut ws = WebSocket::from_raw_socket(
311 s,
312 Role::Server,
313 Some(
314 WebSocketConfig::default()
315 .max_message_size(Some(MAX_MESSAGE))
316 .max_frame_size(Some(MAX_MESSAGE)),
317 ),
318 );
319 if limits.active.fetch_add(1, Ordering::SeqCst) >= limits.max_sessions {
320 limits.active.fetch_sub(1, Ordering::SeqCst);
321 refuse(&mut ws, limits.busy);
322 return;
323 }
324 let _slot = Slot(limits.active);
325 match open() {
326 Ok(pty) => bridge(&mut ws, pty, limits.idle, limits.max_age),
327 Err(e) => refuse(&mut ws, &e.to_string()),
328 }
329 }),
330 )
331 .header("Sec-WebSocket-Accept", accept)
332}
333
334fn control(v: serde_json::Value) -> Message {
335 Message::text(v.to_string())
336}
337
338fn refuse<S: std::io::Read + std::io::Write>(ws: &mut WebSocket<S>, message: &str) {
339 let _ = ws.send(control(json!({"type": "error", "message": message})));
340 let _ = ws.close(Some(CloseFrame {
341 code: CloseCode::Policy,
342 reason: "refused".into(),
343 }));
344 let _ = ws.flush();
345}
346
347fn would_block(e: &tungstenite::Error) -> bool {
348 matches!(e, tungstenite::Error::Io(io) if matches!(io.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut))
349}
350
351#[expect(
354 clippy::too_many_lines,
355 reason = "predates the lint ratchet; split it when next changed"
356)]
357pub fn bridge(
358 ws: &mut WebSocket<&mut dyn Duplex>,
359 mut pty: Box<dyn Pty>,
360 idle: Duration,
361 max_age: Duration,
362) {
363 let _ = ws.get_mut().set_read_timeout(Some(POLL));
364 let started = Instant::now();
365 let mut last = Instant::now();
366 let mut why: Option<&str> = None;
367 'session: loop {
368 if started.elapsed() >= max_age {
369 why = Some("the session reached its time limit");
370 break;
371 }
372 if last.elapsed() >= idle {
373 why = Some("closed after being idle");
374 break;
375 }
376 match ws.read() {
377 Ok(Message::Binary(b)) => {
378 last = Instant::now();
379 if let Err(e) = pty.input(&b) {
380 let _ = ws.send(control(json!({"type": "error", "message": e.to_string()})));
381 break;
382 }
383 }
384 Ok(Message::Text(t)) => {
385 last = Instant::now();
386 if let Ok(v) = serde_json::from_str::<serde_json::Value>(&t) {
387 if v["type"] == "resize" {
388 let n = |k: &str| v[k].as_u64().map(|x| x.clamp(2, 1000) as u16);
389 if let (Some(c), Some(r)) = (n("cols"), n("rows")) {
390 pty.resize(c, r);
391 }
392 }
393 }
394 }
395 Ok(Message::Close(_)) => {
396 pty.close();
397 return;
398 }
399 Ok(_) => {}
400 Err(e) if would_block(&e) => {}
401 Err(_) => {
402 pty.close();
403 return;
404 }
405 }
406 for _ in 0..64 {
408 match pty.output(Duration::ZERO) {
409 PtyOutput::Data(d) => {
410 last = Instant::now();
411 if let Err(e) = ws.send(Message::binary(d)) {
412 if !would_block(&e) {
413 pty.close();
414 return;
415 }
416 }
417 }
418 PtyOutput::Idle => break,
419 PtyOutput::Note(v) => {
420 let _ = ws.send(control(v));
421 }
422 PtyOutput::Failed(m) => {
423 let _ = ws.send(control(json!({"type": "error", "message": m})));
424 let _ = ws.close(Some(CloseFrame {
425 code: CloseCode::Error,
426 reason: "failed".into(),
427 }));
428 let _ = ws.flush();
429 break 'session;
430 }
431 PtyOutput::Exit(code) => {
432 let _ = ws.send(control(json!({"type": "exit", "code": code})));
433 let _ = ws.close(Some(CloseFrame {
434 code: CloseCode::Normal,
435 reason: "exited".into(),
436 }));
437 let _ = ws.flush();
438 break 'session;
439 }
440 }
441 }
442 match ws.flush() {
443 Ok(()) => {}
444 Err(e) if would_block(&e) => {}
445 Err(_) => {
446 pty.close();
447 return;
448 }
449 }
450 }
451 if let Some(w) = why {
452 let _ = ws.send(control(json!({"type": "error", "message": w})));
453 let _ = ws.close(Some(CloseFrame {
454 code: CloseCode::Normal,
455 reason: "closed".into(),
456 }));
457 let _ = ws.flush();
458 }
459 pty.close();
460 let until = Instant::now() + Duration::from_millis(500);
462 while Instant::now() < until {
463 match ws.read() {
464 Ok(_) => {}
465 Err(e) if would_block(&e) => {}
466 Err(_) => break,
467 }
468 }
469}
470
471#[cfg(test)]
472mod tests {
473 use super::*;
474 use crate::server::http::Peer;
475 use std::os::unix::net::UnixStream;
476 use std::sync::Mutex;
477 use std::sync::mpsc::{Receiver, Sender, channel};
478
479 fn req(query: &str, headers: &[(&str, &str)]) -> Request {
480 Request {
481 method: "GET".into(),
482 path: "/orgs/acme/api/v1/terminal".into(),
483 query: Some(query.into()),
484 headers: headers
485 .iter()
486 .map(|(k, v)| (k.to_string(), v.to_string()))
487 .collect(),
488 body: vec![],
489 peer: Peer::Unix { uid: None },
490 }
491 }
492
493 #[test]
494 fn parses_terminal_requests() {
495 let t = term_request(&req("app=web&slot=2&cols=120&rows=40", &[])).unwrap();
496 assert_eq!(
497 t,
498 TermRequest {
499 app: "web".into(),
500 slot: Some(2),
501 instance: None,
502 session: None,
503 cols: 120,
504 rows: 40
505 }
506 );
507 let t = term_request(&req("instance=box&session=Shell%202", &[])).unwrap();
508 assert_eq!(t.session.as_deref(), Some("Shell 2"));
509 assert_eq!(t.query(), "instance=box&cols=80&rows=24&session=Shell%202");
510 assert_eq!(
511 pct_decode(&pct_encode("api: logs #2")).as_deref(),
512 Some("api: logs #2")
513 );
514 let t = term_request(&req("instance=box-1", &[])).unwrap();
515 assert_eq!((t.app.as_str(), t.instance.as_deref()), ("", Some("box-1")));
516 assert_eq!(t.query(), "instance=box-1&cols=80&rows=24");
517 assert_eq!(t.target(), "box-1");
518 let t = term_request(&req("app=web", &[])).unwrap();
519 assert_eq!((t.slot, t.cols, t.rows), (None, 80, 24));
520 for bad in [
521 "",
522 "app=",
523 "app=../x",
524 "app=Web",
525 "app=web&slot=0",
526 "app=web&cols=1",
527 "app=web&rows=x",
528 "instance=Box",
529 "instance=a&app=b",
530 "instance=a&slot=1",
531 "app=web&session=x",
532 "instance=a&session=",
533 "instance=a&session=%0A",
534 "instance=a&session=%zz",
535 ] {
536 assert!(term_request(&req(bad, &[])).is_err(), "{bad}");
537 }
538 }
539
540 #[test]
541 fn origin_must_be_this_site_for_cookies() {
542 let host = ("Host", "isb.example.com");
543 assert!(origin_allowed(&req(
544 "",
545 &[host, ("Origin", "https://isb.example.com")]
546 )));
547 assert!(origin_allowed(&req(
548 "",
549 &[
550 ("Host", "localhost:8092"),
551 ("Origin", "http://localhost:8092")
552 ]
553 )));
554 assert!(!origin_allowed(&req(
555 "",
556 &[host, ("Origin", "https://evil.example")]
557 )));
558 assert!(!origin_allowed(&req(
559 "",
560 &[host, ("Origin", "https://isb.example.com.evil.example")]
561 )));
562 assert!(!origin_allowed(&req("", &[host, ("Origin", "null")])));
563 assert!(!origin_allowed(&req(
565 "",
566 &[host, ("Cookie", "isb_session=x")]
567 )));
568 assert!(origin_allowed(&req(
569 "",
570 &[host, ("Authorization", "Bearer isb_tok_x")]
571 )));
572 }
573
574 #[test]
575 fn recognises_a_websocket_upgrade() {
576 let ok = [
577 ("Upgrade", "websocket"),
578 ("Connection", "keep-alive, Upgrade"),
579 ("Sec-WebSocket-Version", "13"),
580 ("Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ=="),
581 ];
582 assert_eq!(
583 websocket_key(&req("", &ok)).as_deref(),
584 Some("dGhlIHNhbXBsZSBub25jZQ==")
585 );
586 assert_eq!(
587 tungstenite::handshake::derive_accept_key(b"dGhlIHNhbXBsZSBub25jZQ=="),
588 "s3pPLMBiTxaQ9kYGzzhZRbK+xOo="
589 );
590 let without = |skip: &str| {
591 let h: Vec<_> = ok.iter().copied().filter(|(k, _)| *k != skip).collect();
592 websocket_key(&req("", &h))
593 };
594 for h in [
595 "Upgrade",
596 "Connection",
597 "Sec-WebSocket-Version",
598 "Sec-WebSocket-Key",
599 ] {
600 assert!(without(h).is_none(), "{h}");
601 }
602 }
603
604 struct Echo {
606 rx: Receiver<PtyOutput>,
607 tx: Sender<PtyOutput>,
608 resized: Arc<Mutex<Option<(u16, u16)>>>,
609 closed: Arc<Mutex<bool>>,
610 }
611
612 impl Pty for Echo {
613 fn input(&mut self, d: &[u8]) -> crate::Result<()> {
614 if d == b"exit" {
615 self.tx.send(PtyOutput::Exit(Some(3))).unwrap();
616 } else {
617 self.tx
618 .send(PtyOutput::Data(d.to_ascii_uppercase()))
619 .unwrap();
620 }
621 Ok(())
622 }
623 fn resize(&mut self, c: u16, r: u16) {
624 *self.resized.lock().unwrap() = Some((c, r));
625 }
626 fn output(&mut self, wait: Duration) -> PtyOutput {
627 self.rx.recv_timeout(wait).unwrap_or(PtyOutput::Idle)
628 }
629 fn close(&mut self) {
630 *self.closed.lock().unwrap() = true;
631 }
632 }
633
634 type Resized = Arc<Mutex<Option<(u16, u16)>>>;
635
636 fn echo() -> (Box<dyn Pty>, Resized, Arc<Mutex<bool>>) {
637 let (tx, rx) = channel();
638 let resized = Arc::new(Mutex::new(None));
639 let closed = Arc::new(Mutex::new(false));
640 (
641 Box::new(Echo {
642 rx,
643 tx,
644 resized: resized.clone(),
645 closed: closed.clone(),
646 }),
647 resized,
648 closed,
649 )
650 }
651
652 fn pair() -> (UnixStream, WebSocket<UnixStream>) {
653 let (a, b) = UnixStream::pair().unwrap();
654 b.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
655 (a, WebSocket::from_raw_socket(b, Role::Client, None))
656 }
657
658 #[test]
659 fn bridges_bytes_resizes_and_exit() {
660 let (server, mut client) = pair();
661 let (pty, resized, closed) = echo();
662 let t = std::thread::spawn(move || {
663 let mut s = server;
664 let d: &mut dyn Duplex = &mut s;
665 let mut ws = WebSocket::from_raw_socket(d, Role::Server, None);
666 bridge(&mut ws, pty, IDLE, MAX_AGE);
667 });
668 client
669 .send(Message::text(r#"{"type":"resize","cols":100,"rows":30}"#))
670 .unwrap();
671 client.send(Message::binary(b"ls".to_vec())).unwrap();
672 assert_eq!(client.read().unwrap(), Message::binary(b"LS".to_vec()));
673 client.send(Message::binary(b"exit".to_vec())).unwrap();
674 let m = client.read().unwrap();
675 let v: serde_json::Value = serde_json::from_str(m.to_text().unwrap()).unwrap();
676 assert_eq!(v, json!({"type": "exit", "code": 3}));
677 assert!(matches!(client.read(), Ok(Message::Close(_))));
678 t.join().unwrap();
679 assert_eq!(*resized.lock().unwrap(), Some((100, 30)));
680 assert!(*closed.lock().unwrap());
681 }
682
683 #[test]
684 fn idle_sessions_are_closed() {
685 let (server, mut client) = pair();
686 let (pty, _, closed) = echo();
687 let t = std::thread::spawn(move || {
688 let mut s = server;
689 let d: &mut dyn Duplex = &mut s;
690 let mut ws = WebSocket::from_raw_socket(d, Role::Server, None);
691 bridge(&mut ws, pty, Duration::from_millis(100), MAX_AGE);
692 });
693 let m = client.read().unwrap();
694 assert!(m.to_text().unwrap().contains("idle"), "{m:?}");
695 assert!(matches!(client.read(), Ok(Message::Close(_))));
696 t.join().unwrap();
697 assert!(*closed.lock().unwrap());
698 }
699
700 #[test]
701 fn a_dropped_browser_closes_the_terminal() {
702 let (server, client) = pair();
703 let (pty, _, closed) = echo();
704 let t = std::thread::spawn(move || {
705 let mut s = server;
706 let d: &mut dyn Duplex = &mut s;
707 let mut ws = WebSocket::from_raw_socket(d, Role::Server, None);
708 bridge(&mut ws, pty, IDLE, MAX_AGE);
709 });
710 drop(client);
711 t.join().unwrap();
712 assert!(*closed.lock().unwrap());
713 }
714}