diff --git a/crates/fractal-api-client-v0/src/api/client_v0.rs b/crates/fractal-api-client-v0/src/api/client_v0.rs index 2eef97115de..22a3fd1c656 100644 --- a/crates/fractal-api-client-v0/src/api/client_v0.rs +++ b/crates/fractal-api-client-v0/src/api/client_v0.rs @@ -36,6 +36,9 @@ pub use fractal_api_core::api::client_v0::{ClientSession, CollectionSubscribe, E pub use fractal_api_core::api::client_v0::{authenticate, bound_mission_cards, capabilities, conversation_session_id, cut_oversized_timeline_entry, events, fabric_boundary_forbidden, message_provenance, mission_list_cards_with, native_page_cursor_snapshot, now, person_filter}; pub use fractal_api_core::api::client_v0::{published_missions, published_runtimes, require_scope, resources, start_operation_report, timeline_value, validation}; pub mod arrangements; +#[cfg(test)] +#[path = "client_v0/client_control_tests.rs"] +mod client_control_tests; pub mod attention_budget; pub mod assistant; pub mod browser_pairing; @@ -98,6 +101,11 @@ mod stream_timeline_tests; /// The client-v0 routes, added to the API router as a [`RouteFamily`]. pub fn routes(app: Router) -> Router { app + .route("/v1/client/device-commands", post(fractal_api_core::api::client_commands::issue_route) + .layer(axum::extract::DefaultBodyLimit::max(fractal_api_core::api::client_commands::MAX_BODY_BYTES))) + .route("/v1/client/device-commands/{*id}", get(fractal_api_core::api::client_commands::read_route)) + .route("/v1/client/device-results", post(fractal_api_core::api::client_commands::result_route) + .layer(axum::extract::DefaultBodyLimit::max(fractal_api_core::api::client_commands::MAX_BODY_BYTES))) .route("/v1/client/test-clients/enroll", post(fractal_api_core::api::client_control_enrollment::enroll_http)) .route("/v1/client/test-clients/revoke", post(fractal_api_core::api::client_control_enrollment::revoke_http)) .route("/v1/client/control-grants", post(fractal_api_core::api::client_control_grants::issue_grant_route)) @@ -550,6 +558,14 @@ async fn collection_items_with_windows( read_permit: tokio::sync::OwnedSemaphorePermit, windows: Option>, ) -> Result<(ClientSnapshot, Vec, bool, Option), ApiError> { + if request.collection == "commands" { + let state = state.clone(); + let session = session.clone(); + return blocking_store(move || { + let items = fractal_api_core::api::client_commands::collection_items(&state, &session)?; + Ok((new_client_snapshot(&state), items, false, None)) + }).await; + } if !matches!( request.collection.as_str(), "missions" | "attention" | "agents" | "work" | "glasses" | "arrangements" | "ordered-memberships" | "summary" @@ -2115,6 +2131,7 @@ async fn collection_stream_socket_with_admission( let mut views = state.store.subscribe_collection_views(); let mut presence_changes = super::client_presence::subscribe(); let mut push_changes = super::client_push::subscribe(); + let mut command_changes = fractal_api_core::api::client_commands::subscribe(); let mut native_watch: Option = None; let mut native_watch_pending: Option>> = None; let windows = collection_windows::Windows::attach(&state.store); @@ -2435,8 +2452,12 @@ async fn collection_stream_socket_with_admission( } refresh.extend(subscriptions.iter().filter(|(_, s)| s.ivm.is_some()).map(|(id, _)| id.clone())); } + _ = command_changes.recv(), if !command_waiting => { + reread_due.extend(subscriptions.iter().filter(|(_, s)| s.request.collection == "commands").map(|(id, _)| id.clone())); + } result = changed.changed(), if !command_waiting => { if result.is_err() { return; } + reread_due.extend(subscriptions.iter().filter(|(_, s)| s.request.collection == "commands").map(|(id, _)| id.clone())); // Weigh only the commits since the last look: a window is due for a reread when // one of them can change it. A published view rereads when it publishes instead; // a commit that can change the agents roster asks its refresher to publish. diff --git a/crates/fractal-api-client-v0/src/api/client_v0/client_control_tests.rs b/crates/fractal-api-client-v0/src/api/client_v0/client_control_tests.rs new file mode 100644 index 00000000000..7f7a64f1d88 --- /dev/null +++ b/crates/fractal-api-client-v0/src/api/client_v0/client_control_tests.rs @@ -0,0 +1,127 @@ +use super::*; +use base64::Engine as _; +use ring::signature::KeyPair as _; +use fractal_api_core::api::{client_commands, client_control_enrollment, client_control_grants}; +use fractal_schema::client_control::{ClientResultBinding, ClientVisibleState, CLIENT_RESULT_SIGNED_FIELDS, client_result_canonical_bytes}; + +#[tokio::test] +async fn client_control_result_key_and_active_grant_never_authorize_message_send() { + let root = tempfile::tempdir().unwrap(); + let (state, key, grant, issuer) = client_control_grants::test_support::fixture(root.path()); + let client = client_control_enrollment::enrolled(&state, &grant.device).unwrap().session; + // The existing endpoint's deny-unknown-fields wire decoder must accept the + // generated grant field, not just an internally constructed request. + let command = client_commands::issue(&state, &issuer, serde_json::from_value(json!({ + "api_version": CLIENT_API_VERSION, "device": grant.device, "kind": "phone.inspect", + "args": {}, "grant_id": grant.grant_id, + })).unwrap()).unwrap(); + let binding = ClientResultBinding { + purpose: "client.device-result.v1".into(), command_kind: "phone.inspect".into(), device: grant.device.clone(), + session: client.actor.clone(), grant_id: grant.grant_id.clone(), run_id: grant.run_id.clone(), + command_id: command["id"].as_str().unwrap().into(), outcome: "completed".into(), error_code: None, + visible_state: ClientVisibleState { route: "inbox".into(), screen: "NowRoot".into(), visible_item_ids: vec![], opened_item: None }, + }; + let rng = ring::rand::SystemRandom::new(); + let signature = fractal_client::DeviceSignature { + signer: client.authority_actor.clone(), key: format!("p256:{}", base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(key.public_key().as_ref())), + chain: vec![], nonce: "client-control-cross-purpose".into(), signed_at_unix_ms: client_now_ms() as u64, + signature: base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(key.sign(&rng, &client_result_canonical_bytes(&binding).unwrap()).unwrap().as_ref()), + format: "fields-v1".into(), signed_fields: CLIENT_RESULT_SIGNED_FIELDS.iter().map(|field| (*field).into()).collect(), + }; + let result = client_commands::report(&state, &client, serde_json::from_value(json!({ + "command_id": binding.command_id, "outcome":"completed", "error_code":null, + "result": binding.visible_state, "binding": binding, "signature": signature, + })).unwrap()).unwrap(); + assert_eq!(result["state"], "completed"); + assert!(result["receipt_claim_id"].as_str().is_some()); + let snapshot = new_client_snapshot(&state); + let request: ActionRequest = serde_json::from_value(json!({ + "api_version": CLIENT_API_VERSION, "id":"action/client-control-cross-purpose", "type":"message.send", + "idempotency_key":"client-control-cross-purpose", "fence":{"snapshot_id":snapshot.id,"subject_revisions":{}}, + "parameters":{"to":"agent/client-control-fixture/producer","content":"must never be sent", + "signature":signature,"grant_id":grant.grant_id}, + })).unwrap(); + let before = state.store.claims_for_kind_at("message.sent", None, true, 256).unwrap().claims.len(); + let error = action(State(state.clone()), Extension(snapshot), Extension(client), Json(request)).await.unwrap_err(); + assert_eq!(error.status, StatusCode::FORBIDDEN); + assert_eq!(state.store.claims_for_kind_at("message.sent", None, true, 256).unwrap().claims.len(), before); +} + +#[tokio::test] +async fn client_control_lifecycle_uses_existing_commands_collection_before_any_command() { + let root = tempfile::tempdir().unwrap(); + let (state, _, grant, issuer) = client_control_grants::test_support::fixture(root.path()); + let client = client_control_enrollment::enrolled(&state, &grant.device).unwrap().session; + let request: CollectionSubscribe = serde_json::from_value(json!({"kind":"subscribe","id":"commands","collection":"commands"})).unwrap(); + let permit = Arc::new(tokio::sync::Semaphore::new(1)).acquire_owned().await.unwrap(); + let (_, items, more, _) = collection_items_with_windows(&state, &client, &request, permit, None).await.unwrap(); + assert!(!more); + assert_eq!(items.len(), 1); + assert_eq!(items[0]["kind"], "client-control-grant-lifecycle"); + assert_eq!(items[0]["state"], "started"); + assert_eq!(items[0]["session"], client.actor); + client_control_grants::revoke_grant(&state, &issuer, &grant.grant_id).unwrap(); + let permit = Arc::new(tokio::sync::Semaphore::new(1)).acquire_owned().await.unwrap(); + let (_, items, _, _) = collection_items_with_windows(&state, &client, &request, permit, None).await.unwrap(); + assert_eq!(items[0]["state"], "revoked"); +} + +#[tokio::test] +async fn client_control_idle_stream_observes_revocation_and_run_end() { + for run_end in [false, true] { + let root = tempfile::tempdir().unwrap(); + let (state, _, grant, issuer) = client_control_grants::test_support::fixture(root.path()); + let client = client_control_enrollment::enrolled(&state, &grant.device).unwrap().session; + let stream_state = state.clone(); + let app = axum::Router::new().route("/stream", axum::routing::get(move |upgrade: WebSocketUpgrade| { + let state = stream_state.clone(); + let client = client.clone(); + async move { + let windows = collection_windows::Windows::attach(&state.store); + upgrade.on_upgrade(move |socket| collection_stream_socket_with_reader( + socket, state, client, None, + move |state, session, request, permit| { + let windows = windows.clone(); + async move { collection_items_with_windows(&state, &session, &request, permit, windows).await } + }, + )) + } + })); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); + let (mut socket, _) = tokio_tungstenite::connect_async(format!("ws://{address}/stream")).await.unwrap(); + socket.send(tokio_tungstenite::tungstenite::Message::Text( + json!({"kind":"subscribe","id":"commands","collection":"commands"}).to_string().into(), + )).await.unwrap(); + let initial = super::tests::next_data_frame(&mut socket).await; + let started = &initial["items"][0]; + assert_eq!(started["kind"], "client-control-grant-lifecycle"); + assert_eq!(started["state"], "started"); + assert_eq!(started["grant_id"], grant.grant_id); + // Fence the initial send before changing authority. No command, result, + // resubscription or manual lifecycle refresh may wake this idle client. + socket.send(tokio_tungstenite::tungstenite::Message::Ping(Vec::new().into())).await.unwrap(); + super::tests::next_pong(&mut socket).await; + if run_end { + state.store.set_mission_run_state(&grant.run_id, "cancelled", "terminal", Some("idle run ended")).unwrap(); + } else { + client_control_grants::revoke_grant(&state, &issuer, &grant.grant_id).unwrap(); + } + let changed = tokio::time::timeout(Duration::from_secs(10), async { + loop { + let frame = super::tests::next_data_frame(&mut socket).await; + if let Some(event) = frame["upserts"].as_array().and_then(|items| items.iter() + .find(|item| item["kind"] == "client-control-grant-lifecycle" && item["state"] != "started")) { + break event.clone(); + } + } + }).await.expect("idle client must receive the terminal lifecycle transition"); + assert_eq!(changed["id"], started["id"]); + assert_eq!(changed["state"], if run_end { "ended" } else { "revoked" }); + assert_eq!(changed["reason"], if run_end { "run-ended" } else { "revoked" }); + assert!(changed["sequence"].as_u64().unwrap() > started["sequence"].as_u64().unwrap()); + socket.close(None).await.unwrap(); + server.abort(); + } +} diff --git a/crates/fractal-api-core/src/api/client_commands.rs b/crates/fractal-api-core/src/api/client_commands.rs index a4bf9277f96..d1c64c3420b 100644 --- a/crates/fractal-api-core/src/api/client_commands.rs +++ b/crates/fractal-api-core/src/api/client_commands.rs @@ -1,5 +1,5 @@ -//! Foreground phone commands are member-local, bounded, ephemeral correlation records. -//! No claims, durable receipts, replication, or replay participate in this protocol. +//! Foreground device commands are member-local, bounded, ephemeral correlation records. +//! Granted test-client commands additionally produce durable, purpose-bound receipts. use super::client_v0::{acting_party, forbidden, paired_client_session, require_scope, revalidate_session, validation, ClientSession}; use super::*; use std::collections::BTreeMap; @@ -20,14 +20,17 @@ pub struct DeviceCommandRequest { pub kind: String, pub args: Value, pub ttl_ms: Option, + pub grant_id: Option, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Clone, Deserialize)] #[serde(deny_unknown_fields)] pub struct DeviceResultRequest { pub command_id: String, pub outcome: String, pub result: Option, pub error_code: Option, + pub binding: Option, + pub signature: Option, } #[derive(Clone)] struct Command { @@ -64,12 +67,80 @@ impl Command { pub fn subscribe() -> tokio::sync::broadcast::Receiver { CHANGES.subscribe() } fn changed(member: &str) { let _ = CHANGES.send(member.to_owned()); } fn storage_error() -> ApiError { ApiError::internal("local device command memory is unavailable") } +fn persist_receipt(state: &AppState, row: &mut Command) -> Result<(), ApiError> { + if !row.terminal() || row.resource.get("grant_id").is_none() || row.resource.get("receipt_claim_id").is_some() { + return Ok(()); + } + let command_id = row.resource["id"].as_str().ok_or_else(storage_error)?; + let receipt = state.store.append_client_operation_claim(&ClaimInput { + subject: format!("custom/client/command-receipt/{}", command_id.trim_start_matches("device-command/")), + kind: "custom.client.command-receipt".into(), + actor: Some(row.person.clone()), + fields: serde_json::from_value(json!({ + "commandId": command_id, "terminalState": row.resource["state"], + "issuer": row.resource["issued_by"], "device": row.resource["device_id"], + "session": row.resource["session_actor"], "grantId": row.resource["grant_id"], + "runId": row.resource["run_id"], "commandKind": row.resource["command_kind"], + "verifiedResultSignature": row.resource.get("verified_result_signature"), + "signedResult": row.resource.get("signed_result"), + })).map_err(ApiError::internal)?, + evidence: Vec::new(), expected_subject: Some(None), + idempotency_key: Some(format!("client-command-receipt:{command_id}")), + }).map_err(ApiError::internal)?; + row.resource["receipt_claim_id"] = json!(receipt.id); + super::signal_changed(state); + Ok(()) +} +fn control_refusal(code: &'static str) -> ApiError { + ApiError { status: StatusCode::FORBIDDEN, code: code.into(), message: format!("client control refused: {code}"), details: Box::default() } +} +fn verify_result(state: &AppState, session: &ClientSession, row: &Command, request: &DeviceResultRequest) -> Result<(), ApiError> { + use fractal_schema::client_control::{CLIENT_RESULT_SIGNED_FIELDS, client_result_canonical_bytes}; + let Some(grant_id) = row.resource.get("grant_id").and_then(Value::as_str) else { + if request.binding.is_some() || request.signature.is_some() { return Err(forbidden("signed results require a command grant")); } + return Ok(()); + }; + if row.terminal() || client_now_ms() >= row.expires { return Err(forbidden("device result is outside its command window")); } + let binding = request.binding.as_ref().ok_or_else(|| forbidden("device result requires a signed binding"))?; + let signature = request.signature.as_ref().ok_or_else(|| forbidden("device result requires a signature"))?; + let enrollment = super::client_control_enrollment::enrolled(state, &binding.device)?; + let grant = super::client_control_grants::check_result_window( + state, grant_id, row.resource["run_id"].as_str().unwrap_or_default(), + row.resource["device_id"].as_str().unwrap_or_default(), + row.resource["issued_by"].as_str().unwrap_or_default(), + row.resource["command_kind"].as_str().unwrap_or_default(), + &row.resource["args"], &request.command_id, + )?; + if binding.purpose != "client.device-result.v1" + || binding.command_kind != row.resource["command_kind"] + || binding.device != row.resource["device_id"] + || binding.session != session.actor + || binding.grant_id != grant_id || binding.run_id != grant.run_id + || binding.command_id != request.command_id || binding.outcome != request.outcome + || binding.error_code != request.error_code + || enrollment.pairing_subject != row.grant + || signature.signer != session.authority_actor || signature.key != enrollment.public_key + || signature.format != "fields-v1" + || !signature.signed_fields.iter().map(String::as_str).eq(CLIENT_RESULT_SIGNED_FIELDS.iter().copied()) + || request.result.as_ref().is_some_and(|result| serde_json::to_value(&binding.visible_state).as_ref().ok() != Some(result)) + { return Err(forbidden("device result binding does not match its command")); } + super::client_control_grants::check_screen(&grant, &binding.visible_state.screen)?; + if binding.visible_state.screen == "Attention" { + let item = binding.visible_state.opened_item.as_ref().ok_or_else(|| control_refusal("screen-not-granted"))?; + super::client_control_grants::check_item(&grant, &item.id)?; + } + let bytes = client_result_canonical_bytes(binding).map_err(ApiError::internal)?; + if !smallclaims::principal::verify_key_signature(&enrollment.public_key, &bytes, &signature.signature) { + return Err(forbidden("invalid device result signature")); + } + Ok(()) +} fn missing() -> ApiError { ApiError { status: StatusCode::NOT_FOUND, code: "not-found".into(), message: "device command not found".into(), details: Box::default() } } fn capacity() -> ApiError { ApiError { status: StatusCode::TOO_MANY_REQUESTS, code: "resource-exhausted".into(), message: "device command capacity exceeded".into(), details: Box::default() } } pub fn scope(kind: &str) -> Result<&'static str, ApiError> { match kind { "phone.inspect" | "phone.navigate" => Ok("read.projections"), "phone.message.read" => Ok("control.messages"), _ => Err(validation("unknown phone command kind")) } } -pub fn paired(session: &ClientSession) -> bool { session.pairing_grant.is_some() && session.pairing_issued_here && session.transport != "unix" } +pub fn paired(session: &ClientSession) -> bool { session.pairing_grant.is_some() && session.pairing_issued_here } pub fn capable(session: &ClientSession) -> bool { paired(session) || acting_party(session) && (session.allows("read.projections") || session.allows("control.messages")) } fn bounded_string(value: &Value) -> bool { value.as_str().is_some_and(|s| !s.is_empty() && s.len() <= 4096 && !s.chars().any(char::is_control)) } fn fields(value: &Value, required: &[&str], optional: &[&str]) -> bool { @@ -102,13 +173,42 @@ fn sweep(rows: &mut BTreeMap<(String, String), Command>, member: &str, now: u128 for ((node, _), row) in rows.iter_mut().filter(|((node, _), _)| node == member) { row.expire(now, super::client_presence::has_collection_stream(node, row.resource["session_actor"].as_str().unwrap_or_default())); } - rows.retain(|_, row| row.terminal_at.is_none_or(|at| now.saturating_sub(at) < RETENTION_MS)); + rows.retain(|_, row| row.terminal_at.is_none_or(|at| now.saturating_sub(at) < RETENTION_MS) + || row.resource.get("grant_id").is_some() && row.resource.get("receipt_claim_id").is_none()); +} +fn sweep_durable(rows: &mut BTreeMap<(String, String), Command>, state: &AppState, now: u128) -> Result<(), ApiError> { + sweep(rows, &state.node, now); + for (_, row) in rows.iter_mut().filter(|((node, _), _)| node == &state.node) { + if !row.terminal() && let Some(grant_id) = row.resource.get("grant_id").and_then(Value::as_str) { + match super::client_control_grants::check_result_window( + state, grant_id, row.resource["run_id"].as_str().unwrap_or_default(), + row.resource["device_id"].as_str().unwrap_or_default(), + row.resource["issued_by"].as_str().unwrap_or_default(), + row.resource["command_kind"].as_str().unwrap_or_default(), &row.resource["args"], + row.resource["id"].as_str().unwrap_or_default(), + ) { + Ok(_) => {} + Err(error) if error.status == StatusCode::FORBIDDEN => { + row.resource["state"] = json!("refused"); + row.resource["error_code"] = json!(error.code); + row.terminal_at = Some(now); + row.stamp(now); + } + Err(error) => return Err(error), + } + } + persist_receipt(state, row)?; + } + Ok(()) } fn admit(rows: &BTreeMap<(String, String), Command>, member: &str, device: &str) -> Result<(), ApiError> { if rows.len() >= MAX_COMMANDS || rows.iter().filter(|((node, _), row)| node == member && row.resource["device_id"] == device && !row.terminal() ).count() >= MAX_PENDING_PER_DEVICE { Err(capacity()) } else { Ok(()) } } +fn admit_receipt(rows: &BTreeMap<(String, String), Command>) -> Result<(), ApiError> { + if rows.len() >= MAX_COMMANDS { Err(control_refusal("receipt-capacity")) } else { Ok(()) } +} fn visible(row: &Command, session: &ClientSession) -> bool { if session.pairing_grant.is_some() { paired(session) && row.resource["session_actor"] == session.actor && session.pairing_grant.as_deref() == Some(row.grant.as_str()) } else { row.resource["issued_by"] == session.actor && row.issuer_grant.is_none() && (session.authority_actor == row.person || session.transport == "unix" && session.authority_actor.starts_with("agent/") && acting_party(session)) } @@ -122,13 +222,70 @@ fn authorize_read(row: &Command, session: &ClientSession) -> Result<(), ApiError /// Requires a running Tokio runtime to own deadline and retention wakeups. pub fn issue(state: &AppState, session: &ClientSession, request: DeviceCommandRequest) -> Result { let session = revalidate_session(state, session)?; + match issue_inner(state, &session, &request) { + Err(mut error) if session.authority_actor.starts_with("agent/") && acting_party(&session) + && state.local_only() && state.owning_person.as_deref() == Some("person/client-control-fixture") + && matches!(error.code.as_str(), "device-not-test-client" | "grant-missing" | "grant-expired" + | "screen-not-granted" | "verb-not-granted" | "budget-exhausted" | "actor-not-granted") => + { + let now = client_now_ms(); + let mut rows = ROWS.lock().map_err(|_| storage_error())?; + sweep_durable(&mut rows, state, now)?; + admit_receipt(&rows)?; + let mut suffix = crate::model::new_request_id(); + suffix.replace_range(14..15, "8"); + let id = format!("device-command/{suffix}"); + let run_id = request.grant_id.as_deref().and_then(|id| super::client_control_grants::lookup(state, id).ok()).map(|grant| grant.run_id); + let receipt = state.store.append_client_operation_claim(&ClaimInput { + subject: format!("custom/client/command-receipt/{suffix}"), + kind: "custom.client.command-receipt".into(), actor: Some(session.authority_actor.clone()), + fields: serde_json::from_value(json!({ + "commandId": id, "terminalState": "refused", "issuer": session.actor, + "device": request.device, "grantId": request.grant_id, "runId": run_id, + "commandKind": request.kind, "screen": request.args.get("screen"), "reason": error.code, + "verifiedResultSignature": null, "signedResult": null, + })).map_err(ApiError::internal)?, + evidence: Vec::new(), expected_subject: Some(None), + idempotency_key: Some(format!("client-command-receipt:{id}")), + }).map_err(ApiError::internal)?; + let resource = json!({"id":id,"kind":"device-command","revision":format!("{id}:refused"), + "updated_at":client_timestamp(now),"issued_at":client_timestamp(now),"expires_at":client_timestamp(now), + "state":"refused","issued_by":session.actor,"device_id":request.device, + "command_kind":request.kind,"args":request.args,"error_code":error.code,"receipt_claim_id":receipt.id}); + rows.insert((state.node.clone(), id.clone()), Command { resource, expires: now, terminal_at: Some(now), + person: session.authority_actor.clone(), grant: String::new(), issuer_grant: session.pairing_grant.clone() }); + error.details.insert("command_id".into(), json!(id)); + error.details.insert("receipt_claim_id".into(), json!(receipt.id)); + super::signal_changed(state); + Err(error) + } + result => result, + } +} +fn issue_inner(state: &AppState, session: &ClientSession, request: &DeviceCommandRequest) -> Result { if !super::wire_names::api_version(&request.api_version) { return Err(validation("unsupported api_version")); } - if !request.device.starts_with("device/") || request.device.len() <= "device/".len() || request.device.len() > 4096 { return Err(validation("invalid device selector")); } - validate_args(&request.kind, &request.args)?; - require_scope(&session, scope(&request.kind)?)?; + if !request.device.starts_with("device/") || request.device.len() <= "device/".len() + || request.device.len() > 4096 || request.device.chars().any(char::is_control) + { return Err(validation("invalid device selector")); } + if request.grant_id.as_ref().is_some_and(|id| !id.starts_with("client-control-grant/") + || id.len() <= "client-control-grant/".len() || id.len() > 4096 || id.chars().any(char::is_control)) + { return Err(validation("invalid client control grant selector")); } + if request.kind.is_empty() || request.kind.len() > 128 || request.kind.chars().any(char::is_control) + || !request.args.is_object() || serde_json::to_vec(&request.args).map_or(true, |bytes| bytes.len() > 8192) + { return Err(validation("invalid command verb or args")); } if !acting_party(&session) { return Err(forbidden("device commands require a concrete person or local agent")); } let ttl = request.ttl_ms.unwrap_or(30_000); if !(1..=60_000).contains(&ttl) { return Err(validation("ttl_ms must be 1 through 60000")); } + let controlled = session.authority_actor.starts_with("agent/") || request.grant_id.is_some(); + let enrollment = if controlled { Some(super::client_control_enrollment::enrolled(state, &request.device)?) } else { None }; + if controlled && request.grant_id.is_none() { + return Err(control_refusal("grant-missing")); + } + if controlled && !matches!(request.kind.as_str(), "phone.inspect" | "phone.navigate") { + return Err(control_refusal("verb-not-granted")); + } + validate_args(&request.kind, &request.args)?; + require_scope(session, scope(&request.kind)?)?; let mut before = None; let mut target: Option = None; loop { @@ -148,21 +305,40 @@ pub fn issue(state: &AppState, session: &ClientSession, request: DeviceCommandRe // The existing generator supplies random UUID bits; commands use the v8 namespace. suffix.replace_range(14..15, "8"); let id = format!("device-command/{suffix}"); - let resource = json!({"id":id,"kind":"device-command","revision":format!("{id}:pending"),"updated_at":client_timestamp(now),"command_kind":request.kind,"args":request.args,"device_id":request.device,"session_actor":phone.actor,"issued_by":session.actor,"issued_at":client_timestamp(now),"expires_at":client_timestamp(now + u128::from(ttl)),"state":"pending"}); + let mut resource = json!({"id":id,"kind":"device-command","revision":format!("{id}:pending"),"updated_at":client_timestamp(now),"command_kind":request.kind,"args":request.args,"device_id":request.device,"session_actor":phone.actor,"issued_by":session.actor,"issued_at":client_timestamp(now),"expires_at":client_timestamp(now + u128::from(ttl)),"state":"pending"}); { let mut rows = ROWS.lock().map_err(|_| storage_error())?; - sweep(&mut rows, &state.node, now); + sweep_durable(&mut rows, state, now)?; admit(&rows, &state.node, &request.device)?; + if let Some(grant_id) = &request.grant_id { + let grant = super::client_control_grants::lookup(state, grant_id)?; + let enrollment = enrollment.as_ref().ok_or_else(|| control_refusal("device-not-test-client"))?; + if enrollment.pairing_subject != target.subject || grant.session != phone.actor { + return Err(control_refusal("device-not-test-client")); + } + let grant = super::client_control_grants::check_and_consume( + state, grant_id, &grant.run_id, &request.device, &session.authority_actor, + &request.kind, &request.args, &id, + )?; + resource["grant_id"] = json!(grant.grant_id); + resource["run_id"] = json!(grant.run_id); + resource["control_grant"] = grant.resource("active"); + } rows.insert((state.node.clone(), id.clone()), Command { resource: resource.clone(), expires: now + u128::from(ttl), terminal_at: None, person: phone.authority_actor, grant: target.subject, issuer_grant: session.pairing_grant.clone() }); } changed(&state.node); let member = state.node.clone(); + let timer_state = state.clone(); tokio::spawn(async move { tokio::time::sleep(Duration::from_millis(ttl)).await; - if let Ok(mut rows) = ROWS.lock() { sweep(&mut rows, &member, client_now_ms()); } + if let Ok(mut rows) = ROWS.lock() && let Err(error) = sweep_durable(&mut rows, &timer_state, client_now_ms()) { + tracing::error!(message = %error.message, "device command expiry receipt failed"); + } changed(&member); tokio::time::sleep(Duration::from_millis(RETENTION_MS as u64)).await; - if let Ok(mut rows) = ROWS.lock() { rows.remove(&(member.clone(), id)); } + if let Ok(mut rows) = ROWS.lock() && let Err(error) = sweep_durable(&mut rows, &timer_state, client_now_ms()) { + tracing::error!(message = %error.message, "device command retention receipt failed"); + } changed(&member); }); Ok(resource) @@ -170,7 +346,7 @@ pub fn issue(state: &AppState, session: &ClientSession, request: DeviceCommandRe pub fn read(state: &AppState, session: &ClientSession, id: &str) -> Result { let session = revalidate_session(state, session)?; let mut rows = ROWS.lock().map_err(|_| storage_error())?; - sweep(&mut rows, &state.node, client_now_ms()); + sweep_durable(&mut rows, state, client_now_ms())?; let row = rows.get(&(state.node.clone(), id.to_owned())).ok_or_else(missing)?; // Paired callers may also be issuers. GET provides their correlation surface, // while their collection window remains strictly addressed to their own phone. @@ -190,14 +366,23 @@ pub fn report(state: &AppState, session: &ClientSession, request: DeviceResultRe if !paired(&session) { return Err(forbidden("device results require a paired session issued by this member")); } if !matches!(request.outcome.as_str(), "completed" | "not-foreground" | "expired" | "refused" | "failed") || request.error_code.as_ref().is_some_and(|s| s.is_empty() || s.len() > 128 || !s.bytes().all(|b| b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_' | b'.'))) - || request.result.as_ref().is_some_and(|v| serde_json::to_vec(v).map_or(true, |b| b.len() > 8192)) { return Err(validation("invalid device result")); } + || request.result.as_ref().is_some_and(|v| serde_json::to_vec(v).map_or(true, |b| b.len() > 8192)) + || request.binding.as_ref().is_some_and(|v| serde_json::to_vec(v).map_or(true, |b| b.len() > 8192)) + || request.signature.as_ref().is_some_and(|v| serde_json::to_vec(v).map_or(true, |b| b.len() > 4096)) + { return Err(validation("invalid device result")); } let resource = { let mut rows = ROWS.lock().map_err(|_| storage_error())?; let now = client_now_ms(); - sweep(&mut rows, &state.node, now); + sweep_durable(&mut rows, state, now)?; let row = rows.get_mut(&(state.node.clone(), request.command_id.clone())).ok_or_else(missing)?; authorize_read(row, &session)?; + verify_result(state, &session, row, &request)?; + if request.binding.is_some() { + row.resource["verified_result_signature"] = serde_json::to_value(&request.signature).map_err(ApiError::internal)?; + row.resource["signed_result"] = serde_json::to_value(&request.binding).map_err(ApiError::internal)?; + } row.finish(&request, now, super::client_presence::has_collection_stream(&state.node, &session.actor)); + persist_receipt(state, row)?; row.resource.clone() }; changed(&state.node); @@ -209,8 +394,8 @@ pub fn collection_items(state: &AppState, session: &ClientSession) -> Result, Extension(session): Exten } } +#[cfg(test)] +#[path = "client_commands_result_tests.rs"] +mod control_result_tests; + #[cfg(test)] mod tests { use super::*; fn command() -> Command { Command { resource: json!({"state":"pending"}), expires: 100, terminal_at: None, person: "person/ada".into(), grant: "grant/test".into(), issuer_grant: None } } - fn result() -> DeviceResultRequest { DeviceResultRequest { command_id: "test".into(), outcome: "completed".into(), result: Some(json!({"ok":true})), error_code: None } } + fn result() -> DeviceResultRequest { DeviceResultRequest { command_id: "test".into(), outcome: "completed".into(), result: Some(json!({"ok":true})), error_code: None, binding: None, signature: None } } #[test] fn daemon_deadline_and_terminal_precedence() { let mut early = command(); early.finish(&result(), 99, false); early.expire(200, false); assert_eq!(early.resource["state"], "completed"); diff --git a/crates/fractal-api-core/src/api/client_commands_result_tests.rs b/crates/fractal-api-core/src/api/client_commands_result_tests.rs new file mode 100644 index 00000000000..600b4ad4842 --- /dev/null +++ b/crates/fractal-api-core/src/api/client_commands_result_tests.rs @@ -0,0 +1,280 @@ +use super::*; +use base64::Engine as _; +use fractal_schema::client_control::{ClientResultBinding, ClientVisibleState, CLIENT_RESULT_SIGNED_FIELDS, client_result_canonical_bytes}; +use ring::signature::KeyPair as _; + +fn command_request(grant: &super::super::client_control_grants::ControlGrant) -> DeviceCommandRequest { + DeviceCommandRequest { + api_version: "st3.client.v0".into(), device: grant.device.clone(), + kind: "phone.inspect".into(), args: json!({}), ttl_ms: Some(30_000), + grant_id: Some(grant.grant_id.clone()), + } +} +fn signed_request(command: &Value, client: &ClientSession, key: &ring::signature::EcdsaKeyPair) -> DeviceResultRequest { + let binding = ClientResultBinding { + purpose: "client.device-result.v1".into(), command_kind: "phone.inspect".into(), + device: command["device_id"].as_str().unwrap().into(), session: client.actor.clone(), + grant_id: command["grant_id"].as_str().unwrap().into(), run_id: command["run_id"].as_str().unwrap().into(), + command_id: command["id"].as_str().unwrap().into(), outcome: "completed".into(), error_code: None, + visible_state: ClientVisibleState { route: "inbox".into(), screen: "NowRoot".into(), visible_item_ids: vec!["message/seed".into()], opened_item: None }, + }; + let rng = ring::rand::SystemRandom::new(); + let signature = key.sign(&rng, &client_result_canonical_bytes(&binding).unwrap()).unwrap(); + DeviceResultRequest { + command_id: binding.command_id.clone(), outcome: binding.outcome.clone(), + result: Some(serde_json::to_value(&binding.visible_state).unwrap()), error_code: None, + signature: Some(fractal_client::DeviceSignature { + signer: client.authority_actor.clone(), key: format!("p256:{}", base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(key.public_key().as_ref())), + chain: vec![], nonce: "result-test".into(), signed_at_unix_ms: client_now_ms() as u64, + signature: base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(signature.as_ref()), + format: "fields-v1".into(), signed_fields: CLIENT_RESULT_SIGNED_FIELDS.iter().map(|field| (*field).into()).collect(), + }), binding: Some(binding), + } +} + +#[tokio::test] +async fn narrow_grant_accepts_signed_result_but_never_message_authority_and_records_one_receipt() { + let root = tempfile::tempdir().unwrap(); + let (state, key, grant, issuer) = super::super::client_control_grants::test_support::fixture(root.path()); + let client = super::super::client_control_enrollment::enrolled(&state, &grant.device).unwrap().session; + assert!(!client.allows("control.messages")); + // message.send performs this same mandatory scope check before signature authorization. + assert!(require_scope(&client, "control.messages").is_err()); + let command = issue(&state, &issuer, command_request(&grant)).unwrap(); + let request = signed_request(&command, &client, &key); + let result = report(&state, &client, request.clone()).unwrap(); + assert_eq!(result["state"], "completed"); + let receipt_id = result["receipt_claim_id"].as_str().unwrap(); + assert!(!receipt_id.is_empty()); + assert!(report(&state, &client, request).is_err(), "terminal command rejects replay"); + let subject = format!("custom/client/command-receipt/{}", command["id"].as_str().unwrap().trim_start_matches("device-command/")); + let receipts = state.store.claims_for(&subject, Some("custom.client.command-receipt")).unwrap(); + assert_eq!(receipts.len(), 1); + assert_eq!(receipts[0].id, receipt_id); + assert_eq!(receipts[0].body["fields"]["device"], grant.device); + assert_eq!(receipts[0].body["fields"]["grantId"], grant.grant_id); + assert_eq!(receipts[0].body["fields"]["runId"], grant.run_id); + assert_eq!(receipts[0].body["fields"]["verifiedResultSignature"]["key"], result["verified_result_signature"]["key"]); + let stored_binding: ClientResultBinding = serde_json::from_value(receipts[0].body["fields"]["signedResult"].clone()).unwrap(); + let stored_signature: fractal_client::DeviceSignature = serde_json::from_value(receipts[0].body["fields"]["verifiedResultSignature"].clone()).unwrap(); + let enrolled_key = super::super::client_control_enrollment::enrolled(&state, &grant.device).unwrap().public_key; + assert_eq!(stored_signature.key, enrolled_key); + assert!(smallclaims::principal::verify_key_signature( + &enrolled_key, &client_result_canonical_bytes(&stored_binding).unwrap(), &stored_signature.signature, + )); +} + +#[tokio::test] +async fn command_revoke_race_is_fail_closed_in_both_serialization_orders() { + for revoke_first in [false, true] { + let root = tempfile::tempdir().unwrap(); + let (state, key, grant, issuer) = super::super::client_control_grants::test_support::fixture(root.path()); + let client = super::super::client_control_enrollment::enrolled(&state, &grant.device).unwrap().session; + // Two actors are released in a controlled order, without scheduling + // sleeps. Both legal lock orderings must have the same authority fence. + let (first_done, second_may_start) = std::sync::mpsc::channel(); + let attempt = std::thread::scope(|scope| { + let command_state = &state; + let command_issuer = &issuer; + let command_grant = &grant; + let revoke_state = &state; + let revoke_issuer = &issuer; + let revoke_grant = &grant; + if revoke_first { + let revocation = scope.spawn(move || { + super::super::client_control_grants::revoke_grant(revoke_state, revoke_issuer, &revoke_grant.grant_id).unwrap(); + first_done.send(()).unwrap(); + }); + let command = scope.spawn(move || { + second_may_start.recv().unwrap(); + issue(command_state, command_issuer, command_request(command_grant)) + }); + revocation.join().unwrap(); + command.join().unwrap() + } else { + let command = scope.spawn(move || { + let result = issue(command_state, command_issuer, command_request(command_grant)); + first_done.send(()).unwrap(); + result + }); + let revocation = scope.spawn(move || { + second_may_start.recv().unwrap(); + super::super::client_control_grants::revoke_grant(revoke_state, revoke_issuer, &revoke_grant.grant_id).unwrap(); + }); + let result = command.join().unwrap(); + revocation.join().unwrap(); + result + } + }); + if revoke_first { + let error = attempt.unwrap_err(); + assert_eq!(error.code, "grant-expired"); + assert!(error.details["command_id"].is_string()); + assert!(error.details["receipt_claim_id"].is_string()); + } else { + let command = attempt.unwrap(); + assert!(report(&state, &client, signed_request(&command, &client, &key)).is_err()); + let terminal = read(&state, &issuer, command["id"].as_str().unwrap()).unwrap(); + assert_eq!(terminal["state"], "refused"); + assert!(terminal["receipt_claim_id"].is_string()); + assert!(terminal.get("verified_result_signature").is_none()); + } + } +} + +#[tokio::test] +async fn result_refuses_wrong_key_binding_purpose_and_unsigned_visible_state() { + let root = tempfile::tempdir().unwrap(); + let (state, key, grant, issuer) = super::super::client_control_grants::test_support::fixture(root.path()); + let client = super::super::client_control_enrollment::enrolled(&state, &grant.device).unwrap().session; + let command = issue(&state, &issuer, command_request(&grant)).unwrap(); + let original = signed_request(&command, &client, &key); + for field in ["grant", "run", "command", "purpose", "kind", "session", "device", "outcome", "error", "visible"] { + let mut altered = original.clone(); + let binding = altered.binding.as_mut().unwrap(); + match field { + "grant" => binding.grant_id.push_str("-other"), + "run" => binding.run_id.push_str("-other"), + "command" => binding.command_id.push_str("-other"), + "purpose" => binding.purpose = "message.send".into(), + "kind" => binding.command_kind = "phone.navigate".into(), + "session" => binding.session.push_str("-other"), + "device" => binding.device.push_str("-other"), + "outcome" => altered.outcome = "failed".into(), + "error" => altered.error_code = Some("other".into()), + "visible" => altered.result = Some(json!({"route":"other"})), + _ => unreachable!(), + } + assert!(report(&state, &client, altered).is_err(), "must refuse {field}"); + } + let rng = ring::rand::SystemRandom::new(); + let pkcs8 = ring::signature::EcdsaKeyPair::generate_pkcs8(&ring::signature::ECDSA_P256_SHA256_FIXED_SIGNING, &rng).unwrap(); + let wrong_key = ring::signature::EcdsaKeyPair::from_pkcs8(&ring::signature::ECDSA_P256_SHA256_FIXED_SIGNING, pkcs8.as_ref(), &rng).unwrap(); + assert!(report(&state, &client, signed_request(&command, &client, &wrong_key)).is_err()); + let mut wrong_signature_same_enrolled_key = signed_request(&command, &client, &wrong_key); + wrong_signature_same_enrolled_key.signature.as_mut().unwrap().key = original.signature.as_ref().unwrap().key.clone(); + assert!(report(&state, &client, wrong_signature_same_enrolled_key).is_err()); + let mut invalid_signature = original.clone(); + invalid_signature.signature.as_mut().unwrap().signature = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode([0_u8; 64]); + assert!(report(&state, &client, invalid_signature).is_err()); + let mut unsigned = original.clone(); unsigned.signature = None; + assert!(report(&state, &client, unsigned).is_err()); + let other_command = issue(&state, &issuer, command_request(&grant)).unwrap(); + let mut replayed_on_other_command = original.clone(); + replayed_on_other_command.command_id = other_command["id"].as_str().unwrap().into(); + assert!(report(&state, &client, replayed_on_other_command).is_err()); + // Failed attempts do not consume or terminalize the valid command. + assert_eq!(report(&state, &client, original).unwrap()["state"], "completed"); +} + +#[tokio::test] +async fn grant_revocation_and_command_expiry_create_unsigned_terminal_receipts() { + let root = tempfile::tempdir().unwrap(); + let (state, key, grant, issuer) = super::super::client_control_grants::test_support::fixture(root.path()); + let client = super::super::client_control_enrollment::enrolled(&state, &grant.device).unwrap().session; + let command = issue(&state, &issuer, command_request(&grant)).unwrap(); + let request = signed_request(&command, &client, &key); + super::super::client_control_grants::revoke_grant(&state, &issuer, &grant.grant_id).unwrap(); + assert!(report(&state, &client, request).is_err()); + let terminal = read(&state, &issuer, command["id"].as_str().unwrap()).unwrap(); + assert_eq!(terminal["state"], "refused"); + assert!(terminal["receipt_claim_id"].as_str().is_some()); + assert!(terminal.get("verified_result_signature").is_none()); + + let root = tempfile::tempdir().unwrap(); + let (state, key, grant, issuer) = super::super::client_control_grants::test_support::fixture(root.path()); + let client = super::super::client_control_enrollment::enrolled(&state, &grant.device).unwrap().session; + let command = issue(&state, &issuer, command_request(&grant)).unwrap(); + let request = signed_request(&command, &client, &key); + // Expire by the daemon-owned deadline without sleeping or trusting client timestamps. + { + let mut rows = ROWS.lock().unwrap(); + let row = rows.get_mut(&(state.node.clone(), request.command_id.clone())).unwrap(); + row.expires = client_now_ms(); + } + assert!(report(&state, &client, request).is_err()); + let terminal = read(&state, &issuer, command["id"].as_str().unwrap()).unwrap(); + assert_eq!(terminal["state"], "not-foreground"); + assert!(terminal["receipt_claim_id"].as_str().is_some()); +} + +fn assert_rejected_receipt(state: &AppState, error: ApiError, expected: &str) { + assert_eq!(error.code, expected); + let id = error.details["command_id"].as_str().unwrap(); + let receipt_id = error.details["receipt_claim_id"].as_str().unwrap(); + let subject = format!("custom/client/command-receipt/{}", id.trim_start_matches("device-command/")); + let claims = state.store.claims_for(&subject, Some("custom.client.command-receipt")).unwrap(); + assert_eq!(claims.len(), 1); + assert_eq!(claims[0].id, receipt_id); + let fields = &claims[0].body["fields"]; + assert_eq!(fields["terminalState"], "refused"); + assert_eq!(fields["reason"], expected); + assert!(fields["verifiedResultSignature"].is_null()); + assert!(fields["signedResult"].is_null()); +} + +#[tokio::test] +async fn all_admission_refusals_have_durable_fixture_receipt_ids() { + use super::super::client_control_grants::{issue_grant, revoke_grant, test_support}; + let root = tempfile::tempdir().unwrap(); + let (state, _, grant, issuer) = test_support::fixture(root.path()); + let mut request = command_request(&grant); + request.device = "device/not-enrolled".into(); + assert_rejected_receipt(&state, issue(&state, &issuer, request).unwrap_err(), "device-not-test-client"); + let mut request = command_request(&grant); + request.grant_id = None; + assert_rejected_receipt(&state, issue(&state, &issuer, request).unwrap_err(), "grant-missing"); + let other_actor = ClientSession::local(Some("agent/other")).unwrap(); + assert_rejected_receipt(&state, issue(&state, &other_actor, command_request(&grant)).unwrap_err(), "actor-not-granted"); + let mut request = command_request(&grant); + request.kind = "phone.navigate".into(); request.args = json!({"screen":"Conversation","params":{"target":"agent/client-control-fixture/producer"}}); + assert_rejected_receipt(&state, issue(&state, &issuer, request).unwrap_err(), "screen-not-granted"); + for verb in ["phone.message.read", "message.read", "message.close", "message.send", "work.answer", "attention.answer", "attention.close"] { + let mut request = command_request(&grant); + request.kind = verb.into(); request.args = json!({}); + assert_rejected_receipt(&state, issue(&state, &issuer, request).unwrap_err(), "verb-not-granted"); + } + let item_bound = issue_grant(&state, &issuer, test_support::request(&grant, 2)).unwrap(); + let mut wrong_item = command_request(&item_bound); + wrong_item.kind = "phone.navigate".into(); + wrong_item.args = json!({"screen":"Attention","params":{"id":"step-run/ask-other/ask"}}); + assert_rejected_receipt(&state, issue(&state, &issuer, wrong_item).unwrap_err(), "screen-not-granted"); + let bounded = issue_grant(&state, &issuer, test_support::request(&grant, 1)).unwrap(); + issue(&state, &issuer, command_request(&bounded)).unwrap(); + assert_rejected_receipt(&state, issue(&state, &issuer, command_request(&bounded)).unwrap_err(), "budget-exhausted"); + revoke_grant(&state, &issuer, &grant.grant_id).unwrap(); + assert_rejected_receipt(&state, issue(&state, &issuer, command_request(&grant)).unwrap_err(), "grant-expired"); +} + +#[tokio::test] +async fn malformed_unauthorized_and_non_fixture_requests_never_write_refusal_receipts() { + let root = tempfile::tempdir().unwrap(); + let (mut state, _, grant, issuer) = super::super::client_control_grants::test_support::fixture(root.path()); + let count = || state.store.claims_for_kind_at("custom.client.command-receipt", None, true, 256).unwrap().claims.len(); + assert_eq!(count(), 0); + let mut malformed = command_request(&grant); malformed.args = json!({"unexpected":true}); + let error = issue(&state, &issuer, malformed).unwrap_err(); + assert!(!error.details.contains_key("receipt_claim_id")); + let mut malformed = command_request(&grant); malformed.grant_id = Some("malformed".into()); + assert!(!issue(&state, &issuer, malformed).unwrap_err().details.contains_key("receipt_claim_id")); + let mut unauthenticated = issuer.clone(); unauthenticated.authority_actor = "anonymous".into(); + assert!(!issue(&state, &unauthenticated, command_request(&grant)).unwrap_err().details.contains_key("receipt_claim_id")); + assert_eq!(count(), 0); + state.owning_person = Some("person/ordinary".into()); + let error = issue(&state, &issuer, command_request(&grant)).unwrap_err(); + assert!(!error.details.contains_key("receipt_claim_id")); + assert_eq!(state.store.claims_for_kind_at("custom.client.command-receipt", None, true, 256).unwrap().claims.len(), 0); +} + +#[test] +fn rejected_receipt_capacity_has_a_distinct_reason_and_five_minute_window() { + let template = Command { resource: json!({"id":"device-command/test","state":"refused","receipt_claim_id":"claim/test"}), + expires: 100, terminal_at: Some(100), person: "agent/test".into(), grant: String::new(), issuer_grant: None }; + let mut rows = BTreeMap::new(); + for index in 0..MAX_COMMANDS { rows.insert(("receipt-capacity".into(), index.to_string()), template.clone()); } + assert_eq!(admit_receipt(&rows).unwrap_err().code, "receipt-capacity"); + sweep(&mut rows, "receipt-capacity", 100 + RETENTION_MS - 1); + assert_eq!(admit_receipt(&rows).unwrap_err().code, "receipt-capacity"); + sweep(&mut rows, "receipt-capacity", 100 + RETENTION_MS); + assert!(admit_receipt(&rows).is_ok()); +} diff --git a/crates/fractal-api-core/src/api/client_control_grants.rs b/crates/fractal-api-core/src/api/client_control_grants.rs index 9b33942b603..254cf8940fe 100644 --- a/crates/fractal-api-core/src/api/client_control_grants.rs +++ b/crates/fractal-api-core/src/api/client_control_grants.rs @@ -641,8 +641,8 @@ mod mission_binding_tests { } /// Real persisted fixture shared with the signed-result integration tests. -#[cfg(test)] -pub(crate) mod test_support { +#[cfg(any(test, feature = "test-support"))] +pub mod test_support { use super::*; use base64::Engine as _; use crate::PersonWorkStore as _; diff --git a/crates/fractal-api-core/src/api/client_v0.rs b/crates/fractal-api-core/src/api/client_v0.rs index 10e69605f56..d9e376799ec 100644 --- a/crates/fractal-api-core/src/api/client_v0.rs +++ b/crates/fractal-api-core/src/api/client_v0.rs @@ -623,6 +623,7 @@ pub fn capabilities(state: &AppState, session: &ClientSession) -> Vec { }) .collect::>(); capabilities.push(json!({"id":"summary", "version":1, "state":if session.allows("read.projections") {"granted"} else {"ungranted"}})); + capabilities.push(json!({"id":"phone-commands", "version":1, "state":if super::client_commands::capable(session) {"granted"} else {"unavailable"}})); // Attention rows say whether each is an alert and name its agent conversation; `alerts` names // the attention collection and routes, and the summary counts alerts. capabilities.push(json!({"id":"alerts", "version":1, "state":if session.allows("read.projections") {"granted"} else {"ungranted"}})); diff --git a/docs/st3/client-v0/device-commands.md b/docs/st3/client-v0/device-commands.md index a8c7e61b624..03724676b4d 100644 --- a/docs/st3/client-v0/device-commands.md +++ b/docs/st3/client-v0/device-commands.md @@ -1,8 +1,8 @@ -# Foreground phone commands +# Foreground device commands -A device command is one typed, expiring request to a paired phone's foreground instance. -It is not a durable action, background job, push notification, or replay queue. This contract -adds the daemon/client boundary; a phone must implement the handlers before it can execute commands. +A device command is one typed, expiring request to a paired client's foreground instance. +It is not a background job, push notification, or replay queue. The same protocol addresses +enrolled iOS, web and TUI test clients; ordinary person-operated phone commands retain their wire shape. ## Commands and responses @@ -97,9 +97,10 @@ After the capability probe, a phone subscribes on its existing foreground collec ``` The server uses only the existing `snapshot` and `changes` resource-window frames, containing -`device-command` rows. There are no new frame kinds and no second socket. The addressed session -is the delivery boundary; subscriptions cannot override it. Phone windows contain only active -`pending` / `delivered` commands; terminal transitions remove them, never replay them. +`device-command` rows and test-client grant lifecycle resources. There are no new frame kinds +and no second socket. The addressed session is the delivery boundary; subscriptions cannot +override it. Client command windows contain only active `pending` / `delivered` commands; +terminal transitions remove them, never replay them. Nonpaired issuer windows include their terminal outcomes in a newest-first window of at most 32 rows (a requested collection limit can reduce this). GET retains each outcome independently of that window for the correlation period. Phones must never execute terminal or expired rows. @@ -120,13 +121,13 @@ collections stream for the addressed `session_actor`, or `expired` otherwise. Ph client timestamps do not extend that deadline. A late report cannot replace the terminal decision; repeated reports cannot overwrite an already accepted outcome. -Commands live only in memory on the phone pairing's issuing member. They are never claims, -replicated records, persistent receipts, event replay, or restart recovery. Terminal outcomes remain -available for five minutes of ephemeral correlation before eviction. The process retains at most -2048 rows in total and refuses further admission when retention capacity is exhausted; it does not -discard a retained outcome early to make room. A daemon restart or eviction makes outcomes -unavailable. Expired commands are never executed on reconnection. The normal envelope's snapshot -describes the response context, not durable storage of the command. +Command correlation rows live only in memory on the pairing's issuing member; they are not +replicated or replayed after restart. Granted commands and fixture admission refusals have +separate durable receipt claims described below. Terminal correlation outcomes remain +available for five minutes before eviction. The process retains at most 2048 rows and refuses +further admission when capacity is exhausted rather than discarding retained outcomes early. +A restart or eviction removes command correlation, not persisted receipts. Expired commands +are never executed on reconnection. The normal envelope's snapshot describes response context. ## Isolated client-control v1 signed results @@ -224,6 +225,56 @@ of timestamps; new enrollment cannot resurrect old grants. Grant issue, command admission and result acceptance reload enrollment and pairing authority. Operators can revoke even after the pairing expires. +## Grant admission and durable receipts + +`POST /v1/client/control-grants` binds a grant to the issuing agent's currently +claimed step of a running mission run. The exact claim generation, attempt and +incarnation remain authoritative: claim replacement, claim/run end, enrollment +revocation and deadlines end the grant. The default budget is 50 commands, spent +durably before admission; grants last at most two hours. Only `phone.inspect` +and `phone.navigate` to `NowRoot` or `Attention` can be granted. The journey opens +the seeded native ask using `Attention {params: {id}}`, then inspects it. Required +grant `item_id` binds Attention to that existing native ask; missing, unknown or +different asks are `screen-not-granted`. NowRoot-only grants need no item. +`Conversation` is always `screen-not-granted`. Answer, close, read-state and +`message.send` verbs are `verb-not-granted`; no grant authorizes a mutation. + +The existing session-scoped `commands` collection includes generated +`kind: "client-control-grant-lifecycle"` resources before the first command. +Their states are `started`, `ended` and `revoked`; the stable resource ID and +strictly increasing sequence identify the latest durable transition per grant. +Idle subscribers receive revocation and run/claim/deadline termination without +issuing a command or resubscribing. Lifecycle drives only the control indicator +and termination: policy remains in each command's `control_grant` record. +Commands without an active matching lifecycle grant fail closed. Clients ignore +duplicate/out-of-order starts, preserve the grant's `item_id` binding, and never +reactivate a terminal grant ID. A command binds `grant_id` and `run_id`; no second +command transport exists. + +Granted result acceptance verifies the generated canonical bytes directly with +the enrolled P-256 key, not principal message authority. The outer outcome, +error and optional result must agree with the signed binding. The daemon checks +the command and grant windows, pairing session, run, command, verb and visible +screen again, and rejects terminal-command replay. Signature-envelope nonce and +timestamp metadata do not establish freshness; the daemon's clock does. + +Every terminal granted command has one durable +`custom.client.command-receipt` claim. The resource's `receipt_claim_id` identifies +it. Completed signed results record `verifiedResultSignature` and `signedResult`; +unsigned terminal states such as expiry and grant-end refusal record nulls. +Command correlation rows themselves remain ephemeral. + +An authenticated agent's well-formed command refused for a D-3/D-4 reason also +receives one durable rejected-attempt receipt, but only in the local-only fixture +owned by `person/client-control-fixture`. It records an assigned `commandId`, +issuer, device, optional grant/run, verb/screen, typed reason, terminal state +`refused` and no signature. The typed error details include `command_id` and +`receipt_claim_id`. The existing 2,048-row, five-minute command/refusal retention +bound limits receipt admission; saturation returns `receipt-capacity` without a +receipt. Persisted receipt claims are not removed when this window expires. +Unauthenticated or malformed requests, and requests outside the fixture, return +typed errors without receipt writes. + ## Grant lifecycle on the commands collection The existing session-scoped `commands` subscription carries generated diff --git a/release-notes/client-control-results-v1.json b/release-notes/client-control-results-v1.json new file mode 100644 index 00000000000..2c3471b7283 --- /dev/null +++ b/release-notes/client-control-results-v1.json @@ -0,0 +1,11 @@ +{ + "version": 1, + "summary": "Verify narrow-purpose client results and persist terminal and bounded rejected-attempt receipts", + "replay": {"status":"none","detail":"Command rows remain ephemeral; separate trusted receipt claims preserve completed, expired, not-foreground and refused outcomes."}, + "schema": {"status":"changed","detail":"Commands carry grant/run metadata and signed results bind the generated visible-state model. Receipt claims contain command, issuer, device, run/grant and verified signature/binding, with null signatures for unsigned outcomes."}, + "rules": {"status":"changed","detail":"Daemon clock and live grant/claim checks gate enrolled P-256 results. Wrong key, purpose, kind, session, grant, run, command and replay are refused; message.send remains unauthorized with the same key/grant. Authenticated fixture-agent admission refusals receive durable receipt IDs; malformed, unauthenticated and non-fixture attempts do not. The 2048-row five-minute window returns receipt-capacity at saturation."}, + "compatibility": {"status":"changed","detail":"Restores device.command/device.result routes and session-scoped commands collection integration in the split API, including grant lifecycle resources and paired Unix fixture clients. Existing ordinary person-operated command shapes remain supported."}, + "downtime": {"status":"none","detail":"No live setup, enrollment, daemon restart, deployment or journey is performed."}, + "manual": {"status":"none","detail":"Source verification uses local temporary fixtures only; human setup/apply/enrollment gates remain separate."}, + "recovery": {"status":"none","detail":"Receipt writes are idempotent per command and survive correlation eviction; missing persistence never silently drops an unsigned terminal receipt."} +}