diff --git a/cmux-tui/crates/acpmux/spec/acpmux-protocol.json b/cmux-tui/crates/acpmux/spec/acpmux-protocol.json index 2495f27785f9..7e90db6bbc71 100644 --- a/cmux-tui/crates/acpmux/spec/acpmux-protocol.json +++ b/cmux-tui/crates/acpmux/spec/acpmux-protocol.json @@ -1699,6 +1699,12 @@ "params": true, "result": true }, + { + "kind": "request", + "name": "_acpmux/vault_status", + "params": true, + "result": true + }, { "kind": "request", "name": "_acpmux/turn_attestation", diff --git a/cmux-tui/crates/acpmux/src/capabilities.rs b/cmux-tui/crates/acpmux/src/capabilities.rs index 9f523bc96c1a..43eb7a81a591 100644 --- a/cmux-tui/crates/acpmux/src/capabilities.rs +++ b/cmux-tui/crates/acpmux/src/capabilities.rs @@ -95,7 +95,7 @@ pub const CAPABILITIES: &[Capability] = &[ // asks the daemon to pull or forget when the values live only in its memory. Capability { id: crate::vault::CAPABILITY, - wire: &[method::MUX_VAULT_PULL, method::MUX_VAULT_FORGET], + wire: &[method::MUX_VAULT_PULL, method::MUX_VAULT_FORGET, method::MUX_VAULT_STATUS], }, Capability { id: "machine-consent-v1", diff --git a/cmux-tui/crates/acpmux/src/cli/host_vault.rs b/cmux-tui/crates/acpmux/src/cli/host_vault.rs index 7f933fcacd9a..ff87785a40b2 100644 --- a/cmux-tui/crates/acpmux/src/cli/host_vault.rs +++ b/cmux-tui/crates/acpmux/src/cli/host_vault.rs @@ -25,6 +25,10 @@ use crate::cli::output::print_json; use crate::vault::pull::{self, Failure, Outcome}; use crate::vault::store::{OsStore, SecretStore}; +/// The running daemon predates `_acpmux/vault_pull` and the OS store is not +/// usable here, so nothing can keep the values. +pub const DAEMON_OUTDATED: &str = "vault.daemon_outdated"; + fn failure(failure: Failure) -> anyhow::Error { let message = match failure.code.as_str() { "vault.not_joined" => "this machine is not joined; run `acpmux machine join` first", @@ -33,6 +37,9 @@ fn failure(failure: Failure) -> anyhow::Error { "no OS secret store (Keychain or Secret Service) here; vault env is not stored in files" } "vault.busy" => "another vault pull is running", + DAEMON_OUTDATED => { + "the running acpmux daemon is older than this CLI and cannot keep vault env (this machine has no usable OS secret store); run `acpmux daemon restart`, then pull again" + } _ => "vault pull failed", }; let message = if failure.code == pull::PARTIAL { @@ -52,6 +59,11 @@ pub(crate) fn failure_for_test(f: Failure) -> anyhow::Error { failure(f) } +/// The string at `key` of `v`, or "". +fn text_at<'a>(v: &'a Value, key: &str) -> &'a str { + v.get(key).and_then(Value::as_str).unwrap_or_default() +} + /// The outcome JSON with `store`. fn with_store(mut out: Value, store: &str) -> Value { if let Some(map) = out.as_object_mut() { @@ -78,15 +90,16 @@ fn print_outcome(outcome: &Value, store: &str, json_out: bool) { return; } let names = |key: &str| -> Vec { - outcome[key] - .as_array() + outcome + .get(key) + .and_then(Value::as_array) .into_iter() .flatten() .filter_map(|v| v.as_str().map(str::to_owned)) .collect() }; let (env, stored, deleted) = (names("env"), names("stored"), names("deleted")); - match outcome["state"].as_str().unwrap_or_default() { + match outcome.get("state").and_then(Value::as_str).unwrap_or_default() { "not_granted" => { println!( "the vault is not shared with this machine (grant it in cmux, then pull again)" @@ -108,12 +121,8 @@ fn print_outcome(outcome: &Value, store: &str, json_out: bool) { if store == "memory" { " [kept in the daemon's memory]" } else { "" } ), } - for failed in outcome["failed"].as_array().into_iter().flatten() { - eprintln!( - "{}: {}", - failed["name"].as_str().unwrap_or_default(), - failed["code"].as_str().unwrap_or_default() - ); + for failed in outcome.get("failed").and_then(Value::as_array).into_iter().flatten() { + eprintln!("{}: {}", text_at(failed, "name"), text_at(failed, "code")); } } @@ -142,7 +151,8 @@ pub async fn forget_cmd(json_out: bool) -> Result<()> { /// Ask the running daemon to do `method`. With none running (or one too old /// to know it), `here` does it in this process when the OS store is usable; /// otherwise the daemon is started and asked, since only it can keep the -/// values (in memory). +/// values (in memory). A running daemon too old to know `method` is not +/// asked again: that is [`DAEMON_OUTDATED`], which says to restart it. async fn by_daemon(method: &str, here: F) -> Result where F: FnOnce() -> Fut, @@ -150,31 +160,46 @@ where { let ask = |client: Arc| async move { client.request(method, json!({})).await }; - if let Ok(client) = crate::daemon::connect(false).await - && let Ok(reply) = ask(client).await - { - return Ok(reply); + let mut outdated = false; + if let Ok(client) = crate::daemon::connect(false).await { + match ask(client).await { + Ok(reply) => return Ok(reply), + Err(e) => outdated = method_not_found(&e), + } } if tokio::task::spawn_blocking(crate::vault::store::os_usable).await.unwrap_or(false) { return Ok(crate::server::vault_reply(OsStore.kind(), &here().await)); } + if outdated { + return Err(failure(Failure { + code: DAEMON_OUTDATED.into(), + deleted: Vec::new(), + outcome: None, + })); + } let client = crate::daemon::connect(true).await?; ask(client).await } +/// Whether the daemon answered that it has no such method (an older daemon). +pub(crate) fn method_not_found(e: &anyhow::Error) -> bool { + e.downcast_ref::().is_some_and(|d| d.0.code == -32601) +} + /// Print a pull's or forget's reply (`server/vault.rs`). A failed or partial /// one prints its outcome (with `failed`) on stdout when it has one and /// exits 1 with the error envelope. fn finish(reply: Value, json_out: bool) -> Result<()> { - let store = reply["store"].as_str().unwrap_or("unknown").to_owned(); + let store = reply.get("store").and_then(Value::as_str).unwrap_or("unknown").to_owned(); if let Some(outcome) = reply.get("outcome").filter(|o| !o.is_null()) { print_outcome(outcome, &store, json_out); } let Some(f) = reply.get("failure").filter(|f| !f.is_null()) else { return Ok(()) }; Err(failure(Failure { - code: f["code"].as_str().unwrap_or("vault.internal").to_owned(), - deleted: f["deleted"] - .as_array() + code: f.get("code").and_then(Value::as_str).unwrap_or("vault.internal").to_owned(), + deleted: f + .get("deleted") + .and_then(Value::as_array) .into_iter() .flatten() .filter_map(|v| v.as_str().map(str::to_owned)) @@ -183,10 +208,19 @@ fn finish(reply: Value, json_out: bool) -> Result<()> { })) } -/// Where this machine keeps vault values: the OS store's kind, or `memory` -/// when it cannot take a write here. -fn store_kind() -> &'static str { - if crate::vault::store::os_usable() { OsStore.kind() } else { "memory" } +/// Where this machine keeps vault values: the running daemon's store +/// (`_acpmux/vault_status`), since it is the one that keeps them. With no +/// daemon, or one too old to say, what this process finds: the OS store's +/// kind, or `memory` when it cannot take a write here. +async fn store_kind() -> String { + if let Ok(client) = crate::daemon::connect(false).await + && let Ok(reply) = client.request(crate::rpc::method::MUX_VAULT_STATUS, json!({})).await + && let Some(store) = reply.get("store").and_then(Value::as_str) + { + return store.to_owned(); + } + let here = || if crate::vault::store::os_usable() { OsStore.kind() } else { "memory" }; + tokio::task::spawn_blocking(here).await.unwrap_or("unknown").to_owned() } /// Whether a harness is signed in on this machine: a presence check of its own @@ -314,8 +348,8 @@ pub async fn status_cmd(json_out: bool) -> Result<()> { tokio::task::spawn_blocking(|| (claude_signed_in(claude_keychain_item), codex_signed_in())) .await .unwrap_or((SignedIn::Unknown, SignedIn::Unknown)); - let kind = tokio::task::spawn_blocking(store_kind).await.unwrap_or("unknown"); - let out = status_json(joined, &state, kind, claude, codex); + let kind = store_kind().await; + let out = status_json(joined, &state, &kind, claude, codex); if json_out { print_json(&out); return Ok(()); @@ -335,11 +369,9 @@ pub async fn status_cmd(json_out: bool) -> Result<()> { state.env.keys().cloned().collect::>().join(", ") } ); - for h in out["harnesses"].as_array().into_iter().flatten() { - let signed = h["signedIn"].as_str().unwrap_or("unknown"); - let id = h["harness"].as_str().unwrap_or(""); - let step = h["signIn"].as_str().unwrap_or(""); - match signed { + for h in out.get("harnesses").and_then(Value::as_array).into_iter().flatten() { + let (id, step) = (text_at(h, "harness"), text_at(h, "signIn")); + match h.get("signedIn").and_then(Value::as_str).unwrap_or("unknown") { "yes" => println!("{id}: signed in"), "no" => println!("{id}: not signed in here; {step}"), _ => println!("{id}: sign-in not visible here; if it fails, {step}"), diff --git a/cmux-tui/crates/acpmux/src/protocol/mod.rs b/cmux-tui/crates/acpmux/src/protocol/mod.rs index d901aff8b4b5..f683738d219d 100644 --- a/cmux-tui/crates/acpmux/src/protocol/mod.rs +++ b/cmux-tui/crates/acpmux/src/protocol/mod.rs @@ -439,6 +439,7 @@ pub fn registry() -> Vec { untyped(method::MUX_PAIR_REDEEM, Request), untyped(method::MUX_VAULT_PULL, Request), untyped(method::MUX_VAULT_FORGET, Request), + untyped(method::MUX_VAULT_STATUS, Request), untyped(method::MUX_TURN_ATTESTATION, Request), // Notifications. notification::(method::MUX_EVENT), diff --git a/cmux-tui/crates/acpmux/src/rpc.rs b/cmux-tui/crates/acpmux/src/rpc.rs index 1fe9ad6028fa..c598160d22c3 100644 --- a/cmux-tui/crates/acpmux/src/rpc.rs +++ b/cmux-tui/crates/acpmux/src/rpc.rs @@ -704,6 +704,7 @@ pub mod method { // The person's vault on this machine (`crate::vault`, `server/vault.rs`). pub const MUX_VAULT_PULL: &str = "_acpmux/vault_pull"; pub const MUX_VAULT_FORGET: &str = "_acpmux/vault_forget"; + pub const MUX_VAULT_STATUS: &str = "_acpmux/vault_status"; // Notifications from acpmux to attached clients. pub const MUX_EVENT: &str = "_acpmux/event"; /// A persisted session-to-session message, projected separately so chat clients can render diff --git a/cmux-tui/crates/acpmux/src/server/vault.rs b/cmux-tui/crates/acpmux/src/server/vault.rs index b2228bd5903d..76600e0ef096 100644 --- a/cmux-tui/crates/acpmux/src/server/vault.rs +++ b/cmux-tui/crates/acpmux/src/server/vault.rs @@ -1,5 +1,6 @@ //! `_acpmux/vault_pull` and `_acpmux/vault_forget`: `machine vault pull` and -//! `forget` done by this daemon, local socket only. A machine with no usable +//! `forget` done by this daemon, local socket only. `_acpmux/vault_status` +//! answers `{store}`, the store this daemon keeps the values in. A machine with no usable //! OS store keeps the vault env in the daemon's memory (`vault/store.rs`), so //! only the daemon can pull or drop it. The reply holds names and codes only, //! never a value. @@ -10,12 +11,15 @@ use super::{Conn, Origin}; use crate::rpc::{RpcError, method}; pub(super) async fn dispatch(conn: &Conn, m: &str) -> Option> { - if !matches!(m, method::MUX_VAULT_PULL | method::MUX_VAULT_FORGET) { + if !matches!(m, method::MUX_VAULT_PULL | method::MUX_VAULT_FORGET | method::MUX_VAULT_STATUS) { return None; } if conn.origin != Origin::Local { return Some(Err(RpcError::method_not_found(&format!("{m} (local socket only)")))); } + if m == method::MUX_VAULT_STATUS { + return Some(Ok(json!({"store": crate::vault::store_now().await}))); + } let (store, result) = if m == method::MUX_VAULT_PULL { crate::vault::pull_now().await } else { diff --git a/cmux-tui/crates/acpmux/src/vault.rs b/cmux-tui/crates/acpmux/src/vault.rs index 0ba6dc649a1a..70cbbc249bb7 100644 --- a/cmux-tui/crates/acpmux/src/vault.rs +++ b/cmux-tui/crates/acpmux/src/vault.rs @@ -112,7 +112,7 @@ pub fn load_values( state: &State, store: &dyn SecretStore, ) -> (Vec<(String, Zeroizing)>, Vec<(String, &'static str)>) { - load_values_by(state, store, Instant::now() + LOAD_BUDGET) + load_values_by(state, store, store::after(LOAD_BUDGET)) } /// The longest one load reads the store for: entries not read by then are @@ -470,6 +470,12 @@ pub async fn pull_now() -> (&'static str, Result) (kind, result) } +/// The store this daemon keeps vault values in (`_acpmux/vault_status`): +/// the OS store's kind, or `memory`. +pub async fn store_now() -> &'static str { + tokio::task::spawn_blocking(|| store::for_daemon().kind()).await.unwrap_or("unknown") +} + /// `machine vault forget` asked of this daemon (`_acpmux/vault_forget`). pub async fn forget_now() -> (&'static str, Result) { let Ok(store) = tokio::task::spawn_blocking(store::for_daemon).await else { diff --git a/cmux-tui/crates/acpmux/src/vault/store.rs b/cmux-tui/crates/acpmux/src/vault/store.rs index 86ac4d3155fd..8f411e142a16 100644 --- a/cmux-tui/crates/acpmux/src/vault/store.rs +++ b/cmux-tui/crates/acpmux/src/vault/store.rs @@ -210,7 +210,7 @@ impl SecretStore for OsStore { } fn get(&self, name: &str) -> Result>, &'static str> { - self.get_by(name, std::time::Instant::now() + TIMEOUT) + self.get_by(name, after(TIMEOUT)) } fn get_by( @@ -296,16 +296,26 @@ pub fn capture( cmd: Command, from_stderr: bool, ) -> Result<(std::process::ExitStatus, Zeroizing>), &'static str> { - capture_until(cmd, from_stderr, std::time::Instant::now() + TIMEOUT) + capture_until(cmd, from_stderr, after(TIMEOUT)) +} + +/// `now + wait`, or now if that would not fit an `Instant`. +pub fn after(wait: std::time::Duration) -> std::time::Instant { + let now = std::time::Instant::now(); + now.checked_add(wait).unwrap_or(now) } /// [`capture`], killing the command at `deadline` (at most the store timeout). +/// The output is read by the deadline too: a process the command started may +/// hold the pipe open after the command itself exits (`sh -c 'sleep 20 & +/// exit 0'`), and then the read ends at the deadline as a timeout. pub fn capture_until( mut cmd: Command, from_stderr: bool, deadline: std::time::Instant, ) -> Result<(std::process::ExitStatus, Zeroizing>), &'static str> { let wait = deadline.saturating_duration_since(std::time::Instant::now()).min(TIMEOUT); + let end = after(wait); cmd.stdin(Stdio::null()); let mut child = cmd.spawn().map_err(|_| UNAVAILABLE)?; let pipe: Option> = if from_stderr { @@ -313,21 +323,28 @@ pub fn capture_until( } else { child.stdout.take().map(|p| Box::new(p) as Box) }; - let reader = std::thread::spawn(move || drain(pipe)); + // The reader is never joined: a process the command started may still + // hold the pipe open; the reader ends when it closes, and the caller has + // returned at the deadline by then. + let (sent, read) = std::sync::mpsc::sync_channel(1); + std::thread::spawn(move || { + let _ = sent.send(drain(pipe)); + }); let status = match child.wait_timeout(wait) { Ok(Some(status)) => status, _ => { let _ = child.kill(); let _ = child.wait(); - // Not joined: a process the command started may still hold the - // pipe open; the reader ends when it closes, and the caller - // returns at the deadline. - drop(reader); return Err("vault.secret_store_timeout"); } }; - let raw = reader.join().map_err(|_| "vault.secret_store_read_failed")?; - Ok((status, raw)) + match read.recv_timeout(end.saturating_duration_since(std::time::Instant::now())) { + Ok(raw) => Ok((status, raw)), + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => Err("vault.secret_store_timeout"), + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { + Err("vault.secret_store_read_failed") + } + } } /// All of a child's pipe, into a zeroed buffer (at most 1 MiB). @@ -346,8 +363,9 @@ const ITEM_PREFIX: &str = "v1:"; /// The item text of `value`: the marker, then base64, one printable word. pub fn encode(value: &str) -> Zeroizing { - let mut out = - Zeroizing::new(String::with_capacity(ITEM_PREFIX.len() + value.len() * 4 / 3 + 4)); + let mut out = Zeroizing::new(String::with_capacity( + value.len().div_ceil(3).saturating_mul(4).saturating_add(ITEM_PREFIX.len()), + )); out.push_str(ITEM_PREFIX); base64::engine::general_purpose::STANDARD.encode_string(value.as_bytes(), &mut out); out diff --git a/cmux-tui/crates/acpmux/src/vault/tests.rs b/cmux-tui/crates/acpmux/src/vault/tests.rs index 3b0b29192b51..04dae425e041 100644 --- a/cmux-tui/crates/acpmux/src/vault/tests.rs +++ b/cmux-tui/crates/acpmux/src/vault/tests.rs @@ -1224,3 +1224,49 @@ fn a_store_command_crossing_the_deadline_is_stopped_there() { assert_eq!(result.err(), Some("vault.secret_store_timeout")); assert!(start.elapsed() < std::time::Duration::from_secs(3), "stopped at the deadline"); } + +#[test] +#[cfg(unix)] +fn a_store_command_whose_child_keeps_the_pipe_stops_at_the_deadline() { + // The command exits at once; the `sleep` it started holds stdout open. + let mut cmd = std::process::Command::new("sh"); + cmd.args(["-c", "sleep 20 & exit 0"]).stdout(std::process::Stdio::piped()); + let start = std::time::Instant::now(); + let deadline = start + std::time::Duration::from_millis(300); + let result = super::store::capture_until(cmd, false, deadline); + assert_eq!(result.err(), Some("vault.secret_store_timeout")); + assert!(start.elapsed() < std::time::Duration::from_secs(3), "stopped at the deadline"); +} + +#[test] +#[cfg(unix)] +fn a_store_command_that_exits_gives_its_whole_output() { + let mut cmd = std::process::Command::new("sh"); + cmd.args(["-c", "printf done"]).stdout(std::process::Stdio::piped()); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + let (status, out) = super::store::capture_until(cmd, false, deadline).expect("ran"); + assert!(status.success()); + assert_eq!(out.as_slice(), b"done"); +} + +#[test] +fn an_older_daemon_is_told_apart_and_the_refusal_says_to_restart_it() { + use crate::cli::host_vault::{DAEMON_OUTDATED, method_not_found}; + let old: anyhow::Error = + crate::client::DaemonError(crate::rpc::RpcError::method_not_found("_acpmux/vault_pull")) + .into(); + assert!(method_not_found(&old)); + let other: anyhow::Error = + crate::client::DaemonError(crate::rpc::RpcError::new(-32000, "busy")).into(); + assert!(!method_not_found(&other)); + assert!(!method_not_found(&anyhow::anyhow!("connection closed"))); + let shown = format!( + "{:#}", + crate::cli::host_vault::failure_for_test(pull::Failure { + code: DAEMON_OUTDATED.into(), + deleted: Vec::new(), + outcome: None, + }) + ); + assert!(shown.contains("acpmux daemon restart"), "{shown}"); +} diff --git a/cmux-tui/crates/acpmux/src/vault/wire.rs b/cmux-tui/crates/acpmux/src/vault/wire.rs index ae75b105960a..c14e62579b9e 100644 --- a/cmux-tui/crates/acpmux/src/vault/wire.rs +++ b/cmux-tui/crates/acpmux/src/vault/wire.rs @@ -29,7 +29,7 @@ pub const MAX_VAULT_ENV_BYTES: usize = 262_144; /// The bytes `NAME=value` takes in a new process's environment. pub fn env_string_bytes(name: &str, value: &str) -> usize { - name.len() + 1 + value.len() + 1 + name.len().saturating_add(value.len()).saturating_add(2) } /// One env entry of the answer. `value` is `None` for a revoked entry. @@ -85,11 +85,11 @@ pub fn parse(status: u16, body: &[u8]) -> Answer { if items.len() > MAX_ENTRIES { return Answer::Refused("vault.too_many".to_owned()); } - let mut refused = 0; + let mut refused = 0usize; let mut by_name: BTreeMap = BTreeMap::new(); for item in items { let Some(entry) = env_item(item) else { - refused += 1; + refused = refused.saturating_add(1); continue; }; // One entry per name: the newest revision wins. @@ -108,11 +108,11 @@ pub fn parse(status: u16, body: &[u8]) -> Answer { for entry in by_name.into_values() { if let Some(value) = &entry.value { let bytes = env_string_bytes(&entry.name, value); - if bytes > MAX_ENV_STRING_BYTES || total + bytes > MAX_VAULT_ENV_BYTES { - refused += 1; + if bytes > MAX_ENV_STRING_BYTES || total.saturating_add(bytes) > MAX_VAULT_ENV_BYTES { + refused = refused.saturating_add(1); continue; } - total += bytes; + total = total.saturating_add(bytes); } env.push(entry); } @@ -136,7 +136,7 @@ fn env_item(item: &Value) -> Option { /// Base64 (standard, padded or not) of UTF-8 with no NUL, at most /// [`MAX_VALUE_BYTES`]. fn decode_value(text: &str) -> Option> { - if text.len() > MAX_VALUE_BYTES.div_ceil(3) * 4 + 4 { + if text.len() > MAX_VALUE_BYTES.div_ceil(3).saturating_mul(4).saturating_add(4) { return None; } let engine = base64::engine::general_purpose::GeneralPurpose::new( diff --git a/webviews/src/agent-session/acpmux/protocol/generated/methods.ts b/webviews/src/agent-session/acpmux/protocol/generated/methods.ts index 6de85e1ff336..b306a5043afb 100644 --- a/webviews/src/agent-session/acpmux/protocol/generated/methods.ts +++ b/webviews/src/agent-session/acpmux/protocol/generated/methods.ts @@ -262,6 +262,7 @@ export interface AcpmuxRequests { "_acpmux/pair_redeem": { params: unknown; result: unknown }; "_acpmux/vault_pull": { params: unknown; result: unknown }; "_acpmux/vault_forget": { params: unknown; result: unknown }; + "_acpmux/vault_status": { params: unknown; result: unknown }; "_acpmux/turn_attestation": { params: unknown; result: unknown }; } @@ -574,6 +575,7 @@ export const ACPMUX_METHODS: ReadonlyArray<{ name: string; kind: AcpmuxMethodKin { name: "_acpmux/pair_redeem", kind: "request", acp: false, typed: false }, { name: "_acpmux/vault_pull", kind: "request", acp: false, typed: false }, { name: "_acpmux/vault_forget", kind: "request", acp: false, typed: false }, + { name: "_acpmux/vault_status", kind: "request", acp: false, typed: false }, { name: "_acpmux/turn_attestation", kind: "request", acp: false, typed: false }, { name: "_acpmux/event", kind: "notification", acp: false, typed: true }, { name: "_acpmux/message", kind: "notification", acp: false, typed: true },