rs-rich-web 0.0.7

//! The server: [`serve`], [`serve_command`] and [`Server`], behind the
//! crate's `server` feature (on by default). Re-exported at the crate root.

/// What a program runs on: the PTY host trait and the local implementation,
/// from rs-rich-embed.
pub use rich_embed::{Command, ExitStatus, LocalPty, PtyHost};

use std::io::{self, Write};
use std::net::{
    IpAddr, Ipv4Addr, Ipv6Addr, Shutdown, SocketAddr, TcpListener, TcpStream, ToSocketAddrs,
};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};

use intuituive::App;
use tungstenite::WebSocket;
use tungstenite::protocol::{Role, WebSocketConfig};

use crate::http::{self, Head};
use crate::{isolate, pty, session};

pub use crate::isolate::{Isolation, SESSION_ENV};

/// The version of xterm.js the page uses, vendored in this crate.
pub const XTERM_VERSION: &str = "6.0.0";

/// The sessions a server runs at once unless told otherwise.
pub const DEFAULT_MAX_SESSIONS: usize = 8;

/// The output a program's session holds for its page unless told
/// otherwise ([`Server::max_buffered`]): 1 MiB.
pub const DEFAULT_MAX_BUFFERED: usize = 1 << 20;

/// Connections still sending their request at once; more are answered
/// `503`.
const MAX_PENDING: usize = 64;
/// How long a connection may take to send its request, all of it.
const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);
/// How long a write to a browser may take before the session ends.
const WRITE_TIMEOUT: Duration = Duration::from_secs(30);
/// The largest message a page may send (a big paste).
pub(crate) const MAX_MESSAGE: usize = 1 << 20;

const PAGE: &str = include_str!("../assets/index.html");
const PAGE_JS: &str = include_str!("../assets/app.js");
const DOM_PAGE: &str = include_str!("../assets/dom.html");
const DOM_JS: &str = include_str!("../assets/dom.js");
const XTERM_JS: &str = include_str!("../assets/xterm/xterm.js");
const XTERM_CSS: &str = include_str!("../assets/xterm/xterm.css");
const FIT_JS: &str = include_str!("../assets/xterm/addon-fit.js");

/// Serve `app` at `addr`: print the URL to open (with its token), then run
/// one app, made by `app`, per browser tab that opens it, until the process
/// ends. [`Server::bind`] and [`Server::run`] with the defaults.
///
/// `app` is called in each session's own process (or, with
/// [`Isolation::Threads`], on its own thread), so the [`App`] itself never
/// crosses threads.
pub fn serve<A, F>(addr: A, app: F) -> io::Result<()>
where
    A: ToSocketAddrs,
    F: Fn() -> App + Send + Sync + 'static,
{
    Server::bind(addr, app)?.run()
}

/// Serve `command` at `addr`: print the URL to open (with its token), then
/// run the program on a pseudo-terminal of its own for each browser tab
/// that opens it, until the process ends. [`Server::bind_command`] and
/// [`Server::run`] with the defaults.
///
/// A string is a program name; an array or a vector is the program and its
/// arguments; a [`Command`] also sets its environment and directory.
pub fn serve_command<A>(addr: A, command: impl Into<Command>) -> io::Result<()>
where
    A: ToSocketAddrs,
{
    Server::bind_command(addr, command)?.run()
}

/// How an app's page draws it.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum Renderer {
    /// A terminal: xterm.js, fed the app's frames as terminal output.
    #[default]
    Xterm,
    /// The page's own DOM: a grid of styled spans, with the app's
    /// accessibility tree as ARIA for a browser's screen reader. Apps only:
    /// a program on a PTY has no tree, and is always drawn by xterm.js.
    Dom,
}

impl Renderer {
    /// `"xterm"` or `"dom"`, as the page's address names it.
    pub fn name(self) -> &'static str {
        match self {
            Renderer::Xterm => "xterm",
            Renderer::Dom => "dom",
        }
    }

    /// The renderer called `name` (`"xterm"`, `"dom"`).
    pub fn parse(name: &str) -> Option<Renderer> {
        match name {
            "xterm" => Some(Renderer::Xterm),
            "dom" => Some(Renderer::Dom),
            _ => None,
        }
    }
}

/// What each session runs.
enum Mode {
    App(Box<dyn Fn() -> App + Send + Sync>),
    /// A program on a [`LocalPty`], held back by the page's pace.
    Command(Command),
    /// A program on another host.
    Host(Box<dyn Fn() -> Box<dyn PtyHost> + Send + Sync>),
}

/// A server that is bound and ready to [`run`](Self::run) or
/// [`spawn`](Self::spawn), with its options.
pub struct Server {
    listener: TcpListener,
    shared: Shared,
}

/// What every connection's thread reads.
struct Shared {
    mode: Mode,
    token: String,
    max_sessions: usize,
    max_buffered: usize,
    renderer: Renderer,
    /// Compress frames from this size, for pages that ask.
    compress: Option<usize>,
    origins: Vec<String>,
    title: String,
    /// The `Host` values the server answers to, besides each connection's
    /// own address (`http::allowed_hosts`, then `allow_host`'s).
    hosts: Vec<String>,
    /// Sessions running.
    sessions: AtomicUsize,
    /// Connections still sending their request.
    pending: AtomicUsize,
    /// Programs (and sessions' processes) started and not yet ended.
    programs: AtomicUsize,
    stopping: AtomicBool,
    /// Where an app's sessions run, and how their processes start.
    isolation: Isolation,
    spawn: isolate::Spawn,
    /// This server's number among the app servers this program binds, which
    /// a session's process uses to find the same one.
    bind: usize,
}

impl Server {
    /// Listen on `addr` (`"127.0.0.1:0"` picks a free port) with a fresh
    /// random token, to serve the apps `app` makes.
    ///
    /// In a session's own process ([`Isolation::Processes`], the default),
    /// this is where the session runs: the process this program started
    /// for a browser tab reaches this same call, runs the app for that tab,
    /// and exits here, without listening.
    pub fn bind<A, F>(addr: A, app: F) -> io::Result<Server>
    where
        A: ToSocketAddrs,
        F: Fn() -> App + Send + Sync + 'static,
    {
        let bind = isolate::next_bind();
        if let Some(spec) = isolate::session_for(bind) {
            std::process::exit(isolate::run_child(spec, app));
        }
        let mode = Mode::App(Box::new(app));
        let mut server = if isolate::is_session_process() {
            // Another server of this program: a session's process does not
            // take its port.
            Server::with_mode(("127.0.0.1", 0), mode, "intuiTUIve".into())?
        } else {
            Server::with_mode(addr, mode, "intuiTUIve".into())?
        };
        server.shared.bind = bind;
        Ok(server)
    }

    /// Listen on `addr` with a fresh random token, to run `command` on a
    /// pseudo-terminal of its own for each tab (through [`LocalPty`]). The
    /// page's title is the program's name unless set.
    pub fn bind_command<A>(addr: A, command: impl Into<Command>) -> io::Result<Server>
    where
        A: ToSocketAddrs,
    {
        let command = command.into();
        let title = command.program().to_string_lossy().into_owned();
        Server::with_mode(addr, Mode::Command(command), title)
    }

    /// Listen on `addr` with a fresh random token, to run a program on the
    /// [`PtyHost`] `host` makes for each tab: an SSH session, a container's
    /// exec stream, or a test host. It is made, started and dropped on the
    /// session's own thread. [`max_buffered`](Self::max_buffered) bounds
    /// only what the session holds: a host that reads ahead buffers on its
    /// own terms.
    pub fn bind_host<A, F>(addr: A, host: F) -> io::Result<Server>
    where
        A: ToSocketAddrs,
        F: Fn() -> Box<dyn PtyHost> + Send + Sync + 'static,
    {
        Server::with_mode(addr, Mode::Host(Box::new(host)), "terminal".into())
    }

    fn with_mode<A: ToSocketAddrs>(addr: A, mode: Mode, title: String) -> io::Result<Server> {
        let listener = TcpListener::bind(addr)?;
        let hosts = http::allowed_hosts(listener.local_addr()?);
        Ok(Server {
            listener,
            shared: Shared {
                mode,
                token: random_token()?,
                max_sessions: DEFAULT_MAX_SESSIONS,
                max_buffered: DEFAULT_MAX_BUFFERED,
                renderer: Renderer::Xterm,
                compress: None,
                origins: Vec::new(),
                title,
                hosts,
                sessions: AtomicUsize::new(0),
                pending: AtomicUsize::new(0),
                programs: AtomicUsize::new(0),
                stopping: AtomicBool::new(false),
                isolation: Isolation::default(),
                spawn: isolate::Spawn::default(),
                bind: 0,
            },
        })
    }

    /// Where an app's sessions run: each in a process of its own
    /// ([`Isolation::Processes`], the default), or each on a thread of the
    /// server ([`Isolation::Threads`]). A program's sessions (a command, a
    /// host) are processes already, and this does not change them.
    #[must_use]
    pub fn isolation(mut self, isolation: Isolation) -> Server {
        self.shared.isolation = isolation;
        self
    }

    /// Start each session's process as `program` with `args`, rather than
    /// as this program with the arguments it was given. It must reach the
    /// same `serve` or `Server::bind` call: an interpreter running the same
    /// script, say (Python's `rs_rich.tui.serve` passes its own).
    #[must_use]
    pub fn session_command<P, I, S>(mut self, program: P, args: I) -> Server
    where
        P: Into<std::path::PathBuf>,
        I: IntoIterator<Item = S>,
        S: Into<std::ffi::OsString>,
    {
        let args = args.into_iter().map(Into::into).collect();
        self.shared.spawn.command = Some((program.into(), args));
        self
    }

    /// Cap each session's process at `bytes` of memory (its address
    /// space): an allocation past it fails, and the process ends, alone.
    /// On Linux it holds from the moment the process starts; on other Unix
    /// systems from when it reaches `serve`; macOS has no such limit, and
    /// elsewhere (Windows) it is not applied.
    #[must_use]
    pub fn session_memory(mut self, bytes: u64) -> Server {
        self.shared.spawn.memory = Some(bytes);
        self
    }

    /// Cap each session's process at `time` of CPU (whole seconds, at least
    /// one): past it the process is killed, alone. A session that waits for
    /// input uses none. On Linux it holds from the moment the process
    /// starts; on other Unix systems from when it reaches `serve`;
    /// elsewhere (Windows) it is not applied.
    #[must_use]
    pub fn session_cpu(mut self, time: Duration) -> Server {
        self.shared.spawn.cpu = Some(time);
        self
    }

    /// Run at most `n` sessions at once (8 by default); a browser that
    /// opens one more is refused until one ends.
    #[must_use]
    pub fn max_sessions(mut self, n: usize) -> Server {
        self.shared.max_sessions = n;
        self
    }

    /// For a program: hold at most about `bytes` of its output for a page
    /// that has not shown it yet (1 MiB by default). Half is output sent
    /// and not yet acknowledged by the page, half is output read from the
    /// program and not yet sent; past that, the program's writes wait until
    /// the page catches up, as on a slow terminal, so nothing is dropped
    /// and memory stays bounded however fast it writes, at any size (a
    /// [`LocalPty`] with backpressure never drops output, even past its
    /// own 16 MiB limit without it). Apps send whole frames and are not
    /// affected.
    #[must_use]
    pub fn max_buffered(mut self, bytes: usize) -> Server {
        self.shared.max_buffered = bytes.max(2);
        self
    }

    /// Compress frames of `from` bytes or more with deflate, for pages
    /// that can undo it (they have `DecompressionStream`, and ask with
    /// `z=1`): an app's frames, the DOM renderer's messages and a
    /// program's output. Off (`None`) by default. Worth it over a slow
    /// link: measured on a few scenes, frames shrink to 5 to 30% of their
    /// size for 15 to 60 µs each (level 1). On a local machine it only
    /// adds work. 1024 is a fair threshold.
    #[must_use]
    pub fn compress(mut self, from: Option<usize>) -> Server {
        self.shared.compress = from;
        self
    }

    /// How an app's page draws it: [`Renderer::Xterm`] (the default) or
    /// [`Renderer::Dom`]. A page's address can ask for either with
    /// `&renderer=xterm` or `&renderer=dom`. A program is always drawn by
    /// xterm.js.
    #[must_use]
    pub fn renderer(mut self, renderer: Renderer) -> Server {
        self.shared.renderer = renderer;
        self
    }

    /// Use `token` instead of a random one, for an address that stays the
    /// same between runs. Keep it as secret, and as long: anyone with it
    /// can use the app.
    ///
    /// # Panics
    /// When `token` is empty or has characters other than ASCII letters,
    /// digits, `-`, `_`, `.` and `~` (it goes in a URL as it is).
    #[must_use]
    pub fn token(mut self, token: impl Into<String>) -> Server {
        let token = token.into();
        assert!(
            !token.is_empty()
                && token
                    .bytes()
                    .all(|b| b.is_ascii_alphanumeric() || b"-_.~".contains(&b)),
            "a token is ASCII letters, digits, '-', '_', '.' and '~'"
        );
        self.shared.token = token;
        self
    }

    /// Also answer to `host` (say `"box.lan"`, or `"box.lan:9000"`): a name
    /// this machine is reached by that the server does not know itself. The
    /// server's port is added unless `host` names one. The `Host` of a
    /// WebSocket upgrade must be one the server answers to: its bound
    /// address (with `localhost` for a loopback one), or, bound to all
    /// interfaces, the loopback addresses, `localhost`, this machine's own
    /// name and the address the connection arrived on; plus these. That is
    /// what stops a page whose name was pointed here (DNS rebinding). The
    /// token is still required.
    #[must_use]
    pub fn allow_host(mut self, host: impl AsRef<str>) -> Server {
        let port = self.local_addr().port();
        self.shared
            .hosts
            .extend(http::host_entries(host.as_ref(), port));
        self
    }

    /// Also accept WebSocket upgrades from pages served at `origin` (say
    /// `"https://term.example.com"`): the address a reverse proxy serves
    /// this server's page from. The token is still required.
    #[must_use]
    pub fn allow_origin(mut self, origin: impl Into<String>) -> Server {
        self.shared.origins.push(origin.into());
        self
    }

    /// The page's title (`intuiTUIve` for an app, the program's name for a
    /// command, by default).
    #[must_use]
    pub fn title(mut self, title: impl Into<String>) -> Server {
        self.shared.title = title.into();
        self
    }

    /// The address the server listens on.
    pub fn local_addr(&self) -> SocketAddr {
        self.listener
            .local_addr()
            .expect("a bound listener has an address")
    }

    /// The token browsers must give.
    pub fn access_token(&self) -> &str {
        &self.shared.token
    }

    /// The address to open in a browser, with the token.
    pub fn url(&self) -> String {
        url_for(self.local_addr(), &self.shared.token)
    }

    /// Print the URL (and a warning when the address is not loopback),
    /// then serve until the process ends.
    pub fn run(self) -> io::Result<()> {
        self.announce()?;
        accept(self.listener, Arc::new(self.shared));
        Ok(())
    }

    /// Print the URL as [`run`](Self::run) does, then serve until `until`
    /// returns, and stop as [`Handle::stop`] does: every session ends, and
    /// this returns once their programs have. `until` runs on this thread:
    /// a wait for a signal, say, as `rich serve` waits for Ctrl+C, so that
    /// no program outlives the server.
    pub fn run_until(self, until: impl FnOnce()) -> io::Result<()> {
        self.announce()?;
        let handle = self.spawn()?;
        until();
        handle.stop();
        Ok(())
    }

    /// Print the URL, and a warning when the address is not loopback.
    fn announce(&self) -> io::Result<()> {
        let local = self.local_addr();
        let mut out = io::stdout().lock();
        writeln!(out, "Serving {} at {}", self.shared.title, self.url())?;
        out.flush()?;
        drop(out);
        if !local.ip().is_loopback() {
            let what = match self.shared.mode {
                Mode::App(_) => "the app",
                Mode::Command(_) | Mode::Host(_) => "the program",
            };
            eprintln!(
                "warning: listening on {local}, beyond this computer. Anyone who can reach it \
                 with the token can use {what}: there is no other authentication and no TLS. \
                 Put it behind a reverse proxy that has both."
            );
        }
        Ok(())
    }

    /// Serve on a thread of its own, printing nothing, until the
    /// [`Handle`] is stopped or dropped.
    pub fn spawn(self) -> io::Result<Handle> {
        let local = self.local_addr();
        let url = self.url();
        let shared = Arc::new(self.shared);
        let listener = self.listener;
        let thread = {
            let shared = shared.clone();
            thread::Builder::new()
                .name("rich-web accept".into())
                .spawn(move || accept(listener, shared))?
        };
        Ok(Handle {
            local,
            url,
            shared,
            thread: Some(thread),
        })
    }
}

/// A server running in the background, from [`Server::spawn`]. Stopping it
/// (or dropping it) stops accepting connections and ends every session.
pub struct Handle {
    local: SocketAddr,
    url: String,
    shared: Arc<Shared>,
    thread: Option<JoinHandle<()>>,
}

impl Handle {
    /// The address the server listens on.
    pub fn local_addr(&self) -> SocketAddr {
        self.local
    }

    /// The address to open in a browser, with the token.
    pub fn url(&self) -> &str {
        &self.url
    }

    /// The sessions running now.
    pub fn sessions(&self) -> usize {
        self.shared.sessions.load(Ordering::SeqCst)
    }

    /// Stop: no more connections, and every session ends (within one turn
    /// of its loop, at most 50 ms). A program's session ends its program:
    /// it is hung up on, and a [`LocalPty`]'s is killed, with its process
    /// group, if it is still running a second later. This returns once
    /// every program has ended; a connection still sending its request is
    /// not served.
    pub fn stop(mut self) {
        self.shutdown();
    }

    fn shutdown(&mut self) {
        self.shared.stopping.store(true, Ordering::SeqCst);
        if let Some(accepting) = self.thread.take() {
            // Wake the accept loop, which then sees `stopping`.
            let _ = TcpStream::connect_timeout(&reachable(self.local), Duration::from_secs(1));
            let _ = accepting.join();
            // A session counts its program before it looks at `stopping`.
            while self.shared.programs.load(Ordering::SeqCst) > 0 {
                thread::sleep(Duration::from_millis(10));
            }
        }
    }
}

impl Drop for Handle {
    fn drop(&mut self) {
        self.shutdown();
    }
}

/// The address to connect to for `local`: loopback when it is every
/// interface.
fn reachable(local: SocketAddr) -> SocketAddr {
    let ip = match local.ip() {
        IpAddr::V4(v4) if v4.is_unspecified() => IpAddr::V4(Ipv4Addr::LOCALHOST),
        IpAddr::V6(v6) if v6.is_unspecified() => IpAddr::V6(Ipv6Addr::LOCALHOST),
        ip => ip,
    };
    SocketAddr::new(ip, local.port())
}

fn url_for(local: SocketAddr, token: &str) -> String {
    format!("http://{}/?token={token}", reachable(local))
}

/// 128 bits from the operating system's random source, as hex.
pub(crate) fn random_token() -> io::Result<String> {
    let mut bytes = [0u8; 16];
    getrandom::fill(&mut bytes).map_err(|e| io::Error::other(format!("no random token: {e}")))?;
    Ok(bytes.iter().map(|b| format!("{b:02x}")).collect())
}

fn accept(listener: TcpListener, shared: Arc<Shared>) {
    for stream in listener.incoming() {
        if shared.stopping.load(Ordering::SeqCst) {
            break;
        }
        let Ok(stream) = stream else {
            // Out of file descriptors, say: let some go before the next.
            thread::sleep(Duration::from_millis(10));
            continue;
        };
        if shared.pending.fetch_add(1, Ordering::SeqCst) >= MAX_PENDING {
            shared.pending.fetch_sub(1, Ordering::SeqCst);
            busy(stream);
            continue;
        }
        let pending = Pending(shared.clone());
        let spawned = thread::Builder::new()
            .name("rich-web connection".into())
            .spawn(move || connection(stream, pending));
        if spawned.is_err() {
            // The closure, and the `Pending` in it, were dropped.
            continue;
        }
    }
}

/// Answer a connection past [`MAX_PENDING`] with `503`, without waiting
/// for it: the answer is small enough for the socket's buffer.
fn busy(mut stream: TcpStream) {
    if stream.set_nonblocking(true).is_err() {
        return;
    }
    // What the request has sent so far, so closing does not reset the
    // connection before the answer is read.
    let _ = io::Read::read(&mut stream, &mut [0u8; 4096]);
    let _ = http::refuse(
        &mut stream,
        "503 Service Unavailable",
        "Too many connections at once: try again in a moment.",
    );
    let _ = stream.shutdown(Shutdown::Write);
}

/// A connection still sending its request: counted until dropped.
struct Pending(Arc<Shared>);

impl Drop for Pending {
    fn drop(&mut self) {
        self.0.pending.fetch_sub(1, Ordering::SeqCst);
    }
}

/// A running session: counted until dropped.
struct Slot(Arc<Shared>);

impl Slot {
    fn acquire(shared: &Arc<Shared>) -> Option<Slot> {
        let mut n = shared.sessions.load(Ordering::SeqCst);
        loop {
            if n >= shared.max_sessions {
                return None;
            }
            match shared
                .sessions
                .compare_exchange(n, n + 1, Ordering::SeqCst, Ordering::SeqCst)
            {
                Ok(_) => return Some(Slot(shared.clone())),
                Err(now) => n = now,
            }
        }
    }
}

impl Drop for Slot {
    fn drop(&mut self) {
        self.0.sessions.fetch_sub(1, Ordering::SeqCst);
    }
}

/// Serve one connection: the page, an asset, or a session.
fn connection(mut stream: TcpStream, pending: Pending) {
    let _ = stream.set_nodelay(true);
    if stream.set_write_timeout(Some(WRITE_TIMEOUT)).is_err() {
        return;
    }
    // The whole head within the timeout, however slowly it comes.
    let until = Instant::now() + HANDSHAKE_TIMEOUT;
    let Ok(bytes) = http::read_head(&mut http::Deadline::new(&stream, until)) else {
        return;
    };
    // The server stopped while the request came in.
    if pending.0.stopping.load(Ordering::SeqCst) {
        return;
    }
    let Some(head) = Head::parse(&bytes) else {
        let _ = http::refuse(&mut stream, "400 Bad Request", "Bad request");
        return;
    };
    let shared = pending.0.clone();
    if head.method != "GET" {
        let _ = http::refuse(&mut stream, "405 Method Not Allowed", "Only GET");
        return;
    }
    let js = "text/javascript; charset=utf-8";
    let asset = match head.path.as_str() {
        "/ws" => {
            websocket(stream, &head, &shared, pending);
            return;
        }
        "/" | "/index.html" => {
            page(&mut stream, &head, &shared);
            return;
        }
        "/app.js" => (js, PAGE_JS),
        "/dom.js" => (js, DOM_JS),
        "/xterm.js" => (js, XTERM_JS),
        "/addon-fit.js" => (js, FIT_JS),
        "/xterm.css" => ("text/css; charset=utf-8", XTERM_CSS),
        _ => {
            let _ = http::refuse(&mut stream, "404 Not Found", "Not found");
            return;
        }
    };
    let _ = http::respond(&mut stream, "200 OK", asset.0, &[], asset.1.as_bytes());
}

/// The renderer a request asks for: its `renderer` parameter, else the
/// server's. A program is always drawn by xterm.js.
fn renderer_for(head: &Head, shared: &Shared) -> Renderer {
    match shared.mode {
        Mode::App(_) => head
            .param("renderer")
            .and_then(Renderer::parse)
            .unwrap_or(shared.renderer),
        Mode::Command(_) | Mode::Host(_) => Renderer::Xterm,
    }
}

/// The page, for a request with the token.
fn page(stream: &mut TcpStream, head: &Head, shared: &Shared) {
    if !head
        .param("token")
        .is_some_and(|t| http::same_token(t, &shared.token))
    {
        let _ = http::refuse(
            stream,
            "403 Forbidden",
            "Open the address the server printed when it started, with its token.",
        );
        return;
    }
    // The page loads only its own scripts and styles, opens a WebSocket
    // only to this server, and is never framed by another site.
    let mut connect = "'self'".to_string();
    if let Some(host) = head.header("host").filter(|h| {
        h.bytes()
            .all(|b| b.is_ascii_alphanumeric() || b".-:[]".contains(&b))
    }) {
        connect.push_str(&format!(" ws://{host} wss://{host}"));
    }
    let policy = format!(
        "default-src 'none'; script-src 'self'; style-src 'self' 'unsafe-inline'; \
         img-src 'self' data:; font-src 'self' data:; connect-src {connect}; \
         frame-ancestors 'none'; base-uri 'none'; form-action 'none'"
    );
    let template = match renderer_for(head, shared) {
        Renderer::Xterm => PAGE,
        Renderer::Dom => DOM_PAGE,
    };
    let mode = match shared.mode {
        Mode::App(_) => "app",
        Mode::Command(_) | Mode::Host(_) => "program",
    };
    let body = template
        .replace("{{title}}", &http::escape_html(&shared.title))
        .replace("{{mode}}", mode);
    let _ = http::respond(
        stream,
        "200 OK",
        "text/html; charset=utf-8",
        &[
            ("Content-Security-Policy", policy),
            ("X-Frame-Options", "DENY".to_string()),
        ],
        body.as_bytes(),
    );
}

/// Why an upgrade is refused, as a status and a message; `None` when it
/// may go ahead.
fn check_upgrade(
    head: &Head,
    shared: &Shared,
    local: Option<SocketAddr>,
) -> Option<(&'static str, &'static str)> {
    if http::websocket_key(head).is_none() {
        return Some(("400 Bad Request", "Expected a WebSocket upgrade."));
    }
    if !head
        .param("token")
        .is_some_and(|t| http::same_token(t, &shared.token))
    {
        return Some(("403 Forbidden", "Wrong or missing token."));
    }
    let mut hosts = shared.hosts.clone();
    hosts.extend(local.map(http::connection_hosts).unwrap_or_default());
    if !http::origin_allowed(
        head.header("origin"),
        head.header("host"),
        &hosts,
        &shared.origins,
    ) {
        return Some(("403 Forbidden", "This origin may not open a session."));
    }
    None
}

/// Upgrade to a WebSocket and run a session on it, or refuse.
fn websocket(mut stream: TcpStream, head: &Head, shared: &Arc<Shared>, pending: Pending) {
    if let Some((status, why)) = check_upgrade(head, shared, stream.local_addr().ok()) {
        let _ = http::refuse(&mut stream, status, why);
        return;
    }
    let Some(_slot) = Slot::acquire(shared) else {
        let _ = http::refuse(
            &mut stream,
            "503 Service Unavailable",
            "Too many sessions: try again when one has ended.",
        );
        return;
    };
    let key = http::websocket_key(head).expect("checked");
    let accept = tungstenite::handshake::derive_accept_key(key.as_bytes());
    let answer = format!(
        "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\
         Sec-WebSocket-Accept: {accept}\r\n\r\n"
    );
    if stream.write_all(answer.as_bytes()).is_err() || stream.flush().is_err() {
        return;
    }
    drop(pending);
    let size = |name, default| {
        head.param(name)
            .and_then(|n| n.parse::<u16>().ok())
            .filter(|n| (1..=session::MAX_CELLS).contains(n))
            .unwrap_or(default)
    };
    let (columns, rows) = (size("cols", 80), size("rows", 24));
    // Framed when the page can undo compression, compressed only when the
    // server allows it too.
    let z = session::Compress::negotiate(shared.compress, head.param("z") == Some("1"));
    if let (Mode::App(_), Isolation::Processes) = (&shared.mode, shared.isolation) {
        let session = isolate::Session {
            spawn: &shared.spawn,
            bind: shared.bind,
            columns,
            rows,
            renderer: renderer_for(head, shared),
            compress: z.0,
            stopping: &shared.stopping,
            running: &shared.programs,
        };
        isolate::run_parent(stream, session);
        return;
    }
    let config = WebSocketConfig::default()
        .max_message_size(Some(MAX_MESSAGE))
        .max_frame_size(Some(MAX_MESSAGE));
    let mut socket = WebSocket::from_raw_socket(stream, Role::Server, Some(config));
    let (stopping, programs) = (&shared.stopping, &shared.programs);
    let _ = match &shared.mode {
        Mode::App(app) => {
            let renderer = renderer_for(head, shared);
            session::run(&mut socket, app(), columns, rows, stopping, renderer, z)
        }
        Mode::Command(command) => {
            let (window, held) = pty::split(shared.max_buffered);
            let host = LocalPty::new(command.clone()).backpressure(held);
            let host = Box::new(host);
            pty::run(
                &mut socket,
                host,
                (columns, rows),
                stopping,
                programs,
                window,
                z,
            )
        }
        Mode::Host(host) => {
            let (window, _) = pty::split(shared.max_buffered);
            let host = host();
            pty::run(
                &mut socket,
                host,
                (columns, rows),
                stopping,
                programs,
                window,
                z,
            )
        }
    };
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn tokens_are_random_hex() {
        let a = random_token().unwrap();
        let b = random_token().unwrap();
        assert_eq!(a.len(), 32);
        assert!(a.bytes().all(|c| c.is_ascii_hexdigit()));
        assert_ne!(a, b);
    }

    #[test]
    fn urls_point_somewhere_reachable() {
        let token = "t0k";
        assert_eq!(
            url_for("127.0.0.1:8080".parse().unwrap(), token),
            "http://127.0.0.1:8080/?token=t0k"
        );
        assert_eq!(
            url_for("0.0.0.0:80".parse().unwrap(), token),
            "http://127.0.0.1:80/?token=t0k"
        );
        assert_eq!(
            url_for("[::]:80".parse().unwrap(), token),
            "http://[::1]:80/?token=t0k"
        );
    }

    #[test]
    #[should_panic(expected = "a token is")]
    fn tokens_must_be_url_safe() {
        let _ = Server::bind("127.0.0.1:0", || App::new(|| intuituive::label("x")))
            .unwrap()
            .token("has space");
    }

    #[test]
    fn renderers_have_names() {
        for renderer in [Renderer::Xterm, Renderer::Dom] {
            assert_eq!(Renderer::parse(renderer.name()), Some(renderer));
        }
        assert_eq!(Renderer::parse("canvas"), None);
        assert_eq!(Renderer::default(), Renderer::Xterm);
    }

    #[test]
    fn connections_past_the_pending_cap_are_answered() {
        use std::io::Read;

        let handle = Server::bind("127.0.0.1:0", || App::new(|| intuituive::label("x")))
            .unwrap()
            .spawn()
            .unwrap();
        let addr = handle.local_addr();
        // Connections that send nothing hold every place for a request.
        let idle: Vec<TcpStream> = (0..MAX_PENDING)
            .map(|_| TcpStream::connect(addr).unwrap())
            .collect();
        let start = Instant::now();
        while handle.shared.pending.load(Ordering::SeqCst) < MAX_PENDING {
            assert!(
                start.elapsed() < Duration::from_secs(5),
                "the idle connections"
            );
            thread::sleep(Duration::from_millis(10));
        }
        // One more is told so, not closed without a word.
        let mut late = TcpStream::connect(addr).unwrap();
        late.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
        let mut answer = String::new();
        let _ = late.read_to_string(&mut answer);
        assert!(answer.starts_with("HTTP/1.1 503 "), "{answer:?}");
        drop(idle);
    }

    #[test]
    fn a_command_is_titled_by_its_program() {
        let server = Server::bind_command("127.0.0.1:0", ["htop", "-d", "10"]).unwrap();
        assert_eq!(server.shared.title, "htop");
        assert_eq!(server.shared.max_buffered, DEFAULT_MAX_BUFFERED);
    }
}