haematite-cli 0.33.0

//! ACCESS-003 R4: every served write records who made it, and history
//! answers that author beside the version it was written at.

use std::error::Error;
use std::num::{NonZeroU32, NonZeroUsize};
use std::os::unix::fs::PermissionsExt;
use std::path::{Path, PathBuf};
use std::sync::Arc;
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 serde_json::value::RawValue;
use serde_json::{Value, json};

use crate::client::{Client, Reply};
use crate::serve::pass::{AUDIENCE, Gate, Unreachable, now};
use crate::serve::service::Service;
use crate::serve::storage::Storage;
use crate::serve::transport::{Limits, Listener};

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

const PAGE: NonZeroU32 = NonZeroU32::MIN.saturating_add(9);
/// The bound on each step of the client's connection, as every client sets one.
const STEP: Duration = Duration::from_secs(60);
const ISSUER: &str = "https://lys.example";
const KID: &str = "haematite-author-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` of `kind`, answering to `responsible`, that may read,
/// query and write database `wards`, its collection `patients` and its
/// branch `main`, good for ten minutes from the real clock the store reads.
fn pass(holder: &str, kind: &str, responsible: Option<&str>) -> String {
    let issued = now();
    let rights: Vec<Value> = [
        ("haematite.database", "wards"),
        ("haematite.collection", "wards.patients"),
        ("haematite.branch", "wards.main"),
    ]
    .into_iter()
    .map(|(kind, id)| {
        json!({"resource": {"kind": kind, "id": id}, "actions": ["read", "query", "write"],
            "mode": "outright", "grant": "author-grant"})
    })
    .collect();
    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": kind, "responsible": responsible},
            "rights": rights,
        })
        .to_string(),
    );
    let signature = URL_SAFE_NO_PAD.encode(
        signer()
            .sign(format!("{header}.{claims}").as_bytes())
            .to_bytes(),
    );
    format!("{header}.{claims}.{signature}")
}

/// A store of one shard on a socket in a private folder under `directory`,
/// served until the test's process ends; through Lys when `lys` is set.
fn served(directory: &Path, lys: bool) -> Result<PathBuf, Box<dyn Error>> {
    let private = directory.join("store");
    std::fs::create_dir(&private)?;
    std::fs::set_permissions(&private, std::fs::Permissions::from_mode(0o700))?;
    let storage = Storage::create(
        DatabaseConfig {
            data_dir: private.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)?;
    let service = Arc::new(if lys {
        service.with_lys(Gate::new(&keys(), ISSUER, "wards", Arc::new(Unreachable))?)
    } else {
        service
    });
    let socket = private.join("s.sock");
    let listener = Listener::bind(&socket)?;
    let limits = Limits {
        max_frame: NonZeroU32::MIN.saturating_add(65_535),
        max_clients: NonZeroUsize::MIN.saturating_add(3),
        frame_timeout: Duration::from_secs(60),
    };
    std::thread::spawn(move || listener.run(&service, limits));
    Ok(socket)
}

/// The login name of this process's effective user, as `id` says it.
fn login() -> Result<String, Box<dyn Error>> {
    let output = std::process::Command::new("/usr/bin/id")
        .arg("-un")
        .output()?;
    assert!(output.status.success(), "id -un failed");
    Ok(String::from_utf8(output.stdout)?.trim_end().to_owned())
}

fn raw(value: &Value) -> Result<Box<RawValue>, Box<dyn Error>> {
    Ok(serde_json::value::to_raw_value(value)?)
}

/// The answer to `method` with `params`, or the store's refusal as an error.
fn answered(client: &mut Client, method: &str, params: &Value) -> Result<Value, Box<dyn Error>> {
    match client.call(method, Some(&raw(params)?))? {
        Reply::Answered(answer) => Ok(serde_json::from_str(answer.get())?),
        Reply::Refused(refusal) => {
            Err(format!("{method} refused: {}: {}", refusal.code, refusal.message).into())
        }
    }
}

fn head(client: &mut Client) -> Result<Value, Box<dyn Error>> {
    let answer = answered(client, "branches.list", &json!({"limit": 10}))?;
    answer["entries"]
        .as_array()
        .ok_or("branches has no entries")?
        .iter()
        .find(|row| row["branch"] == "main")
        .map(|row| row["version"].clone())
        .ok_or_else(|| "main is missing".into())
}

/// Apply `change` to main at its head as `request_id`, answering the version
/// the receipt says it was written at.
fn apply(client: &mut Client, request_id: &str, change: &Value) -> Result<Value, Box<dyn Error>> {
    let expected = head(client)?;
    let receipt = answered(
        client,
        "mutation.apply",
        &json!({"request_id": request_id, "branch": "main", "expected": expected, "change": change}),
    )?;
    Ok(receipt["version"].clone())
}

fn patients(client: &mut Client) -> Result<Value, Box<dyn Error>> {
    apply(
        client,
        "make-patients",
        &json!({"kind": "collection.put", "value": {"collection": "patients", "schema": null}}),
    )
}

fn put(
    client: &mut Client,
    request_id: &str,
    id: &str,
    document: &Value,
) -> Result<Value, Box<dyn Error>> {
    apply(
        client,
        request_id,
        &json!({"kind": "document.put", "value": {"collection": "patients", "document_id": id, "document": document}}),
    )
}

/// The authors history answers beside `version` on main.
fn authors(client: &mut Client, version: &Value) -> Result<Value, Box<dyn Error>> {
    let history = answered(
        client,
        "history.list",
        &json!({"branch": "main", "after": null, "limit": 10}),
    )?;
    let entry = history["entries"]
        .as_array()
        .ok_or("history has no entries")?
        .iter()
        .find(|entry| &entry["version"] == version)
        .ok_or_else(|| format!("history does not hold {version}: {history}"))?;
    entry
        .get("authors")
        .cloned()
        .ok_or_else(|| format!("history answers no author beside {version}: {entry}").into())
}

/// Fails while history answers no author for a put through a store with no
/// [lys] block, or names anyone but the owner of the store's socket.
#[test]
fn a_put_through_a_store_with_no_lys_records_the_socket_owner() -> TestResult {
    let directory = tempfile::tempdir()?;
    let socket = served(directory.path(), false)?;
    let mut client = Client::connect(&socket, STEP)?;
    patients(&mut client)?;
    let written = put(&mut client, "put-p1", "p1", &json!({"name": "Ada"}))?;
    assert_eq!(
        authors(&mut client, &written)?,
        json!([{"request_id": "put-p1", "caller": {"issuer": "local", "subject": login()?}}])
    );
    Ok(())
}

/// Fails while a put through a Lys-served store records anyone but the
/// accepted pass's holder, or an agent's put loses its responsible person.
#[test]
fn a_put_through_lys_records_the_pass_holder_and_an_agents_person() -> TestResult {
    let directory = tempfile::tempdir()?;
    let socket = served(directory.path(), true)?;
    let mut person = Client::connect(&socket, STEP)?;
    answered(
        &mut person,
        "session.pass",
        &json!({"pass": pass("clinician", "person", None)}),
    )?;
    patients(&mut person)?;
    let by_person = put(&mut person, "put-person", "p1", &json!({"name": "Ada"}))?;
    assert_eq!(
        authors(&mut person, &by_person)?,
        json!([{"request_id": "put-person", "caller": {"issuer": ISSUER, "subject": "clinician"}}])
    );
    let mut agent = Client::connect(&socket, STEP)?;
    answered(
        &mut agent,
        "session.pass",
        &json!({"pass": pass("agent-7", "agent", Some("clinician"))}),
    )?;
    let by_agent = put(&mut agent, "put-agent", "p2", &json!({"name": "Grace"}))?;
    assert_eq!(
        authors(&mut agent, &by_agent)?,
        json!([{"request_id": "put-agent", "caller":
            {"issuer": ISSUER, "subject": "agent-7", "responsible": "clinician"}}])
    );
    Ok(())
}

/// Fails while a name the request carries becomes the recorded author: a
/// document naming an author is content, and the store's own author is
/// recorded; a change naming its own caller is refused and records nothing.
#[test]
fn a_request_naming_an_author_never_becomes_the_author() -> TestResult {
    let directory = tempfile::tempdir()?;
    let socket = served(directory.path(), false)?;
    let mut client = Client::connect(&socket, STEP)?;
    patients(&mut client)?;
    let named = json!({"name": "Ada", "author": {"issuer": "local", "subject": "mallory"},
        "caller": {"issuer": "local", "subject": "mallory"}});
    let written = put(&mut client, "put-named", "p1", &named)?;
    assert_eq!(
        authors(&mut client, &written)?,
        json!([{"request_id": "put-named", "caller": {"issuer": "local", "subject": login()?}}])
    );
    let before = head(&mut client)?;
    let claimed = client.call(
        "mutation.apply",
        Some(&raw(
            &json!({"request_id": "put-claimed", "branch": "main", "expected": before,
            "caller": {"issuer": "local", "subject": "mallory"},
            "change": {"kind": "document.put", "value": {"collection": "patients",
                "document_id": "p2", "document": {"name": "Grace"}}}}),
        )?),
    );
    assert!(
        !matches!(claimed, Ok(Reply::Answered(_))),
        "a change naming its caller was applied: {claimed:?}"
    );
    let mut after = Client::connect(&socket, STEP)?;
    assert_eq!(head(&mut after)?, before, "a refused change moved main");
    Ok(())
}

/// The page bound of the service the largest-batch test runs: the walk's and
/// README's, whose batches reach `max_page × 64` = 6,400 ids.
const MOST_PAGE: NonZeroU32 = NonZeroU32::MIN.saturating_add(99);

/// Dispatch `method` with `params` on `service` as its socket's owner,
/// answering the result or the refusal as an error.
fn dispatched(service: &Service, method: &str, params: &Value) -> Result<Value, Box<dyn Error>> {
    let operation = serde_json::from_value(json!({"method": method, "params": params}))?;
    service.dispatch(operation).map_err(|failure| {
        serde_json::to_string(&failure)
            .unwrap_or_else(|error| error.to_string())
            .into()
    })
}

fn applied(service: &Service, request_id: &str, change: &Value) -> Result<Value, Box<dyn Error>> {
    let branches = dispatched(service, "branches.list", &json!({"limit": 10}))?;
    let expected = branches["entries"][0]["version"].clone();
    let receipt = dispatched(
        service,
        "mutation.apply",
        &json!({"request_id": request_id, "branch": "main", "expected": expected, "change": change}),
    )?;
    Ok(receipt["version"].clone())
}

/// Fails while a delete of the most ids the service takes in one batch is
/// refused for the size of its author row, or history does not name its
/// author: an author row records who made the change, not what it changed.
#[test]
fn a_delete_of_the_most_ids_a_service_takes_commits_and_names_its_author() -> TestResult {
    let directory = tempfile::tempdir()?;
    let storage = Storage::create(
        DatabaseConfig {
            data_dir: directory.path().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, MOST_PAGE)?;
    let most = usize::try_from(MOST_PAGE.get())? * 64;
    let ids: Vec<String> = (0..most)
        .map(|index| format!("record-{index:05}"))
        .collect();
    applied(
        &service,
        "make-patients",
        &json!({"kind": "collection.put", "value": {"collection": "patients", "schema": null}}),
    )?;
    let documents: Vec<Value> = ids
        .iter()
        .map(|id| json!({"document_id": id, "document": {"name": id}}))
        .collect();
    applied(
        &service,
        "import",
        &json!({"kind": "documents.put_many", "value": {"collection": "patients", "documents": documents}}),
    )?;
    let deleted = applied(
        &service,
        "delete-most",
        &json!({"kind": "documents.delete_many", "value": {"collection": "patients", "document_ids": ids}}),
    )?;
    let history = dispatched(
        &service,
        "history.list",
        &json!({"branch": "main", "after": null, "limit": 10}),
    )?;
    let entry = history["entries"]
        .as_array()
        .ok_or("history has no entries")?
        .iter()
        .find(|entry| entry["version"] == deleted)
        .ok_or_else(|| format!("history does not hold {deleted}: {history}"))?;
    assert_eq!(
        entry["authors"],
        json!([{"request_id": "delete-most", "caller": {"issuer": "local", "subject": login()?}}])
    );
    Ok(())
}