haematite-cli 0.21.12

//! HAEMNET-001 R1 and R2: the haem/1 door over an explicitly configured
//! network endpoint.
//!
//! The same four-byte big-endian frames, hello, correlation ids, limits and
//! service dispatch as the Unix socket (`connection`), over TLS, or over
//! plain TCP on a loopback address only, for a TLS proxy in front (the rule
//! mcp-http and the door follow). Ruled 19 Oct by Apollo: a network
//! connection is decided through Lys exactly as a Unix-socket connection is,
//! by the pass it presents with `session.pass`; a store served without Lys
//! binds no network endpoint, so there is never an owner Standalone over the
//! network. Network serving starts only when service.json names it.

use std::io::{self, Read, Write};
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, TcpListener, TcpStream};
use std::num::NonZeroUsize;
use std::os::fd::OwnedFd;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use rustls::pki_types::{CertificateDer, PrivateKeyDer, pem::PemObject};
use rustls::{ServerConfig, ServerConnection, StreamOwned};
use serde::Deserialize;

use super::super::service::Service;
use super::clients::Clients;
use super::listener::{Accepted, admit, spawn_client};
use super::{Limits, Wire};

/// service.json's `network`: where the endpoint listens, and the TLS
/// certificate and key it serves with. Absent, no network listener exists.
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct NetworkConfig {
    /// The one address bound. Without `tls` it must be a loopback address.
    pub listen: SocketAddr,
    #[serde(default)]
    pub tls: Option<TlsFiles>,
    /// The most network connections served at once; `max_clients` when
    /// absent. A further one is refused `busy_max_clients`.
    #[serde(default)]
    pub max_connections: Option<NonZeroUsize>,
}

/// The PEM files the endpoint's TLS is served with, both absolute paths:
/// the certificate chain, leaf first, and its private key.
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TlsFiles {
    pub certificate_file: PathBuf,
    pub key_file: PathBuf,
}

/// A bound network endpoint, serving once [`Network::run`] is called.
pub struct Network {
    listener: TcpListener,
    address: SocketAddr,
    tls: Option<Arc<ServerConfig>>,
    max_connections: NonZeroUsize,
    stopping: Arc<AtomicBool>,
    active: Arc<Clients>,
}

impl Network {
    /// Binds `config`'s endpoint for `service`, its connections bounded by
    /// `max_connections`, else by `max_clients`.
    ///
    /// # Errors
    ///
    /// `network_requires_lys` for a service without Lys,
    /// `network_config_invalid` for plain TCP on an address that is not
    /// loopback, `network_tls_invalid` for a certificate or key that cannot
    /// be read or served, and `network_bind_failed` with the OS's words.
    pub fn bind(
        config: &NetworkConfig,
        service: &Service,
        max_clients: NonZeroUsize,
    ) -> Result<Self, String> {
        if !service.decides() {
            return Err("network_requires_lys: a network connection is admitted only by the Lys pass it presents; add a lys section to the service configuration".to_owned());
        }
        let tls = match &config.tls {
            Some(files) => Some(Arc::new(server_config(files)?)),
            None if config.listen.ip().is_loopback() => None,
            None => {
                return Err(format!(
                    "network_config_invalid: listen {} is not a loopback address, so it needs tls with a certificate_file and key_file; plain TCP is served only on loopback, behind a TLS proxy",
                    config.listen
                ));
            }
        };
        let listener = TcpListener::bind(config.listen)
            .map_err(|error| format!("network_bind_failed: {}: {error}", config.listen))?;
        let address = listener
            .local_addr()
            .map_err(|error| format!("network_bind_failed: {}: {error}", config.listen))?;
        Ok(Self {
            listener,
            address,
            tls,
            max_connections: config.max_connections.unwrap_or(max_clients),
            stopping: Arc::new(AtomicBool::new(false)),
            active: Arc::new(Clients::new(0)),
        })
    }

    /// The address bound, its port as the OS chose it for port 0.
    #[must_use]
    pub const fn address(&self) -> SocketAddr {
        self.address
    }

    /// Whether connections are served under TLS, not plain on loopback.
    #[must_use]
    pub const fn secured(&self) -> bool {
        self.tls.is_some()
    }

    /// A handle that stops [`Network::run`] from another thread.
    #[must_use]
    pub fn stopper(&self) -> Stop {
        Stop {
            stopping: Arc::clone(&self.stopping),
            wake: wake_address(self.address),
        }
    }

    #[cfg(test)]
    pub(super) fn occupancy(&self) -> Occupancy {
        Occupancy(Arc::clone(&self.active))
    }

    /// Serves every connection the endpoint accepts, each on a thread of its
    /// own, until [`Stop::stop`]; then waits for the admitted ones to end.
    ///
    /// # Errors
    ///
    /// The frame timeout is zero, or waiting for the clients failed.
    pub fn run(&self, service: &Arc<Service>, limits: Limits) -> io::Result<()> {
        if limits.frame_timeout.is_zero() {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "frame timeout must be positive",
            ));
        }
        let limits = Limits {
            max_clients: self.max_connections,
            ..limits
        };
        let incoming = self
            .listener
            .incoming()
            .map_while(|accepted| (!self.stopping.load(Ordering::Acquire)).then_some(accepted))
            .map(|accepted| {
                accepted.map(|tcp| Peer {
                    tcp,
                    tls: self.tls.clone(),
                })
            });
        admit(incoming, &self.active, service, limits, spawn_client);
        self.active.drain()
    }
}

/// Stops a running [`Network`]: no connection is accepted after it.
#[derive(Clone)]
pub struct Stop {
    stopping: Arc<AtomicBool>,
    wake: SocketAddr,
}

impl Stop {
    /// Marks the endpoint stopping and wakes its accept with one connection
    /// of its own, which is never served.
    pub fn stop(&self) {
        self.stopping.store(true, Ordering::Release);
        if let Err(error) = TcpStream::connect_timeout(&self.wake, Duration::from_secs(1)) {
            eprintln!("haematite network_stop_wake_failed: {}: {error}", self.wake);
        }
    }
}

/// How many network connections are admitted, for a test to wait on.
#[cfg(test)]
pub(super) struct Occupancy(Arc<Clients>);

#[cfg(test)]
impl Occupancy {
    pub(super) fn wait_at_most(&self, count: usize, bound: Duration) -> bool {
        self.0.wait_at_most(count, bound)
    }
}

/// Where the endpoint's own stop connection goes: the bound address, or the
/// loopback address of its family for an unspecified one.
const fn wake_address(address: SocketAddr) -> SocketAddr {
    let ip = match address.ip() {
        IpAddr::V4(ip) if ip.is_unspecified() => IpAddr::V4(Ipv4Addr::LOCALHOST),
        IpAddr::V6(ip) if ip.is_unspecified() => IpAddr::V6(Ipv6Addr::LOCALHOST),
        ip => ip,
    };
    SocketAddr::new(ip, address.port())
}

/// The TLS configuration `files` name, through the ring provider the graph
/// already builds.
pub(crate) fn server_config(files: &TlsFiles) -> Result<ServerConfig, String> {
    let invalid =
        |what: &str, error: &dyn std::fmt::Display| format!("network_tls_invalid: {what}: {error}");
    if !files.certificate_file.is_absolute() || !files.key_file.is_absolute() {
        return Err(
            "network_tls_invalid: certificate_file and key_file must be absolute paths".to_owned(),
        );
    }
    let certificate = files.certificate_file.display().to_string();
    let chain = CertificateDer::pem_file_iter(&files.certificate_file)
        .map_err(|error| invalid(&certificate, &error))?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|error| invalid(&certificate, &error))?;
    if chain.is_empty() {
        return Err(invalid(&certificate, &"holds no certificate"));
    }
    let key_file = files.key_file.display().to_string();
    let key = PrivateKeyDer::from_pem_file(&files.key_file)
        .map_err(|error| invalid(&key_file, &error))?;
    ServerConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider()))
        .with_safe_default_protocol_versions()
        .map_err(|error| invalid("protocol versions", &error))?
        .with_no_client_auth()
        .with_single_cert(chain, key)
        .map_err(|error| invalid("certificate and key", &error))
}

/// A network connection as accepted, before its TLS handshake.
struct Peer {
    tcp: TcpStream,
    tls: Option<Arc<ServerConfig>>,
}

impl Accepted for Peer {
    type Wire = NetworkWire;

    fn duplicate(&self) -> io::Result<Self> {
        Ok(Self {
            tcp: self.tcp.try_clone()?,
            tls: self.tls.clone(),
        })
    }

    /// The handshake is not run here: rustls runs it on the connection's
    /// first read, under its frame deadline.
    fn into_wire(self) -> io::Result<NetworkWire> {
        Ok(match self.tls {
            None => NetworkWire::Plain(self.tcp),
            Some(config) => NetworkWire::Tls(Box::new(StreamOwned::new(
                ServerConnection::new(config).map_err(io::Error::other)?,
                self.tcp,
            ))),
        })
    }
}

/// One network connection's byte stream.
pub(super) enum NetworkWire {
    Plain(TcpStream),
    Tls(Box<StreamOwned<ServerConnection, TcpStream>>),
}

impl NetworkWire {
    fn tcp(&self) -> &TcpStream {
        match self {
            Self::Plain(tcp) => tcp,
            Self::Tls(tls) => &tls.sock,
        }
    }
}

impl Read for NetworkWire {
    fn read(&mut self, bytes: &mut [u8]) -> io::Result<usize> {
        match self {
            Self::Plain(tcp) => tcp.read(bytes),
            // A peer that closes its socket without a TLS close_notify has
            // still closed: between frames that is the end of the
            // connection, and inside one the frame reader names the cut.
            Self::Tls(tls) => match tls.read(bytes) {
                Err(error) if error.kind() == io::ErrorKind::UnexpectedEof => Ok(0),
                read => read,
            },
        }
    }
}

impl Write for NetworkWire {
    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
        match self {
            Self::Plain(tcp) => tcp.write(bytes),
            Self::Tls(tls) => tls.write(bytes),
        }
    }

    fn flush(&mut self) -> io::Result<()> {
        match self {
            Self::Plain(tcp) => tcp.flush(),
            Self::Tls(tls) => tls.flush(),
        }
    }
}

impl Wire for NetworkWire {
    fn set_read_timeout(&self, timeout: Option<Duration>) -> io::Result<()> {
        self.tcp().set_read_timeout(timeout)
    }
    fn set_write_timeout(&self, timeout: Option<Duration>) -> io::Result<()> {
        self.tcp().set_write_timeout(timeout)
    }
    fn set_nonblocking(&self, nonblocking: bool) -> io::Result<()> {
        self.tcp().set_nonblocking(nonblocking)
    }
    fn peer(&self) -> io::Result<OwnedFd> {
        self.tcp().try_clone().map(OwnedFd::from)
    }
}