haematite-cli 0.21.12

//! HAEMNET-003 R1: a door store entry that names a HAEMNET-001 network
//! endpoint is reached over verified TLS with the signed-in person's pass,
//! and every route answers it as it answers a socket entry.
//!
//! One served store is reached both ways: on its Unix socket (`local`) and
//! at its TLS endpoint (`play`), from one service decided through Lys, so a
//! verb succeeds only when the person's pass was presented and verified.
//! The certificate fixtures are serve's test-only CA and its `localhost`
//! server certificate (`serve/network_fixtures`).

use std::error::Error;
use std::io::Cursor;
use std::net::SocketAddr;
use std::num::{NonZeroU32, NonZeroUsize};
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 lys_pass::Error as LysError;
use serde_json::{Value, json};

use super::config::Config;
use super::http::{Request, read};
use super::route::{Site, answer};
use super::signin::{Door, Issuer, SignedIn, Started};
use crate::client::network::Trust;
use crate::serve::pass::{AUDIENCE, Gate, Unreachable, now};
use crate::serve::service::Service;
use crate::serve::storage::Storage;
use crate::serve::transport::network::{Network, NetworkConfig, Stop, TlsFiles};
use crate::serve::transport::{Limits, Listener};

/// One route of the door, answering a store's name with its status and body.
type Route<'a> = dyn Fn(&str) -> Result<(u16, Value), Box<dyn Error>> + 'a;

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

const ORIGIN: &str = "https://studio.example";
const ISSUER: &str = "https://lys.example";
const KID: &str = "haematite-door-network-test";
const PAGE: NonZeroU32 = NonZeroU32::MIN.saturating_add(9);

fn signer() -> SigningKey {
    SigningKey::from_bytes(&[29; 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 person's pass 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": "door-network-grant"},
                {"resource": {"kind": "haematite.branch", "id": "wards.main"},
                    "actions": ["read", "write"], "mode": "outright", "grant": "door-network-grant"},
            ],
        })
        .to_string(),
    );
    let signature = URL_SAFE_NO_PAD.encode(
        signer()
            .sign(format!("{header}.{claims}").as_bytes())
            .to_bytes(),
    );
    format!("{header}.{claims}.{signature}")
}

/// Signs every browser in as `ward-nurse` with a pass the store verifies.
struct Lys;

impl Issuer for Lys {
    fn start(&self) -> Result<Started, LysError> {
        Ok(Started {
            url: format!("{ISSUER}/oauth/authorize?state=s1"),
            state: "s1".to_owned(),
            verifier: "v1".to_owned(),
            browser: "b1".to_owned(),
        })
    }

    fn finish(&self, _: &str, _: &str, _: &str, _: &Started) -> Result<SignedIn, LysError> {
        Ok(SignedIn {
            token: pass("ward-nurse"),
            holder: "ward-nurse".to_owned(),
            expires_at: now() + 600,
            refresh: None,
        })
    }

    fn refresh(&self, _: &str) -> Result<SignedIn, LysError> {
        Err(LysError::CannotAsk("no refresh in this double"))
    }
}

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

fn limits() -> Result<Limits, Box<dyn Error>> {
    Ok(Limits {
        max_frame: NonZeroU32::MIN.saturating_add(65_535),
        max_clients: NonZeroUsize::new(8).ok_or("zero clients")?,
        frame_timeout: Duration::from_secs(60),
    })
}

/// A served store of one shard, decided through Lys.
fn service(directory: &Path) -> 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(service.with_lys(Gate::new(
        &keys(),
        ISSUER,
        "wards",
        Arc::new(Unreachable),
    )?)))
}

/// The store's TLS endpoint, serving until [`Endpoint::stop`].
struct Endpoint {
    address: SocketAddr,
    stop: Stop,
    worker: Option<JoinHandle<std::io::Result<()>>>,
}

impl Endpoint {
    fn start(service: &Arc<Service>) -> Result<Self, Box<dyn Error>> {
        let limits = limits()?;
        let config = NetworkConfig {
            listen: "127.0.0.1:0".parse()?,
            tls: Some(TlsFiles {
                certificate_file: fixture("server.pem"),
                key_file: fixture("server.key"),
            }),
            max_connections: None,
        };
        let network = Network::bind(&config, service, limits.max_clients)?;
        let address = network.address();
        let stop = network.stopper();
        let served = Arc::clone(service);
        let worker = std::thread::Builder::new()
            .name("door-network-store".to_owned())
            .spawn(move || network.run(&served, limits))?;
        Ok(Self {
            address,
            stop,
            worker: Some(worker),
        })
    }

    /// Stop accepting and wait for every connection to end: after this
    /// nothing listens at the address.
    fn stop(&mut self) -> TestResult {
        self.stop.stop();
        let worker = self.worker.take().ok_or("stopped twice")?;
        worker.join().map_err(|_| "network thread panicked")??;
        Ok(())
    }
}

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

/// The store's Unix socket, served on a thread of its own for the test,
/// in a private folder as serve requires.
fn socket(service: &Arc<Service>, path: &Path) -> TestResult {
    use std::os::unix::fs::PermissionsExt as _;
    let folder = path.parent().ok_or("the socket has no folder")?;
    std::fs::set_permissions(folder, std::fs::Permissions::from_mode(0o700))?;
    let listener = Listener::bind(path)?;
    let served = Arc::clone(service);
    let limits = limits()?;
    std::thread::Builder::new()
        .name("door-socket-store".to_owned())
        .spawn(move || {
            if let Err(error) = listener.run(&served, limits) {
                eprintln!("socket store ended: {error}");
            }
        })?;
    Ok(())
}

/// The door's configuration with `stores`, parsed as the door parses it.
fn config(files: &Path, stores: &Value) -> Result<Config, String> {
    std::fs::write(files.join("secret"), "fixture").map_err(|error| error.to_string())?;
    Config::parse(
        json!({
            "listen": "127.0.0.1:0", "origin": ORIGIN, "files": files, "stores": stores,
            "lys": {"issuer": ISSUER, "client_id": "fixture",
                "secret_file": files.join("secret"), "redirect": format!("{ORIGIN}/callback")},
        })
        .to_string()
        .as_bytes(),
    )
}

fn network_entry(address: SocketAddr) -> Value {
    json!({"endpoint": address.to_string(), "server_name": "localhost"})
}

/// The door's site, its network stores verified against the test CA alone.
fn site(config: &Config) -> Result<Site, Box<dyn Error>> {
    Ok(config.site(&Trust::only(&fixture("ca.pem"))?)?)
}

fn signed_in(door: &Door<Lys>) -> Result<String, Box<dyn Error>> {
    let begun = door.begin(now())?;
    Ok(door.callback("code", "s1", &begun.browser, now())?)
}

/// The door's status and JSON body for `text`, a whole request.
fn ask(door: &Door<Lys>, site: &Site, text: &str) -> Result<(u16, Value), Box<dyn Error>> {
    let request: Request = read(&mut Cursor::new(text.as_bytes().to_vec()), 16_384, 1 << 20)?;
    let mut out = Vec::new();
    answer(door, site, &request, now()).write(&mut out)?;
    let written = String::from_utf8(out)?;
    let status = written
        .split(' ')
        .nth(1)
        .ok_or("no status")?
        .parse::<u16>()?;
    let body = written.split("\r\n\r\n").nth(1).ok_or("no body")?;
    Ok((status, serde_json::from_str(body)?))
}

fn verb(
    door: &Door<Lys>,
    site: &Site,
    session: &str,
    store: &str,
    method: &str,
    params: &Value,
) -> Result<(u16, Value), Box<dyn Error>> {
    let body = json!({"method": method, "params": params}).to_string();
    ask(
        door,
        site,
        &format!(
            "POST /api/haem/{store} HTTP/1.1\r\nOrigin: {ORIGIN}\r\nCookie: haem_session={session}\r\nContent-Length: {}\r\n\r\n{body}",
            body.len()
        ),
    )
}

fn follow(
    door: &Door<Lys>,
    site: &Site,
    session: &str,
    store: &str,
) -> Result<(u16, Value), Box<dyn Error>> {
    ask(
        door,
        site,
        &format!(
            "GET /api/haem/{store}/follow?branch=main HTTP/1.1\r\nCookie: haem_session={session}\r\n\r\n"
        ),
    )
}

fn start_assistant(
    door: &Door<Lys>,
    site: &Site,
    session: &str,
    store: &str,
) -> Result<(u16, Value), Box<dyn Error>> {
    ask(
        door,
        site,
        &format!(
            "POST /api/assistant/{store}/start HTTP/1.1\r\nOrigin: {ORIGIN}\r\nCookie: haem_session={session}\r\nContent-Length: 0\r\n\r\n"
        ),
    )
}

fn collections() -> Value {
    json!({"selection": {"branch": "main", "version": null}, "after": null, "limit": 10})
}

/// R1: the `play` entry names the store's TLS endpoint, and the door lists,
/// reads and writes through it exactly as through the `local` socket entry
/// to the same store. The store is decided through Lys, so the write is
/// carried only because the person's pass was presented on the connection
/// and verified there. Fails today: a network entry is refused at parse.
#[test]
fn a_network_store_lists_reads_and_writes_through_the_door_as_its_socket_does() -> TestResult {
    let directory = tempfile::tempdir()?;
    let service = service(directory.path())?;
    let local = directory.path().join("store.sock");
    socket(&service, &local)?;
    let endpoint = Endpoint::start(&service)?;
    let config = config(
        directory.path(),
        &json!({"local": local, "play": network_entry(endpoint.address)}),
    )?;
    let site = site(&config)?;
    let door = Door::new(Lys);
    let session = signed_in(&door)?;

    let (status, hello) = verb(&door, &site, &session, "play", "hello", &Value::Null)?;
    assert_eq!(status, 200, "{hello}");
    assert_eq!(hello["protocol"], "haem/1");

    let listed = |store: &str| {
        verb(
            &door,
            &site,
            &session,
            store,
            "branches.list",
            &json!({"limit": 10}),
        )
    };
    let (local_status, over_socket) = listed("local")?;
    let (play_status, over_network) = listed("play")?;
    assert_eq!((local_status, play_status), (200, 200), "{over_network}");
    assert_eq!(over_network, over_socket);
    let head = over_network["entries"]
        .as_array()
        .ok_or("branches has no entries")?
        .iter()
        .find(|row| row["branch"] == "main")
        .ok_or("branch main missing")?["version"]
        .clone();

    let (status, written) = verb(
        &door,
        &site,
        &session,
        "play",
        "mutation.apply",
        &json!({
            "request_id": "door-network-write", "branch": "main", "expected": head,
            "change": {"kind": "collection.put", "value": {"collection": "patients", "schema": null}},
        }),
    )?;
    assert_eq!(
        status, 200,
        "the write through the network store: {written}"
    );
    assert!(written.get("code").is_none(), "{written}");

    let (local_status, read_socket) = verb(
        &door,
        &site,
        &session,
        "local",
        "collections.list",
        &collections(),
    )?;
    let (play_status, read_network) = verb(
        &door,
        &site,
        &session,
        "play",
        "collections.list",
        &collections(),
    )?;
    assert_eq!((local_status, play_status), (200, 200), "{read_network}");
    assert_eq!(read_network, read_socket);
    assert!(
        read_network["entries"]
            .as_array()
            .ok_or("collections has no entries")?
            .iter()
            .any(|entry| entry["name"] == "patients"),
        "the collection written through the network store is not read back: {read_network}"
    );
    Ok(())
}

/// R1: the network store stopped, every route the door carries to a store
/// answers it with the same status and name as a stopped socket store (one
/// whose socket no longer exists, as serve leaves it when it stops), never a
/// 502 without a name, and the words name the endpoint. Fails today: a
/// network entry is refused at parse.
#[test]
fn a_stopped_network_store_answers_as_a_stopped_socket_store() -> TestResult {
    let directory = tempfile::tempdir()?;
    let service = service(directory.path())?;
    let mut endpoint = Endpoint::start(&service)?;
    let address = endpoint.address;
    endpoint.stop()?;
    let config = config(
        directory.path(),
        &json!({"local": directory.path().join("stopped.sock"), "play": network_entry(address)}),
    )?;
    let site = site(&config)?;
    let door = Door::new(Lys);
    let session = signed_in(&door)?;
    let routes: [(&str, &Route<'_>); 3] = [
        ("verb", &|store| {
            verb(
                &door,
                &site,
                &session,
                store,
                "branches.list",
                &json!({"limit": 1}),
            )
        }),
        ("follow", &|store| follow(&door, &site, &session, store)),
        ("assistant start", &|store| {
            start_assistant(&door, &site, &session, store)
        }),
    ];
    for (route, asked) in routes {
        let (socket_status, over_socket) = asked("local")?;
        let (network_status, over_network) = asked("play")?;
        assert_eq!(network_status, socket_status, "{route}: {over_network}");
        assert_eq!(
            over_network["code"], over_socket["code"],
            "{route}: {over_network}"
        );
        assert!(
            over_network["code"]
                .as_str()
                .is_some_and(|code| !code.is_empty()),
            "{route}: no name: {over_network}"
        );
        assert!(
            over_network["message"]
                .as_str()
                .is_some_and(|message| message.contains(&address.to_string())),
            "{route}: the endpoint is not named: {over_network}"
        );
    }
    Ok(())
}