haematite-cli 0.21.12

//! HAEMNET-003 R2: the door's assistant route over the network.
//!
//! `GET /assistant` on the door's own origin, upgraded to a WebSocket
//! (RFC 6455). The door listens on loopback behind its TLS front, so the bytes
//! here are plain HTTP/1.1. Each binary message carries exactly one assistant
//! frame, bounded by `frame_bytes` as on the socket. [`Messages`] shows the
//! socket's newline-delimited frames to `assistant_wire` one message at a
//! time, and [`Lines`] sends each frame the hub writes as one message, so the
//! admission and the frame loop are the socket's
//! ([`super::assistant_socket::serve_link`]), unchanged. On the network the
//! pass check is the whole admission: no peer is trusted by its address.
//! A text message, a fragmented message, an unmasked client frame, a message
//! over the bound or one holding a newline is refused by name and the
//! connection closed.

use std::io::{self, BufReader, Read, Write};
use std::net::TcpStream;
use std::sync::{Arc, Mutex};

use base64::Engine as _;
use base64::engine::general_purpose::STANDARD;
use serde_json::json;
use sha1::{Digest as _, Sha1};

use super::assistant_socket::{Link, serve_link};
use super::assistant_state::bound;
use super::http::{Request, Response};
use super::route::Site;
use super::session::lock;
use super::signin::{Door, Issuer};
use crate::serve::storage::Failure;

/// RFC 6455 section 1.3: appended to the client's key before the SHA-1.
const ACCEPT_GUID: &str = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";

const OPCODE_CONTINUATION: u8 = 0x0;
const OPCODE_TEXT: u8 = 0x1;
/// A binary message: the one kind that carries an assistant frame.
pub const OPCODE_BINARY: u8 = 0x2;
/// The close frame, from either side.
pub const OPCODE_CLOSE: u8 = 0x8;
const OPCODE_PING: u8 = 0x9;
const OPCODE_PONG: u8 = 0xA;

/// A control frame's payload is at most 125 bytes (RFC 6455 section 5.5).
const CONTROL_BYTES: usize = 125;

/// Whether `request` is the assistant route's upgrade.
pub(super) fn is_route(request: &Request) -> bool {
    request.method == "GET" && request.path == "/assistant"
}

/// Answer the assistant route on `stream`: refuse the upgrade by name with
/// an HTTP answer, or switch protocols and serve the seat until its EOF.
/// `reader` is the reader the request was read through, which may already
/// hold the seat's first bytes.
///
/// # Errors
///
/// The stream's own error.
pub(super) fn serve<I: Issuer>(
    door: &Door<I>,
    site: &Site,
    request: &Request,
    reader: BufReader<&TcpStream>,
    stream: &TcpStream,
) -> io::Result<()> {
    let hub = match door.assistant() {
        Ok(hub) => hub,
        Err(failure) => return refuse_upgrade(stream, 404, &failure),
    };
    let accept = match check_upgrade(
        request
            .headers
            .iter()
            .map(|(name, value)| (name.as_str(), value.as_str())),
        &site.origin,
    ) {
        Ok(accept) => accept,
        Err(failure) => {
            let status = if failure.code == "origin_refused" {
                403
            } else {
                400
            };
            return refuse_upgrade(stream, status, &failure);
        }
    };
    let mut output = stream;
    output.write_all(&switching(&accept))?;
    output.flush()?;
    let lines = Arc::new(Mutex::new(Lines::new(stream.try_clone()?)));
    let writer: Arc<Mutex<dyn Write + Send>> = lines.clone();
    let bound_bytes = hub.config.frame_bytes.get();
    let mut messages = BufReader::new(Messages::new(reader, Arc::clone(&lines), bound_bytes));
    let set_read_timeout = |timeout| stream.set_read_timeout(timeout);
    let served = serve_link(
        door,
        &hub,
        Link {
            reader: &mut messages,
            writer: Arc::clone(&writer),
            set_read_timeout: &set_read_timeout,
        },
        None,
    );
    let refused = messages.get_mut().refusal.take();
    if let Err(failure) = served {
        let failure = refused.unwrap_or(failure);
        let mut lines = lock(&lines).map_err(|failure| io::Error::other(failure.code))?;
        super::assistant_wire::refuse(&mut *lines, &failure)?;
        lines.close()?;
    }
    Ok(())
}

fn refuse_upgrade(stream: &TcpStream, status: u16, failure: &Failure) -> io::Result<()> {
    Response::json(
        status,
        &json!({"code": failure.code, "message": failure.message}),
    )
    .write_to(stream)
}

/// The `Sec-WebSocket-Accept` value for the client's `Sec-WebSocket-Key`.
#[must_use]
pub fn accept(key: &str) -> String {
    let mut hasher = Sha1::new();
    hasher.update(key.as_bytes());
    hasher.update(ACCEPT_GUID.as_bytes());
    STANDARD.encode(hasher.finalize())
}

/// Check an upgrade request's headers, given as (lower-case name, value)
/// pairs, against the door's `origin`; the accept value on success.
///
/// # Errors
///
/// The refusal naming the first header that does not upgrade.
pub(super) fn check_upgrade<'a>(
    headers: impl IntoIterator<Item = (&'a str, &'a str)>,
    origin: &str,
) -> Result<String, Failure> {
    let (mut upgrade, mut connection, mut version, mut key, mut from) =
        (false, false, false, None, None);
    for (name, value) in headers {
        match name {
            "upgrade" => upgrade = value.trim().eq_ignore_ascii_case("websocket"),
            "connection" => {
                connection = value
                    .split(',')
                    .any(|token| token.trim().eq_ignore_ascii_case("upgrade"));
            }
            "sec-websocket-version" => version = value.trim() == "13",
            "sec-websocket-key" => key = Some(value.trim()),
            "origin" => from = Some(value.trim()),
            _ => {}
        }
    }
    if !(upgrade && connection) {
        return Err(Failure::new(
            "assistant_upgrade_required",
            "the assistant route is a WebSocket upgrade",
        ));
    }
    if !version {
        return Err(Failure::new(
            "assistant_upgrade_version",
            "the assistant route speaks WebSocket version 13",
        ));
    }
    if from != Some(origin) {
        return Err(Failure::new(
            "origin_refused",
            "the upgrade's Origin is not the door's origin",
        ));
    }
    let key = key
        .filter(|key| STANDARD.decode(key).is_ok_and(|bytes| bytes.len() == 16))
        .ok_or_else(|| {
            Failure::new(
                "assistant_upgrade_key",
                "Sec-WebSocket-Key must be 16 bytes in base64",
            )
        })?;
    Ok(accept(key))
}

/// The 101 answer that completes the upgrade.
#[must_use]
pub(super) fn switching(accept: &str) -> Vec<u8> {
    format!(
        "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: {accept}\r\n\r\n"
    )
    .into_bytes()
}

/// One frame read from the wire.
pub(super) struct Frame {
    pub fin: bool,
    pub opcode: u8,
    pub payload: Vec<u8>,
}

/// Read one frame, requiring the client's mask when `masked` is set and
/// refusing a data payload over `bound` bytes; `None` at the end of stream.
///
/// # Errors
///
/// The read's own error, or the refusal by name.
pub(super) fn read_frame(
    input: &mut impl Read,
    bound: usize,
    masked: bool,
) -> Result<Option<Frame>, Failure> {
    let broken = |_: io::Error| {
        Failure::new(
            "assistant_socket_read_failed",
            "the seat frame could not be read",
        )
    };
    let mut head = [0_u8; 2];
    if input.read(&mut head[..1]).map_err(broken)? == 0 {
        return Ok(None);
    }
    input.read_exact(&mut head[1..]).map_err(broken)?;
    let fin = head[0] & 0x80 != 0;
    if head[0] & 0x70 != 0 {
        return Err(Failure::new(
            "assistant_frame_invalid",
            "reserved bits are set",
        ));
    }
    let opcode = head[0] & 0x0F;
    if (head[1] & 0x80 != 0) != masked {
        return Err(Failure::new(
            "assistant_frame_invalid",
            "a client frame is masked and a door frame is not",
        ));
    }
    let length = match head[1] & 0x7F {
        126 => {
            let mut bytes = [0_u8; 2];
            input.read_exact(&mut bytes).map_err(broken)?;
            u64::from(u16::from_be_bytes(bytes))
        }
        127 => {
            let mut bytes = [0_u8; 8];
            input.read_exact(&mut bytes).map_err(broken)?;
            u64::from_be_bytes(bytes)
        }
        short => u64::from(short),
    };
    let limit = if opcode & 0x8 == 0 {
        bound
    } else {
        CONTROL_BYTES
    };
    let length = usize::try_from(length)
        .ok()
        .filter(|length| *length <= limit)
        .ok_or_else(|| bound_failure(limit, length))?;
    let mut mask = [0_u8; 4];
    if masked {
        input.read_exact(&mut mask).map_err(broken)?;
    }
    let mut payload = vec![0_u8; length];
    input.read_exact(&mut payload).map_err(broken)?;
    if masked {
        for (index, byte) in payload.iter_mut().enumerate() {
            *byte ^= mask[index % 4];
        }
    }
    Ok(Some(Frame {
        fin,
        opcode,
        payload,
    }))
}

fn bound_failure(limit: usize, length: u64) -> Failure {
    bound(
        "frame_bytes",
        limit,
        usize::try_from(length).unwrap_or(usize::MAX),
    )
}

/// Write one frame; a client's carries `mask`, a door's none.
///
/// # Errors
///
/// The writer's error.
pub(super) fn write_frame(
    output: &mut impl Write,
    opcode: u8,
    payload: &[u8],
    mask: Option<[u8; 4]>,
) -> io::Result<()> {
    output.write_all(&encode(opcode, payload, mask)?)?;
    output.flush()
}

/// One whole frame's bytes; a client's carries `mask`, a door's none.
///
/// # Errors
///
/// A payload whose length does not fit the frame's 64 bits.
pub fn encode(opcode: u8, payload: &[u8], mask: Option<[u8; 4]>) -> io::Result<Vec<u8>> {
    let mut bytes = Vec::with_capacity(payload.len().saturating_add(14));
    bytes.push(0x80 | opcode);
    let bit = if mask.is_some() { 0x80 } else { 0 };
    if let Ok(short @ 0..=125) = u8::try_from(payload.len()) {
        bytes.push(bit | short);
    } else if let Ok(medium) = u16::try_from(payload.len()) {
        bytes.push(bit | 0x7E);
        bytes.extend_from_slice(&medium.to_be_bytes());
    } else {
        let long = u64::try_from(payload.len()).map_err(io::Error::other)?;
        bytes.push(bit | 0x7F);
        bytes.extend_from_slice(&long.to_be_bytes());
    }
    match mask {
        Some(mask) => {
            bytes.extend_from_slice(&mask);
            bytes.extend(
                payload
                    .iter()
                    .enumerate()
                    .map(|(index, byte)| byte ^ mask[index % 4]),
            );
        }
        None => bytes.extend_from_slice(payload),
    }
    Ok(bytes)
}

/// The door's reader on an upgraded connection: each binary message, then
/// one newline, so `assistant_wire::read` sees the socket's frames. A
/// refusal is kept by name in [`Self::refusal`], since the reader's error
/// reaches `assistant_wire` only as a failed read.
pub(super) struct Messages<R> {
    input: R,
    output: Arc<Mutex<Lines<TcpStream>>>,
    bound: usize,
    buffer: Vec<u8>,
    at: usize,
    closed: bool,
    pub refusal: Option<Failure>,
}

impl<R: Read> Messages<R> {
    /// Messages from `input`, each at most `bound` bytes; a ping is answered
    /// and a close returned through `output`, the seat's own writer.
    pub(super) const fn new(input: R, output: Arc<Mutex<Lines<TcpStream>>>, bound: usize) -> Self {
        Self {
            input,
            output,
            bound,
            buffer: Vec::new(),
            at: 0,
            closed: false,
            refusal: None,
        }
    }

    /// Fill the buffer with the next message and its newline; false at the
    /// peer's close or the end of stream.
    fn next_message(&mut self) -> Result<bool, Failure> {
        loop {
            let Some(frame) = read_frame(&mut self.input, self.bound, true)? else {
                return Ok(false);
            };
            match frame.opcode {
                OPCODE_BINARY if frame.fin => {
                    if frame.payload.contains(&b'\n') {
                        return Err(Failure::new(
                            "assistant_frame_invalid",
                            "a message holds exactly one frame, with no newline",
                        ));
                    }
                    self.buffer = frame.payload;
                    self.buffer.push(b'\n');
                    self.at = 0;
                    return Ok(true);
                }
                OPCODE_BINARY | OPCODE_CONTINUATION => {
                    return Err(Failure::new(
                        "assistant_frame_invalid",
                        "a message is one frame, never fragmented",
                    ));
                }
                OPCODE_TEXT => {
                    return Err(Failure::new(
                        "assistant_frame_invalid",
                        "assistant messages are binary",
                    ));
                }
                OPCODE_PING => lock(&self.output)?
                    .control(OPCODE_PONG, &frame.payload)
                    .map_err(|_| written())?,
                OPCODE_PONG => {}
                OPCODE_CLOSE => {
                    if !self.closed {
                        self.closed = true;
                        lock(&self.output)?.close().map_err(|_| written())?;
                    }
                    return Ok(false);
                }
                _ => {
                    return Err(Failure::new(
                        "assistant_frame_invalid",
                        "an unknown WebSocket opcode",
                    ));
                }
            }
        }
    }
}

fn written() -> Failure {
    Failure::new(
        "assistant_socket_failed",
        "a control frame could not be written",
    )
}

impl<R: Read> Read for Messages<R> {
    fn read(&mut self, out: &mut [u8]) -> io::Result<usize> {
        if self.at == self.buffer.len() {
            match self.next_message() {
                Ok(true) => {}
                Ok(false) => return Ok(0),
                Err(failure) => {
                    let error = io::Error::new(io::ErrorKind::InvalidData, failure.code.clone());
                    self.refusal = Some(failure);
                    return Err(error);
                }
            }
        }
        let count = out.len().min(self.buffer.len() - self.at);
        out[..count].copy_from_slice(&self.buffer[self.at..self.at + count]);
        self.at += count;
        Ok(count)
    }
}

/// The door's writer on an upgraded connection: each newline-terminated
/// frame `assistant_wire` writes goes out as one binary message.
pub(super) struct Lines<W> {
    output: W,
    line: Vec<u8>,
    closed: bool,
}

impl<W: Write> Lines<W> {
    pub(super) const fn new(output: W) -> Self {
        Self {
            output,
            line: Vec::new(),
            closed: false,
        }
    }

    /// One control frame, never inside a message.
    fn control(&mut self, opcode: u8, payload: &[u8]) -> io::Result<()> {
        write_frame(&mut self.output, opcode, payload, None)
    }

    /// The door's close frame, once; nothing is written after it.
    pub(super) fn close(&mut self) -> io::Result<()> {
        if std::mem::replace(&mut self.closed, true) {
            return Ok(());
        }
        self.control(OPCODE_CLOSE, &[])
    }
}

impl<W: Write> Write for Lines<W> {
    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
        if self.closed {
            return Err(io::Error::from(io::ErrorKind::BrokenPipe));
        }
        for &byte in bytes {
            if byte == b'\n' {
                write_frame(&mut self.output, OPCODE_BINARY, &self.line, None)?;
                self.line.clear();
            } else {
                self.line.push(byte);
            }
        }
        Ok(bytes.len())
    }

    fn flush(&mut self) -> io::Result<()> {
        self.output.flush()
    }
}

#[cfg(test)]
#[path = "assistant_network_tests.rs"]
mod tests;