haematite-cli 0.21.12

//! HAEMNET-001 R1 and R2: haem/1 over the explicitly configured network
//! endpoint, decided through Lys exactly as the Unix socket is.
//!
//! The certificate fixtures in `network_fixtures/` are a test-only CA and a
//! server certificate for `localhost`, `127.0.0.1` and `::1`, with its key. They
//! sign nothing outside these tests.

use std::error::Error;
use std::io::{self, Read, Write};
use std::net::{SocketAddr, TcpStream};
use std::num::{NonZeroU32, NonZeroUsize};
use std::os::unix::net::UnixStream;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::thread::JoinHandle;
use std::time::Duration;

use base64::Engine;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use ed25519_dalek::{Signer, SigningKey};
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use rustls::pki_types::{CertificateDer, ServerName, pem::PemObject};
use rustls::{ClientConfig, ClientConnection, RootCertStore, StreamOwned};
use serde_json::{Value, json};

use super::network::{Network, NetworkConfig, Occupancy, Stop, TlsFiles};
use super::{Limits, connection};
use crate::serve::pass::{AUDIENCE, Gate, PASS_MISSING, Unreachable, now};
use crate::serve::protocol::{BranchPage, Operation, PROTOCOL, RECEIPT_CAPABILITY, Version};
use crate::serve::{frame, service::Service, storage::Storage};

type TestResult = Result<(), Box<dyn Error>>;

const PAGE: NonZeroU32 = NonZeroU32::MIN.saturating_add(9);
const MAX_FRAME: NonZeroU32 = NonZeroU32::MIN.saturating_add(65_535);
/// Bounds a read of a frame that is due, so a broken endpoint fails the
/// test rather than hanging it; no frame that is due waits on it.
const WAIT: Duration = Duration::from_secs(30);
const ISSUER: &str = "https://lys.example";
const KID: &str = "haematite-network-test";

fn signer() -> SigningKey {
    SigningKey::from_bytes(&[13; 32])
}

fn keys() -> String {
    let x = URL_SAFE_NO_PAD.encode(signer().verifying_key().as_bytes());
    json!({"keys": [{"kty": "OKP", "crv": "Ed25519", "kid": KID, "x": x, "alg": "EdDSA", "use": "sig"}]})
        .to_string()
}

/// A pass for `holder` with read and write on database `wards` and its
/// branch `main`, good for ten
/// minutes from the real clock the service reads.
fn pass(holder: &str) -> String {
    let issued = now();
    let header =
        URL_SAFE_NO_PAD.encode(json!({"alg": "EdDSA", "typ": "JWT", "kid": KID}).to_string());
    let claims = URL_SAFE_NO_PAD.encode(
        json!({
            "iss": ISSUER, "sub": holder, "aud": AUDIENCE, "iat": issued - 10, "exp": issued + 600,
            "holder": {"id": holder, "kind": "person", "responsible": null},
            "rights": [
                {"resource": {"kind": "haematite.database", "id": "wards"},
                    "actions": ["read", "write"], "mode": "outright", "grant": "network-grant"},
                {"resource": {"kind": "haematite.branch", "id": "wards.main"},
                    "actions": ["read", "write"], "mode": "outright", "grant": "network-grant"},
            ],
        })
        .to_string(),
    );
    let signature = URL_SAFE_NO_PAD.encode(
        signer()
            .sign(format!("{header}.{claims}").as_bytes())
            .to_bytes(),
    );
    format!("{header}.{claims}.{signature}")
}

fn fixture(name: &str) -> PathBuf {
    Path::new(env!("CARGO_MANIFEST_DIR"))
        .join("src/serve/network_fixtures")
        .join(name)
}

fn tls_files() -> TlsFiles {
    TlsFiles {
        certificate_file: fixture("server.pem"),
        key_file: fixture("server.key"),
    }
}

fn limits(max_clients: usize) -> Result<Limits, Box<dyn Error>> {
    Ok(Limits {
        max_frame: MAX_FRAME,
        max_clients: NonZeroUsize::new(max_clients).ok_or("zero clients")?,
        frame_timeout: Duration::from_secs(60),
    })
}

/// A served store of one shard, decided through Lys when `lys` is set.
fn service(directory: &Path, lys: bool) -> Result<Arc<Service>, Box<dyn Error>> {
    let storage = Storage::create(
        DatabaseConfig {
            data_dir: directory.join("database"),
            shard_count: 1,
            distributed: None,
            executor_threads: Some(1),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::KeepAll,
        },
        NodeCacheBudget::Unlimited,
        Duration::from_secs(1),
    )?;
    let service = Service::new(storage, PAGE)?;
    Ok(Arc::new(if lys {
        service.with_lys(Gate::new(&keys(), ISSUER, "wards", Arc::new(Unreachable))?)
    } else {
        service
    }))
}

fn head(service: &Service) -> Result<Version, Box<dyn Error>> {
    let answer = service.dispatch(Operation::Branches(BranchPage {
        after: None,
        limit: PAGE,
    }))?;
    let row = answer["entries"]
        .as_array()
        .ok_or("branches has no entries")?
        .iter()
        .find(|row| row["branch"] == "main")
        .ok_or("branch main missing")?;
    Ok(serde_json::from_value(row["version"].clone())?)
}

/// A network endpoint serving on a thread of its own until [`Served::finish`].
struct Served {
    address: SocketAddr,
    stop: Stop,
    occupancy: Occupancy,
    worker: Option<JoinHandle<io::Result<()>>>,
}

impl Served {
    fn start(
        service: &Arc<Service>,
        config: &NetworkConfig,
        limits: Limits,
    ) -> Result<Self, Box<dyn Error>> {
        let network = Network::bind(config, service, limits.max_clients)?;
        let address = network.address();
        let stop = network.stopper();
        let occupancy = network.occupancy();
        let service = Arc::clone(service);
        let worker = std::thread::Builder::new()
            .name("network-test".to_owned())
            .spawn(move || network.run(&service, limits))?;
        Ok(Self {
            address,
            stop,
            occupancy,
            worker: Some(worker),
        })
    }

    /// Stops accepting and waits for every client to leave; every client of
    /// the test is dropped before this.
    fn finish(mut self) -> TestResult {
        self.stop.stop();
        let worker = self.worker.take().ok_or("finished twice")?;
        worker.join().map_err(|_| "network thread panicked")??;
        Ok(())
    }
}

impl Drop for Served {
    fn drop(&mut self) {
        if let Some(worker) = self.worker.take() {
            self.stop.stop();
            if worker.join().is_err() {
                eprintln!("network thread panicked");
            }
        }
    }
}

type Tls = StreamOwned<ClientConnection, TcpStream>;

/// A client that trusts only the fixture CA, speaking TLS to `address` as
/// `localhost`.
fn tls_client(address: SocketAddr) -> Result<Tls, Box<dyn Error>> {
    let mut roots = RootCertStore::empty();
    for certificate in CertificateDer::pem_file_iter(fixture("ca.pem"))? {
        roots.add(certificate?)?;
    }
    let config =
        ClientConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider()))
            .with_safe_default_protocol_versions()?
            .with_root_certificates(roots)
            .with_no_client_auth();
    let connection = ClientConnection::new(Arc::new(config), ServerName::try_from("localhost")?)?;
    let tcp = TcpStream::connect(address)?;
    tcp.set_read_timeout(Some(WAIT))?;
    tcp.set_write_timeout(Some(WAIT))?;
    Ok(StreamOwned::new(connection, tcp))
}

fn plain_client(address: SocketAddr) -> Result<TcpStream, Box<dyn Error>> {
    let tcp = TcpStream::connect(address)?;
    tcp.set_read_timeout(Some(WAIT))?;
    tcp.set_write_timeout(Some(WAIT))?;
    Ok(tcp)
}

fn ask(client: &mut (impl Read + Write), request: &Value) -> Result<Value, Box<dyn Error>> {
    frame::write(client, &serde_json::to_vec(request)?, MAX_FRAME)?;
    let bytes = frame::read(client, MAX_FRAME)?.ok_or("connection closed")?;
    Ok(serde_json::from_slice(&bytes)?)
}

fn hello() -> Value {
    json!({"id":"hello", "operation":{"method":"hello", "params":{
        "protocol": PROTOCOL,
        "capabilities": [RECEIPT_CAPABILITY],
        "max_frame": MAX_FRAME,
    }}})
}

fn present(token: &str) -> Value {
    json!({"id":"pass", "operation":{"method":"session.pass", "params":{"pass": token}}})
}

fn branches() -> Value {
    json!({"id":"read", "operation":{"method":"branches.list", "params":{"limit":1}}})
}

fn collection(request_id: &str, expected: Version) -> Value {
    json!({"id":"write", "operation":{"method":"mutation.apply", "params":{
        "request_id": request_id, "branch": "main", "expected": expected,
        "change": {"kind": "collection.put", "value": {"collection": "patients", "schema": null}},
    }}})
}

fn code(answer: &Value) -> Option<&str> {
    answer["error"]["code"].as_str()
}

/// The answers to hello, a pass and one read over `client`.
fn exchange(client: &mut (impl Read + Write), token: &str) -> Result<Vec<Value>, Box<dyn Error>> {
    [hello(), present(token), branches()]
        .iter()
        .map(|request| ask(client, request))
        .collect()
}

fn network_tls() -> Result<NetworkConfig, Box<dyn Error>> {
    Ok(NetworkConfig {
        listen: "127.0.0.1:0".parse()?,
        tls: Some(tls_files()),
        max_connections: None,
    })
}

/// R1: the same hello, pass and read answer alike over the Unix door and the
/// TLS endpoint, under their own correlation ids, from one service.
#[test]
fn hello_and_request_answer_alike_over_unix_and_tls() -> TestResult {
    let directory = tempfile::tempdir()?;
    let service = service(directory.path(), true)?;
    let token = pass("ward-nurse");
    let (mut unix, server) = UnixStream::pair()?;
    unix.set_read_timeout(Some(WAIT))?;
    let local = Arc::clone(&service);
    let limits = limits(4)?;
    let unix_worker = std::thread::spawn(move || connection(server, &local, limits));
    let over_unix = exchange(&mut unix, &token)?;
    drop(unix);
    unix_worker
        .join()
        .map_err(|_| "unix connection panicked")??;

    let endpoint = Served::start(&service, &network_tls()?, limits)?;
    let mut tls = tls_client(endpoint.address)?;
    let over_tls = exchange(&mut tls, &token)?;
    drop(tls);
    endpoint.finish()?;

    for answer in &over_tls {
        assert!(answer.get("error").is_none(), "over TLS: {answer}");
    }
    assert_eq!(over_tls, over_unix);
    assert_eq!(over_tls[0]["id"], "hello");
    assert_eq!(over_tls[1]["id"], "pass");
    assert_eq!(over_tls[1]["result"]["holder"], "ward-nurse");
    assert_eq!(over_tls[2]["id"], "read");
    Ok(())
}

/// R1: before hello every operation is `hello_required` over TLS, as on the
/// Unix door.
#[test]
fn a_request_before_hello_is_hello_required_over_tls() -> TestResult {
    let directory = tempfile::tempdir()?;
    let service = service(directory.path(), true)?;
    let endpoint = Served::start(&service, &network_tls()?, limits(4)?)?;
    let mut tls = tls_client(endpoint.address)?;
    let answer = ask(&mut tls, &branches())?;
    assert_eq!(answer["id"], "read");
    assert_eq!(code(&answer), Some("hello_required"));
    drop(tls);
    endpoint.finish()
}

/// R1: a zero-length frame and a frame whose header claims more than the
/// limit each end the connection at once, with no dispatch: the store's
/// head does not move, and the server never waits for the claimed body,
/// which it would only do had it read the header as an allocation.
#[test]
fn zero_and_over_limit_frames_end_the_connection_without_dispatch() -> TestResult {
    let directory = tempfile::tempdir()?;
    let service = service(directory.path(), true)?;
    let before = head(&service)?;
    let endpoint = Served::start(&service, &network_tls()?, limits(4)?)?;
    for header in [0_u32, u32::MAX] {
        let mut tls = tls_client(endpoint.address)?;
        ask(&mut tls, &hello())?;
        tls.write_all(&header.to_be_bytes())?;
        tls.flush()?;
        let mut rest = Vec::new();
        // The connection ends: end of file, or a reset, and never a frame.
        let ended = tls.read_to_end(&mut rest);
        assert!(
            rest.is_empty(),
            "header {header}: answered {} bytes",
            rest.len()
        );
        if let Err(error) = ended {
            assert_ne!(
                error.kind(),
                io::ErrorKind::WouldBlock,
                "header {header}: the server waited for a body"
            );
            assert_ne!(
                error.kind(),
                io::ErrorKind::TimedOut,
                "header {header}: the server waited for a body"
            );
        }
    }
    endpoint.finish()?;
    assert_eq!(head(&service)?, before);
    Ok(())
}

/// R1: with the one connection the endpoint allows held, another client is
/// refused `busy_max_clients` by name; when the first leaves, the endpoint
/// says so and a third client is endpoint.
#[test]
fn a_full_endpoint_refuses_by_name_and_a_leaving_client_frees_it() -> TestResult {
    let directory = tempfile::tempdir()?;
    let service = service(directory.path(), true)?;
    let endpoint = Served::start(&service, &network_tls()?, limits(1)?)?;
    let mut first = tls_client(endpoint.address)?;
    assert!(ask(&mut first, &hello())?.get("error").is_none());
    let mut second = tls_client(endpoint.address)?;
    let refused = ask(&mut second, &hello())?;
    assert_eq!(refused["id"], "hello");
    assert_eq!(code(&refused), Some("busy_max_clients"));
    drop(second);
    drop(first);
    assert!(
        endpoint.occupancy.wait_at_most(0, WAIT),
        "the first client's connection was never released"
    );
    let mut third = tls_client(endpoint.address)?;
    assert!(ask(&mut third, &hello())?.get("error").is_none());
    drop(third);
    endpoint.finish()
}

/// R1, ruled 19 Oct by Apollo: plain TCP is bound only on a loopback
/// address, for a TLS proxy in front; anywhere else needs the certificate.
#[test]
fn plain_tcp_is_served_only_on_loopback() -> TestResult {
    let directory = tempfile::tempdir()?;
    let service = service(directory.path(), true)?;
    let limits = limits(4)?;
    for listen in ["0.0.0.0:0", "[::]:0"] {
        let open = NetworkConfig {
            listen: listen.parse()?,
            tls: None,
            max_connections: None,
        };
        let refused = Network::bind(&open, &service, limits.max_clients)
            .err()
            .ok_or("an open plain bind was accepted")?;
        assert!(
            refused.starts_with("network_config_invalid: "),
            "{listen}: {refused}"
        );
    }
    let loopback = NetworkConfig {
        listen: "127.0.0.1:0".parse()?,
        tls: None,
        max_connections: None,
    };
    let endpoint = Served::start(&service, &loopback, limits)?;
    let mut plain = plain_client(endpoint.address)?;
    let answers = exchange(&mut plain, &pass("ward-nurse"))?;
    for answer in &answers {
        assert!(answer.get("error").is_none(), "plain loopback: {answer}");
    }
    drop(plain);
    endpoint.finish()
}

/// R1: a certificate or key that cannot be read stops the endpoint by name
/// before it binds.
#[test]
fn unreadable_tls_files_are_refused_by_name() -> TestResult {
    let directory = tempfile::tempdir()?;
    let service = service(directory.path(), true)?;
    let max = limits(4)?.max_clients;
    for files in [
        TlsFiles {
            certificate_file: fixture("absent.pem"),
            key_file: fixture("server.key"),
        },
        TlsFiles {
            certificate_file: fixture("server.pem"),
            key_file: fixture("ca.pem"),
        },
        TlsFiles {
            certificate_file: PathBuf::from("server.pem"),
            key_file: fixture("server.key"),
        },
    ] {
        let config = NetworkConfig {
            listen: "127.0.0.1:0".parse()?,
            tls: Some(files),
            max_connections: None,
        };
        let refused = Network::bind(&config, &service, max)
            .err()
            .ok_or("unreadable TLS files were accepted")?;
        assert!(refused.starts_with("network_tls_invalid: "), "{refused}");
    }
    Ok(())
}

/// R2: a store served by its owner alone, with no Lys, never binds a network
/// endpoint: there is no owner Standalone over the network.
#[test]
fn a_store_served_without_lys_binds_no_network_endpoint() -> TestResult {
    let directory = tempfile::tempdir()?;
    let service = service(directory.path(), false)?;
    for config in [
        network_tls()?,
        NetworkConfig {
            listen: "127.0.0.1:0".parse()?,
            tls: None,
            max_connections: None,
        },
    ] {
        let refused = Network::bind(&config, &service, limits(4)?.max_clients)
            .err()
            .ok_or("a standalone store bound a network endpoint")?;
        assert!(refused.starts_with("network_requires_lys: "), "{refused}");
    }
    Ok(())
}

/// R2: over the network, a request before a verified pass is refused by the
/// socket path's own name, a pass that does not verify is refused by its
/// name and leaves the connection without one, and neither moves the store;
/// the verified pass then reaches the act it grants.
#[test]
fn a_network_request_needs_a_verified_pass() -> TestResult {
    let directory = tempfile::tempdir()?;
    let service = service(directory.path(), true)?;
    let before = head(&service)?;
    let endpoint = Served::start(&service, &network_tls()?, limits(4)?)?;
    let mut tls = tls_client(endpoint.address)?;
    assert!(ask(&mut tls, &hello())?.get("error").is_none());

    let unpassed = ask(&mut tls, &collection("write-unpassed", before))?;
    assert_eq!(code(&unpassed), Some(PASS_MISSING), "{unpassed}");
    let read = ask(&mut tls, &branches())?;
    assert_eq!(code(&read), Some(PASS_MISSING), "{read}");

    let mut forged = pass("ward-nurse");
    forged.push('A');
    let refused = ask(&mut tls, &present(&forged))?;
    let name = code(&refused).ok_or("a forged pass was accepted")?;
    assert_ne!(name, PASS_MISSING);
    let still = ask(&mut tls, &collection("write-forged", before))?;
    assert_eq!(code(&still), Some(PASS_MISSING), "{still}");
    assert_eq!(head(&service)?, before);

    assert!(
        ask(&mut tls, &present(&pass("ward-nurse")))?
            .get("error")
            .is_none()
    );
    let written = ask(&mut tls, &collection("write-passed", before))?;
    assert!(written.get("error").is_none(), "{written}");
    assert_ne!(head(&service)?, before);
    drop(tls);
    endpoint.finish()
}

/// R1: with no `network` section no network listener is made; the Unix
/// socket is served as before (every other test in this crate).
#[test]
fn no_network_section_binds_no_network_listener() -> TestResult {
    let directory = tempfile::tempdir()?;
    let service = service(directory.path(), true)?;
    let bound = crate::serve::cli::network_endpoint(None, &service, limits(4)?.max_clients)?;
    assert!(bound.is_none());
    let named: NetworkConfig = serde_json::from_value(json!({
        "listen": "127.0.0.1:0",
        "tls": {"certificate_file": "/etc/haematite/cert.pem", "key_file": "/etc/haematite/key.pem"},
    }))?;
    assert!(named.tls.is_some() && named.max_connections.is_none());
    let unknown = serde_json::from_value::<NetworkConfig>(json!({
        "listen": "127.0.0.1:0", "standalone": true,
    }));
    assert!(unknown.is_err(), "an unknown network setting was accepted");
    Ok(())
}