geth/crates/geth/tests/bootstrap.rs

use std::process::{Child, Command};
use std::time::{Duration, Instant};

fn skip_iroh_integration_tests() -> bool {
    std::env::var_os("GETH_TEST_SKIP_IROH").is_some()
}

fn unix_sockets_available(home: &std::path::Path) -> bool {
    let probe = home.join("probe.sock");
    match std::os::unix::net::UnixListener::bind(&probe) {
        Ok(listener) => {
            drop(listener);
            let _ = std::fs::remove_file(probe);
            true
        }
        Err(error) => {
            eprintln!("skipping daemon socket test; Unix sockets unavailable: {error}");
            false
        }
    }
}

fn geth_bin() -> &'static str {
    env!("CARGO_BIN_EXE_geth")
}

fn run_geth(home: &std::path::Path, args: &[&str]) -> std::process::Output {
    Command::new(geth_bin())
        .env("GETH_HOME", home)
        .args(args)
        .output()
        .expect("run geth")
}

fn json_fixture(name: &str) -> serde_json::Value {
    let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
        .join("tests/fixtures/json")
        .join(name);
    serde_json::from_str(&std::fs::read_to_string(&path).expect("read json fixture"))
        .expect("decode json fixture")
}

fn command_json(output: &std::process::Output) -> serde_json::Value {
    assert!(
        output.status.success(),
        "stderr: {}",
        String::from_utf8_lossy(&output.stderr)
    );
    serde_json::from_slice(&output.stdout).expect("decode command json")
}

fn normalize_node_id_json(mut value: serde_json::Value) -> serde_json::Value {
    let object = value.as_object_mut().expect("node id json object");
    object.insert(
        "agent_id".to_owned(),
        serde_json::Value::String("agent:<normalized>".to_owned()),
    );
    object.insert(
        "node_id".to_owned(),
        serde_json::Value::String("node:<normalized>".to_owned()),
    );
    object.insert(
        "endpoint_id".to_owned(),
        serde_json::Value::String("<endpoint-id>".to_owned()),
    );
    value
}

fn spawn_daemon(home: &std::path::Path) -> Child {
    Command::new(geth_bin())
        .env("GETH_HOME", home)
        .args(["daemon", "run"])
        .spawn()
        .expect("spawn daemon")
}

fn ssh_keygen_available() -> bool {
    Command::new("ssh-keygen").arg("-?").output().is_ok()
}

fn generate_ssh_key(path: &std::path::Path) {
    let status = Command::new("ssh-keygen")
        .arg("-q")
        .arg("-t")
        .arg("ed25519")
        .arg("-N")
        .arg("")
        .arg("-f")
        .arg(path)
        .status()
        .expect("generate ssh key");
    assert!(status.success());
}

fn sign_payload_with_ssh_key(
    dir: &std::path::Path,
    stem: &str,
    private_key_path: &std::path::Path,
    namespace: &str,
    payload: &[u8],
) -> Vec<u8> {
    let payload_path = dir.join(format!("{stem}.payload"));
    std::fs::write(&payload_path, payload).expect("write signing payload");
    let output = geth_ssh_identity::sign_command(private_key_path, namespace, &payload_path)
        .output()
        .expect("run ssh-keygen sign");
    assert!(
        output.status.success(),
        "ssh-keygen sign stderr: {}",
        String::from_utf8_lossy(&output.stderr)
    );
    let signature_path =
        std::path::Path::new(&format!("{}.sig", payload_path.display())).to_path_buf();
    std::fs::read(signature_path).expect("read signature")
}

fn wait_for_socket(path: &std::path::Path) {
    let started = Instant::now();
    while started.elapsed() < Duration::from_secs(5) {
        if path.exists() {
            return;
        }
        std::thread::sleep(Duration::from_millis(50));
    }
    panic!("socket did not appear: {}", path.display());
}

#[test]
fn geth_init_in_temp_home() {
    let home = tempfile::tempdir().expect("tempdir");
    let output = run_geth(home.path(), &["init"]);
    assert!(
        output.status.success(),
        "stderr: {}",
        String::from_utf8_lossy(&output.stderr)
    );
    assert!(home.path().join("geth.sqlite").exists());
    assert!(home.path().join("identity/agent.ed25519").exists());
    assert!(home.path().join("config.toml").exists());
    assert!(
        std::fs::read_to_string(home.path().join("config.toml"))
            .expect("read config")
            .contains("relay_mode = \"default\"")
    );
    assert!(
        std::fs::read_to_string(home.path().join("config.toml"))
            .expect("read config")
            .contains("local_discovery = true")
    );
}

#[test]
fn backup_create_and_restore_use_separate_home_without_private_identity_keys() {
    let home = tempfile::tempdir().expect("tempdir");
    let backup = tempfile::tempdir().expect("backup");
    let restore = tempfile::tempdir().expect("restore");
    let restore_home = restore.path().join("restored-home");

    let init = run_geth(home.path(), &["init"]);
    assert!(
        init.status.success(),
        "stderr: {}",
        String::from_utf8_lossy(&init.stderr)
    );
    let create = run_geth(
        home.path(),
        &[
            "--json",
            "backup",
            "create",
            "--out",
            backup.path().to_str().expect("backup path"),
        ],
    );
    let create_json = command_json(&create);
    assert_eq!(create_json["type"], "backup-created");
    assert!(backup.path().join("manifest.json").exists());
    assert!(backup.path().join("home/config.toml").exists());
    assert!(backup.path().join("home/geth.sqlite").exists());
    assert!(!backup.path().join("home/identity/agent.ed25519").exists());
    assert!(
        create_json["manifest"]["excludes"]
            .as_array()
            .expect("excludes")
            .iter()
            .any(|entry| entry
                .as_str()
                .expect("exclude")
                .contains("private SSH admin keys"))
    );

    let restore_output = run_geth(
        home.path(),
        &[
            "--json",
            "backup",
            "restore",
            backup.path().to_str().expect("backup path"),
            "--target-home",
            restore_home.to_str().expect("restore path"),
        ],
    );
    let restore_json = command_json(&restore_output);
    assert_eq!(restore_json["type"], "backup-restored");
    assert!(restore_home.join("config.toml").exists());
    assert!(restore_home.join("geth.sqlite").exists());
    assert!(!restore_home.join("identity/agent.ed25519").exists());
}

#[test]
fn cli_help_documents_owner_init_keys() {
    let home = tempfile::tempdir().expect("tempdir");
    let output = run_geth(home.path(), &["init", "--help"]);
    assert!(
        output.status.success(),
        "stderr: {}",
        String::from_utf8_lossy(&output.stderr)
    );
    let stdout = String::from_utf8_lossy(&output.stdout);
    assert!(stdout.contains("--admin-key <OPENSSH_PUBLIC_KEY>"));
    assert!(stdout.contains("--signing-key <OPENSSH_PRIVATE_KEY>"));
    assert!(stdout.contains("OpenSSH public key"));
    assert!(stdout.contains("YubiKey"));
    assert!(stdout.contains("both `--admin-key` and `--signing-key` are"));
    assert!(stdout.contains("geth guide owner-setup"));
}

#[test]
fn guide_command_explains_key_roles() {
    let home = tempfile::tempdir().expect("tempdir");
    let output = run_geth(home.path(), &["guide", "keys"]);
    assert!(
        output.status.success(),
        "stderr: {}",
        String::from_utf8_lossy(&output.stderr)
    );
    let stdout = String::from_utf8_lossy(&output.stdout);
    assert!(stdout.contains("--admin-key"));
    assert!(stdout.contains("trust anchor"));
    assert!(stdout.contains("--signing-key"));
    assert!(stdout.contains("ssh-keygen -Y sign"));
}

#[test]
fn guide_command_explains_overlay_boundaries() {
    let home = tempfile::tempdir().expect("tempdir");
    let output = run_geth(home.path(), &["guide", "overlay"]);
    assert!(
        output.status.success(),
        "stderr: {}",
        String::from_utf8_lossy(&output.stderr)
    );
    let stdout = String::from_utf8_lossy(&output.stdout);
    assert!(stdout.contains("geth overlay plan home"));
    assert!(stdout.contains("TUN/Wintun"));
    assert!(stdout.contains("all overlay packets must be carried over Iroh"));
}

#[test]
fn completions_are_generated_without_daemon() {
    let home = tempfile::tempdir().expect("tempdir");
    let output = run_geth(home.path(), &["completions", "bash"]);
    assert!(
        output.status.success(),
        "stderr: {}",
        String::from_utf8_lossy(&output.stderr)
    );
    let stdout = String::from_utf8_lossy(&output.stdout);
    assert!(stdout.contains("_geth"));
    assert!(stdout.contains("overlay"));
    assert!(stdout.contains("completions"));
}

#[test]
fn guide_command_explains_completion_installation() {
    let home = tempfile::tempdir().expect("tempdir");
    let output = run_geth(home.path(), &["guide", "completions"]);
    assert!(
        output.status.success(),
        "stderr: {}",
        String::from_utf8_lossy(&output.stderr)
    );
    let stdout = String::from_utf8_lossy(&output.stdout);
    assert!(stdout.contains("geth completions bash"));
    assert!(stdout.contains("PowerShell"));
    assert!(stdout.contains("fish"));
}

#[test]
fn owner_init_requires_admin_key_and_signing_key() {
    let home = tempfile::tempdir().expect("tempdir");
    let output = run_geth(home.path(), &["init", "--node-name", "laptop"]);
    assert!(!output.status.success());
    assert!(
        String::from_utf8_lossy(&output.stderr).contains("owner init requires --admin-key"),
        "stderr: {}",
        String::from_utf8_lossy(&output.stderr)
    );
    assert!(!home.path().join("geth.sqlite").exists());
}

#[test]
fn geth_status_against_running_daemon() {
    let home = tempfile::tempdir().expect("tempdir");
    if !unix_sockets_available(home.path()) {
        return;
    }
    assert!(run_geth(home.path(), &["init"]).status.success());
    std::fs::write(
        home.path().join("config.toml"),
        "[iroh]\nrelay_mode = \"disabled\"\nlocal_discovery = false\n",
    )
    .expect("write config");
    let mut daemon = spawn_daemon(home.path());
    wait_for_socket(&home.path().join("run/geth.sock"));

    let output = run_geth(home.path(), &["status"]);
    let json_output = run_geth(home.path(), &["--json", "status"]);
    let wait_output = run_geth(
        home.path(),
        &[
            "--json",
            "wait",
            "daemon",
            "--timeout-ms",
            "1000",
            "--interval-ms",
            "10",
        ],
    );
    let _ = daemon.kill();
    let _ = daemon.wait();

    assert!(
        output.status.success(),
        "stderr: {}",
        String::from_utf8_lossy(&output.stderr)
    );
    let stdout = String::from_utf8_lossy(&output.stdout);
    assert!(stdout.contains("geth daemon: running"));
    assert!(stdout.contains("agent:"));
    assert!(stdout.contains("uptime seconds:"));
    assert!(stdout.contains("store schema:"));
    assert!(stdout.contains("store journal: wal"));
    assert!(stdout.contains("store synchronous: normal"));
    assert!(stdout.contains("store status: ok"));
    assert!(stdout.contains("endpoint:"));
    assert!(stdout.contains("iroh relay: disabled"));
    assert!(stdout.contains("iroh discovery: local-network disabled"));
    assert!(stdout.contains(
        "native backend cas: shared-iroh-endpoint-ready target iroh-blobs 0.103.0 (wired)"
    ));
    assert!(stdout.contains(
        "native backend kv: shared-iroh-endpoint-ready target iroh-docs 0.101.0 (wired)"
    ));
    assert!(stdout.contains(
        "native backend pubsub: shared-iroh-endpoint-ready target iroh-gossip 0.101.0 (wired)"
    ));

    let status_json = command_json(&json_output);
    assert_eq!(status_json["type"], "status");
    assert!(
        status_json["daemon_started_at_ms"]
            .as_i64()
            .expect("daemon started at ms")
            > 0
    );
    assert!(
        status_json["daemon_uptime_seconds"]
            .as_u64()
            .expect("daemon uptime seconds")
            < 60
    );
    assert_eq!(status_json["store_journal_mode"], "wal");
    assert_eq!(status_json["store_synchronous"], "normal");
    assert_eq!(status_json["store_status"], "ok");
    assert!(status_json["native_backends"].is_array());

    let wait_json = command_json(&wait_output);
    assert_eq!(wait_json["type"], "wait");
    assert_eq!(wait_json["target"], "daemon");
    assert_eq!(wait_json["ready"], true);
}

#[test]
fn stable_json_output_matches_fixtures() {
    let home = tempfile::tempdir().expect("tempdir");
    if !unix_sockets_available(home.path()) {
        return;
    }
    assert!(run_geth(home.path(), &["init"]).status.success());
    std::fs::write(
        home.path().join("config.toml"),
        "[iroh]\nrelay_mode = \"disabled\"\nlocal_discovery = false\n",
    )
    .expect("write config");

    let mut daemon = spawn_daemon(home.path());
    wait_for_socket(&home.path().join("run/geth.sock"));

    let node_id = run_geth(home.path(), &["--json", "node", "id"]);
    let resource_list = run_geth(home.path(), &["--json", "resource", "list"]);
    let sync_status = run_geth(home.path(), &["--json", "sync", "status"]);

    let _ = daemon.kill();
    let _ = daemon.wait();

    assert_eq!(
        normalize_node_id_json(command_json(&node_id)),
        json_fixture("node-id.json")
    );
    assert_eq!(
        command_json(&resource_list),
        json_fixture("resource-list-fresh.json")
    );
    assert_eq!(
        command_json(&sync_status),
        json_fixture("sync-status-empty.json")
    );
}

#[test]
fn json_errors_include_stable_code_for_common_failures() {
    let home = tempfile::tempdir().expect("tempdir");
    let output = run_geth(home.path(), &["--json", "status"]);

    assert!(!output.status.success());
    let error = serde_json::from_slice::<serde_json::Value>(&output.stdout)
        .expect("decode json error output");
    assert_eq!(error["type"], "error");
    assert_eq!(error["code"], "daemon_unavailable");
    assert!(
        error["message"]
            .as_str()
            .expect("error message")
            .contains("connect to daemon")
    );
}

#[test]
fn wait_daemon_reports_json_timeout_without_daemon() {
    let home = tempfile::tempdir().expect("tempdir");
    let output = run_geth(
        home.path(),
        &[
            "--json",
            "wait",
            "daemon",
            "--timeout-ms",
            "1",
            "--interval-ms",
            "1",
        ],
    );

    assert!(!output.status.success());
    let report =
        serde_json::from_slice::<serde_json::Value>(&output.stdout).expect("decode wait report");
    assert_eq!(report["type"], "wait");
    assert_eq!(report["target"], "daemon");
    assert_eq!(report["ready"], false);
    assert!(report["attempts"].as_u64().expect("attempts") >= 1);
}

#[test]
fn peer_ping_uses_daemon_owned_iroh_endpoint() {
    if skip_iroh_integration_tests() {
        eprintln!("skipping Iroh integration test because GETH_TEST_SKIP_IROH is set");
        return;
    }
    let left_home = tempfile::tempdir().expect("left tempdir");
    let right_home = tempfile::tempdir().expect("right tempdir");
    if !unix_sockets_available(left_home.path()) || !unix_sockets_available(right_home.path()) {
        return;
    }
    assert!(run_geth(left_home.path(), &["init"]).status.success());
    assert!(run_geth(right_home.path(), &["init"]).status.success());
    for home in [left_home.path(), right_home.path()] {
        std::fs::write(
            home.join("config.toml"),
            "[iroh]\nrelay_mode = \"disabled\"\nlocal_discovery = false\n",
        )
        .expect("write config");
    }

    let mut left_daemon = spawn_daemon(left_home.path());
    let mut right_daemon = spawn_daemon(right_home.path());
    wait_for_socket(&left_home.path().join("run/geth.sock"));
    wait_for_socket(&right_home.path().join("run/geth.sock"));

    let left_card = left_home.path().join("left-peer-card.json");
    let right_card = right_home.path().join("right-peer-card.json");
    let left_export = run_geth(
        left_home.path(),
        &["peer", "export", "--out", left_card.to_str().unwrap()],
    );
    let right_export = run_geth(
        right_home.path(),
        &["peer", "export", "--out", right_card.to_str().unwrap()],
    );
    assert!(
        left_export.status.success(),
        "left export stderr: {}",
        String::from_utf8_lossy(&left_export.stderr)
    );
    assert!(
        right_export.status.success(),
        "right export stderr: {}",
        String::from_utf8_lossy(&right_export.stderr)
    );
    let import = run_geth(
        left_home.path(),
        &["peer", "import", right_card.to_str().unwrap()],
    );
    assert!(
        import.status.success(),
        "import stderr: {}",
        String::from_utf8_lossy(&import.stderr)
    );
    let right_card_json =
        std::fs::read_to_string(&right_card).expect("read right exported peer card");
    let right_card: geth_discovery::PeerCard =
        serde_json::from_str(&right_card_json).expect("decode right peer card");
    assert!(
        !right_card.endpoints[0].direct_addresses.is_empty(),
        "exported peer card should include at least one direct Iroh address"
    );

    let ping = run_geth(
        left_home.path(),
        &["peer", "ping", right_card.node_id.as_str()],
    );
    let _ = left_daemon.kill();
    let _ = right_daemon.kill();
    let _ = left_daemon.wait();
    let _ = right_daemon.wait();

    assert!(
        ping.status.success(),
        "ping stderr: {}",
        String::from_utf8_lossy(&ping.stderr)
    );
    let stdout = String::from_utf8_lossy(&ping.stdout);
    assert!(stdout.contains("peer pong:"));
    assert!(stdout.contains("candidate status does not grant resource capabilities"));
}

#[test]
fn sync_now_completes_owner_approved_node_enrollment_flow() {
    if skip_iroh_integration_tests() {
        eprintln!("skipping Iroh integration test because GETH_TEST_SKIP_IROH is set");
        return;
    }
    if !ssh_keygen_available() {
        return;
    }

    let owner_home = tempfile::tempdir().expect("owner tempdir");
    let requester_home = tempfile::tempdir().expect("requester tempdir");
    if !unix_sockets_available(owner_home.path()) || !unix_sockets_available(requester_home.path())
    {
        return;
    }

    let owner_key = owner_home.path().join("owner_ed25519");
    generate_ssh_key(&owner_key);
    let owner_pub = owner_key.with_extension("pub");
    let owner_pub_arg = owner_pub.to_str().expect("owner pub path");
    let owner_key_arg = owner_key.to_str().expect("owner key path");

    let owner_init = run_geth(
        owner_home.path(),
        &[
            "init",
            "--admin-key",
            owner_pub_arg,
            "--signing-key",
            owner_key_arg,
            "--owner",
            "Eric",
            "--node-name",
            "owner-laptop",
        ],
    );
    assert!(
        owner_init.status.success(),
        "owner init stderr: {}",
        String::from_utf8_lossy(&owner_init.stderr)
    );
    let requester_init = run_geth(requester_home.path(), &["init"]);
    assert!(
        requester_init.status.success(),
        "requester init stderr: {}",
        String::from_utf8_lossy(&requester_init.stderr)
    );
    for home in [owner_home.path(), requester_home.path()] {
        std::fs::write(
            home.join("config.toml"),
            "[iroh]\nrelay_mode = \"disabled\"\nlocal_discovery = false\n",
        )
        .expect("write config");
    }

    let mut owner_daemon = spawn_daemon(owner_home.path());
    let mut requester_daemon = spawn_daemon(requester_home.path());
    wait_for_socket(&owner_home.path().join("run/geth.sock"));
    wait_for_socket(&requester_home.path().join("run/geth.sock"));

    let requester_trust = run_geth(
        requester_home.path(),
        &["keychain", "init", "--admin-key", owner_pub_arg],
    );
    assert!(
        requester_trust.status.success(),
        "requester trust stderr: {}",
        String::from_utf8_lossy(&requester_trust.stderr)
    );

    let owner_card_path = owner_home.path().join("owner-peer-card.json");
    let requester_card_path = requester_home.path().join("requester-peer-card.json");
    let owner_export = run_geth(
        owner_home.path(),
        &[
            "peer",
            "export",
            "--out",
            owner_card_path.to_str().expect("owner card path"),
        ],
    );
    let requester_export = run_geth(
        requester_home.path(),
        &[
            "peer",
            "export",
            "--out",
            requester_card_path.to_str().expect("requester card path"),
        ],
    );
    assert!(
        owner_export.status.success(),
        "owner export stderr: {}",
        String::from_utf8_lossy(&owner_export.stderr)
    );
    assert!(
        requester_export.status.success(),
        "requester export stderr: {}",
        String::from_utf8_lossy(&requester_export.stderr)
    );
    let owner_card_json = std::fs::read_to_string(&owner_card_path).expect("read owner card");
    let owner_card: geth_discovery::PeerCard =
        serde_json::from_str(&owner_card_json).expect("decode owner card");
    let requester_card_json =
        std::fs::read_to_string(&requester_card_path).expect("read requester card");
    let requester_card: geth_discovery::PeerCard =
        serde_json::from_str(&requester_card_json).expect("decode requester card");

    let import_owner = run_geth(
        requester_home.path(),
        &[
            "peer",
            "import",
            owner_card_path.to_str().expect("owner card path"),
        ],
    );
    let import_requester = run_geth(
        owner_home.path(),
        &[
            "peer",
            "import",
            requester_card_path.to_str().expect("requester card path"),
        ],
    );
    assert!(
        import_owner.status.success(),
        "requester import owner stderr: {}",
        String::from_utf8_lossy(&import_owner.stderr)
    );
    assert!(
        import_requester.status.success(),
        "owner import requester stderr: {}",
        String::from_utf8_lossy(&import_requester.stderr)
    );

    let enroll_request = run_geth(
        requester_home.path(),
        &[
            "node",
            "enroll",
            "request",
            "--node-name",
            "workstation",
            "--capability",
            "resource:ssh-proxy:local=ssh_proxy.connect",
        ],
    );
    assert!(
        enroll_request.status.success(),
        "enroll request stderr: {}",
        String::from_utf8_lossy(&enroll_request.stderr)
    );
    let enroll_submit = run_geth(
        requester_home.path(),
        &["node", "enroll", "submit", owner_card.node_id.as_str()],
    );
    assert!(
        enroll_submit.status.success(),
        "enroll submit stderr: {}",
        String::from_utf8_lossy(&enroll_submit.stderr)
    );
    assert!(String::from_utf8_lossy(&enroll_submit.stdout).contains("accepted: true"));

    let enroll_list = run_geth(owner_home.path(), &["node", "enroll", "list"]);
    assert!(
        enroll_list.status.success(),
        "enroll list stderr: {}",
        String::from_utf8_lossy(&enroll_list.stderr)
    );
    let enroll_list_stdout = String::from_utf8_lossy(&enroll_list.stdout);
    let request_id = enroll_list_stdout
        .lines()
        .find_map(|line| {
            let mut fields = line.split('\t');
            let id = fields.next()?;
            if id.starts_with("node-enrollment:") {
                Some(id.to_owned())
            } else {
                None
            }
        })
        .expect("request id in enrollment list");

    let approve = run_geth(
        owner_home.path(),
        &[
            "node",
            "enroll",
            "approve",
            &request_id,
            "--signing-key",
            owner_key_arg,
            "--capability",
            "resource:cas:local=cas.fetch",
        ],
    );
    assert!(
        approve.status.success(),
        "approve stderr: {}",
        String::from_utf8_lossy(&approve.stderr)
    );

    let sync = run_geth(
        requester_home.path(),
        &["sync", "now", owner_card.node_id.as_str()],
    );
    assert!(
        sync.status.success(),
        "sync now stderr: {}",
        String::from_utf8_lossy(&sync.stderr)
    );
    let sync_stdout = String::from_utf8_lossy(&sync.stdout);
    assert!(sync_stdout.contains("keychain\tok"));
    assert!(sync_stdout.contains("auth\tok"));

    let requester_nodes = run_geth(requester_home.path(), &["node", "list"]);
    assert!(
        requester_nodes.status.success(),
        "requester node list stderr: {}",
        String::from_utf8_lossy(&requester_nodes.stderr)
    );
    let requester_nodes_stdout = String::from_utf8_lossy(&requester_nodes.stdout);
    assert!(requester_nodes_stdout.contains("owner-laptop"));
    assert!(requester_nodes_stdout.contains("workstation"));

    let explain = run_geth(
        requester_home.path(),
        &[
            "auth",
            "explain",
            requester_card.node_id.as_str(),
            "resource:ssh-proxy:local",
            "ssh_proxy.connect",
        ],
    );
    assert!(
        explain.status.success(),
        "auth explain stderr: {}",
        String::from_utf8_lossy(&explain.stderr)
    );
    assert!(String::from_utf8_lossy(&explain.stdout).contains("allowed: true"));

    let sync_status = run_geth(requester_home.path(), &["sync", "status"]);
    let _ = owner_daemon.kill();
    let _ = requester_daemon.kill();
    let _ = owner_daemon.wait();
    let _ = requester_daemon.wait();

    assert!(
        sync_status.status.success(),
        "sync status stderr: {}",
        String::from_utf8_lossy(&sync_status.stderr)
    );
    let sync_status_stdout = String::from_utf8_lossy(&sync_status.stdout);
    assert!(sync_status_stdout.contains("peer:"));
    assert!(sync_status_stdout.contains("keychain"));
    assert!(sync_status_stdout.contains("auth"));
}

#[test]
fn denied_remote_operations_do_not_mutate_serving_node_state() {
    if skip_iroh_integration_tests() {
        eprintln!("skipping Iroh integration test because GETH_TEST_SKIP_IROH is set");
        return;
    }
    let left_home = tempfile::tempdir().expect("left tempdir");
    let right_home = tempfile::tempdir().expect("right tempdir");
    if !unix_sockets_available(left_home.path()) || !unix_sockets_available(right_home.path()) {
        return;
    }
    assert!(run_geth(left_home.path(), &["init"]).status.success());
    assert!(run_geth(right_home.path(), &["init"]).status.success());
    for home in [left_home.path(), right_home.path()] {
        std::fs::write(
            home.join("config.toml"),
            "[iroh]\nrelay_mode = \"disabled\"\nlocal_discovery = false\n",
        )
        .expect("write config");
    }

    let mut left_daemon = spawn_daemon(left_home.path());
    let mut right_daemon = spawn_daemon(right_home.path());
    wait_for_socket(&left_home.path().join("run/geth.sock"));
    wait_for_socket(&right_home.path().join("run/geth.sock"));

    let left_card_path = left_home.path().join("left-peer-card.json");
    let right_card_path = right_home.path().join("right-peer-card.json");
    let left_export = run_geth(
        left_home.path(),
        &[
            "peer",
            "export",
            "--out",
            left_card_path.to_str().expect("left card path"),
        ],
    );
    let right_export = run_geth(
        right_home.path(),
        &[
            "peer",
            "export",
            "--out",
            right_card_path.to_str().expect("right card path"),
        ],
    );
    assert!(
        left_export.status.success(),
        "left export stderr: {}",
        String::from_utf8_lossy(&left_export.stderr)
    );
    assert!(
        right_export.status.success(),
        "right export stderr: {}",
        String::from_utf8_lossy(&right_export.stderr)
    );
    let right_card_json = std::fs::read_to_string(&right_card_path).expect("read right card");
    let right_card: geth_discovery::PeerCard =
        serde_json::from_str(&right_card_json).expect("decode right card");

    let import_right = run_geth(
        left_home.path(),
        &[
            "peer",
            "import",
            right_card_path.to_str().expect("right card path"),
        ],
    );
    let import_left = run_geth(
        right_home.path(),
        &[
            "peer",
            "import",
            left_card_path.to_str().expect("left card path"),
        ],
    );
    assert!(
        import_right.status.success(),
        "left import right stderr: {}",
        String::from_utf8_lossy(&import_right.stderr)
    );
    assert!(
        import_left.status.success(),
        "right import left stderr: {}",
        String::from_utf8_lossy(&import_left.stderr)
    );

    let denied_publish = run_geth(
        left_home.path(),
        &[
            "pubsub",
            "pub",
            "ops",
            "should-not-stick",
            "--node",
            right_card.node_id.as_str(),
        ],
    );
    assert!(
        denied_publish.status.success(),
        "denied publish stderr: {}",
        String::from_utf8_lossy(&denied_publish.stderr)
    );
    let denied_publish_stdout = String::from_utf8_lossy(&denied_publish.stdout);
    assert!(denied_publish_stdout.contains("allowed: false"));
    assert!(denied_publish_stdout.contains("pubsub publish denied"));

    let right_pubsub = run_geth(right_home.path(), &["pubsub", "sub", "ops"]);
    assert!(
        right_pubsub.status.success(),
        "right pubsub sub stderr: {}",
        String::from_utf8_lossy(&right_pubsub.stderr)
    );
    assert!(String::from_utf8_lossy(&right_pubsub.stdout).contains("messages: 0"));

    let denied_pipe_listen = run_geth(
        left_home.path(),
        &[
            "pipe",
            "listen",
            "inbox",
            "--node",
            right_card.node_id.as_str(),
        ],
    );
    assert!(
        denied_pipe_listen.status.success(),
        "denied pipe listen stderr: {}",
        String::from_utf8_lossy(&denied_pipe_listen.stderr)
    );
    let denied_pipe_stdout = String::from_utf8_lossy(&denied_pipe_listen.stdout);
    assert!(denied_pipe_stdout.contains("allowed: false"));
    assert!(denied_pipe_stdout.contains("pipe listen denied"));

    let right_pipe_connect = run_geth(right_home.path(), &["pipe", "connect", "inbox"]);
    assert!(
        right_pipe_connect.status.success(),
        "right pipe connect stderr: {}",
        String::from_utf8_lossy(&right_pipe_connect.stderr)
    );
    assert!(
        String::from_utf8_lossy(&right_pipe_connect.stdout).contains("local_listener_found: false")
    );

    let denied_admin_shell = run_geth(
        left_home.path(),
        &["ssh", "admin-shell", right_card.node_id.as_str(), "status"],
    );
    let _ = left_daemon.kill();
    let _ = right_daemon.kill();
    let _ = left_daemon.wait();
    let _ = right_daemon.wait();

    assert!(
        denied_admin_shell.status.success(),
        "denied admin shell stderr: {}",
        String::from_utf8_lossy(&denied_admin_shell.stderr)
    );
    let denied_admin_stdout = String::from_utf8_lossy(&denied_admin_shell.stdout);
    assert!(denied_admin_stdout.contains("allowed: false"));
    assert!(denied_admin_stdout.contains("ssh admin shell denied"));
}

#[test]
fn unsigned_keychain_and_auth_ops_are_rejected_during_peer_sync() {
    if skip_iroh_integration_tests() {
        eprintln!("skipping Iroh integration test because GETH_TEST_SKIP_IROH is set");
        return;
    }
    let left_home = tempfile::tempdir().expect("left tempdir");
    let right_home = tempfile::tempdir().expect("right tempdir");
    if !unix_sockets_available(left_home.path()) || !unix_sockets_available(right_home.path()) {
        return;
    }
    assert!(run_geth(left_home.path(), &["init"]).status.success());
    assert!(run_geth(right_home.path(), &["init"]).status.success());
    for home in [left_home.path(), right_home.path()] {
        std::fs::write(
            home.join("config.toml"),
            "[iroh]\nrelay_mode = \"disabled\"\nlocal_discovery = false\n",
        )
        .expect("write config");
    }

    let mut left_daemon = spawn_daemon(left_home.path());
    let mut right_daemon = spawn_daemon(right_home.path());
    wait_for_socket(&left_home.path().join("run/geth.sock"));
    wait_for_socket(&right_home.path().join("run/geth.sock"));

    let left_card_path = left_home.path().join("left-peer-card.json");
    let right_card_path = right_home.path().join("right-peer-card.json");
    assert!(
        run_geth(
            left_home.path(),
            &[
                "peer",
                "export",
                "--out",
                left_card_path.to_str().expect("left card path"),
            ],
        )
        .status
        .success()
    );
    assert!(
        run_geth(
            right_home.path(),
            &[
                "peer",
                "export",
                "--out",
                right_card_path.to_str().expect("right card path"),
            ],
        )
        .status
        .success()
    );
    let right_card_json = std::fs::read_to_string(&right_card_path).expect("read right card");
    let right_card: geth_discovery::PeerCard =
        serde_json::from_str(&right_card_json).expect("decode right card");
    assert!(
        run_geth(
            left_home.path(),
            &[
                "peer",
                "import",
                right_card_path.to_str().expect("right card path"),
            ],
        )
        .status
        .success()
    );
    assert!(
        run_geth(
            right_home.path(),
            &[
                "peer",
                "import",
                left_card_path.to_str().expect("left card path"),
            ],
        )
        .status
        .success()
    );

    let unsigned_keychain_op = geth_keychain::KeychainOp {
        id: geth_types::AuthOpId::new("keychain-op:unsigned-user-add"),
        created_at: geth_types::UnixMillis(10),
        kind: geth_keychain::KeychainOpKind::UserAdd {
            user: geth_types::UserId::new("user:unsigned"),
            name: "Unsigned".to_owned(),
        },
    };
    let unsigned_auth_op = geth_auth::AuthOp {
        id: geth_types::AuthOpId::new("auth-op:unsigned-grant"),
        resource: geth_types::ResourceId::new("resource:ssh-proxy:local"),
        created_at: geth_types::UnixMillis(11),
        kind: geth_auth::AuthOpKind::GrantCreate {
            grant_id: "grant:unsigned".to_owned(),
            principal: geth_types::PrincipalId::new("node:unsigned"),
            capabilities: vec![geth_types::Capability::new("ssh_proxy.connect")],
        },
    };
    let right_paths = geth_config::GethPaths::from_home(right_home.path());
    let right_store =
        geth_store::Store::open(&right_paths.metadata_db()).expect("open right store");
    right_store
        .insert_keychain_op(&geth_store::StoredKeychainOp {
            op_id: unsigned_keychain_op.id.to_string(),
            op_json: serde_json::to_string(&unsigned_keychain_op).expect("encode keychain op"),
            created_at_ms: unsigned_keychain_op.created_at.0,
        })
        .expect("insert unsigned keychain op");
    right_store
        .insert_auth_op(&geth_store::StoredAuthOp {
            op_id: unsigned_auth_op.id.to_string(),
            resource_id: unsigned_auth_op.resource.to_string(),
            op_json: serde_json::to_string(&unsigned_auth_op).expect("encode auth op"),
            created_at_ms: unsigned_auth_op.created_at.0,
        })
        .expect("insert unsigned auth op");

    let keychain_sync = run_geth(
        left_home.path(),
        &["keychain", "sync", right_card.node_id.as_str()],
    );
    let auth_sync = run_geth(
        left_home.path(),
        &["auth", "sync", right_card.node_id.as_str()],
    );
    let _ = left_daemon.kill();
    let _ = right_daemon.kill();
    let _ = left_daemon.wait();
    let _ = right_daemon.wait();

    assert!(
        keychain_sync.status.success(),
        "keychain sync stderr: {}",
        String::from_utf8_lossy(&keychain_sync.stderr)
    );
    assert!(
        auth_sync.status.success(),
        "auth sync stderr: {}",
        String::from_utf8_lossy(&auth_sync.stderr)
    );
    assert!(String::from_utf8_lossy(&keychain_sync.stdout).contains("invalid_ops_rejected: 1"));
    assert!(String::from_utf8_lossy(&auth_sync.stdout).contains("invalid_ops_rejected: 1"));

    let left_paths = geth_config::GethPaths::from_home(left_home.path());
    let left_store = geth_store::Store::open(&left_paths.metadata_db()).expect("open left store");
    assert!(
        !left_store
            .list_keychain_ops()
            .expect("list left keychain ops")
            .iter()
            .any(|op| op.op_id == unsigned_keychain_op.id.as_str())
    );
    assert!(
        !left_store
            .list_auth_ops()
            .expect("list left auth ops")
            .iter()
            .any(|op| op.op_id == unsigned_auth_op.id.as_str())
    );
}

#[test]
fn invalidly_signed_keychain_and_auth_ops_are_rejected_during_peer_sync() {
    if skip_iroh_integration_tests() {
        eprintln!("skipping Iroh integration test because GETH_TEST_SKIP_IROH is set");
        return;
    }
    let left_home = tempfile::tempdir().expect("left tempdir");
    let right_home = tempfile::tempdir().expect("right tempdir");
    if !unix_sockets_available(left_home.path()) || !unix_sockets_available(right_home.path()) {
        return;
    }
    assert!(run_geth(left_home.path(), &["init"]).status.success());
    assert!(run_geth(right_home.path(), &["init"]).status.success());
    for home in [left_home.path(), right_home.path()] {
        std::fs::write(
            home.join("config.toml"),
            "[iroh]\nrelay_mode = \"disabled\"\nlocal_discovery = false\n",
        )
        .expect("write config");
    }
    let fake_admin_public_key =
        "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIFakeAdminKeyMaterialForNegativeTests test@geth\n";
    let fake_admin_public_key_path = left_home.path().join("fake-admin.pub");
    std::fs::write(&fake_admin_public_key_path, fake_admin_public_key)
        .expect("write fake admin public key");

    let mut left_daemon = spawn_daemon(left_home.path());
    let mut right_daemon = spawn_daemon(right_home.path());
    wait_for_socket(&left_home.path().join("run/geth.sock"));
    wait_for_socket(&right_home.path().join("run/geth.sock"));

    let trust_admin = run_geth(
        left_home.path(),
        &[
            "keychain",
            "init",
            "--admin-key",
            fake_admin_public_key_path
                .to_str()
                .expect("fake admin public key path"),
        ],
    );
    assert!(
        trust_admin.status.success(),
        "trust admin stderr: {}",
        String::from_utf8_lossy(&trust_admin.stderr)
    );

    let left_card_path = left_home.path().join("left-peer-card.json");
    let right_card_path = right_home.path().join("right-peer-card.json");
    assert!(
        run_geth(
            left_home.path(),
            &[
                "peer",
                "export",
                "--out",
                left_card_path.to_str().expect("left card path"),
            ],
        )
        .status
        .success()
    );
    assert!(
        run_geth(
            right_home.path(),
            &[
                "peer",
                "export",
                "--out",
                right_card_path.to_str().expect("right card path"),
            ],
        )
        .status
        .success()
    );
    let right_card_json = std::fs::read_to_string(&right_card_path).expect("read right card");
    let right_card: geth_discovery::PeerCard =
        serde_json::from_str(&right_card_json).expect("decode right card");
    assert!(
        run_geth(
            left_home.path(),
            &[
                "peer",
                "import",
                right_card_path.to_str().expect("right card path"),
            ],
        )
        .status
        .success()
    );
    assert!(
        run_geth(
            right_home.path(),
            &[
                "peer",
                "import",
                left_card_path.to_str().expect("left card path"),
            ],
        )
        .status
        .success()
    );

    let signer = geth_types::KeyId::new(geth_ssh_identity::ssh_public_key_fingerprint(
        fake_admin_public_key,
    ));
    let keychain_op = geth_keychain::KeychainOp {
        id: geth_types::AuthOpId::new("keychain-op:invalid-signature-user-add"),
        created_at: geth_types::UnixMillis(20),
        kind: geth_keychain::KeychainOpKind::UserAdd {
            user: geth_types::UserId::new("user:invalid-signature"),
            name: "Invalid Signature".to_owned(),
        },
    };
    let auth_op = geth_auth::AuthOp {
        id: geth_types::AuthOpId::new("auth-op:invalid-signature-grant"),
        resource: geth_types::ResourceId::new("resource:ssh-proxy:local"),
        created_at: geth_types::UnixMillis(21),
        kind: geth_auth::AuthOpKind::GrantCreate {
            grant_id: "grant:invalid-signature".to_owned(),
            principal: geth_types::PrincipalId::new("node:invalid-signature"),
            capabilities: vec![geth_types::Capability::new("ssh_proxy.connect")],
        },
    };
    let right_paths = geth_config::GethPaths::from_home(right_home.path());
    let right_store =
        geth_store::Store::open(&right_paths.metadata_db()).expect("open right store");
    right_store
        .insert_keychain_op(&geth_store::StoredKeychainOp {
            op_id: keychain_op.id.to_string(),
            op_json: serde_json::to_string(&keychain_op).expect("encode keychain op"),
            created_at_ms: keychain_op.created_at.0,
        })
        .expect("insert invalidly signed keychain op");
    right_store
        .insert_keychain_signature(&geth_store::StoredKeychainSignature {
            op_id: keychain_op.id.to_string(),
            signer: signer.to_string(),
            signer_public_key: fake_admin_public_key.to_owned(),
            namespace: geth_keychain::KEYCHAIN_SIGNATURE_NAMESPACE.to_owned(),
            signature: b"not a valid openssh signature".to_vec(),
            created_at_ms: 22,
        })
        .expect("insert invalid keychain signature");
    right_store
        .insert_auth_op(&geth_store::StoredAuthOp {
            op_id: auth_op.id.to_string(),
            resource_id: auth_op.resource.to_string(),
            op_json: serde_json::to_string(&auth_op).expect("encode auth op"),
            created_at_ms: auth_op.created_at.0,
        })
        .expect("insert invalidly signed auth op");
    right_store
        .insert_auth_signature(&geth_store::StoredAuthSignature {
            op_id: auth_op.id.to_string(),
            signer: signer.to_string(),
            signer_public_key: fake_admin_public_key.to_owned(),
            namespace: geth_auth::AUTH_SIGNATURE_NAMESPACE.to_owned(),
            signature: b"not a valid openssh signature".to_vec(),
            created_at_ms: 23,
        })
        .expect("insert invalid auth signature");

    let keychain_sync = run_geth(
        left_home.path(),
        &["keychain", "sync", right_card.node_id.as_str()],
    );
    let auth_sync = run_geth(
        left_home.path(),
        &["auth", "sync", right_card.node_id.as_str()],
    );
    let _ = left_daemon.kill();
    let _ = right_daemon.kill();
    let _ = left_daemon.wait();
    let _ = right_daemon.wait();

    assert!(
        keychain_sync.status.success(),
        "keychain sync stderr: {}",
        String::from_utf8_lossy(&keychain_sync.stderr)
    );
    assert!(
        auth_sync.status.success(),
        "auth sync stderr: {}",
        String::from_utf8_lossy(&auth_sync.stderr)
    );
    assert!(String::from_utf8_lossy(&keychain_sync.stdout).contains("invalid_ops_rejected: 1"));
    assert!(String::from_utf8_lossy(&auth_sync.stdout).contains("invalid_ops_rejected: 1"));

    let left_paths = geth_config::GethPaths::from_home(left_home.path());
    let left_store = geth_store::Store::open(&left_paths.metadata_db()).expect("open left store");
    assert!(
        !left_store
            .list_keychain_ops()
            .expect("list left keychain ops")
            .iter()
            .any(|op| op.op_id == keychain_op.id.as_str())
    );
    assert!(
        !left_store
            .list_auth_ops()
            .expect("list left auth ops")
            .iter()
            .any(|op| op.op_id == auth_op.id.as_str())
    );
}

#[test]
fn conflicting_keychain_and_auth_ops_are_rejected_during_peer_sync() {
    if skip_iroh_integration_tests() {
        eprintln!("skipping Iroh integration test because GETH_TEST_SKIP_IROH is set");
        return;
    }
    if !ssh_keygen_available() {
        return;
    }

    let left_home = tempfile::tempdir().expect("left tempdir");
    let right_home = tempfile::tempdir().expect("right tempdir");
    if !unix_sockets_available(left_home.path()) || !unix_sockets_available(right_home.path()) {
        return;
    }
    assert!(run_geth(left_home.path(), &["init"]).status.success());
    assert!(run_geth(right_home.path(), &["init"]).status.success());
    for home in [left_home.path(), right_home.path()] {
        std::fs::write(
            home.join("config.toml"),
            "[iroh]\nrelay_mode = \"disabled\"\nlocal_discovery = false\n",
        )
        .expect("write config");
    }

    let admin_key_path = left_home.path().join("admin_ed25519");
    generate_ssh_key(&admin_key_path);
    let admin_public_key_path = admin_key_path.with_extension("pub");
    let admin_public_key =
        std::fs::read_to_string(&admin_public_key_path).expect("read admin public key");
    let signer = geth_types::KeyId::new(geth_ssh_identity::ssh_public_key_fingerprint(
        &admin_public_key,
    ));

    let mut left_daemon = spawn_daemon(left_home.path());
    let mut right_daemon = spawn_daemon(right_home.path());
    wait_for_socket(&left_home.path().join("run/geth.sock"));
    wait_for_socket(&right_home.path().join("run/geth.sock"));

    let trust_admin = run_geth(
        left_home.path(),
        &[
            "keychain",
            "init",
            "--admin-key",
            admin_public_key_path
                .to_str()
                .expect("admin public key path"),
        ],
    );
    assert!(
        trust_admin.status.success(),
        "trust admin stderr: {}",
        String::from_utf8_lossy(&trust_admin.stderr)
    );

    let left_card_path = left_home.path().join("left-peer-card.json");
    let right_card_path = right_home.path().join("right-peer-card.json");
    assert!(
        run_geth(
            left_home.path(),
            &[
                "peer",
                "export",
                "--out",
                left_card_path.to_str().expect("left card path"),
            ],
        )
        .status
        .success()
    );
    assert!(
        run_geth(
            right_home.path(),
            &[
                "peer",
                "export",
                "--out",
                right_card_path.to_str().expect("right card path"),
            ],
        )
        .status
        .success()
    );
    let right_card_json = std::fs::read_to_string(&right_card_path).expect("read right card");
    let right_card: geth_discovery::PeerCard =
        serde_json::from_str(&right_card_json).expect("decode right card");
    assert!(
        run_geth(
            left_home.path(),
            &[
                "peer",
                "import",
                right_card_path.to_str().expect("right card path"),
            ],
        )
        .status
        .success()
    );
    assert!(
        run_geth(
            right_home.path(),
            &[
                "peer",
                "import",
                left_card_path.to_str().expect("left card path"),
            ],
        )
        .status
        .success()
    );

    let keychain_op_id = geth_types::AuthOpId::new("keychain-op:conflict");
    let local_keychain_op = geth_keychain::KeychainOp {
        id: keychain_op_id.clone(),
        created_at: geth_types::UnixMillis(30),
        kind: geth_keychain::KeychainOpKind::UserAdd {
            user: geth_types::UserId::new("user:local-conflict"),
            name: "Local Conflict".to_owned(),
        },
    };
    let remote_keychain_op = geth_keychain::KeychainOp {
        id: keychain_op_id.clone(),
        created_at: geth_types::UnixMillis(31),
        kind: geth_keychain::KeychainOpKind::UserAdd {
            user: geth_types::UserId::new("user:remote-conflict"),
            name: "Remote Conflict".to_owned(),
        },
    };
    let auth_op_id = geth_types::AuthOpId::new("auth-op:conflict");
    let local_auth_op = geth_auth::AuthOp {
        id: auth_op_id.clone(),
        resource: geth_types::ResourceId::new("resource:ssh-proxy:local"),
        created_at: geth_types::UnixMillis(32),
        kind: geth_auth::AuthOpKind::GrantCreate {
            grant_id: "grant:local-conflict".to_owned(),
            principal: geth_types::PrincipalId::new("node:local-conflict"),
            capabilities: vec![geth_types::Capability::new("ssh_proxy.connect")],
        },
    };
    let remote_auth_op = geth_auth::AuthOp {
        id: auth_op_id.clone(),
        resource: geth_types::ResourceId::new("resource:ssh-proxy:local"),
        created_at: geth_types::UnixMillis(33),
        kind: geth_auth::AuthOpKind::GrantCreate {
            grant_id: "grant:remote-conflict".to_owned(),
            principal: geth_types::PrincipalId::new("node:remote-conflict"),
            capabilities: vec![geth_types::Capability::new("ssh_proxy.admin_shell")],
        },
    };

    let left_paths = geth_config::GethPaths::from_home(left_home.path());
    let right_paths = geth_config::GethPaths::from_home(right_home.path());
    let left_store = geth_store::Store::open(&left_paths.metadata_db()).expect("open left store");
    let right_store =
        geth_store::Store::open(&right_paths.metadata_db()).expect("open right store");
    left_store
        .insert_keychain_op(&geth_store::StoredKeychainOp {
            op_id: local_keychain_op.id.to_string(),
            op_json: serde_json::to_string(&local_keychain_op).expect("encode local keychain op"),
            created_at_ms: local_keychain_op.created_at.0,
        })
        .expect("insert local keychain op");
    left_store
        .insert_auth_op(&geth_store::StoredAuthOp {
            op_id: local_auth_op.id.to_string(),
            resource_id: local_auth_op.resource.to_string(),
            op_json: serde_json::to_string(&local_auth_op).expect("encode local auth op"),
            created_at_ms: local_auth_op.created_at.0,
        })
        .expect("insert local auth op");
    right_store
        .insert_keychain_op(&geth_store::StoredKeychainOp {
            op_id: remote_keychain_op.id.to_string(),
            op_json: serde_json::to_string(&remote_keychain_op).expect("encode remote keychain op"),
            created_at_ms: remote_keychain_op.created_at.0,
        })
        .expect("insert remote keychain op");
    let keychain_signature = sign_payload_with_ssh_key(
        right_home.path(),
        "conflicting-keychain-op",
        &admin_key_path,
        geth_keychain::KEYCHAIN_SIGNATURE_NAMESPACE,
        &geth_keychain::keychain_signing_payload(&remote_keychain_op)
            .expect("keychain signing payload"),
    );
    right_store
        .insert_keychain_signature(&geth_store::StoredKeychainSignature {
            op_id: remote_keychain_op.id.to_string(),
            signer: signer.to_string(),
            signer_public_key: admin_public_key.clone(),
            namespace: geth_keychain::KEYCHAIN_SIGNATURE_NAMESPACE.to_owned(),
            signature: keychain_signature,
            created_at_ms: 34,
        })
        .expect("insert remote keychain signature");
    right_store
        .insert_auth_op(&geth_store::StoredAuthOp {
            op_id: remote_auth_op.id.to_string(),
            resource_id: remote_auth_op.resource.to_string(),
            op_json: serde_json::to_string(&remote_auth_op).expect("encode remote auth op"),
            created_at_ms: remote_auth_op.created_at.0,
        })
        .expect("insert remote auth op");
    let auth_signature = sign_payload_with_ssh_key(
        right_home.path(),
        "conflicting-auth-op",
        &admin_key_path,
        geth_auth::AUTH_SIGNATURE_NAMESPACE,
        &geth_auth::auth_signing_payload(&remote_auth_op).expect("auth signing payload"),
    );
    right_store
        .insert_auth_signature(&geth_store::StoredAuthSignature {
            op_id: remote_auth_op.id.to_string(),
            signer: signer.to_string(),
            signer_public_key: admin_public_key.clone(),
            namespace: geth_auth::AUTH_SIGNATURE_NAMESPACE.to_owned(),
            signature: auth_signature,
            created_at_ms: 35,
        })
        .expect("insert remote auth signature");

    let keychain_sync = run_geth(
        left_home.path(),
        &["keychain", "sync", right_card.node_id.as_str()],
    );
    let auth_sync = run_geth(
        left_home.path(),
        &["auth", "sync", right_card.node_id.as_str()],
    );
    let _ = left_daemon.kill();
    let _ = right_daemon.kill();
    let _ = left_daemon.wait();
    let _ = right_daemon.wait();

    assert!(
        keychain_sync.status.success(),
        "keychain sync stderr: {}",
        String::from_utf8_lossy(&keychain_sync.stderr)
    );
    assert!(
        auth_sync.status.success(),
        "auth sync stderr: {}",
        String::from_utf8_lossy(&auth_sync.stderr)
    );
    assert!(String::from_utf8_lossy(&keychain_sync.stdout).contains("invalid_ops_rejected: 1"));
    assert!(String::from_utf8_lossy(&auth_sync.stdout).contains("invalid_ops_rejected: 1"));

    let left_store = geth_store::Store::open(&left_paths.metadata_db()).expect("open left store");
    let stored_keychain_op = left_store
        .list_keychain_ops()
        .expect("list left keychain ops")
        .into_iter()
        .find(|op| op.op_id == keychain_op_id.as_str())
        .expect("local keychain op remains");
    assert_eq!(
        stored_keychain_op.op_json,
        serde_json::to_string(&local_keychain_op).expect("encode expected local keychain op")
    );
    let stored_auth_op = left_store
        .list_auth_ops()
        .expect("list left auth ops")
        .into_iter()
        .find(|op| op.op_id == auth_op_id.as_str())
        .expect("local auth op remains");
    assert_eq!(
        stored_auth_op.op_json,
        serde_json::to_string(&local_auth_op).expect("encode expected local auth op")
    );
}

#[test]
fn peer_card_export_import_and_list_are_candidate_only() {
    let source_home = tempfile::tempdir().expect("source tempdir");
    let source_paths = geth_config::GethPaths::from_home(source_home.path());
    let mut source = geth_node::init_node(&source_paths).expect("init source");
    source.iroh_status = geth_iroh::EndpointStatus {
        enabled: true,
        endpoint_id: Some("endpoint:test-source".to_owned()),
        relay_mode: "disabled".to_owned(),
        local_discovery: false,
        note: "test endpoint".to_owned(),
    };
    let peer_card_path = source_home.path().join("peer-card.json");

    let exported = geth_node::handle_request(
        &source,
        geth_control::ControlRequest::PeerCardExport {
            out: Some(peer_card_path.clone()),
        },
    )
    .expect("export peer card");
    let exported_card = match exported {
        geth_control::ControlResponse::PeerCardExported { card, out, note } => {
            assert_eq!(out, Some(peer_card_path.clone()));
            assert!(note.contains("never grants trust"));
            card.validate_candidate().expect("valid exported card");
            card
        }
        other => panic!("unexpected response: {other:?}"),
    };
    assert!(peer_card_path.exists());

    let target_home = tempfile::tempdir().expect("target tempdir");
    let target_paths = geth_config::GethPaths::from_home(target_home.path());
    let target = geth_node::init_node(&target_paths).expect("init target");
    let imported = geth_node::handle_request(
        &target,
        geth_control::ControlRequest::PeerCardImport {
            path: peer_card_path,
        },
    )
    .expect("import peer card");
    match imported {
        geth_control::ControlResponse::PeerCardImported { peer, note } => {
            assert_eq!(peer.card, exported_card);
            assert_eq!(
                peer.trust_state,
                geth_discovery::CandidateTrustState::CandidateOnly
            );
            assert!(note.contains("never grants trust"));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let listed = geth_node::handle_request(&target, geth_control::ControlRequest::PeerCardList)
        .expect("list peers");
    match listed {
        geth_control::ControlResponse::PeerCardList { peers, note } => {
            assert_eq!(peers.len(), 1);
            assert_eq!(peers[0].card, exported_card);
            assert!(note.contains("never grants trust"));
        }
        other => panic!("unexpected response: {other:?}"),
    }
}

#[test]
fn initialized_node_can_roundtrip_cas_blob() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    assert_eq!(node.paths.home(), home.path());

    let cas = geth_cas::LocalCas::new(paths.cas_dir());
    let output_path = home.path().join("output.txt");
    let added = cas.add_bytes(b"hello geth integration").expect("add blob");
    assert!(cas.has(&added.hash).expect("has blob"));
    cas.get_to_path(&added.hash, &output_path)
        .expect("get blob");
    assert_eq!(
        std::fs::read(output_path).expect("read output"),
        b"hello geth integration"
    );
}

#[test]
fn overlay_plan_is_available_without_starting_packet_runtime() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::OverlayPlan {
            name: "home-lan".to_owned(),
            cidr: None,
        },
    )
    .expect("overlay plan");
    match response {
        geth_control::ControlResponse::OverlayPlanned { plan } => {
            assert_eq!(plan.resource.to_string(), "resource:overlay:home-lan");
            assert_eq!(plan.alpn, "/geth/overlay/1");
            assert!(plan.capabilities.contains(&"overlay.join".to_owned()));
            assert!(plan.runtime.contains("overlay up"));
        }
        other => panic!("unexpected response: {other:?}"),
    }
}

#[test]
fn overlay_join_persists_membership_resource_and_leave_state() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");

    let joined = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::OverlayJoin {
            name: "home-lan".to_owned(),
            secret: "local-bootstrap-secret".to_owned(),
            cidr: Some("172.22.0.0/24".to_owned()),
        },
    )
    .expect("overlay join");
    match joined {
        geth_control::ControlResponse::OverlayJoined { join } => {
            assert!(join.enabled);
            assert_eq!(join.plan.resource.to_string(), "resource:overlay:home-lan");
            assert_eq!(join.network.state, geth_overlay::OverlayState::Joined);
            assert!(join.network.virtual_ip.is_some());
            assert!(join.note.contains("local bootstrap overlay"));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let resources = geth_node::handle_request(&node, geth_control::ControlRequest::ResourceList)
        .expect("resource list");
    match resources {
        geth_control::ControlResponse::ResourceList { resources } => {
            assert!(resources.iter().any(|resource| {
                resource.id.to_string() == "resource:overlay:home-lan"
                    && resource.kind == geth_types::ResourceKind::Overlay
            }));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let status = geth_node::handle_request(&node, geth_control::ControlRequest::OverlayStatus)
        .expect("overlay status");
    match status {
        geth_control::ControlResponse::OverlayStatus { networks, note } => {
            assert_eq!(networks.len(), 1);
            assert_eq!(networks[0].state, geth_overlay::OverlayState::Joined);
            assert!(note.contains("persisted locally"));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::OverlayLeave {
            name: "home-lan".to_owned(),
        },
    )
    .expect("overlay leave");
    let status = geth_node::handle_request(&node, geth_control::ControlRequest::OverlayStatus)
        .expect("overlay status after leave");
    match status {
        geth_control::ControlResponse::OverlayStatus { networks, .. } => {
            assert_eq!(networks.len(), 1);
            assert_eq!(networks[0].state, geth_overlay::OverlayState::Stopped);
        }
        other => panic!("unexpected response: {other:?}"),
    }
}

#[test]
fn overlay_join_uses_bearer_invite_when_one_exists() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    let resource = "resource:overlay:locked".to_owned();

    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::ResourceCreate {
            kind: "overlay".to_owned(),
            name: "locked".to_owned(),
        },
    )
    .expect("create overlay resource");
    let access = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SecretBearerCreate {
            resource: resource.clone(),
            capabilities: vec!["overlay.join".to_owned()],
            expires_at_ms: None,
        },
    )
    .expect("create bearer invite");
    let token = match access {
        geth_control::ControlResponse::SecretBearerCreated { access } => {
            access.token.expect("private bearer token").to_string()
        }
        other => panic!("unexpected response: {other:?}"),
    };

    let denied = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::OverlayJoin {
            name: "locked".to_owned(),
            secret: "wrong-token".to_owned(),
            cidr: None,
        },
    )
    .expect_err("bad bearer should be rejected");
    assert!(denied.to_string().contains("overlay join requires"));

    let joined = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::OverlayJoin {
            name: "locked".to_owned(),
            secret: token,
            cidr: None,
        },
    )
    .expect("authorized overlay join");
    match joined {
        geth_control::ControlResponse::OverlayJoined { join } => {
            assert!(join.enabled);
            assert_eq!(join.network.state, geth_overlay::OverlayState::Joined);
            assert!(join.note.contains("bearer overlay.join authorization"));
        }
        other => panic!("unexpected response: {other:?}"),
    }
}

#[test]
fn private_cas_blob_uses_resource_secret_epoch() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    let resource = "resource:cas:local".to_owned();
    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SecretCreate {
            resource: resource.clone(),
        },
    )
    .expect("create resource secret");
    let input = home.path().join("private.txt");
    std::fs::write(&input, b"private blob bytes").expect("write private input");

    let added = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasAddPrivate {
            resource: resource.clone(),
            path: input,
        },
    )
    .expect("add private blob");
    let encrypted_hash = match added {
        geth_control::ControlResponse::CasPrivateAdded {
            encrypted_hash,
            plaintext_hash,
            epoch,
            ..
        } => {
            assert_eq!(epoch, 1);
            assert_eq!(plaintext_hash, geth_cas::hash_bytes(b"private blob bytes"));
            encrypted_hash
        }
        other => panic!("unexpected private add response: {other:?}"),
    };

    let out = home.path().join("private-out.txt");
    let got = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasGetPrivate {
            resource,
            hash: encrypted_hash,
            out: out.clone(),
        },
    )
    .expect("get private blob");
    match got {
        geth_control::ControlResponse::CasPrivateGot {
            plaintext_hash,
            size_bytes,
            ..
        } => {
            assert_eq!(plaintext_hash, geth_cas::hash_bytes(b"private blob bytes"));
            assert_eq!(size_bytes, b"private blob bytes".len() as u64);
        }
        other => panic!("unexpected private get response: {other:?}"),
    }
    assert_eq!(
        std::fs::read(out).expect("read private out"),
        b"private blob bytes"
    );
}

#[test]
fn cas_pin_unpin_updates_local_pin_metadata() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    let blob_path = home.path().join("blob.txt");
    std::fs::write(&blob_path, b"pinned bytes").expect("write blob");

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasAdd { path: blob_path },
    )
    .expect("add cas blob");
    let hash = match response {
        geth_control::ControlResponse::CasAdded { hash, .. } => hash,
        other => panic!("unexpected response: {other:?}"),
    };

    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasPin { hash: hash.clone() },
    )
    .expect("pin blob");
    let response = geth_node::handle_request(&node, geth_control::ControlRequest::CasList)
        .expect("list blobs");
    match response {
        geth_control::ControlResponse::CasList { blobs } => {
            assert_eq!(blobs.len(), 1);
            assert!(blobs[0].pinned);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    geth_node::handle_request(&node, geth_control::ControlRequest::CasUnpin { hash })
        .expect("unpin blob");
    let response = geth_node::handle_request(&node, geth_control::ControlRequest::CasList)
        .expect("list blobs");
    match response {
        geth_control::ControlResponse::CasList { blobs } => {
            assert_eq!(blobs.len(), 1);
            assert!(!blobs[0].pinned);
        }
        other => panic!("unexpected response: {other:?}"),
    }
}

#[test]
fn cas_file_root_add_list_scan_detects_changes() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    let root_path = home.path().join("notes-root");
    std::fs::create_dir_all(&root_path).expect("create root");
    std::fs::write(root_path.join("a.txt"), b"alpha").expect("write a");

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasRootAdd {
            name: "notes".to_owned(),
            path: root_path.clone(),
        },
    )
    .expect("add root");
    match response {
        geth_control::ControlResponse::CasRootAdded { root } => {
            assert_eq!(root.name, "notes");
            assert!(root.latest_tree.is_none());
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(&node, geth_control::ControlRequest::CasRootList)
        .expect("list roots");
    match response {
        geth_control::ControlResponse::CasRootList { roots } => {
            assert_eq!(roots.len(), 1);
            assert_eq!(roots[0].name, "notes");
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasRootScan {
            name: "notes".to_owned(),
        },
    )
    .expect("initial scan");
    let first_tree = match response {
        geth_control::ControlResponse::CasRootScanned { scan } => {
            assert_eq!(scan.changes.len(), 1);
            assert!(scan.root.latest_tree.is_some());
            assert!(scan.note.contains("never overwrites"));
            scan.tree.hash
        }
        other => panic!("unexpected response: {other:?}"),
    };

    std::fs::rename(root_path.join("a.txt"), root_path.join("b.txt")).expect("rename a to b");
    std::fs::write(root_path.join("c.txt"), b"charlie").expect("write c");
    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasRootScan {
            name: "notes".to_owned(),
        },
    )
    .expect("second scan");
    match response {
        geth_control::ControlResponse::CasRootScanned { scan } => {
            assert_ne!(scan.tree.hash, first_tree);
            assert!(scan.changes.iter().any(|change| {
                matches!(
                    change,
                    geth_cas::FileRootChange::Renamed { from, to }
                        if from == "a.txt" && to == "b.txt"
                )
            }));
            assert!(scan.changes.iter().any(|change| {
                matches!(change, geth_cas::FileRootChange::Created { path } if path == "c.txt")
            }));
        }
        other => panic!("unexpected response: {other:?}"),
    }
}

#[test]
fn cas_file_root_apply_writes_missing_files_and_records_conflicts() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    let source_path = home.path().join("source-root");
    std::fs::create_dir_all(source_path.join("dir")).expect("create source root");
    std::fs::write(source_path.join("a.txt"), b"remote").expect("write remote a");
    std::fs::write(source_path.join("dir").join("b.txt"), b"remote b").expect("write remote b");

    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasRootAdd {
            name: "remote-node-peer-shared".to_owned(),
            path: source_path,
        },
    )
    .expect("add source root");
    let tree = match geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasRootScan {
            name: "remote-node-peer-shared".to_owned(),
        },
    )
    .expect("scan source root")
    {
        geth_control::ControlResponse::CasRootScanned { scan } => scan.tree.hash,
        other => panic!("unexpected response: {other:?}"),
    };

    let target_path = home.path().join("target-root");
    std::fs::create_dir_all(&target_path).expect("create target root");
    std::fs::write(target_path.join("a.txt"), b"local edit").expect("write local a");

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasRootApply {
            source: "remote-node-peer-shared".to_owned(),
            target: target_path.clone(),
            dry_run: false,
        },
    )
    .expect("apply source root");
    match response {
        geth_control::ControlResponse::CasRootApplied {
            files_written,
            dirs_created,
            conflicts,
            note,
            ..
        } => {
            assert_eq!(files_written, 1);
            assert_eq!(dirs_created, 1);
            assert_eq!(conflicts.len(), 1);
            assert_eq!(conflicts[0].path, "a.txt");
            assert_eq!(conflicts[0].remote_tree, Some(tree));
            assert!(note.contains("never deletes"));
        }
        other => panic!("unexpected response: {other:?}"),
    }
    assert_eq!(
        std::fs::read(target_path.join("dir").join("b.txt")).expect("read applied b"),
        b"remote b"
    );
    assert_eq!(
        std::fs::read(target_path.join("a.txt")).expect("read local a"),
        b"local edit"
    );

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasConflictList {
            root: Some("remote-node-peer-shared".to_owned()),
        },
    )
    .expect("list apply conflicts");
    match response {
        geth_control::ControlResponse::CasConflictList { conflicts } => {
            assert_eq!(conflicts.len(), 1);
            assert_eq!(conflicts[0].path, "a.txt");
        }
        other => panic!("unexpected response: {other:?}"),
    }
}

#[test]
fn cas_file_root_apply_three_way_updates_deletes_and_renames_safe_changes() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    let target_path = home.path().join("target-root");
    std::fs::create_dir_all(&target_path).expect("create target root");
    std::fs::write(target_path.join("update.txt"), b"base").expect("write base update");
    std::fs::write(target_path.join("delete.txt"), b"base delete").expect("write base delete");
    std::fs::write(target_path.join("rename.txt"), b"base rename").expect("write base rename");

    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasRootAdd {
            name: "shared".to_owned(),
            path: target_path.clone(),
        },
    )
    .expect("add local root");
    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasRootScan {
            name: "shared".to_owned(),
        },
    )
    .expect("scan local base");

    let source_path = home.path().join("source-root");
    std::fs::create_dir_all(&source_path).expect("create source root");
    std::fs::write(source_path.join("update.txt"), b"remote").expect("write remote update");
    std::fs::write(source_path.join("rename-new.txt"), b"base rename")
        .expect("write remote rename");
    std::fs::write(source_path.join("created.txt"), b"created").expect("write remote create");
    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasRootAdd {
            name: "remote-node-peer-shared".to_owned(),
            path: source_path,
        },
    )
    .expect("add remote root");
    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasRootScan {
            name: "remote-node-peer-shared".to_owned(),
        },
    )
    .expect("scan remote root");

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasRootApply {
            source: "remote-node-peer-shared".to_owned(),
            target: target_path.clone(),
            dry_run: false,
        },
    )
    .expect("three-way apply");
    match response {
        geth_control::ControlResponse::CasRootApplied {
            files_written,
            conflicts,
            note,
            ..
        } => {
            assert_eq!(files_written, 2);
            assert!(conflicts.is_empty());
            assert!(note.contains("three-way"));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    assert_eq!(
        std::fs::read(target_path.join("update.txt")).expect("read updated"),
        b"remote"
    );
    assert!(
        !target_path.join("delete.txt").exists(),
        "safe remote delete should remove unchanged local base file"
    );
    assert!(
        !target_path.join("rename.txt").exists(),
        "safe remote rename should remove old path"
    );
    assert_eq!(
        std::fs::read(target_path.join("rename-new.txt")).expect("read renamed"),
        b"base rename"
    );
    assert_eq!(
        std::fs::read(target_path.join("created.txt")).expect("read created"),
        b"created"
    );
}

#[test]
fn cas_file_root_apply_three_way_keeps_ambiguous_changes_as_conflicts() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    let target_path = home.path().join("target-root");
    std::fs::create_dir_all(&target_path).expect("create target root");
    std::fs::write(target_path.join("note.txt"), b"base").expect("write base");

    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasRootAdd {
            name: "shared".to_owned(),
            path: target_path.clone(),
        },
    )
    .expect("add local root");
    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasRootScan {
            name: "shared".to_owned(),
        },
    )
    .expect("scan local base");
    std::fs::write(target_path.join("note.txt"), b"local edit").expect("write local edit");

    let source_path = home.path().join("source-root");
    std::fs::create_dir_all(&source_path).expect("create source root");
    std::fs::write(source_path.join("note.txt"), b"remote edit").expect("write remote edit");
    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasRootAdd {
            name: "remote-node-peer-shared".to_owned(),
            path: source_path,
        },
    )
    .expect("add remote root");
    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasRootScan {
            name: "remote-node-peer-shared".to_owned(),
        },
    )
    .expect("scan remote root");

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasRootApply {
            source: "remote-node-peer-shared".to_owned(),
            target: target_path.clone(),
            dry_run: false,
        },
    )
    .expect("three-way conflict apply");
    match response {
        geth_control::ControlResponse::CasRootApplied { conflicts, .. } => {
            assert_eq!(conflicts.len(), 1);
            assert_eq!(conflicts[0].path, "note.txt");
            assert_eq!(
                conflicts[0].kind,
                geth_cas::FileConflictKind::ConcurrentEdit
            );
            assert!(conflicts[0].base_tree.is_some());
            assert!(conflicts[0].local_tree.is_some());
            assert!(conflicts[0].remote_tree.is_some());
        }
        other => panic!("unexpected response: {other:?}"),
    }
    assert_eq!(
        std::fs::read(target_path.join("note.txt")).expect("read local after conflict"),
        b"local edit"
    );
}

#[test]
fn cas_file_conflict_record_list_and_resolve() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    let root_path = home.path().join("notes-root");
    std::fs::create_dir_all(&root_path).expect("create root");
    std::fs::write(root_path.join("todo.md"), b"local").expect("write file");

    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasRootAdd {
            name: "notes".to_owned(),
            path: root_path,
        },
    )
    .expect("add root");
    let local_tree = match geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasRootScan {
            name: "notes".to_owned(),
        },
    )
    .expect("scan root")
    {
        geth_control::ControlResponse::CasRootScanned { scan } => scan.tree.hash,
        other => panic!("unexpected response: {other:?}"),
    };
    let remote_tree = geth_types::BlobHash::new(
        "fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210",
    );

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasConflictRecord {
            root: "notes".to_owned(),
            path: "todo.md".to_owned(),
            kind: "concurrent-edit".to_owned(),
            detail: "local and remote edits touched todo.md".to_owned(),
            base_tree: None,
            local_tree: Some(local_tree.clone()),
            remote_tree: Some(remote_tree.clone()),
        },
    )
    .expect("record conflict");
    let conflict_id = match response {
        geth_control::ControlResponse::CasConflictRecorded { conflict } => {
            assert_eq!(conflict.root, "notes");
            assert_eq!(conflict.path, "todo.md");
            assert_eq!(conflict.kind, geth_cas::FileConflictKind::ConcurrentEdit);
            assert_eq!(conflict.status, geth_cas::FileConflictStatus::Open);
            assert_eq!(conflict.local_tree, Some(local_tree));
            assert_eq!(conflict.remote_tree, Some(remote_tree));
            conflict.id
        }
        other => panic!("unexpected response: {other:?}"),
    };

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasConflictList {
            root: Some("notes".to_owned()),
        },
    )
    .expect("list conflicts");
    match response {
        geth_control::ControlResponse::CasConflictList { conflicts } => {
            assert_eq!(conflicts.len(), 1);
            assert_eq!(conflicts[0].id, conflict_id);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasConflictResolve {
            conflict_id: conflict_id.clone(),
            resolution: "keep-local".to_owned(),
            note: Some("local file is authoritative".to_owned()),
        },
    )
    .expect("resolve conflict");
    match response {
        geth_control::ControlResponse::CasConflictResolved { conflict } => {
            assert_eq!(conflict.id, conflict_id);
            assert_eq!(conflict.status, geth_cas::FileConflictStatus::Resolved);
            assert_eq!(
                conflict.resolution,
                Some(geth_cas::FileConflictResolution::KeepLocal)
            );
            assert_eq!(
                conflict.resolution_note,
                Some("local file is authoritative".to_owned())
            );
            assert!(conflict.resolved_at_ms.is_some());
        }
        other => panic!("unexpected response: {other:?}"),
    }
}

#[test]
fn cas_cleanup_removes_unpinned_and_retains_pinned_blobs() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    let pinned_path = home.path().join("pinned.txt");
    let unpinned_path = home.path().join("unpinned.txt");
    std::fs::write(&pinned_path, b"keep me").expect("write pinned");
    std::fs::write(&unpinned_path, b"remove me").expect("write unpinned");

    let pinned_hash = match geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasAdd { path: pinned_path },
    )
    .expect("add pinned")
    {
        geth_control::ControlResponse::CasAdded { hash, .. } => hash,
        other => panic!("unexpected response: {other:?}"),
    };
    let unpinned_hash = match geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasAdd {
            path: unpinned_path,
        },
    )
    .expect("add unpinned")
    {
        geth_control::ControlResponse::CasAdded { hash, .. } => hash,
        other => panic!("unexpected response: {other:?}"),
    };
    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasPin {
            hash: pinned_hash.clone(),
        },
    )
    .expect("pin blob");

    let dry_run = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasCleanup { dry_run: true },
    )
    .expect("dry-run cleanup");
    match dry_run {
        geth_control::ControlResponse::CasCleanup {
            removed,
            retained_pinned,
            dry_run,
        } => {
            assert!(dry_run);
            assert_eq!(removed, vec![unpinned_hash.clone()]);
            assert_eq!(retained_pinned, vec![pinned_hash.clone()]);
        }
        other => panic!("unexpected response: {other:?}"),
    }
    assert!(
        geth_cas::LocalCas::new(paths.cas_dir())
            .has(&unpinned_hash)
            .expect("dry-run keeps unpinned")
    );

    let cleanup = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::CasCleanup { dry_run: false },
    )
    .expect("cleanup");
    match cleanup {
        geth_control::ControlResponse::CasCleanup {
            removed,
            retained_pinned,
            dry_run,
        } => {
            assert!(!dry_run);
            assert_eq!(removed, vec![unpinned_hash.clone()]);
            assert_eq!(retained_pinned, vec![pinned_hash.clone()]);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let cas = geth_cas::LocalCas::new(paths.cas_dir());
    assert!(cas.has(&pinned_hash).expect("pinned retained"));
    assert!(!cas.has(&unpinned_hash).expect("unpinned removed"));
}

#[test]
fn auth_explain_distinguishes_discovered_peer_candidates() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    let store = geth_store::Store::open(&paths.metadata_db()).expect("open store");
    store
        .upsert_peer_card(&geth_store::StoredPeerCard {
            peer_id: "node:discovered".to_owned(),
            card_json: r#"{"node_id":"node:discovered"}"#.to_owned(),
            updated_at_ms: 1,
        })
        .expect("insert peer card");

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::AuthExplain {
            subject: "node:discovered".to_owned(),
            resource: "resource:cas:local".to_owned(),
            capability: "cas.fetch".to_owned(),
        },
    )
    .expect("auth explain");
    match response {
        geth_control::ControlResponse::AuthExplain(explanation) => {
            assert!(!explanation.allowed);
            assert!(explanation.reason.contains("discovered peer candidate"));
            assert!(
                explanation
                    .diagnostics
                    .contains(&"subject:discovered-only".to_owned())
            );
        }
        other => panic!("unexpected response: {other:?}"),
    }
}

#[test]
fn auth_explain_distinguishes_endpoint_binding_state() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    let store = geth_store::Store::open(&paths.metadata_db()).expect("open store");
    let created_at = geth_store::now_ms();

    for op in [
        geth_keychain::KeychainOp {
            id: "op:keychain:init".into(),
            created_at: geth_types::UnixMillis(created_at),
            kind: geth_keychain::KeychainOpKind::KeychainInit,
        },
        geth_keychain::KeychainOp {
            id: "op:user:add".into(),
            created_at: geth_types::UnixMillis(created_at + 1),
            kind: geth_keychain::KeychainOpKind::UserAdd {
                user: "user:owner".into(),
                name: "Owner".to_owned(),
            },
        },
        geth_keychain::KeychainOp {
            id: "op:device:add".into(),
            created_at: geth_types::UnixMillis(created_at + 2),
            kind: geth_keychain::KeychainOpKind::DeviceAdd {
                device: "device:laptop".into(),
                user: "user:owner".into(),
            },
        },
        geth_keychain::KeychainOp {
            id: "op:node:add".into(),
            created_at: geth_types::UnixMillis(created_at + 3),
            kind: geth_keychain::KeychainOpKind::NodeAdd {
                node: "node:trusted".into(),
                device: "device:laptop".into(),
                name: "trusted".to_owned(),
            },
        },
    ] {
        store
            .insert_keychain_op(&geth_store::StoredKeychainOp {
                op_id: op.id.to_string(),
                op_json: serde_json::to_string(&op).expect("keychain op json"),
                created_at_ms: op.created_at.0,
            })
            .expect("insert keychain op");
    }

    store
        .upsert_peer_card(&geth_store::StoredPeerCard {
            peer_id: "node:trusted".to_owned(),
            card_json: r#"{
                "node_id":"node:trusted",
                "agent_id":"agent:test",
                "endpoints":[{"endpoint_id":"endpoint:card-only","relay_url":null,"direct_addresses":[],"source":"manual"}],
                "issued_at":1,
                "signature":{"namespace":"","signer":"","public_key":"","signature":""}
            }"#
            .to_owned(),
            updated_at_ms: created_at + 4,
        })
        .expect("insert peer card");

    let missing = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::AuthExplain {
            subject: "node:trusted".to_owned(),
            resource: "resource:cas:local".to_owned(),
            capability: "cas.fetch".to_owned(),
        },
    )
    .expect("explain missing endpoint binding");
    match missing {
        geth_control::ControlResponse::AuthExplain(explanation) => {
            assert!(!explanation.allowed);
            assert!(
                explanation
                    .diagnostics
                    .contains(&"endpoint-binding:missing".to_owned())
            );
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let endpoint_op = geth_keychain::KeychainOp {
        id: "op:node:endpoint:add".into(),
        created_at: geth_types::UnixMillis(created_at + 5),
        kind: geth_keychain::KeychainOpKind::NodeEndpointAdd {
            node: "node:trusted".into(),
            endpoint: "endpoint:card-only".to_owned(),
        },
    };
    store
        .insert_keychain_op(&geth_store::StoredKeychainOp {
            op_id: endpoint_op.id.to_string(),
            op_json: serde_json::to_string(&endpoint_op).expect("endpoint op json"),
            created_at_ms: endpoint_op.created_at.0,
        })
        .expect("insert endpoint op");

    let matched = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::AuthExplain {
            subject: "node:trusted".to_owned(),
            resource: "resource:cas:local".to_owned(),
            capability: "cas.fetch".to_owned(),
        },
    )
    .expect("explain matched endpoint binding");
    match matched {
        geth_control::ControlResponse::AuthExplain(explanation) => {
            assert!(
                explanation
                    .diagnostics
                    .contains(&"endpoint-binding:matched:endpoint:card-only".to_owned())
            );
        }
        other => panic!("unexpected response: {other:?}"),
    }
}

#[test]
fn auth_grant_revoke_and_explain_use_local_auth_log() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::AuthGrant {
            subject: "node:laptop".to_owned(),
            resource: "resource:cas:local".to_owned(),
            capability: "cas.fetch".to_owned(),
            grant_id: Some("grant:test-fetch".to_owned()),
            signing_key_path: None,
            admin_key_path: None,
        },
    )
    .expect("grant capability");
    match response {
        geth_control::ControlResponse::AuthOpRecorded { op, signatures } => {
            assert_eq!(op.resource.to_string(), "resource:cas:local");
            assert!(signatures.is_empty());
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::AuthExplain {
            subject: "node:laptop".to_owned(),
            resource: "resource:cas:local".to_owned(),
            capability: "cas.fetch".to_owned(),
        },
    )
    .expect("explain grant");
    match response {
        geth_control::ControlResponse::AuthExplain(explanation) => {
            assert!(explanation.allowed);
            assert_eq!(explanation.evaluated_ops, 1);
            assert!(explanation.reason.contains("grant:test-fetch"));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::AuthRevoke {
            resource: "resource:cas:local".to_owned(),
            grant_id: "grant:test-fetch".to_owned(),
            signing_key_path: None,
            admin_key_path: None,
        },
    )
    .expect("revoke grant");

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::AuthExplain {
            subject: "node:laptop".to_owned(),
            resource: "resource:cas:local".to_owned(),
            capability: "cas.fetch".to_owned(),
        },
    )
    .expect("explain revoke");
    match response {
        geth_control::ControlResponse::AuthExplain(explanation) => {
            assert!(!explanation.allowed);
            assert_eq!(explanation.evaluated_ops, 2);
            assert!(explanation.reason.contains("revoked"));
            assert!(
                explanation
                    .diagnostics
                    .contains(&"grant:revoked:grant:test-fetch".to_owned())
            );
        }
        other => panic!("unexpected response: {other:?}"),
    }
}

#[test]
fn keychain_init_and_status_use_local_keychain_log() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    let admin_key_path = home.path().join("admin.pub");
    std::fs::write(&admin_key_path, "ssh-ed25519 AAAAADMIN eric@geth\n").expect("write admin key");

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::KeychainInit {
            admin_key_path: Some(admin_key_path),
            signing_key_path: None,
        },
    )
    .expect("init keychain");
    match response {
        geth_control::ControlResponse::KeychainInitialized { ops, signatures } => {
            assert_eq!(ops.len(), 2);
            assert!(signatures.is_empty());
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(&node, geth_control::ControlRequest::KeychainStatus)
        .expect("keychain status");
    match response {
        geth_control::ControlResponse::KeychainStatus(status) => {
            assert!(status.initialized);
            assert_eq!(status.admin_keys, 1);
            assert_eq!(status.signatures, 0);
            assert_eq!(status.verified_signatures, 0);
            assert_eq!(status.failed_signatures, 0);
            assert_eq!(status.users, 0);
        }
        other => panic!("unexpected response: {other:?}"),
    }
}

#[test]
fn keychain_init_can_record_openssh_signatures() {
    if Command::new("ssh-keygen").arg("-?").output().is_err() {
        return;
    }

    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    let admin_key_path = home.path().join("admin_ed25519");
    let status = Command::new("ssh-keygen")
        .arg("-q")
        .arg("-t")
        .arg("ed25519")
        .arg("-N")
        .arg("")
        .arg("-f")
        .arg(&admin_key_path)
        .status()
        .expect("generate admin ssh key");
    assert!(status.success());

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::KeychainInit {
            admin_key_path: Some(admin_key_path.with_extension("pub")),
            signing_key_path: Some(admin_key_path),
        },
    )
    .expect("init signed keychain");
    match response {
        geth_control::ControlResponse::KeychainInitialized { ops, signatures } => {
            assert_eq!(ops.len(), 2);
            assert_eq!(signatures.len(), 2);
            assert!(signatures.iter().all(|signature| {
                signature.namespace == "geth.keychain.v1@geth.local"
                    && signature.signer_public_key.starts_with("ssh-ed25519 ")
                    && !signature.signature.is_empty()
            }));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let signatures = geth_store::Store::open(&paths.metadata_db())
        .expect("open store")
        .list_keychain_signatures()
        .expect("list signatures");
    assert_eq!(signatures.len(), 2);

    let response = geth_node::handle_request(&node, geth_control::ControlRequest::KeychainStatus)
        .expect("keychain status");
    match response {
        geth_control::ControlResponse::KeychainStatus(status) => {
            assert_eq!(status.signatures, 2);
            assert_eq!(status.verified_signatures, 2);
            assert_eq!(status.failed_signatures, 0);
        }
        other => panic!("unexpected response: {other:?}"),
    }
}

#[test]
fn keychain_admin_sigchain_exports_allowed_signers_and_verifies() {
    if Command::new("ssh-keygen").arg("-?").output().is_err() {
        return;
    }

    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    let admin_key_path = home.path().join("admin_ed25519");
    generate_ssh_key(&admin_key_path);

    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::KeychainInit {
            admin_key_path: Some(admin_key_path.with_extension("pub")),
            signing_key_path: Some(admin_key_path.clone()),
        },
    )
    .expect("init signed keychain");

    let second_admin_key_path = home.path().join("second_admin_ed25519");
    generate_ssh_key(&second_admin_key_path);
    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::KeychainAdminAdd {
            admin_key_path: second_admin_key_path.with_extension("pub"),
            signing_key_path: admin_key_path.clone(),
            principal: Some("second-admin".to_owned()),
            valid_after_ms: None,
            valid_before_ms: None,
        },
    )
    .expect("admin add");
    match response {
        geth_control::ControlResponse::KeychainAdminUpdated { op, signatures, .. } => {
            assert_eq!(signatures.len(), 1);
            match op.kind {
                geth_keychain::KeychainOpKind::AdminKeyAdd {
                    public_key,
                    principal,
                    ..
                } => {
                    assert!(public_key.expect("public key").starts_with("ssh-ed25519 "));
                    assert_eq!(principal.as_deref(), Some("second-admin"));
                }
                other => panic!("unexpected op kind: {other:?}"),
            }
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::KeychainAllowedSigners { out: None },
    )
    .expect("allowed signers");
    match response {
        geth_control::ControlResponse::KeychainAllowedSigners {
            entries,
            allowed_signers,
            ..
        } => {
            assert_eq!(entries.len(), 2);
            assert!(allowed_signers.contains("second-admin ssh-ed25519 "));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let allowed_signers_path = home.path().join("allowed_signers");
    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::KeychainAllowedSigners {
            out: Some(allowed_signers_path.clone()),
        },
    )
    .expect("allowed signers file");
    match response {
        geth_control::ControlResponse::KeychainAllowedSigners { out, .. } => {
            assert_eq!(out.as_deref(), Some(allowed_signers_path.as_path()));
            let allowed_signers_file =
                std::fs::read_to_string(&allowed_signers_path).expect("read allowed_signers");
            assert!(allowed_signers_file.contains("second-admin ssh-ed25519 "));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let authorized_keys_path = home.path().join("authorized_keys");
    std::fs::write(
        &authorized_keys_path,
        "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIGexample external-login-key\n",
    )
    .expect("write authorized_keys snapshot");
    let authorized_keys_signature_path = home.path().join("authorized_keys.sig");
    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::KeychainSignFile {
            input: authorized_keys_path.clone(),
            out: Some(authorized_keys_signature_path.clone()),
            namespace: None,
            signing_key_path: Some(admin_key_path.clone()),
            admin_key_path: Some(admin_key_path.with_extension("pub")),
        },
    )
    .expect("sign authorized_keys snapshot");
    match response {
        geth_control::ControlResponse::KeychainFileSigned {
            input,
            out,
            namespace,
            signer,
            ..
        } => {
            assert_eq!(input, authorized_keys_path);
            assert_eq!(
                out.as_deref(),
                Some(authorized_keys_signature_path.as_path())
            );
            assert_eq!(namespace, geth_keychain::AUTHORIZED_KEYS_NAMESPACE);
            assert!(!signer.is_empty());
            assert!(authorized_keys_signature_path.exists());
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::KeychainVerifyFile {
            input: authorized_keys_path.clone(),
            signature: authorized_keys_signature_path.clone(),
            namespace: None,
            allowed_signers_path: Some(allowed_signers_path.clone()),
            principal: None,
        },
    )
    .expect("verify authorized_keys snapshot");
    match response {
        geth_control::ControlResponse::KeychainFileVerified {
            verified,
            principal,
            ..
        } => {
            assert!(verified);
            assert_eq!(principal.as_deref(), Some("admin"));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let sigchain_path = home.path().join("keychain.sigchain.jsonl");
    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::KeychainSigchainExport {
            out: Some(sigchain_path.clone()),
        },
    )
    .expect("sigchain export");
    match response {
        geth_control::ControlResponse::KeychainSigchainExported { entries, out, .. } => {
            assert_eq!(out.as_deref(), Some(sigchain_path.as_path()));
            assert!(entries.len() >= 3);
            let jsonl = std::fs::read_to_string(&sigchain_path).expect("read sigchain");
            let decoded = geth_keychain::decode_sigchain_jsonl(&jsonl).expect("decode sigchain");
            assert_eq!(decoded, entries);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::KeychainVerifySigchain {
            input: sigchain_path.clone(),
        },
    )
    .expect("verify sigchain file");
    match response {
        geth_control::ControlResponse::KeychainSigchainFileVerified { report, .. } => {
            assert_eq!(report.rejected_ops, 0);
            assert_eq!(report.active_admin_keys, 2);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let bundle_dir = home.path().join("public-bundle");
    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::KeychainPublishBundle {
            out: bundle_dir.clone(),
            base_url: None,
            signing_key_path: admin_key_path.clone(),
            admin_key_path: Some(admin_key_path.with_extension("pub")),
            snapshots: vec![format!(
                "authorized_keys={}",
                authorized_keys_path.display()
            )],
        },
    )
    .expect("publish bundle");
    match response {
        geth_control::ControlResponse::KeychainBundlePublished {
            base_url,
            allowed_signers_path,
            sigchain_path: bundle_sigchain_path,
            checkpoint_path,
            checkpoint_signature_path,
            checkpoint,
            snapshots,
            ..
        } => {
            assert_eq!(base_url, geth_keychain::DEFAULT_SSH_SIGCHAIN_DISCOVERY_URL);
            assert!(allowed_signers_path.exists());
            assert!(bundle_sigchain_path.exists());
            assert!(checkpoint_path.exists());
            assert!(checkpoint_signature_path.exists());
            assert_eq!(
                checkpoint.base_url,
                geth_keychain::DEFAULT_SSH_SIGCHAIN_DISCOVERY_URL
            );
            assert_eq!(snapshots.len(), 1);
            assert_eq!(snapshots[0].name, "authorized_keys");
            assert!(snapshots[0].path.exists());
            assert!(snapshots[0].signature_path.exists());
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::KeychainVerifyCheckpoint {
            checkpoint: bundle_dir.join("geth.sigchain.checkpoint.json"),
            signature: bundle_dir.join("geth.sigchain.checkpoint.json.sig"),
            sigchain: bundle_dir.join("geth.sigchain.jsonl"),
            allowed_signers: bundle_dir.join("allowed_signers"),
            base_url: Some(geth_keychain::DEFAULT_SSH_SIGCHAIN_DISCOVERY_URL.to_owned()),
            principal: None,
        },
    )
    .expect("verify checkpoint");
    match response {
        geth_control::ControlResponse::KeychainCheckpointVerified {
            verified,
            principal,
            ..
        } => {
            assert!(verified);
            assert_eq!(principal.as_deref(), Some("admin"));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let fetched_home = tempfile::tempdir().expect("fetch tempdir");
    let fetched_paths = geth_config::GethPaths::from_home(fetched_home.path());
    let fetched_node = geth_node::init_node(&fetched_paths).expect("init fetched node");
    let response = geth_node::handle_request(
        &fetched_node,
        geth_control::ControlRequest::KeychainFetch {
            url: format!("file://{}", bundle_dir.display()),
            out: Some(fetched_home.path().join("bundle")),
            import: true,
        },
    )
    .expect("fetch bundle");
    match response {
        geth_control::ControlResponse::KeychainFetched {
            imported: Some(imported),
            checkpoint,
            ..
        } => {
            assert!(imported.ops_imported >= 3);
            assert!(imported.signatures_imported >= 3);
            assert_eq!(imported.invalid_ops_rejected, 0);
            assert_eq!(
                checkpoint.base_url,
                geth_keychain::DEFAULT_SSH_SIGCHAIN_DISCOVERY_URL
            );
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let jsonl = std::fs::read_to_string(&sigchain_path).expect("read sigchain for explain");
    let decoded =
        geth_keychain::decode_sigchain_jsonl(&jsonl).expect("decode sigchain for explain");
    let explain_op_id = decoded.last().expect("last sigchain op").op.id.to_string();
    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::KeychainExplain {
            op_id: explain_op_id.clone(),
        },
    )
    .expect("explain op");
    match response {
        geth_control::ControlResponse::KeychainExplained { subject, lines } => {
            assert_eq!(subject, explain_op_id);
            assert!(lines.iter().any(|line| line.contains("accepted_by_replay")));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let admin_public_key =
        std::fs::read_to_string(admin_key_path.with_extension("pub")).expect("admin pub");
    let admin_key = geth_keychain::admin_key_fingerprint(&admin_public_key);
    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::KeychainExplainSigner { key: admin_key },
    )
    .expect("explain signer");
    match response {
        geth_control::ControlResponse::KeychainExplained { lines, .. } => {
            assert!(lines.iter().any(|line| line == "active_admin_signer: true"));
            assert!(
                lines
                    .iter()
                    .any(|line| line.starts_with("signed_operations:"))
            );
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let import_home = tempfile::tempdir().expect("import tempdir");
    let import_paths = geth_config::GethPaths::from_home(import_home.path());
    let import_node = geth_node::init_node(&import_paths).expect("init import node");
    let response = geth_node::handle_request(
        &import_node,
        geth_control::ControlRequest::KeychainImportSigchain {
            input: sigchain_path.clone(),
        },
    )
    .expect("import sigchain file");
    match response {
        geth_control::ControlResponse::KeychainSigchainImported {
            ops_imported,
            signatures_imported,
            invalid_ops_rejected,
            ..
        } => {
            assert!(ops_imported >= 3);
            assert!(signatures_imported >= 3);
            assert_eq!(invalid_ops_rejected, 0);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(&node, geth_control::ControlRequest::KeychainVerify)
        .expect("verify sigchain");
    match response {
        geth_control::ControlResponse::KeychainVerified { report } => {
            assert_eq!(report.rejected_ops, 0);
            assert_eq!(report.active_admin_keys, 2);
            assert!(report.note.contains("git-skm"));
        }
        other => panic!("unexpected response: {other:?}"),
    }
}

#[test]
fn init_owned_node_records_signed_owner_device_and_node() {
    if Command::new("ssh-keygen").arg("-?").output().is_err() {
        return;
    }

    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let admin_key_path = home.path().join("owner_ed25519");
    let status = Command::new("ssh-keygen")
        .arg("-q")
        .arg("-t")
        .arg("ed25519")
        .arg("-N")
        .arg("")
        .arg("-f")
        .arg(&admin_key_path)
        .status()
        .expect("generate owner ssh key");
    assert!(status.success());

    let node = geth_node::init_owned_node(
        &paths,
        geth_node::InitOwnerOptions {
            admin_key_path: Some(admin_key_path.with_extension("pub")),
            signing_key_path: Some(admin_key_path.clone()),
            owner_name: "Eric".to_owned(),
            node_name: "laptop".to_owned(),
            capabilities: vec!["resource:ssh-proxy:local=ssh_proxy.admin_shell".to_owned()],
        },
    )
    .expect("init owned node");

    match geth_node::handle_request(&node, geth_control::ControlRequest::KeychainStatus)
        .expect("keychain status")
    {
        geth_control::ControlResponse::KeychainStatus(status) => {
            assert!(status.initialized);
            assert_eq!(status.admin_keys, 1);
            assert_eq!(status.users, 1);
            assert_eq!(status.devices, 1);
            assert_eq!(status.nodes, 1);
            assert_eq!(status.verified_signatures, status.signatures);
            assert!(status.signatures >= 6);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    match geth_node::handle_request(&node, geth_control::ControlRequest::NodeList)
        .expect("node list")
    {
        geth_control::ControlResponse::NodeList { nodes, .. } => {
            assert_eq!(nodes.len(), 1);
            assert_eq!(nodes[0].name, "laptop");
            assert_eq!(nodes[0].id.as_str(), node.node_id);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let renamed = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::NodeRename {
            node: "laptop".to_owned(),
            name: "work-laptop".to_owned(),
            signing_key_path: Some(admin_key_path.clone()),
        },
    )
    .expect("rename node");
    match renamed {
        geth_control::ControlResponse::NodeKeychainUpdated { signatures, .. } => {
            assert_eq!(signatures.len(), 1);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let endpoint_added = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::NodeEndpointAdd {
            node: "work-laptop".to_owned(),
            endpoint: "endpoint:test-rotated".to_owned(),
            signing_key_path: Some(admin_key_path.clone()),
        },
    )
    .expect("add endpoint");
    match endpoint_added {
        geth_control::ControlResponse::NodeKeychainUpdated { signatures, .. } => {
            assert_eq!(signatures.len(), 1);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let grant = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::NodeGrant {
            node: "work-laptop".to_owned(),
            resource: "resource:cas:local".to_owned(),
            capability: "cas.fetch".to_owned(),
            grant_id: Some("grant:work-laptop-cas-fetch".to_owned()),
            signing_key_path: Some(admin_key_path.clone()),
            admin_key_path: None,
        },
    )
    .expect("signed node grant");
    match grant {
        geth_control::ControlResponse::NodeGrantUpdated { signatures, .. } => {
            assert_eq!(signatures.len(), 1);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let explained = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::AuthExplain {
            subject: node.node_id.clone(),
            resource: "resource:ssh-proxy:local".to_owned(),
            capability: "ssh_proxy.admin_shell".to_owned(),
        },
    )
    .expect("explain init grant");
    match explained {
        geth_control::ControlResponse::AuthExplain(explanation) => {
            assert!(explanation.allowed);
            assert!(explanation.reason.contains("direct grant"));
        }
        other => panic!("unexpected response: {other:?}"),
    }
}

#[test]
fn node_enrollment_request_import_and_approval_records_signed_ops() {
    if Command::new("ssh-keygen").arg("-?").output().is_err() {
        return;
    }

    let owner_home = tempfile::tempdir().expect("owner tempdir");
    let requester_home = tempfile::tempdir().expect("requester tempdir");
    let owner_paths = geth_config::GethPaths::from_home(owner_home.path());
    let requester_paths = geth_config::GethPaths::from_home(requester_home.path());
    let admin_key_path = owner_home.path().join("owner_ed25519");
    let status = Command::new("ssh-keygen")
        .arg("-q")
        .arg("-t")
        .arg("ed25519")
        .arg("-N")
        .arg("")
        .arg("-f")
        .arg(&admin_key_path)
        .status()
        .expect("generate owner ssh key");
    assert!(status.success());

    let owner = geth_node::init_owned_node(
        &owner_paths,
        geth_node::InitOwnerOptions {
            admin_key_path: Some(admin_key_path.with_extension("pub")),
            signing_key_path: Some(admin_key_path.clone()),
            owner_name: "Eric".to_owned(),
            node_name: "owner-laptop".to_owned(),
            capabilities: Vec::new(),
        },
    )
    .expect("init owner");
    let requester = geth_node::init_node(&requester_paths).expect("init requester");
    let request_path = requester_home.path().join("enrollment.json");

    let response = geth_node::handle_request(
        &requester,
        geth_control::ControlRequest::NodeEnrollRequest {
            node_name: "workstation".to_owned(),
            capabilities: vec!["resource:ssh-proxy:local=ssh_proxy.connect".to_owned()],
            reason: Some("new test machine".to_owned()),
            out: Some(request_path.clone()),
        },
    )
    .expect("create enrollment request");
    let request_id = match response {
        geth_control::ControlResponse::NodeEnrollmentRequested { request, .. } => {
            assert_eq!(request.requested_node_name, "workstation");
            assert!(request.provenance.is_some());
            request.id.to_string()
        }
        other => panic!("unexpected response: {other:?}"),
    };
    assert!(request_path.exists());

    let imported = geth_node::handle_request(
        &owner,
        geth_control::ControlRequest::NodeEnrollImport { path: request_path },
    )
    .expect("import request");
    match imported {
        geth_control::ControlResponse::NodeEnrollmentImported { request, .. } => {
            assert_eq!(request.id.as_str(), request_id);
            assert_eq!(request.status.to_string(), "pending");
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let approved = geth_node::handle_request(
        &owner,
        geth_control::ControlRequest::NodeEnrollApprove {
            request_id: request_id.clone(),
            signing_key_path: admin_key_path,
            admin_key_path: None,
            node_name: None,
            capabilities: vec!["resource:cas:local=cas.fetch".to_owned()],
        },
    )
    .expect("approve request");
    match approved {
        geth_control::ControlResponse::NodeEnrollmentApproved {
            keychain_ops,
            keychain_signatures,
            auth_ops,
            auth_signatures,
            ..
        } => {
            assert!(keychain_ops.len() >= 3);
            assert_eq!(keychain_signatures.len(), keychain_ops.len());
            assert_eq!(auth_ops.len(), 2);
            assert_eq!(auth_signatures.len(), 2);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    match geth_node::handle_request(&owner, geth_control::ControlRequest::NodeList)
        .expect("node list")
    {
        geth_control::ControlResponse::NodeList { nodes, .. } => {
            assert!(nodes.iter().any(|node| node.name == "workstation"));
        }
        other => panic!("unexpected response: {other:?}"),
    }
    match geth_node::handle_request(
        &owner,
        geth_control::ControlRequest::AuthExplain {
            subject: requester.node_id.clone(),
            resource: "resource:ssh-proxy:local".to_owned(),
            capability: "ssh_proxy.connect".to_owned(),
        },
    )
    .expect("explain enrollment grant")
    {
        geth_control::ControlResponse::AuthExplain(explanation) => assert!(explanation.allowed),
        other => panic!("unexpected response: {other:?}"),
    }
    let store = geth_store::Store::open(&owner_paths.metadata_db()).expect("open store");
    assert_eq!(
        store
            .list_auth_signatures()
            .expect("list auth signatures")
            .len(),
        2
    );
}

#[test]
fn db_add_and_status_register_local_db_metadata() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    let db_path = home.path().join("notes.sqlite");
    let conn = rusqlite::Connection::open(&db_path).expect("open sqlite");
    conn.execute(
        "CREATE TABLE notes(id INTEGER PRIMARY KEY, body TEXT NOT NULL)",
        [],
    )
    .expect("create notes table");
    conn.execute(
        r#"CREATE TABLE crsql_changes(
            table_name TEXT NOT NULL,
            pk TEXT NOT NULL,
            cid TEXT NOT NULL,
            val BLOB,
            col_version INTEGER NOT NULL,
            db_version INTEGER NOT NULL
        )"#,
        [],
    )
    .expect("create crsql_changes");
    conn.execute(
        "INSERT INTO crsql_changes(table_name, pk, cid, val, col_version, db_version) VALUES ('notes', '1', 'body', 'hello', 1, 3)",
        [],
    )
    .expect("insert crsql change");
    drop(conn);

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::DbAdd {
            name: "notes".to_owned(),
            path: db_path.clone(),
        },
    )
    .expect("add db");
    match response {
        geth_control::ControlResponse::DbAdded { db } => {
            assert_eq!(db.name, "notes");
            assert_eq!(db.sync_status, "local-only");
            assert!(db.path_exists);
            assert!(db.size_bytes.unwrap_or_default() > 0);
            assert!(db.schema_metadata.contains("tables=2"));
            assert!(db.schema_metadata.contains("schema_hash="));
            assert!(db.crsqlite_changes.available);
            assert_eq!(db.crsqlite_changes.change_count, Some(1));
            assert_eq!(db.crsqlite_changes.max_db_version, Some(3));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::DbStatus {
            name: "notes".to_owned(),
        },
    )
    .expect("db status");
    match response {
        geth_control::ControlResponse::DbStatus { db } => {
            assert_eq!(db.name, "notes");
            assert!(db.path.ends_with("notes.sqlite"));
            assert!(db.schema_metadata.contains("tables=2"));
            assert!(db.schema_metadata.contains("schema_hash="));
            assert!(
                db.crsqlite_changes
                    .columns
                    .contains(&"db_version".to_owned())
            );
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::DbChanges {
            name: "notes".to_owned(),
            after_db_version: None,
            limit: 10,
        },
    )
    .expect("db changes");
    match response {
        geth_control::ControlResponse::DbChanges { db, batch } => {
            assert_eq!(db.name, "notes");
            assert_eq!(batch.changes.len(), 1);
            assert_eq!(batch.max_db_version, Some(3));
            assert_eq!(batch.changes[0].table_name, "notes");
            assert_eq!(batch.changes[0].column_id, "body");
            assert!(batch.schema_metadata.contains("schema_hash="));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::DbChanges {
            name: "notes".to_owned(),
            after_db_version: Some(3),
            limit: 10,
        },
    )
    .expect("db changes after latest");
    match response {
        geth_control::ControlResponse::DbChanges { batch, .. } => {
            assert!(batch.changes.is_empty());
            assert_eq!(batch.max_db_version, None);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    assert!(
        geth_node::handle_request(
            &node,
            geth_control::ControlRequest::DbAdd {
                name: "../bad".to_owned(),
                path: db_path,
            },
        )
        .is_err()
    );
    assert!(
        geth_node::handle_request(
            &node,
            geth_control::ControlRequest::DbAdd {
                name: "missing".to_owned(),
                path: home.path().join("missing.sqlite"),
            },
        )
        .is_err()
    );
}

#[test]
fn kv_create_set_get_use_local_store() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::KvCreate {
            name: "prefs".to_owned(),
        },
    )
    .expect("create kv");
    match response {
        geth_control::ControlResponse::KvCreated { kv } => {
            assert_eq!(kv.name, "prefs");
            assert_eq!(kv.sync_status, "local-only");
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::KvSet {
            name: "prefs".to_owned(),
            key: "apps/foo/theme".to_owned(),
            value: "dark".to_owned(),
            subject: None,
        },
    )
    .expect("set kv");
    match response {
        geth_control::ControlResponse::KvSet { entry } => {
            assert_eq!(entry.store.to_string(), "kv:prefs");
            assert_eq!(entry.key, "apps/foo/theme");
            assert_eq!(entry.value, "dark");
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::KvGet {
            name: "prefs".to_owned(),
            key: "apps/foo/theme".to_owned(),
        },
    )
    .expect("get kv");
    match response {
        geth_control::ControlResponse::KvGet { entry } => {
            assert_eq!(entry.expect("entry").value, "dark");
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::KvGet {
            name: "prefs".to_owned(),
            key: "apps/foo/missing".to_owned(),
        },
    )
    .expect("get missing kv");
    match response {
        geth_control::ControlResponse::KvGet { entry } => {
            assert!(entry.is_none());
        }
        other => panic!("unexpected response: {other:?}"),
    }

    assert!(
        geth_node::handle_request(
            &node,
            geth_control::ControlRequest::KvSet {
                name: "prefs".to_owned(),
                key: "apps/foo/accent".to_owned(),
                value: "blue".to_owned(),
                subject: Some("node:tablet".to_owned()),
            },
        )
        .is_err()
    );
    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::AuthGrant {
            subject: "node:tablet".to_owned(),
            resource: "resource:kv:prefs".to_owned(),
            capability: "kv.write_prefix:apps/foo/".to_owned(),
            grant_id: None,
            signing_key_path: None,
            admin_key_path: None,
        },
    )
    .expect("grant prefix write");
    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::KvSet {
            name: "prefs".to_owned(),
            key: "apps/foo/accent".to_owned(),
            value: "blue".to_owned(),
            subject: Some("node:tablet".to_owned()),
        },
    )
    .expect("authorized prefix write");
    assert!(
        geth_node::handle_request(
            &node,
            geth_control::ControlRequest::KvSet {
                name: "prefs".to_owned(),
                key: "apps/bar/accent".to_owned(),
                value: "red".to_owned(),
                subject: Some("node:tablet".to_owned()),
            },
        )
        .is_err()
    );

    assert!(
        geth_node::handle_request(
            &node,
            geth_control::ControlRequest::KvCreate {
                name: "../bad".to_owned(),
            },
        )
        .is_err()
    );
    assert!(
        geth_node::handle_request(
            &node,
            geth_control::ControlRequest::KvSet {
                name: "missing".to_owned(),
                key: "apps/foo/theme".to_owned(),
                value: "dark".to_owned(),
                subject: None,
            },
        )
        .is_err()
    );
}

#[test]
fn document_create_and_status_use_local_store() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::DocumentCreate {
            name: "notes".to_owned(),
        },
    )
    .expect("create document");
    match response {
        geth_control::ControlResponse::DocumentCreated { document } => {
            assert_eq!(document.name, "notes");
            assert_eq!(document.sync_status, "automerge-local");
            assert!(document.state_bytes > 2);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::DocumentStatus {
            name: "notes".to_owned(),
        },
    )
    .expect("document status");
    match response {
        geth_control::ControlResponse::DocumentStatus { document } => {
            assert_eq!(document.id.to_string(), "document:notes");
            assert_eq!(document.resource.to_string(), "resource:document:notes");
            assert!(document.state_bytes > 2);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::DocumentSet {
            name: "notes".to_owned(),
            state_json: r#"{ "title": "notes", "items": [1, 2] }"#.to_owned(),
        },
    )
    .expect("document set");
    match response {
        geth_control::ControlResponse::DocumentSet { state } => {
            assert_eq!(state.document.name, "notes");
            assert_eq!(state.state_json, r#"{"items":[1,2],"title":"notes"}"#);
            assert!(state.document.state_bytes > 2);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let reopened = geth_node::open_node(&paths).expect("reopen node");
    let response = geth_node::handle_request(
        &reopened,
        geth_control::ControlRequest::DocumentGet {
            name: "notes".to_owned(),
        },
    )
    .expect("document get");
    match response {
        geth_control::ControlResponse::DocumentGet { state } => {
            assert_eq!(state.state_json, r#"{"items":[1,2],"title":"notes"}"#);
            assert!(state.updated_at.0 > 0);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    assert!(
        geth_node::handle_request(
            &node,
            geth_control::ControlRequest::DocumentSet {
                name: "notes".to_owned(),
                state_json: "{".to_owned(),
            },
        )
        .is_err()
    );

    assert!(
        geth_node::handle_request(
            &node,
            geth_control::ControlRequest::DocumentCreate {
                name: "../bad".to_owned(),
            },
        )
        .is_err()
    );
    assert!(
        geth_node::handle_request(
            &node,
            geth_control::ControlRequest::DocumentStatus {
                name: "missing".to_owned(),
            },
        )
        .is_err()
    );
}

#[test]
fn secret_create_rotate_and_status_track_resource_epochs() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");

    let first = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SecretCreate {
            resource: "resource:cas:local".to_owned(),
        },
    )
    .expect("create secret");
    match first {
        geth_control::ControlResponse::SecretCreated { secret } => {
            assert_eq!(secret.resource.to_string(), "resource:cas:local");
            assert_eq!(secret.epoch, 1);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let second = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SecretRotate {
            resource: "resource:cas:local".to_owned(),
        },
    )
    .expect("rotate secret");
    match second {
        geth_control::ControlResponse::SecretCreated { secret } => {
            assert_eq!(secret.resource.to_string(), "resource:cas:local");
            assert_eq!(secret.epoch, 2);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let status = geth_node::handle_request(&node, geth_control::ControlRequest::SecretStatus)
        .expect("secret status");
    match status {
        geth_control::ControlResponse::SecretStatus { secrets } => {
            assert_eq!(secrets.len(), 2);
            assert_eq!(secrets[0].epoch, 1);
            assert_eq!(secrets[1].epoch, 2);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    assert!(
        geth_node::handle_request(
            &node,
            geth_control::ControlRequest::SecretCreate {
                resource: "resource:missing".to_owned(),
            },
        )
        .is_err()
    );
}

#[test]
fn bearer_access_create_list_revoke_uses_resource_scoped_auth_ops() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SecretBearerCreate {
            resource: "resource:cas:local".to_owned(),
            capabilities: vec!["cas.fetch".to_owned(), "cas.pin".to_owned()],
            expires_at_ms: Some(4_102_444_800_000),
        },
    )
    .expect("create bearer");
    let (secret, token) = match response {
        geth_control::ControlResponse::SecretBearerCreated { access } => {
            assert_eq!(access.resource.to_string(), "resource:cas:local");
            assert_eq!(access.capabilities.len(), 2);
            assert_eq!(
                access.expires_at.map(|expires_at| expires_at.0),
                Some(4_102_444_800_000)
            );
            assert!(!access.may_delegate);
            let token = access.token.expect("private bearer token on create");
            assert_ne!(access.secret, token);
            assert!(access.token_hash.is_none());
            (access.secret.to_string(), token.to_string())
        }
        other => panic!("unexpected response: {other:?}"),
    };

    let response = geth_node::handle_request(&node, geth_control::ControlRequest::SecretBearerList)
        .expect("list bearer");
    match response {
        geth_control::ControlResponse::SecretBearerList { access } => {
            assert_eq!(access.len(), 1);
            assert_eq!(access[0].secret.to_string(), secret);
            assert!(access[0].token.is_none());
            assert!(access[0].token_hash.is_some());
            assert!(!access[0].may_delegate);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SecretBearerChallenge {
            resource: "resource:cas:local".to_owned(),
            capabilities: vec!["cas.fetch".to_owned()],
        },
    )
    .expect("create bearer challenge");
    let nonce = match response {
        geth_control::ControlResponse::SecretBearerChallenge { challenge } => {
            assert_eq!(challenge.resource.to_string(), "resource:cas:local");
            assert_eq!(challenge.capabilities[0].to_string(), "cas.fetch");
            challenge.nonce
        }
        other => panic!("unexpected response: {other:?}"),
    };
    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SecretBearerProve {
            secret: token.clone(),
            resource: "resource:cas:local".to_owned(),
            capabilities: vec!["cas.fetch".to_owned()],
            nonce: nonce.clone(),
        },
    )
    .expect("prove bearer challenge");
    let proof_response = match response {
        geth_control::ControlResponse::SecretBearerProof { proof } => {
            assert_eq!(proof.secret.to_string(), secret);
            assert_eq!(proof.resource.to_string(), "resource:cas:local");
            assert!(!proof.response.is_empty());
            proof.response
        }
        other => panic!("unexpected response: {other:?}"),
    };
    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SecretBearerVerify {
            secret: token.clone(),
            resource: "resource:cas:local".to_owned(),
            capabilities: vec!["cas.fetch".to_owned()],
            nonce: nonce.clone(),
            response: proof_response.clone(),
        },
    )
    .expect("verify bearer proof");
    match response {
        geth_control::ControlResponse::SecretBearerVerified { verified, .. } => {
            assert!(verified);
        }
        other => panic!("unexpected response: {other:?}"),
    }
    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SecretBearerVerify {
            secret: token,
            resource: "resource:cas:local".to_owned(),
            capabilities: vec!["cas.pin".to_owned()],
            nonce,
            response: proof_response,
        },
    )
    .expect("verify bearer proof for wrong capability");
    match response {
        geth_control::ControlResponse::SecretBearerVerified {
            verified, reason, ..
        } => {
            assert!(!verified);
            assert!(reason.contains("did not match"));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SecretBearerRevoke {
            resource: "resource:cas:local".to_owned(),
            secret: secret.clone(),
        },
    )
    .expect("revoke bearer");
    let response = geth_node::handle_request(&node, geth_control::ControlRequest::SecretBearerList)
        .expect("list revoked bearer");
    match response {
        geth_control::ControlResponse::SecretBearerList { access } => {
            assert!(access.is_empty());
        }
        other => panic!("unexpected response: {other:?}"),
    }

    assert!(
        geth_node::handle_request(
            &node,
            geth_control::ControlRequest::SecretBearerCreate {
                resource: "resource:cas:local".to_owned(),
                capabilities: vec!["auth.delegate".to_owned()],
                expires_at_ms: None,
            },
        )
        .is_err()
    );
}

#[test]
fn pubsub_pub_sub_uses_lossy_in_memory_runtime() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::PubsubPub {
            topic: "presence/laptop".to_owned(),
            message: "online".to_owned(),
            node: None,
            bearer_secret: None,
        },
    )
    .expect("publish");
    match response {
        geth_control::ControlResponse::PubsubPublished { message } => {
            assert_eq!(message.topic.to_string(), "presence/laptop");
            assert_eq!(message.message, "online");
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::PubsubSub {
            topic: "presence/laptop".to_owned(),
            node: None,
            bearer_secret: None,
        },
    )
    .expect("subscribe snapshot");
    match response {
        geth_control::ControlResponse::PubsubMessages {
            topic,
            messages,
            note,
        } => {
            assert_eq!(topic, "presence/laptop");
            assert_eq!(messages.len(), 1);
            assert_eq!(messages[0].message, "online");
            assert!(note.contains("not authoritative storage"));
            assert!(note.contains("CAS, KV, documents, or DB resources"));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let reopened = geth_node::open_node(&paths).expect("reopen node");
    let response = geth_node::handle_request(
        &reopened,
        geth_control::ControlRequest::PubsubSub {
            topic: "presence/laptop".to_owned(),
            node: None,
            bearer_secret: None,
        },
    )
    .expect("subscribe reopened snapshot");
    match response {
        geth_control::ControlResponse::PubsubMessages { messages, .. } => {
            assert!(messages.is_empty());
        }
        other => panic!("unexpected response: {other:?}"),
    }

    assert!(
        geth_node::handle_request(
            &node,
            geth_control::ControlRequest::PubsubPub {
                topic: "presence/laptop".to_owned(),
                message: String::new(),
                node: None,
                bearer_secret: None,
            },
        )
        .is_err()
    );
}

#[test]
fn pipe_listen_connect_uses_local_runtime_registry() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::PipeListen {
            name: "inbox".to_owned(),
            node: None,
            bearer_secret: None,
        },
    )
    .expect("listen pipe");
    match response {
        geth_control::ControlResponse::PipeListening { listener } => {
            assert_eq!(listener.id.to_string(), "pipe:inbox");
            assert_eq!(listener.name, "inbox");
            assert!(listener.note.contains("daemon-lifetime pipe runtime"));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::PipeConnect {
            target: "inbox".to_owned(),
            node: None,
            bearer_secret: None,
        },
    )
    .expect("connect pipe");
    match response {
        geth_control::ControlResponse::PipeConnected { connection } => {
            assert_eq!(connection.target, "inbox");
            assert!(connection.local_listener_found);
            assert!(connection.note.contains("pipe runtime"));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::PipeSend {
            target: "inbox".to_owned(),
            data_base64: "aGVsbG8gcGlwZQ==".to_owned(),
            node: None,
            bearer_secret: None,
        },
    )
    .expect("send pipe message");
    match response {
        geth_control::ControlResponse::PipeSent {
            message,
            listener_found,
            ..
        } => {
            assert!(listener_found);
            assert_eq!(message.expect("message").data_base64, "aGVsbG8gcGlwZQ==");
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::PipeRecv {
            name: "inbox".to_owned(),
            peek: false,
        },
    )
    .expect("receive pipe messages");
    match response {
        geth_control::ControlResponse::PipeMessages {
            messages, drained, ..
        } => {
            assert!(drained);
            assert_eq!(messages.len(), 1);
            assert_eq!(messages[0].data_base64, "aGVsbG8gcGlwZQ==");
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::PipeRecv {
            name: "inbox".to_owned(),
            peek: false,
        },
    )
    .expect("receive drained pipe messages");
    match response {
        geth_control::ControlResponse::PipeMessages { messages, .. } => {
            assert!(messages.is_empty());
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let reopened = geth_node::open_node(&paths).expect("reopen node");
    let response = geth_node::handle_request(
        &reopened,
        geth_control::ControlRequest::PipeConnect {
            target: "inbox".to_owned(),
            node: None,
            bearer_secret: None,
        },
    )
    .expect("connect after reopen");
    match response {
        geth_control::ControlResponse::PipeConnected { connection } => {
            assert!(!connection.local_listener_found);
        }
        other => panic!("unexpected response: {other:?}"),
    }

    assert!(
        geth_node::handle_request(
            &node,
            geth_control::ControlRequest::PipeListen {
                name: "../bad".to_owned(),
                node: None,
                bearer_secret: None,
            },
        )
        .is_err()
    );
}

#[test]
fn ssh_cert_request_approval_and_revocation_export_use_local_state() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    let public_key_path = home.path().join("id_ed25519.pub");
    std::fs::write(&public_key_path, "ssh-ed25519 AAAATEST eric@geth\n").expect("write pubkey");

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SshCertRequest {
            public_key_path: public_key_path.clone(),
            cert_kind: "user".to_owned(),
            principals: vec!["eric".to_owned()],
            requested_validity: Some("+52w".to_owned()),
            renewal_of: None,
            reason: Some("renewal".to_owned()),
            subject: None,
        },
    )
    .expect("request cert");
    let request_id = match response {
        geth_control::ControlResponse::SshCertRequested { request } => request.id.to_string(),
        other => panic!("unexpected response: {other:?}"),
    };

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SshCertApprove {
            request_id: request_id.clone(),
            ca_key_path: home.path().join("ca_sk"),
            valid_for: Some("+4w".to_owned()),
            serial: Some(42),
            out: None,
            sign: false,
            subject: None,
        },
    )
    .expect("approve cert");
    match response {
        geth_control::ControlResponse::SshCertApproved { approval } => {
            assert_eq!(approval.request_id.to_string(), request_id);
            assert!(approval.signing_command.contains(&"ssh-keygen".to_owned()));
            assert!(approval.signing_command.contains(&"42".to_owned()));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let export_path = home.path().join("revocations.jsonl");
    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SshRevocationAdd {
            kind: "public-key".to_owned(),
            target: "ssh:blake3:test".to_owned(),
            reason: Some("lost key".to_owned()),
            subject: None,
        },
    )
    .expect("add revocation");
    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SshRevocationExport {
            out: export_path.clone(),
            format: "jsonl".to_owned(),
            ca_public: None,
            subject: None,
        },
    )
    .expect("export revocations");
    assert!(
        std::fs::read_to_string(export_path)
            .expect("read revocations")
            .contains("lost key")
    );

    let krl_spec_path = home.path().join("revocations.krl-spec");
    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SshRevocationExport {
            out: krl_spec_path.clone(),
            format: "openssh-krl-spec".to_owned(),
            ca_public: None,
            subject: None,
        },
    )
    .expect("export revocation krl spec");
    match response {
        geth_control::ControlResponse::SshRevocationExported {
            format,
            count,
            note,
            ..
        } => {
            assert_eq!(format, "openssh-krl-spec");
            assert_eq!(count, 1);
            assert!(note.contains("ssh-keygen -k"));
        }
        other => panic!("unexpected response: {other:?}"),
    }
    assert!(
        std::fs::read_to_string(&krl_spec_path)
            .expect("read krl spec")
            .contains("key: ssh:blake3:test")
    );

    let import_home = tempfile::tempdir().expect("import tempdir");
    let import_paths = geth_config::GethPaths::from_home(import_home.path());
    let import_node = geth_node::init_node(&import_paths).expect("init import node");
    let response = geth_node::handle_request(
        &import_node,
        geth_control::ControlRequest::SshRevocationImport {
            path: krl_spec_path,
            format: "openssh-krl-spec".to_owned(),
            subject: None,
        },
    )
    .expect("import krl spec");
    match response {
        geth_control::ControlResponse::SshRevocationImported {
            count,
            revocations,
            note,
            ..
        } => {
            assert_eq!(count, 1);
            assert_eq!(
                revocations[0].kind,
                geth_ssh_identity::SshRevocationKind::PublicKey
            );
            assert!(note.contains("binary OpenSSH KRL files cannot be enumerated"));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &import_node,
        geth_control::ControlRequest::SshRevocationImport {
            path: home.path().join("revocations.jsonl"),
            format: "jsonl".to_owned(),
            subject: None,
        },
    )
    .expect("import jsonl revocations");
    match response {
        geth_control::ControlResponse::SshRevocationImported { count, .. } => {
            assert_eq!(count, 1);
        }
        other => panic!("unexpected response: {other:?}"),
    }
}

#[test]
fn ssh_cert_and_revocation_commands_check_subject_capabilities() {
    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    let public_key_path = home.path().join("id_ed25519.pub");
    std::fs::write(&public_key_path, "ssh-ed25519 AAAATEST eric@geth\n").expect("write pubkey");

    assert!(
        geth_node::handle_request(
            &node,
            geth_control::ControlRequest::SshCertRequest {
                public_key_path: public_key_path.clone(),
                cert_kind: "user".to_owned(),
                principals: vec!["eric".to_owned()],
                requested_validity: Some("+52w".to_owned()),
                renewal_of: None,
                reason: Some("unauthorized".to_owned()),
                subject: Some("node:ssh-operator".to_owned()),
            },
        )
        .is_err()
    );

    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::AuthGrant {
            subject: "node:ssh-operator".to_owned(),
            resource: "resource:ssh:certs".to_owned(),
            capability: "ssh_cert.request".to_owned(),
            grant_id: Some("grant:ssh-cert-request".to_owned()),
            signing_key_path: None,
            admin_key_path: None,
        },
    )
    .expect("grant cert request");
    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SshCertRequest {
            public_key_path,
            cert_kind: "user".to_owned(),
            principals: vec!["eric".to_owned()],
            requested_validity: Some("+52w".to_owned()),
            renewal_of: None,
            reason: Some("authorized".to_owned()),
            subject: Some("node:ssh-operator".to_owned()),
        },
    )
    .expect("authorized cert request");

    assert!(
        geth_node::handle_request(
            &node,
            geth_control::ControlRequest::SshRevocationAdd {
                kind: "key-id".to_owned(),
                target: "old-key".to_owned(),
                reason: Some("unauthorized".to_owned()),
                subject: Some("node:ssh-operator".to_owned()),
            },
        )
        .is_err()
    );

    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::AuthGrant {
            subject: "node:ssh-operator".to_owned(),
            resource: "resource:ssh:revocations".to_owned(),
            capability: "ssh_revocation.publish".to_owned(),
            grant_id: Some("grant:ssh-revocation-publish".to_owned()),
            signing_key_path: None,
            admin_key_path: None,
        },
    )
    .expect("grant revocation publish");
    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SshRevocationAdd {
            kind: "key-id".to_owned(),
            target: "old-key".to_owned(),
            reason: Some("authorized".to_owned()),
            subject: Some("node:ssh-operator".to_owned()),
        },
    )
    .expect("authorized revocation add");
}

#[test]
fn ssh_cert_approve_can_sign_and_import_with_openssh_key() {
    if Command::new("ssh-keygen").arg("-?").output().is_err() {
        return;
    }

    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    let ca_key_path = home.path().join("ca_ed25519");
    let user_key_path = home.path().join("user_ed25519");
    for key_path in [&ca_key_path, &user_key_path] {
        let status = Command::new("ssh-keygen")
            .arg("-q")
            .arg("-t")
            .arg("ed25519")
            .arg("-N")
            .arg("")
            .arg("-f")
            .arg(key_path)
            .status()
            .expect("generate ssh key");
        assert!(status.success());
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SshCertRequest {
            public_key_path: user_key_path.with_extension("pub"),
            cert_kind: "user".to_owned(),
            principals: vec!["eric".to_owned()],
            requested_validity: Some("+1w".to_owned()),
            renewal_of: None,
            reason: Some("sign now".to_owned()),
            subject: None,
        },
    )
    .expect("request cert");
    let request_id = match response {
        geth_control::ControlResponse::SshCertRequested { request } => request.id.to_string(),
        other => panic!("unexpected response: {other:?}"),
    };

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SshCertApprove {
            request_id: request_id.clone(),
            ca_key_path,
            valid_for: Some("+1w".to_owned()),
            serial: Some(7),
            out: None,
            sign: true,
            subject: None,
        },
    )
    .expect("approve and sign cert");
    match response {
        geth_control::ControlResponse::SshCertApproved { approval } => {
            assert!(approval.signed);
            assert!(approval.certificate_id.is_some());
            assert!(approval.note.contains("signed with ssh-keygen"));
        }
        other => panic!("unexpected response: {other:?}"),
    }

    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SshCertList { subject: None },
    )
    .expect("list certs");
    match response {
        geth_control::ControlResponse::SshCertList {
            requests,
            certificates,
        } => {
            assert_eq!(certificates.len(), 1);
            assert_eq!(certificates[0].request_id.to_string(), request_id);
            assert_eq!(
                requests[0].status,
                geth_ssh_identity::SshCertRequestStatus::Signed
            );
        }
        other => panic!("unexpected response: {other:?}"),
    }
}

#[test]
fn ssh_revocation_export_can_write_binary_openssh_krl() {
    if Command::new("ssh-keygen").arg("-?").output().is_err() {
        return;
    }

    let home = tempfile::tempdir().expect("tempdir");
    let paths = geth_config::GethPaths::from_home(home.path());
    let node = geth_node::init_node(&paths).expect("init node");
    let key_path = home.path().join("revoked");
    let status = Command::new("ssh-keygen")
        .arg("-q")
        .arg("-t")
        .arg("ed25519")
        .arg("-N")
        .arg("")
        .arg("-f")
        .arg(&key_path)
        .status()
        .expect("generate ssh key");
    assert!(status.success());
    let public_key_path = key_path.with_extension("pub");
    let public_key = std::fs::read_to_string(&public_key_path).expect("read public key");

    geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SshRevocationAdd {
            kind: "public-key".to_owned(),
            target: public_key,
            reason: Some("test binary krl".to_owned()),
            subject: None,
        },
    )
    .expect("add revocation");
    let krl_path = home.path().join("revocations.krl");
    let response = geth_node::handle_request(
        &node,
        geth_control::ControlRequest::SshRevocationExport {
            out: krl_path.clone(),
            format: "openssh-krl".to_owned(),
            ca_public: None,
            subject: None,
        },
    )
    .expect("export binary krl");
    match response {
        geth_control::ControlResponse::SshRevocationExported {
            format,
            count,
            note,
            ..
        } => {
            assert_eq!(format, "openssh-krl");
            assert_eq!(count, 1);
            assert!(note.contains("binary KRL"));
        }
        other => panic!("unexpected response: {other:?}"),
    }
    assert!(krl_path.exists());

    let query = Command::new("ssh-keygen")
        .arg("-Q")
        .arg("-f")
        .arg(&krl_path)
        .arg(&public_key_path)
        .output()
        .expect("query krl");
    assert!(!query.status.success());
    assert!(
        String::from_utf8_lossy(&query.stdout)
            .to_ascii_lowercase()
            .contains("revoked")
    );
}