isb_server/server/
ssh.rs

1//! SSH without open ports: `GET /orgs/<org>/api/v1/ssh?instance=NAME`
2//! upgrades to a websocket whose binary frames are an SSH connection's
3//! bytes, both ways, to an sshd the embedder starts inside the instance
4//! (`isb serve`: `sshd -i` through incus exec, docs/guides/ssh.md). Nothing in the
5//! instance listens, and nothing on the host opens a port: the websocket is
6//! the daemon's own, behind its usual authentication.
7//!
8//! The gate is the web terminal's ([`super::terminal`]): the caller
9//! authenticates as for any tool and is admitted as if calling
10//! `sandbox_exec` in the org, so viewers, `read`/`deploy` tokens and
11//! `--deny-tools sandbox_exec` are refused. A cookie needs an `Origin`
12//! naming this site; the unix socket and bearer tokens need none.
13//!
14//! Text frames are control messages from the server only: `{"type":
15//! "exit", "code"}` and `{"type": "error", "message"}` (why the session
16//! ended, for the person running `ssh`).
17//!
18//! The client half is here too: [`Remote`] (the unix socket, or a URL and a
19//! token) and [`pump`], which `isb ssh-proxy` runs between its stdio and the
20//! websocket for `ProxyCommand`.
21
22use std::io::{ErrorKind, Read, Write};
23use std::net::{TcpStream, ToSocketAddrs};
24use std::os::unix::net::UnixStream;
25use std::path::PathBuf;
26use std::sync::Arc;
27use std::sync::atomic::AtomicUsize;
28use std::sync::mpsc::{TryRecvError, sync_channel};
29use std::time::Duration;
30
31use serde_json::Value;
32use tungstenite::{Message, WebSocket};
33
34use super::http::{Peer, Request};
35use super::mcp::Caller;
36use super::terminal::{Limits, Pty, plain_name, query_param};
37use crate::error::{Error, Result};
38
39/// What the client asked for.
40#[derive(Debug, Clone, PartialEq, Eq)]
41pub struct SshRequest {
42    /// The instance in the org.
43    pub instance: String,
44    /// Whose keys to let in, for a caller with no isb account of its own
45    /// (the unix socket, a superadmin token): an account's email. Anyone
46    /// else may only name themselves.
47    pub keys_of: Option<String>,
48}
49
50impl SshRequest {
51    pub fn query(&self) -> String {
52        let mut q = format!("instance={}", self.instance);
53        if let Some(e) = &self.keys_of {
54            q.push_str(&format!("&as={}", encode(e)));
55        }
56        q
57    }
58}
59
60/// Opens the SSH session for `caller` in `org`; refusals become an error
61/// frame. The [`Pty`] carries the connection's bytes (no resize).
62pub type Ssh =
63    Arc<dyn Fn(&Caller, &crate::org::OrgId, &SshRequest) -> Result<Box<dyn Pty>> + Send + Sync>;
64
65static ACTIVE: AtomicUsize = AtomicUsize::new(0);
66
67/// SSH sessions: more than terminals (an editor or herdr holds several),
68/// longer-lived, and kept alive by the client's `ServerAliveInterval`.
69pub static LIMITS: Limits = Limits {
70    active: &ACTIVE,
71    max_sessions: 64,
72    idle: Duration::from_secs(2 * 3600),
73    max_age: Duration::from_secs(24 * 3600),
74    busy: "too many SSH sessions are open on this server; close one and try again",
75};
76
77/// The request's parameters, or what is wrong with them.
78pub fn ssh_request(req: &Request) -> std::result::Result<SshRequest, String> {
79    let instance = query_param(req, "instance").ok_or("instance= is required")?;
80    if !plain_name(&instance) {
81        return Err("instance= is not an instance name".into());
82    }
83    let keys_of = match query_param(req, "as") {
84        None => None,
85        Some(e) => {
86            let e = decode(&e).ok_or("as= is not an email")?;
87            if e.is_empty() || e.len() > 254 || !e.contains('@') || e.contains(char::is_control) {
88                return Err("as= is not an email".into());
89            }
90            Some(e)
91        }
92    };
93    Ok(SshRequest { instance, keys_of })
94}
95
96/// The unix socket and bearer tokens need no `Origin`; a cookie needs one
97/// naming this site, as for the terminal. An `Origin` that is there must
98/// name this site, whatever the transport.
99pub fn origin_allowed(req: &Request) -> bool {
100    match req.header("origin") {
101        Some(_) => super::terminal::origin_allowed(req),
102        None => matches!(req.peer, Peer::Unix { .. }) || super::terminal::origin_allowed(req),
103    }
104}
105
106fn encode(s: &str) -> String {
107    s.bytes()
108        .map(|b| match b {
109            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' | b'@' => {
110                (b as char).to_string()
111            }
112            _ => format!("%{b:02X}"),
113        })
114        .collect()
115}
116
117fn decode(s: &str) -> Option<String> {
118    let b = s.as_bytes();
119    let mut out = Vec::with_capacity(b.len());
120    let mut i = 0;
121    while i < b.len() {
122        match b[i] {
123            b'%' => {
124                let h = std::str::from_utf8(b.get(i + 1..i + 3)?).ok()?;
125                out.push(u8::from_str_radix(h, 16).ok()?);
126                i += 3;
127            }
128            b'+' => {
129                out.push(b' ');
130                i += 1;
131            }
132            c => {
133                out.push(c);
134                i += 1;
135            }
136        }
137    }
138    String::from_utf8(out).ok()
139}
140
141// ---- the client half ----
142
143/// Where `isb serve` is: its unix socket (as the local user), or a URL and
144/// an API token (`ISB_URL`, `ISB_TOKEN`).
145#[derive(Clone)]
146pub enum Remote {
147    Socket(PathBuf),
148    Url { base: String, token: Option<String> },
149}
150
151impl std::fmt::Debug for Remote {
152    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
153        match self {
154            Remote::Socket(p) => write!(f, "Socket({})", p.display()),
155            // Never the token.
156            Remote::Url { base, token } => write!(
157                f,
158                "Url({base}, token: {})",
159                if token.is_some() { "set" } else { "none" }
160            ),
161        }
162    }
163}
164
165/// A connection the websocket runs over.
166pub trait Conn: Read + Write + Send {
167    fn set_timeout(&self, t: Option<Duration>) -> std::io::Result<()>;
168}
169
170impl Conn for TcpStream {
171    fn set_timeout(&self, t: Option<Duration>) -> std::io::Result<()> {
172        self.set_read_timeout(t)
173    }
174}
175
176impl Conn for UnixStream {
177    fn set_timeout(&self, t: Option<Duration>) -> std::io::Result<()> {
178        self.set_read_timeout(t)
179    }
180}
181
182impl Conn for rustls::StreamOwned<rustls::ClientConnection, TcpStream> {
183    fn set_timeout(&self, t: Option<Duration>) -> std::io::Result<()> {
184        self.sock.set_read_timeout(t)
185    }
186}
187
188/// A `http(s)://host[:port][/prefix]` base: TLS, host, port, prefix.
189pub fn split_base(base: &str) -> Result<(bool, String, u16, String)> {
190    let (tls, rest) = if let Some(r) = base.strip_prefix("https://") {
191        (true, r)
192    } else if let Some(r) = base.strip_prefix("http://") {
193        (false, r)
194    } else {
195        return Err(Error::invalid(format!(
196            "{base:?}: want an http:// or https:// URL"
197        )));
198    };
199    let (authority, prefix) = match rest.find('/') {
200        Some(i) => (&rest[..i], rest[i..].trim_end_matches('/')),
201        None => (rest, ""),
202    };
203    let (host, port) = match authority.rsplit_once(':') {
204        Some((h, p)) if !h.ends_with(']') || h.starts_with('[') => match p.parse::<u16>() {
205            Ok(p) => (h, p),
206            Err(_) => (authority, if tls { 443 } else { 80 }),
207        },
208        _ => (authority, if tls { 443 } else { 80 }),
209    };
210    if host.is_empty() {
211        return Err(Error::invalid(format!("{base:?}: no host")));
212    }
213    Ok((tls, host.to_string(), port, prefix.to_string()))
214}
215
216impl Remote {
217    /// Call a tool in `org`: over the socket's MCP, or the URL's REST
218    /// surface (`/orgs/<org>/api/v1/tools/<tool>`).
219    pub fn call_tool(&self, org: &str, tool: &str, mut args: Value) -> Result<Value> {
220        match self {
221            Remote::Socket(p) => {
222                args["org"] = Value::String(org.to_string());
223                super::client::call_tool(p, tool, args, Duration::from_secs(60))
224            }
225            Remote::Url { .. } => {
226                let (status, v) = self.http(
227                    "POST",
228                    &format!("/orgs/{org}/api/v1/tools/{tool}"),
229                    Some(&args),
230                )?;
231                if status == 200 {
232                    return Ok(v.get("result").cloned().unwrap_or(Value::Null));
233                }
234                Err(answer_error(status, &v))
235            }
236        }
237    }
238
239    /// One JSON request to the URL (`ISB_URL`): its status and body.
240    pub fn http(&self, method: &str, path: &str, body: Option<&Value>) -> Result<(u16, Value)> {
241        let Remote::Url { base, token } = self else {
242            return Err(Error::invalid(
243                "this needs isb serve's URL: pass --url or set ISB_URL",
244            ));
245        };
246        let url = format!("{}{path}", base.trim_end_matches('/'));
247        let agent: ureq::Agent = ureq::Agent::config_builder()
248            .timeout_global(Some(Duration::from_secs(60)))
249            .http_status_as_error(false)
250            .user_agent(concat!("isb/", env!("CARGO_PKG_VERSION")))
251            .build()
252            .into();
253        let auth = token.as_ref().map(|t| format!("Bearer {}", t.trim()));
254        let payload = match body {
255            Some(b) => serde_json::to_vec(b)?,
256            None => Vec::new(),
257        };
258        macro_rules! go {
259            ($req:expr) => {{
260                let mut r = $req.header("X-Isb-Csrf", "1");
261                if let Some(a) = &auth {
262                    r = r.header("Authorization", a);
263                }
264                r
265            }};
266        }
267        let resp = match method {
268            "GET" => go!(agent.get(&url)).call(),
269            "DELETE" => go!(agent.delete(&url)).call(),
270            "POST" => go!(agent.post(&url))
271                .header("Content-Type", "application/json")
272                .send(&payload[..]),
273            m => return Err(Error::invalid(format!("unsupported method {m}"))),
274        };
275        let mut resp = resp.map_err(|e| Error::invalid(format!("{method} {url}: {e}")))?;
276        let status = resp.status().as_u16();
277        let text = resp
278            .body_mut()
279            .with_config()
280            .limit(16 << 20)
281            .read_to_string()
282            .unwrap_or_default();
283        Ok((status, serde_json::from_str(&text).unwrap_or(Value::Null)))
284    }
285
286    /// Open the websocket at `path` (with its query) on this server.
287    pub fn websocket(&self, path: &str) -> Result<WebSocket<Box<dyn Conn>>> {
288        use tungstenite::client::IntoClientRequest;
289        let timeout = Duration::from_secs(15);
290        let (conn, url, token): (Box<dyn Conn>, String, Option<&String>) = match self {
291            Remote::Socket(p) => {
292                let s = UnixStream::connect(p).map_err(|e| {
293                    Error::invalid(format!(
294                        "cannot connect to isb serve at {}: {e}",
295                        p.display()
296                    ))
297                })?;
298                (Box::new(s), format!("ws://localhost{path}"), None)
299            }
300            Remote::Url { base, token } => {
301                let (tls, host, port, prefix) = split_base(base)?;
302                let addr = (host.trim_start_matches('[').trim_end_matches(']'), port)
303                    .to_socket_addrs()
304                    .map_err(|e| Error::invalid(format!("{host}: {e}")))?
305                    .next()
306                    .ok_or_else(|| Error::invalid(format!("{host} does not resolve")))?;
307                let sock = TcpStream::connect_timeout(&addr, timeout)
308                    .map_err(|e| Error::invalid(format!("cannot connect to {base}: {e}")))?;
309                let _ = sock.set_nodelay(true);
310                sock.set_read_timeout(Some(timeout))?;
311                let authority = if (tls && port == 443) || (!tls && port == 80) {
312                    host.clone()
313                } else {
314                    format!("{host}:{port}")
315                };
316                let scheme = if tls { "wss" } else { "ws" };
317                let url = format!("{scheme}://{authority}{prefix}{path}");
318                let conn: Box<dyn Conn> = if tls {
319                    let name = rustls::pki_types::ServerName::try_from(
320                        host.trim_start_matches('[')
321                            .trim_end_matches(']')
322                            .to_string(),
323                    )
324                    .map_err(|e| Error::invalid(format!("{host}: {e}")))?;
325                    let c = rustls::ClientConnection::new(crate::net::default_tls(), name)
326                        .map_err(|e| Error::invalid(format!("TLS: {e}")))?;
327                    Box::new(rustls::StreamOwned::new(c, sock))
328                } else {
329                    Box::new(sock)
330                };
331                (conn, url, token.as_ref())
332            }
333        };
334        conn.set_timeout(Some(timeout))?;
335        let mut req = url
336            .into_client_request()
337            .map_err(|e| Error::WebSocket(e.to_string()))?;
338        if let Some(t) = token {
339            req.headers_mut().insert(
340                "authorization",
341                format!("Bearer {}", t.trim())
342                    .parse()
343                    .map_err(|_| Error::invalid("the token is not a valid header value"))?,
344            );
345        }
346        let (ws, _) = tungstenite::client(req, conn).map_err(|e| match e {
347            tungstenite::HandshakeError::Failure(tungstenite::Error::Http(r)) => {
348                let status = r.status().as_u16();
349                let body = r
350                    .body()
351                    .as_ref()
352                    .map(|b| String::from_utf8_lossy(b).into_owned())
353                    .unwrap_or_default();
354                let v: Value = serde_json::from_str(&body).unwrap_or_else(|_| match body.trim() {
355                    // The handshake often ends before the body is read:
356                    // say what the status means here.
357                    "" => serde_json::json!({ "message": match status {
358                        401 => "the API token was refused (unknown, expired or revoked)",
359                        403 => "refused: SSH needs exec in the org (members and up; not viewers, and not read- or deploy-scoped tokens), from this site",
360                        404 => "no SSH here: no such org, or SSH is off on this server (--deny-tools sandbox_exec)",
361                        _ => "the server refused the connection",
362                    }}),
363                    b => serde_json::json!({ "message": b }),
364                });
365                answer_error(status, &v)
366            }
367            e => Error::WebSocket(e.to_string()),
368        })?;
369        Ok(ws)
370    }
371}
372
373/// A REST error answer (`{"error", "message"}`) as an isb error.
374pub fn answer_error(status: u16, v: &Value) -> Error {
375    let message = v["message"]
376        .as_str()
377        .map(String::from)
378        .unwrap_or_else(|| format!("HTTP {status}"));
379    match status {
380        401 | 403 => Error::Forbidden(message),
381        404 => Error::NotFound(message),
382        _ => Error::Remote {
383            code: v["error"].as_str().unwrap_or("server_error").into(),
384            message,
385            data: v.get("data").cloned().unwrap_or(Value::Null),
386        },
387    }
388}
389
390fn would_block(e: &tungstenite::Error) -> bool {
391    matches!(e, tungstenite::Error::Io(i) if matches!(i.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut))
392}
393
394/// Shuttle bytes between `input`/`output` (ssh's end of a `ProxyCommand`)
395/// and the websocket until either side ends. `Ok` when the session ended
396/// normally; the server's reason otherwise.
397pub fn pump<S: Conn + ?Sized>(
398    ws: &mut WebSocket<Box<S>>,
399    input: impl Read + Send + 'static,
400    mut output: impl Write,
401) -> Result<()> {
402    const CHUNK: usize = 32 * 1024;
403    let (tx, rx) = sync_channel::<Option<Vec<u8>>>(64);
404    std::thread::spawn(move || {
405        let mut input = input;
406        let mut buf = vec![0u8; CHUNK];
407        loop {
408            match input.read(&mut buf) {
409                Ok(0) | Err(_) => {
410                    let _ = tx.send(None);
411                    return;
412                }
413                Ok(n) => {
414                    if tx.send(Some(buf[..n].to_vec())).is_err() {
415                        return;
416                    }
417                }
418            }
419        }
420    });
421    ws.get_ref().set_timeout(Some(Duration::from_millis(10)))?;
422    loop {
423        match ws.read() {
424            Ok(Message::Binary(b)) => {
425                output.write_all(&b)?;
426                output.flush()?;
427            }
428            Ok(Message::Text(t)) => {
429                let v: Value = serde_json::from_str(t.as_str()).unwrap_or(Value::Null);
430                match v["type"].as_str() {
431                    Some("error") => {
432                        return Err(Error::Remote {
433                            code: "ssh".into(),
434                            message: v["message"].as_str().unwrap_or("refused").to_string(),
435                            data: Value::Null,
436                        });
437                    }
438                    Some("exit") => return Ok(()),
439                    _ => {}
440                }
441            }
442            Ok(Message::Close(_)) => return Ok(()),
443            Ok(_) => {}
444            Err(e) if would_block(&e) => {}
445            Err(tungstenite::Error::ConnectionClosed | tungstenite::Error::AlreadyClosed) => {
446                return Ok(());
447            }
448            Err(e) => return Err(Error::WebSocket(e.to_string())),
449        }
450        // What ssh wrote meanwhile, all of it.
451        loop {
452            match rx.try_recv() {
453                Ok(Some(d)) => match ws.send(Message::binary(d)) {
454                    Ok(()) => {}
455                    Err(e) if would_block(&e) => {}
456                    Err(e) => return Err(Error::WebSocket(e.to_string())),
457                },
458                Ok(None) | Err(TryRecvError::Disconnected) => {
459                    // ssh is done with us.
460                    let _ = ws.close(None);
461                    let _ = ws.flush();
462                    return Ok(());
463                }
464                Err(TryRecvError::Empty) => break,
465            }
466        }
467    }
468}
469
470#[cfg(test)]
471mod tests {
472    use super::*;
473
474    fn req(query: &str, peer: Peer, headers: &[(&str, &str)]) -> Request {
475        Request {
476            method: "GET".into(),
477            path: "/orgs/acme/api/v1/ssh".into(),
478            query: Some(query.into()),
479            headers: headers
480                .iter()
481                .map(|(k, v)| (k.to_string(), v.to_string()))
482                .collect(),
483            body: vec![],
484            peer,
485        }
486    }
487
488    fn tcp() -> Peer {
489        Peer::Tcp("127.0.0.1:5000".parse().unwrap())
490    }
491
492    #[test]
493    fn parses_requests() {
494        let r = ssh_request(&req("instance=box", tcp(), &[])).unwrap();
495        assert_eq!(
496            r,
497            SshRequest {
498                instance: "box".into(),
499                keys_of: None,
500            }
501        );
502        let r = ssh_request(&req("instance=box&as=a%2Bb%40example.com", tcp(), &[])).unwrap();
503        assert_eq!(r.keys_of.as_deref(), Some("a+b@example.com"));
504        assert_eq!(r.query(), "instance=box&as=a%2Bb@example.com");
505        assert_eq!(ssh_request(&req(&r.query(), tcp(), &[])).unwrap(), r);
506        for bad in [
507            "",
508            "instance=",
509            "instance=Box",
510            "instance=../x",
511            "instance=a&as=nobody",
512            "instance=a&as=%zz",
513        ] {
514            assert!(ssh_request(&req(bad, tcp(), &[])).is_err(), "{bad}");
515        }
516    }
517
518    #[test]
519    fn origin_rules() {
520        let host = ("Host", "isb.example.com");
521        // The unix socket: no Origin needed, but a foreign one is refused.
522        assert!(origin_allowed(&req("", Peer::Unix { uid: None }, &[])));
523        assert!(!origin_allowed(&req(
524            "",
525            Peer::Unix { uid: None },
526            &[host, ("Origin", "https://evil.example")]
527        )));
528        // A bearer token over TCP: none needed either.
529        assert!(origin_allowed(&req(
530            "",
531            tcp(),
532            &[host, ("Authorization", "Bearer x")]
533        )));
534        // A cookie: only from this site.
535        assert!(!origin_allowed(&req(
536            "",
537            tcp(),
538            &[host, ("Cookie", "isb_session=x")]
539        )));
540        assert!(!origin_allowed(&req(
541            "",
542            tcp(),
543            &[
544                host,
545                ("Cookie", "isb_session=x"),
546                ("Origin", "https://evil.example")
547            ]
548        )));
549        assert!(origin_allowed(&req(
550            "",
551            tcp(),
552            &[
553                host,
554                ("Cookie", "isb_session=x"),
555                ("Origin", "https://isb.example.com")
556            ]
557        )));
558    }
559
560    #[test]
561    fn splits_bases() {
562        assert_eq!(
563            split_base("https://isb.example.com").unwrap(),
564            (true, "isb.example.com".into(), 443, "".into())
565        );
566        assert_eq!(
567            split_base("http://127.0.0.1:8092/").unwrap(),
568            (false, "127.0.0.1".into(), 8092, "".into())
569        );
570        assert_eq!(
571            split_base("https://h.example:8443/isb/").unwrap(),
572            (true, "h.example".into(), 8443, "/isb".into())
573        );
574        assert!(split_base("isb.example.com").is_err());
575        assert!(split_base("https://").is_err());
576        let r = Remote::Url {
577            base: "https://x".into(),
578            token: Some("isb_tok_secret".into()),
579        };
580        assert!(!format!("{r:?}").contains("secret"));
581    }
582
583    /// The client's pump against a server bridge over a socket pair: bytes
584    /// both ways, and the server's error frame becomes the error.
585    #[test]
586    fn pumps_bytes_both_ways() {
587        use super::super::http::Duplex;
588        use super::super::terminal::{PtyOutput, bridge};
589        use std::sync::mpsc::{Receiver, Sender, channel};
590
591        struct Upper(Sender<PtyOutput>, Receiver<PtyOutput>);
592        impl Pty for Upper {
593            fn input(&mut self, d: &[u8]) -> crate::Result<()> {
594                if d == b"bye" {
595                    self.0
596                        .send(PtyOutput::Failed("the key was removed".into()))
597                        .unwrap();
598                } else {
599                    self.0
600                        .send(PtyOutput::Data(d.to_ascii_uppercase()))
601                        .unwrap();
602                }
603                Ok(())
604            }
605            fn resize(&mut self, _: u16, _: u16) {}
606            fn output(&mut self, w: Duration) -> PtyOutput {
607                self.1.recv_timeout(w).unwrap_or(PtyOutput::Idle)
608            }
609            fn close(&mut self) {}
610        }
611        let (a, b) = UnixStream::pair().unwrap();
612        let server = std::thread::spawn(move || {
613            let mut s = a;
614            let d: &mut dyn Duplex = &mut s;
615            let mut ws = WebSocket::from_raw_socket(d, tungstenite::protocol::Role::Server, None);
616            let (tx, rx) = channel();
617            bridge(
618                &mut ws,
619                Box::new(Upper(tx, rx)),
620                Duration::from_secs(5),
621                Duration::from_secs(5),
622            );
623        });
624        let conn: Box<UnixStream> = Box::new(b);
625        let mut ws = WebSocket::from_raw_socket(conn, tungstenite::protocol::Role::Client, None);
626        // Input: "hello", then "bye" a little later (separate frames).
627        let (mut w, r) = UnixStream::pair().unwrap();
628        let feeder = std::thread::spawn(move || {
629            w.write_all(b"hello").unwrap();
630            std::thread::sleep(Duration::from_millis(200));
631            w.write_all(b"bye").unwrap();
632            std::thread::sleep(Duration::from_millis(2000));
633        });
634        let mut out = Vec::new();
635        let e = pump(&mut ws, r, &mut out).unwrap_err();
636        assert_eq!(out, b"HELLO");
637        assert!(e.to_string().contains("the key was removed"), "{e}");
638        server.join().unwrap();
639        feeder.join().unwrap();
640    }
641}
isb_server/server/ssh.rs

isb_server/server/
ssh.rs
