isb_server/server/
terminal.rs

1//! A terminal over a websocket: `GET /orgs/<org>/api/v1/terminal?app=NAME`
2//! (or `?instance=NAME`) upgrades to a websocket bridged to a
3//! pseudo-terminal the embedder opens (`isb serve`: a shell in one of the
4//! app's replicas, or in an instance of the org).
5//!
6//! The gate is the REST surface's: the caller authenticates as for any tool
7//! (session, API token, Access), and is admitted as if calling
8//! `sandbox_exec` in the org, so the org rules and `--deny-tools` apply.
9//! A browser's cookie rides along on a cross-site websocket too, so a
10//! cookie-authenticated upgrade must come from this site (`Origin` naming
11//! the request's `Host`).
12//!
13//! On the wire: binary frames are terminal bytes both ways; text frames are
14//! JSON control messages: from the browser `{"type": "resize", "cols",
15//! "rows"}`, from the server `{"type": "exit", "code"}` and `{"type":
16//! "error", "message"}`.
17//!
18//! Bounded: at most [`MAX_SESSIONS`] at once, [`IDLE`] without a byte either
19//! way, [`MAX_AGE`] in all, messages up to [`MAX_MESSAGE`].
20
21use 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;
36/// How long one turn of the bridge waits for the browser before polling the
37/// terminal's output.
38const POLL: Duration = Duration::from_millis(15);
39
40/// What the browser asked for.
41#[derive(Debug, Clone, PartialEq, Eq)]
42pub struct TermRequest {
43    /// The app, or empty when `instance` names the target.
44    pub app: String,
45    /// A replica's slot; `None` picks one.
46    pub slot: Option<u32>,
47    /// An instance of the org (a workspace, a sandbox) instead of an app.
48    pub instance: Option<String>,
49    /// A workspace's named terminal session (herdr-backed), which outlives
50    /// the websocket; `None` is a plain shell that ends with it.
51    pub session: Option<String>,
52    pub cols: u16,
53    pub rows: u16,
54}
55
56impl TermRequest {
57    /// The query string that asks for this terminal again (forwarding).
58    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    /// What it opens on, for logs: the app or the instance.
74    pub fn target(&self) -> &str {
75        self.instance.as_deref().unwrap_or(&self.app)
76    }
77}
78
79/// How many sessions of one kind may be open at once, and for how long.
80#[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    /// Said when `max_sessions` are open.
87    pub busy: &'static str,
88}
89
90/// The web terminal's limits.
91pub 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/// One terminal session's output, polled.
100#[derive(Debug, PartialEq, Eq)]
101pub enum PtyOutput {
102    Data(Vec<u8>),
103    /// Nothing within the wait.
104    Idle,
105    /// The program ended (with its exit code, when known).
106    Exit(Option<i32>),
107    /// It could not run, or the exec broke: shown to the person, then closed.
108    Failed(String),
109}
110
111/// A pseudo-terminal the bridge drives.
112pub trait Pty: Send {
113    fn input(&mut self, data: &[u8]) -> crate::Result<()>;
114    fn resize(&mut self, cols: u16, rows: u16);
115    fn output(&mut self, wait: Duration) -> PtyOutput;
116    /// End the program; the session is over.
117    fn close(&mut self);
118    /// What it runs in (an instance name), for the audit log.
119    fn target(&self) -> Option<String> {
120        None
121    }
122    /// What it learnt while running that the audit log's closing row should
123    /// keep (an SSH session's user and key fingerprint). Never content.
124    fn details(&self) -> Option<serde_json::Map<String, serde_json::Value>> {
125        None
126    }
127}
128
129/// Opens a terminal for `caller` in `org`; refusals become an error frame.
130pub type Terminal = Arc<
131    dyn Fn(&Caller, &crate::org::OrgId, &TermRequest) -> crate::Result<Box<dyn Pty>> + Send + Sync,
132>;
133
134static ACTIVE: AtomicUsize = AtomicUsize::new(0);
135
136struct Slot(&'static AtomicUsize);
137
138impl Drop for Slot {
139    fn drop(&mut self) {
140        self.0.fetch_sub(1, Ordering::SeqCst);
141    }
142}
143
144fn param(req: &Request, key: &str) -> Option<String> {
145    req.query.as_deref()?.split('&').find_map(|kv| {
146        let (k, v) = kv.split_once('=').unwrap_or((kv, ""));
147        (k == key).then(|| v.to_string())
148    })
149}
150
151/// A lower-case name of letters, digits and dashes, at most 64 long (an
152/// app's or an instance's).
153pub fn plain_name(s: &str) -> bool {
154    !s.is_empty()
155        && s.len() <= 64
156        && s.bytes()
157            .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-')
158}
159
160/// A query parameter's raw value.
161pub fn query_param(req: &Request, key: &str) -> Option<String> {
162    param(req, key)
163}
164
165/// Percent-encode everything but unreserved characters.
166pub fn pct_encode(s: &str) -> String {
167    s.bytes()
168        .map(|b| match b {
169            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => {
170                (b as char).to_string()
171            }
172            b => format!("%{b:02X}"),
173        })
174        .collect()
175}
176
177/// Undo [`pct_encode`] (and a form's `+` for a space); `None` when it is not
178/// valid UTF-8 or a `%` is not followed by two hex digits.
179pub fn pct_decode(s: &str) -> Option<String> {
180    let b = s.as_bytes();
181    let mut out = Vec::with_capacity(b.len());
182    let mut i = 0;
183    while i < b.len() {
184        match b[i] {
185            b'%' => {
186                let hex = std::str::from_utf8(b.get(i + 1..i + 3)?).ok()?;
187                out.push(u8::from_str_radix(hex, 16).ok()?);
188                i += 3;
189            }
190            b'+' => {
191                out.push(b' ');
192                i += 1;
193            }
194            c => {
195                out.push(c);
196                i += 1;
197            }
198        }
199    }
200    String::from_utf8(out).ok()
201}
202
203/// The request's terminal parameters, or what is wrong with them.
204pub fn term_request(req: &Request) -> std::result::Result<TermRequest, String> {
205    let (app, instance) = match (param(req, "app"), param(req, "instance")) {
206        (Some(a), None) if plain_name(&a) => (a, None),
207        (Some(_), None) => return Err("app= is not an app name".into()),
208        (None, Some(i)) if plain_name(&i) => (String::new(), Some(i)),
209        (None, Some(_)) => return Err("instance= is not an instance name".into()),
210        (Some(_), Some(_)) => return Err("give app= or instance=, not both".into()),
211        (None, None) => return Err("app= (or instance=) is required".into()),
212    };
213    let num = |k: &str, lo: u32, hi: u32, def: u32| -> std::result::Result<u32, String> {
214        match param(req, k) {
215            None => Ok(def),
216            Some(v) => v
217                .parse::<u32>()
218                .ok()
219                .filter(|n| (lo..=hi).contains(n))
220                .ok_or_else(|| format!("{k}= must be {lo}-{hi}")),
221        }
222    };
223    let slot = match param(req, "slot") {
224        None => None,
225        Some(_) if instance.is_some() => return Err("slot= is for apps".into()),
226        Some(_) => Some(num("slot", 1, 1000, 1)?),
227    };
228    let session = match param(req, "session") {
229        None => None,
230        Some(_) if instance.is_none() => return Err("session= is for instances".into()),
231        Some(v) => match pct_decode(&v) {
232            Some(s)
233                if !s.is_empty() && s.chars().count() <= 40 && !s.chars().any(char::is_control) =>
234            {
235                Some(s)
236            }
237            _ => return Err("session= is not a session name".into()),
238        },
239    };
240    Ok(TermRequest {
241        app,
242        slot,
243        instance,
244        session,
245        cols: num("cols", 2, 1000, 80)? as u16,
246        rows: num("rows", 2, 1000, 24)? as u16,
247    })
248}
249
250/// `Origin`'s host and port, when it is an http(s) origin.
251fn origin_authority(origin: &str) -> Option<&str> {
252    let rest = origin
253        .strip_prefix("https://")
254        .or_else(|| origin.strip_prefix("http://"))?;
255    (!rest.is_empty() && !rest.contains('/')).then_some(rest)
256}
257
258/// A cookie-authenticated upgrade must come from a page on this site. A
259/// bearer token is not sent by browsers on their own, so it needs no Origin.
260pub fn origin_allowed(req: &Request) -> bool {
261    let bearer = req.header("authorization").is_some();
262    match req.header("origin") {
263        None => bearer,
264        Some(o) => match (origin_authority(o), req.header("host")) {
265            (Some(a), Some(h)) => a.eq_ignore_ascii_case(h.trim()),
266            _ => false,
267        },
268    }
269}
270
271/// The client's websocket key, if this is a websocket upgrade request.
272pub fn websocket_key(req: &Request) -> Option<String> {
273    let has = |name: &str, token: &str| {
274        req.header(name)
275            .is_some_and(|v| v.split(',').any(|t| t.trim().eq_ignore_ascii_case(token)))
276    };
277    if req.method != "GET" || !has("upgrade", "websocket") || !has("connection", "upgrade") {
278        return None;
279    }
280    if req.header("sec-websocket-version").map(str::trim) != Some("13") {
281        return None;
282    }
283    let key = req.header("sec-websocket-key")?.trim();
284    (key.len() == 24).then(|| key.to_string())
285}
286
287/// The 101 that hands the connection to a session opened by `open`. The
288/// opening happens after the upgrade, so its errors reach the browser as an
289/// error frame rather than a bare failed handshake.
290pub fn upgrade<F>(key: &str, open: F) -> Response
291where
292    F: FnOnce() -> crate::Result<Box<dyn Pty>> + Send + 'static,
293{
294    upgrade_with(key, &TERMINALS, open)
295}
296
297/// [`upgrade`] under `limits` (the SSH bridge has its own).
298pub fn upgrade_with<F>(key: &str, limits: &'static Limits, open: F) -> Response
299where
300    F: FnOnce() -> crate::Result<Box<dyn Pty>> + Send + 'static,
301{
302    let accept = tungstenite::handshake::derive_accept_key(key.as_bytes());
303    Response::upgrade(
304        "websocket",
305        Box::new(move |s: &mut dyn Duplex| {
306            let mut ws = WebSocket::from_raw_socket(
307                s,
308                Role::Server,
309                Some(
310                    WebSocketConfig::default()
311                        .max_message_size(Some(MAX_MESSAGE))
312                        .max_frame_size(Some(MAX_MESSAGE)),
313                ),
314            );
315            if limits.active.fetch_add(1, Ordering::SeqCst) >= limits.max_sessions {
316                limits.active.fetch_sub(1, Ordering::SeqCst);
317                refuse(&mut ws, limits.busy);
318                return;
319            }
320            let _slot = Slot(limits.active);
321            match open() {
322                Ok(pty) => bridge(&mut ws, pty, limits.idle, limits.max_age),
323                Err(e) => refuse(&mut ws, &e.to_string()),
324            }
325        }),
326    )
327    .header("Sec-WebSocket-Accept", accept)
328}
329
330fn control(v: serde_json::Value) -> Message {
331    Message::text(v.to_string())
332}
333
334fn refuse<S: std::io::Read + std::io::Write>(ws: &mut WebSocket<S>, message: &str) {
335    let _ = ws.send(control(json!({"type": "error", "message": message})));
336    let _ = ws.close(Some(CloseFrame {
337        code: CloseCode::Policy,
338        reason: "refused".into(),
339    }));
340    let _ = ws.flush();
341}
342
343fn would_block(e: &tungstenite::Error) -> bool {
344    matches!(e, tungstenite::Error::Io(io) if matches!(io.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut))
345}
346
347/// Shuttle bytes between the websocket and the terminal until either ends,
348/// the session idles for `idle`, or it reaches `max_age`.
349#[expect(
350    clippy::too_many_lines,
351    reason = "predates the lint ratchet; split it when next changed"
352)]
353pub fn bridge(
354    ws: &mut WebSocket<&mut dyn Duplex>,
355    mut pty: Box<dyn Pty>,
356    idle: Duration,
357    max_age: Duration,
358) {
359    let _ = ws.get_mut().set_read_timeout(Some(POLL));
360    let started = Instant::now();
361    let mut last = Instant::now();
362    let mut why: Option<&str> = None;
363    'session: loop {
364        if started.elapsed() >= max_age {
365            why = Some("the session reached its time limit");
366            break;
367        }
368        if last.elapsed() >= idle {
369            why = Some("closed after being idle");
370            break;
371        }
372        match ws.read() {
373            Ok(Message::Binary(b)) => {
374                last = Instant::now();
375                if let Err(e) = pty.input(&b) {
376                    let _ = ws.send(control(json!({"type": "error", "message": e.to_string()})));
377                    break;
378                }
379            }
380            Ok(Message::Text(t)) => {
381                last = Instant::now();
382                if let Ok(v) = serde_json::from_str::<serde_json::Value>(&t) {
383                    if v["type"] == "resize" {
384                        let n = |k: &str| v[k].as_u64().map(|x| x.clamp(2, 1000) as u16);
385                        if let (Some(c), Some(r)) = (n("cols"), n("rows")) {
386                            pty.resize(c, r);
387                        }
388                    }
389                }
390            }
391            Ok(Message::Close(_)) => {
392                pty.close();
393                return;
394            }
395            Ok(_) => {}
396            Err(e) if would_block(&e) => {}
397            Err(_) => {
398                pty.close();
399                return;
400            }
401        }
402        // Pass on what the terminal wrote, a bounded amount per turn.
403        for _ in 0..64 {
404            match pty.output(Duration::ZERO) {
405                PtyOutput::Data(d) => {
406                    last = Instant::now();
407                    if let Err(e) = ws.send(Message::binary(d)) {
408                        if !would_block(&e) {
409                            pty.close();
410                            return;
411                        }
412                    }
413                }
414                PtyOutput::Idle => break,
415                PtyOutput::Failed(m) => {
416                    let _ = ws.send(control(json!({"type": "error", "message": m})));
417                    let _ = ws.close(Some(CloseFrame {
418                        code: CloseCode::Error,
419                        reason: "failed".into(),
420                    }));
421                    let _ = ws.flush();
422                    break 'session;
423                }
424                PtyOutput::Exit(code) => {
425                    let _ = ws.send(control(json!({"type": "exit", "code": code})));
426                    let _ = ws.close(Some(CloseFrame {
427                        code: CloseCode::Normal,
428                        reason: "exited".into(),
429                    }));
430                    let _ = ws.flush();
431                    break 'session;
432                }
433            }
434        }
435        match ws.flush() {
436            Ok(()) => {}
437            Err(e) if would_block(&e) => {}
438            Err(_) => {
439                pty.close();
440                return;
441            }
442        }
443    }
444    if let Some(w) = why {
445        let _ = ws.send(control(json!({"type": "error", "message": w})));
446        let _ = ws.close(Some(CloseFrame {
447            code: CloseCode::Normal,
448            reason: "closed".into(),
449        }));
450        let _ = ws.flush();
451    }
452    pty.close();
453    // Let the close handshake finish, briefly.
454    let until = Instant::now() + Duration::from_millis(500);
455    while Instant::now() < until {
456        match ws.read() {
457            Ok(_) => {}
458            Err(e) if would_block(&e) => {}
459            Err(_) => break,
460        }
461    }
462}
463
464#[cfg(test)]
465mod tests {
466    use super::*;
467    use crate::server::http::Peer;
468    use std::os::unix::net::UnixStream;
469    use std::sync::Mutex;
470    use std::sync::mpsc::{Receiver, Sender, channel};
471
472    fn req(query: &str, headers: &[(&str, &str)]) -> Request {
473        Request {
474            method: "GET".into(),
475            path: "/orgs/acme/api/v1/terminal".into(),
476            query: Some(query.into()),
477            headers: headers
478                .iter()
479                .map(|(k, v)| (k.to_string(), v.to_string()))
480                .collect(),
481            body: vec![],
482            peer: Peer::Unix { uid: None },
483        }
484    }
485
486    #[test]
487    fn parses_terminal_requests() {
488        let t = term_request(&req("app=web&slot=2&cols=120&rows=40", &[])).unwrap();
489        assert_eq!(
490            t,
491            TermRequest {
492                app: "web".into(),
493                slot: Some(2),
494                instance: None,
495                session: None,
496                cols: 120,
497                rows: 40
498            }
499        );
500        let t = term_request(&req("instance=box&session=Shell%202", &[])).unwrap();
501        assert_eq!(t.session.as_deref(), Some("Shell 2"));
502        assert_eq!(t.query(), "instance=box&cols=80&rows=24&session=Shell%202");
503        assert_eq!(
504            pct_decode(&pct_encode("api: logs #2")).as_deref(),
505            Some("api: logs #2")
506        );
507        let t = term_request(&req("instance=box-1", &[])).unwrap();
508        assert_eq!((t.app.as_str(), t.instance.as_deref()), ("", Some("box-1")));
509        assert_eq!(t.query(), "instance=box-1&cols=80&rows=24");
510        assert_eq!(t.target(), "box-1");
511        let t = term_request(&req("app=web", &[])).unwrap();
512        assert_eq!((t.slot, t.cols, t.rows), (None, 80, 24));
513        for bad in [
514            "",
515            "app=",
516            "app=../x",
517            "app=Web",
518            "app=web&slot=0",
519            "app=web&cols=1",
520            "app=web&rows=x",
521            "instance=Box",
522            "instance=a&app=b",
523            "instance=a&slot=1",
524            "app=web&session=x",
525            "instance=a&session=",
526            "instance=a&session=%0A",
527            "instance=a&session=%zz",
528        ] {
529            assert!(term_request(&req(bad, &[])).is_err(), "{bad}");
530        }
531    }
532
533    #[test]
534    fn origin_must_be_this_site_for_cookies() {
535        let host = ("Host", "isb.example.com");
536        assert!(origin_allowed(&req(
537            "",
538            &[host, ("Origin", "https://isb.example.com")]
539        )));
540        assert!(origin_allowed(&req(
541            "",
542            &[
543                ("Host", "localhost:8092"),
544                ("Origin", "http://localhost:8092")
545            ]
546        )));
547        assert!(!origin_allowed(&req(
548            "",
549            &[host, ("Origin", "https://evil.example")]
550        )));
551        assert!(!origin_allowed(&req(
552            "",
553            &[host, ("Origin", "https://isb.example.com.evil.example")]
554        )));
555        assert!(!origin_allowed(&req("", &[host, ("Origin", "null")])));
556        // No Origin: only a bearer token, which a browser never adds by itself.
557        assert!(!origin_allowed(&req(
558            "",
559            &[host, ("Cookie", "isb_session=x")]
560        )));
561        assert!(origin_allowed(&req(
562            "",
563            &[host, ("Authorization", "Bearer isb_tok_x")]
564        )));
565    }
566
567    #[test]
568    fn recognises_a_websocket_upgrade() {
569        let ok = [
570            ("Upgrade", "websocket"),
571            ("Connection", "keep-alive, Upgrade"),
572            ("Sec-WebSocket-Version", "13"),
573            ("Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ=="),
574        ];
575        assert_eq!(
576            websocket_key(&req("", &ok)).as_deref(),
577            Some("dGhlIHNhbXBsZSBub25jZQ==")
578        );
579        assert_eq!(
580            tungstenite::handshake::derive_accept_key(b"dGhlIHNhbXBsZSBub25jZQ=="),
581            "s3pPLMBiTxaQ9kYGzzhZRbK+xOo="
582        );
583        let without = |skip: &str| {
584            let h: Vec<_> = ok.iter().copied().filter(|(k, _)| *k != skip).collect();
585            websocket_key(&req("", &h))
586        };
587        for h in [
588            "Upgrade",
589            "Connection",
590            "Sec-WebSocket-Version",
591            "Sec-WebSocket-Key",
592        ] {
593            assert!(without(h).is_none(), "{h}");
594        }
595    }
596
597    /// A terminal that echoes input upper-cased, and exits on "exit".
598    struct Echo {
599        rx: Receiver<PtyOutput>,
600        tx: Sender<PtyOutput>,
601        resized: Arc<Mutex<Option<(u16, u16)>>>,
602        closed: Arc<Mutex<bool>>,
603    }
604
605    impl Pty for Echo {
606        fn input(&mut self, d: &[u8]) -> crate::Result<()> {
607            if d == b"exit" {
608                self.tx.send(PtyOutput::Exit(Some(3))).unwrap();
609            } else {
610                self.tx
611                    .send(PtyOutput::Data(d.to_ascii_uppercase()))
612                    .unwrap();
613            }
614            Ok(())
615        }
616        fn resize(&mut self, c: u16, r: u16) {
617            *self.resized.lock().unwrap() = Some((c, r));
618        }
619        fn output(&mut self, wait: Duration) -> PtyOutput {
620            self.rx.recv_timeout(wait).unwrap_or(PtyOutput::Idle)
621        }
622        fn close(&mut self) {
623            *self.closed.lock().unwrap() = true;
624        }
625    }
626
627    type Resized = Arc<Mutex<Option<(u16, u16)>>>;
628
629    fn echo() -> (Box<dyn Pty>, Resized, Arc<Mutex<bool>>) {
630        let (tx, rx) = channel();
631        let resized = Arc::new(Mutex::new(None));
632        let closed = Arc::new(Mutex::new(false));
633        (
634            Box::new(Echo {
635                rx,
636                tx,
637                resized: resized.clone(),
638                closed: closed.clone(),
639            }),
640            resized,
641            closed,
642        )
643    }
644
645    fn pair() -> (UnixStream, WebSocket<UnixStream>) {
646        let (a, b) = UnixStream::pair().unwrap();
647        b.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
648        (a, WebSocket::from_raw_socket(b, Role::Client, None))
649    }
650
651    #[test]
652    fn bridges_bytes_resizes_and_exit() {
653        let (server, mut client) = pair();
654        let (pty, resized, closed) = echo();
655        let t = std::thread::spawn(move || {
656            let mut s = server;
657            let d: &mut dyn Duplex = &mut s;
658            let mut ws = WebSocket::from_raw_socket(d, Role::Server, None);
659            bridge(&mut ws, pty, IDLE, MAX_AGE);
660        });
661        client
662            .send(Message::text(r#"{"type":"resize","cols":100,"rows":30}"#))
663            .unwrap();
664        client.send(Message::binary(b"ls".to_vec())).unwrap();
665        assert_eq!(client.read().unwrap(), Message::binary(b"LS".to_vec()));
666        client.send(Message::binary(b"exit".to_vec())).unwrap();
667        let m = client.read().unwrap();
668        let v: serde_json::Value = serde_json::from_str(m.to_text().unwrap()).unwrap();
669        assert_eq!(v, json!({"type": "exit", "code": 3}));
670        assert!(matches!(client.read(), Ok(Message::Close(_))));
671        t.join().unwrap();
672        assert_eq!(*resized.lock().unwrap(), Some((100, 30)));
673        assert!(*closed.lock().unwrap());
674    }
675
676    #[test]
677    fn idle_sessions_are_closed() {
678        let (server, mut client) = pair();
679        let (pty, _, closed) = echo();
680        let t = std::thread::spawn(move || {
681            let mut s = server;
682            let d: &mut dyn Duplex = &mut s;
683            let mut ws = WebSocket::from_raw_socket(d, Role::Server, None);
684            bridge(&mut ws, pty, Duration::from_millis(100), MAX_AGE);
685        });
686        let m = client.read().unwrap();
687        assert!(m.to_text().unwrap().contains("idle"), "{m:?}");
688        assert!(matches!(client.read(), Ok(Message::Close(_))));
689        t.join().unwrap();
690        assert!(*closed.lock().unwrap());
691    }
692
693    #[test]
694    fn a_dropped_browser_closes_the_terminal() {
695        let (server, client) = pair();
696        let (pty, _, closed) = echo();
697        let t = std::thread::spawn(move || {
698            let mut s = server;
699            let d: &mut dyn Duplex = &mut s;
700            let mut ws = WebSocket::from_raw_socket(d, Role::Server, None);
701            bridge(&mut ws, pty, IDLE, MAX_AGE);
702        });
703        drop(client);
704        t.join().unwrap();
705        assert!(*closed.lock().unwrap());
706    }
707}
isb_server/server/terminal.rs

isb_server/server/
terminal.rs
