diff --git a/Cargo.lock b/Cargo.lock index a33e7df65..d52648fff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4137,6 +4137,7 @@ dependencies = [ "glam", "hmac", "jian-ops-schema", + "libc", "memmap2", "op-acp", "op-ai", diff --git a/crates/op-host-services/Cargo.toml b/crates/op-host-services/Cargo.toml index 6170f369a..93cf6750e 100644 --- a/crates/op-host-services/Cargo.toml +++ b/crates/op-host-services/Cargo.toml @@ -76,6 +76,12 @@ base64 = "0.22" getrandom = "0.3" hmac = "0.12" sha2 = "0.10" + +# Unix signal handling for the online daemon's controlled shutdown. Already in +# the dependency graph transitively; named here so the SIGTERM handler can use +# it directly. Unix-only — the online daemon is a container deployment. +[target.'cfg(unix)'.dependencies] +libc = "0.2" zip = "2" memmap2 = "0.9" glam = { version = "0.29", default-features = false, features = ["std"] } diff --git a/crates/op-host-services/src/mcp_serve/tool_profile.rs b/crates/op-host-services/src/mcp_serve/tool_profile.rs index 9eb0be009..8246ed62c 100644 --- a/crates/op-host-services/src/mcp_serve/tool_profile.rs +++ b/crates/op-host-services/src/mcp_serve/tool_profile.rs @@ -163,19 +163,16 @@ impl McpScopes { /// Derive from the scope list on a hub token. /// - /// A token that names NO `mcp:*` scope at all is treated as unrestricted. - /// That is the compatibility default for a token store that does not yet - /// issue scopes: the alternative would refuse every token ever minted - /// before scopes existed. Once a token names ANY `mcp:*` scope, the list - /// is authoritative and anything absent from it is denied — so an - /// explicitly read-only token is genuinely read-only. + /// **Fail-closed.** A token that names no `mcp:*` scope gets nothing: it + /// can neither read nor write. An earlier version treated an unscoped + /// token as unrestricted, for compatibility with a token store that did + /// not yet issue scopes — but that store still issues no tokens, so + /// tightening this costs nothing today and removes a default that would + /// have been very hard to tighten later. + /// + /// **op-hub must issue explicit `mcp:read` / `mcp:write` scopes** on every + /// token intended to drive the canvas; one without them is inert. pub fn from_scope_list>(scopes: &[S]) -> Self { - let mentions_mcp = scopes - .iter() - .any(|scope| scope.as_ref().trim().starts_with("mcp:")); - if !mentions_mcp { - return Self::FULL; - } let has = |wanted: &str| scopes.iter().any(|scope| scope.as_ref().trim() == wanted); Self { read: has(MCP_READ_SCOPE), @@ -183,6 +180,16 @@ impl McpScopes { } } + /// A credential that may do nothing. + pub const NONE: Self = Self { + read: false, + write: false, + }; + + pub const fn can_read(self) -> bool { + self.read + } + pub const fn allows(self, access: ToolAccess) -> bool { match access { ToolAccess::Read => self.read, diff --git a/crates/op-host-services/src/mcp_serve/tool_profile_tests.rs b/crates/op-host-services/src/mcp_serve/tool_profile_tests.rs index a6aaed96a..439ccad16 100644 --- a/crates/op-host-services/src/mcp_serve/tool_profile_tests.rs +++ b/crates/op-host-services/src/mcp_serve/tool_profile_tests.rs @@ -187,12 +187,14 @@ fn scopes_are_derived_from_the_hub_scope_list() { // Write without read is honoured as written rather than normalised. assert!(McpScopes::from_scope_list(&["mcp:write"]).can_write()); assert!(!McpScopes::from_scope_list(&["mcp:write"]).allows(ToolAccess::Read)); - // A list that names no mcp scope predates scoping — see the doc comment. + // Fail-closed: a token that names no mcp scope may do nothing. op-hub + // MUST issue explicit `mcp:read` / `mcp:write` on any token intended to + // drive the canvas — one without them is inert by design. assert_eq!( McpScopes::from_scope_list(&["billing:read"]), - McpScopes::FULL + McpScopes::NONE ); - assert_eq!(McpScopes::from_scope_list::<&str>(&[]), McpScopes::FULL); + assert_eq!(McpScopes::from_scope_list::<&str>(&[]), McpScopes::NONE); // An unknown mcp scope does NOT grant anything. assert_eq!( McpScopes::from_scope_list(&["mcp:admin"]), diff --git a/crates/op-host-services/src/web_canvas_server.rs b/crates/op-host-services/src/web_canvas_server.rs index 54e8c13bb..45075da2c 100644 --- a/crates/op-host-services/src/web_canvas_server.rs +++ b/crates/op-host-services/src/web_canvas_server.rs @@ -19,7 +19,6 @@ use std::io::{Read, Write}; use std::net::TcpListener; use std::path::PathBuf; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::mpsc::{self, Receiver, RecvTimeoutError}; use std::sync::{Arc, Mutex}; use std::thread; use std::time::Duration; @@ -61,45 +60,6 @@ pub struct SseTick { pub collab_seq: u64, } -/// Broadcast hub for SSE subscribers. Each `GET /api/mcp/events` connection -/// registers a channel; a document mutation broadcasts the new tick to all -/// of them, and each SSE connection thread writes it to its socket. Senders to -/// disconnected clients are pruned on the next broadcast. -#[derive(Default)] -pub struct SseHub { - subscribers: Mutex>>, -} - -impl SseHub { - /// Register a subscriber; the SSE connection thread blocks on the returned - /// receiver for ticks. - pub(crate) fn subscribe(&self) -> Receiver { - let (tx, rx) = mpsc::channel(); - self.subscribers - .lock() - .unwrap_or_else(|p| p.into_inner()) - .push(tx); - rx - } - - /// Broadcast a tick to all live subscribers, pruning any whose receiver was - /// dropped (client disconnected). - pub(crate) fn broadcast(&self, tick: SseTick) { - self.subscribers - .lock() - .unwrap_or_else(|p| p.into_inner()) - .retain(|tx| tx.send(tick).is_ok()); - } - - #[cfg(test)] - pub(crate) fn subscriber_count(&self) -> usize { - self.subscribers - .lock() - .unwrap_or_else(|p| p.into_inner()) - .len() - } -} - /// RAII decrement for the connection counter — `Drop` runs on normal exit AND /// panic unwind, so a panicking connection can't leak its `MAX_CONNS` slot. struct ConnGuard(Arc); @@ -347,15 +307,34 @@ impl WebCanvasState { // Inside the session's local-edit capture: the session core diffs // the document and emits precise node operations, so pushing an // unchanged document produces no transaction at all. - if !self.ingest_document_in_session(prepared) { - return Err(WebCanvasError::Collab( - collab_state::DaemonMutationRefusal::Collab( - op_editor_core::CollabGateReason::PendingEdit, - ), - )); + match self.ingest_document_in_session(prepared) { + // Sequenced: publish it. + collab_state::IngestOutcome::Committed => { + self.version = self.version.saturating_add(1); + self.version + } + // Accepted, but identical. Answering 200 is right — the + // client's document IS the daemon's — but bumping the version + // would make every peer refetch a document that did not + // change. + collab_state::IngestOutcome::NoChange => self.version, + // The session did not take it. Reporting success here (which + // is what the old `bool` did for `Failed`) left the browser + // believing a write landed that had been discarded, with no + // signal to resync. + collab_state::IngestOutcome::Rejected => { + return Err(WebCanvasError::Collab( + collab_state::DaemonMutationRefusal::Collab( + op_editor_core::CollabGateReason::PendingEdit, + ), + )); + } + collab_state::IngestOutcome::Failed => { + return Err(WebCanvasError::IngestRejected( + collab_state::IngestOutcome::Failed, + )); + } } - self.version = self.version.saturating_add(1); - self.version } else { self.replace_document(prepared.into_document()) }; @@ -637,7 +616,26 @@ pub fn handle_web_canvas_request( { online_policy::refusal_reply(online_policy::OnlineRouteRefusal::LocalFileAccess) } - ("POST", "/api/file/save") => save_current_file(body, state), + // A save reply REPLACES `state.editor` (it installs the document it + // just wrote, plus its metadata), so it is a whole-document swap and + // has to clear the same gate `open-recent` does. Swapping the document + // out from under a live session would leave the peers editing a + // document this daemon no longer has. + ("POST", "/api/file/save") => match state.gate_daemon_mutation( + op_editor_core::CollabGateAction::ReplaceDocument, + op_editor_core::CollabEditSource::ExternalSync, + ) { + Ok(()) => save_current_file(body, state), + Err(refusal) => WebReply { + status: refusal.http_status(), + body: serde_json::json!({ + "ok": false, + "error": refusal.code(), + "message": refusal.to_string(), + }) + .to_string(), + }, + }, ("POST", "/api/file/open-recent") => open_recent_file(body, state), ("POST", "/api/export/pdf") => export_pdf_download(body, state), ("POST", "/api/export/raster") => export_raster_download(body, state), @@ -726,6 +724,9 @@ fn is_device_login_route(path: &str) -> bool { } mod collab_driver; +mod sse_hub; +pub use sse_hub::SseHub; +pub(crate) use sse_hub::SseSlot; mod collab_routes; pub(crate) mod collab_state; mod connect_routes; @@ -743,8 +744,9 @@ mod share_routes; pub mod tenant; pub mod tenant_auth; pub mod tenant_store; +mod tool_scopes; -pub use collab_state::DaemonMutationRefusal; +pub use collab_state::{DaemonMutationRefusal, IngestOutcome}; pub use connect_routes::*; use connection::*; use doc_routes::*; diff --git a/crates/op-host-services/src/web_canvas_server/collab_driver.rs b/crates/op-host-services/src/web_canvas_server/collab_driver.rs index ed05a2f6f..1dec1bda2 100644 --- a/crates/op-host-services/src/web_canvas_server/collab_driver.rs +++ b/crates/op-host-services/src/web_canvas_server/collab_driver.rs @@ -154,7 +154,7 @@ mod tests { // The first pass may publish the initial availability projection. assert!(!tick(&state, &hub)); let settled = state.lock().unwrap_or_else(|p| p.into_inner()).collab.seq(); - while sub.try_recv().is_ok() {} + sub.pending(); // Every pass after that is silent, which is what makes a 250 ms idle // cadence free: no bump, no broadcast, no browser refetch. @@ -165,7 +165,7 @@ mod tests { state.lock().unwrap_or_else(|p| p.into_inner()).collab.seq(), settled ); - assert!(sub.try_recv().is_err(), "an idle daemon must stay quiet"); + assert!(sub.pending().is_none(), "an idle daemon must stay quiet"); } #[test] @@ -174,7 +174,7 @@ mod tests { let hub = SseHub::default(); let sub = hub.subscribe(); tick(&state, &hub); - while sub.try_recv().is_ok() {} + sub.pending(); let (version_before, seq_before) = { let mut guard = state.lock().unwrap_or_else(|p| p.into_inner()); @@ -194,7 +194,7 @@ mod tests { every cursor move would make the browser refetch the document" ); assert_eq!(guard.collab.seq(), seq_before + 1); - let tick_out = sub.try_recv().expect("the change was broadcast"); + let tick_out = sub.pending().expect("the change was broadcast"); assert_eq!(tick_out.version, version_before); assert_eq!(tick_out.collab_seq, seq_before + 1); } @@ -205,7 +205,7 @@ mod tests { let hub = SseHub::default(); let sub = hub.subscribe(); tick(&state, &hub); - while sub.try_recv().is_ok() {} + sub.pending(); // A REST push mutated the document and published its own version while // the driver was asleep. @@ -225,7 +225,7 @@ mod tests { state.lock().unwrap_or_else(|p| p.into_inner()).version, version_before ); - assert!(sub.try_recv().is_err(), "nothing new to announce"); + assert!(sub.pending().is_none(), "nothing new to announce"); } #[test] diff --git a/crates/op-host-services/src/web_canvas_server/collab_state.rs b/crates/op-host-services/src/web_canvas_server/collab_state.rs index 8b6cb0414..b6a399852 100644 --- a/crates/op-host-services/src/web_canvas_server/collab_state.rs +++ b/crates/op-host-services/src/web_canvas_server/collab_state.rs @@ -152,6 +152,69 @@ impl CollabHost for DaemonCollabHost<'_> { } } +/// What a session did with a whole-document push. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum IngestOutcome { + /// Installed and sequenced. The only outcome that publishes a version. + Committed, + /// Accepted, but the document was identical — no transaction, no version. + NoChange, + /// No capture could be opened (busy, or this peer is read-only). + Rejected, + /// The capture opened but the session would not commit it. The browser's + /// copy is now definitively behind and must be refetched. + Failed, +} + +impl IngestOutcome { + /// Stable machine-readable code, for the outcomes that are errors. + pub const fn error_code(self) -> Option<&'static str> { + match self { + Self::Committed | Self::NoChange => None, + Self::Rejected => Some("collab-busy"), + // Shares the existing conflict code so the browser's established + // refetch path handles it without a new client branch. + Self::Failed => Some("version-conflict"), + } + } +} + +/// RAII close for a session's local-edit capture. +/// +/// `begin_local_edit` and `finish_local_edit` must pair on every path. Holding +/// the pair in a guard means an early return — or a panic in the install — +/// cannot leave the capture open and the session wedged. +struct LocalEditCapture<'a> { + state: Option<&'a mut WebCanvasState>, +} + +impl LocalEditCapture<'_> { + fn state_mut(&mut self) -> &mut WebCanvasState { + self.state.as_mut().expect("capture is open") + } + + /// Close the capture, reporting whether the session committed it. + fn finish(mut self) -> bool { + let Some(state) = self.state.take() else { + return false; + }; + let (runtime, mut host) = state.collab_runtime_and_host(); + runtime.finish_local_edit(&mut host) + } +} + +impl Drop for LocalEditCapture<'_> { + fn drop(&mut self) { + // Only reached when `finish` was not called — an unwind. Close the + // capture anyway; its result is unobservable on this path, but a + // session left mid-edit would refuse every later push. + if let Some(state) = self.state.take() { + let (runtime, mut host) = state.collab_runtime_and_host(); + let _ = runtime.finish_local_edit(&mut host); + } + } +} + /// Why the daemon refused a mutation. /// /// The bool acks the older applier paths returned could not say *why* a write @@ -270,22 +333,38 @@ impl WebCanvasState { /// transaction at all. [`EditorState::install_prepared_document`] keeps the /// document generation, which is what lets the matching finish succeed. /// - /// Returns `false` when no capture could be opened (busy or read-only); the - /// caller then reports a conflict rather than writing behind the session's - /// back. - pub(crate) fn ingest_document_in_session(&mut self, prepared: PreparedDocument) -> bool { + /// Reports what the session did with the push — see [`IngestOutcome`]. + /// The previous `bool` could not distinguish "the session accepted this" + /// from "the session threw it away", so a rejected push was answered 200 + /// with a version bump and the browser never learned to resync. + pub(crate) fn ingest_document_in_session( + &mut self, + prepared: PreparedDocument, + ) -> IngestOutcome { + let revision_before = self.editor.document_revision(); let (runtime, mut host) = self.collab_runtime_and_host(); if !runtime.begin_local_edit(&mut host) { - return false; + return IngestOutcome::Rejected; } - // From here the capture is open and MUST be closed on every path; - // `finish_local_edit` is the only close, so nothing fallible may run - // between the two calls. - host.editor_state_mut() + // The capture is now open and MUST be closed on every path. The guard + // is what makes that true even if the install below panics: an + // unclosed capture leaves the session permanently unable to accept + // another edit, which is worse than the failed push itself. + let mut capture = LocalEditCapture { state: Some(self) }; + capture + .state_mut() + .editor .install_prepared_document(prepared, EditOrigin::Local); - let (runtime, mut host) = self.collab_runtime_and_host(); - runtime.finish_local_edit(&mut host); - true + let committed = capture.finish(); + if !committed { + return IngestOutcome::Failed; + } + if self.editor.document_revision() == revision_before { + // The session diffed the document and found nothing to send. The + // push succeeded; there is simply no new version to publish. + return IngestOutcome::NoChange; + } + IngestOutcome::Committed } } diff --git a/crates/op-host-services/src/web_canvas_server/collab_state_tests.rs b/crates/op-host-services/src/web_canvas_server/collab_state_tests.rs index 559dfd97c..046567dc2 100644 --- a/crates/op-host-services/src/web_canvas_server/collab_state_tests.rs +++ b/crates/op-host-services/src/web_canvas_server/collab_state_tests.rs @@ -228,7 +228,10 @@ fn ingest_refuses_rather_than_writing_when_no_capture_can_open() { ) .expect("validates"); - assert!(!state.ingest_document_in_session(prepared)); + assert_eq!( + state.ingest_document_in_session(prepared), + crate::web_canvas_server::IngestOutcome::Rejected + ); assert_eq!(state.editor.doc, before, "a refused ingest writes nothing"); } @@ -244,3 +247,75 @@ fn the_projection_sequence_is_independent_of_the_document_version() { assert_eq!(state.sse_tick().version, 0); assert_eq!(state.sse_tick().collab_seq, 1); } + +// --------------------------------------------------------------------------- +// H3: `/api/file/save` replaces the editor, so it needs the same gate +// `open-recent` passes. +// --------------------------------------------------------------------------- + +/// Drive the REST route table directly, which is where the gate lives. +fn save_reply(state: &mut WebCanvasState) -> crate::web_canvas_server::WebReply { + crate::web_canvas_server::handle_web_canvas_request( + "POST", + "/api/file/save", + r#"{"document":{"version":"1.0.0","children":[]}}"#, + state, + ) +} + +#[test] +fn saving_is_refused_for_an_owner_in_a_live_session() { + // A save reply REPLACES `state.editor`, so it is a whole-document swap: + // letting it through would leave the peers editing a document this daemon + // no longer has. + let mut state = daemon(); + in_session( + &mut state, + CollabConnectionPhase::Active, + CollabUiRole::Owner, + ); + let reply = save_reply(&mut state); + assert_eq!(reply.status, "409 Conflict", "{}", reply.body); + assert!(reply.body.contains("collab-active"), "{}", reply.body); +} + +#[test] +fn saving_is_refused_for_a_guest_in_a_live_session() { + let mut state = daemon(); + in_session( + &mut state, + CollabConnectionPhase::Active, + CollabUiRole::Editor, + ); + let reply = save_reply(&mut state); + assert_eq!(reply.status, "409 Conflict", "{}", reply.body); +} + +#[test] +fn saving_outside_a_session_is_not_gated() { + // The local daemon's normal path must be untouched: no session, no gate. + // (It then fails on having no backing path, which is a different answer + // from the collaboration refusal and is what this asserts.) + let mut state = daemon(); + let reply = save_reply(&mut state); + assert_ne!(reply.status, "409 Conflict", "{}", reply.body); + assert!(!reply.body.contains("collab-active"), "{}", reply.body); +} + +// --------------------------------------------------------------------------- +// H4: a rejected ingest must not read as success. +// --------------------------------------------------------------------------- + +#[test] +fn a_rejected_ingest_is_distinguishable_from_an_accepted_one() { + // The `bool` this replaces could not say whether the session took the + // document, so a discarded push was answered 200 with a version bump and + // the browser never learned to resync. + use crate::web_canvas_server::IngestOutcome; + assert_eq!(IngestOutcome::Committed.error_code(), None); + assert_eq!(IngestOutcome::NoChange.error_code(), None); + assert_eq!(IngestOutcome::Rejected.error_code(), Some("collab-busy")); + // Shares the browser's existing conflict code so its established refetch + // path handles it with no new client branch. + assert_eq!(IngestOutcome::Failed.error_code(), Some("version-conflict")); +} diff --git a/crates/op-host-services/src/web_canvas_server/connection.rs b/crates/op-host-services/src/web_canvas_server/connection.rs index babe5b3ca..628fe68fa 100644 --- a/crates/op-host-services/src/web_canvas_server/connection.rs +++ b/crates/op-host-services/src/web_canvas_server/connection.rs @@ -25,6 +25,14 @@ pub(super) struct ConnCtx<'a> { /// the local and managed daemons — the whole catalog, full authority, /// exactly as before capability profiles existed. pub(super) mcp_profile: crate::mcp_serve::tool_profile::McpAccessProfile, + /// How this connection's caller authenticated, and what it may do. + /// + /// `None` for the local and managed daemons, which have no per-request + /// identity and are unrestricted — the REST scope gate is skipped whole. + pub(super) rest_identity: Option<( + super::tenant_auth::IdentityVia, + crate::mcp_serve::tool_profile::McpScopes, + )>, } /// Handle one connection against the single-user document authority. @@ -76,6 +84,7 @@ pub(super) fn serve_one_in_mode( hub, mode, mcp_profile: crate::mcp_serve::tool_profile::McpAccessProfile::UNRESTRICTED, + rest_identity: None, }, ) } @@ -246,9 +255,9 @@ pub(super) fn dispatch( // version so no broadcast is missed (a duplicate is harmless — versions are // monotonic). The state lock is released before the long SSE wait. if req.method == "GET" && req.path == "/api/mcp/events" { - let rx = hub.subscribe(); + let slot = hub.subscribe(); let current = state.lock().unwrap_or_else(|p| p.into_inner()).sse_tick(); - return serve_sse(stream, rx, current, cors_origin).map(|()| false); + return serve_sse(stream, &slot, current, cors_origin).map(|()| false); } // The AI / image routes, which parse under the lock and then run long // network on this connection thread. See `connection_ai_routes.rs`. @@ -276,6 +285,27 @@ pub(super) fn dispatch( // daemon doesn't implement yet, which it answers with 404 rather than // mis-routing them into the JSON-RPC dispatch below. if req.path.starts_with("/api/") { + // Scopes apply to REST exactly as they apply to `/mcp`. Without this + // a read-only token could replace the whole document here — strictly + // more damage than any tool call it is refused. + if let Some((via, scopes)) = ctx.rest_identity { + if let Some(refusal) = + super::tool_scopes::check_rest_scope(via, scopes, &req.method, &req.path) + { + crate::mcp_serve::write_mcp_http_response_with_origin( + stream, + refusal.http_status(), + &serde_json::json!({ + "ok": false, + "error": refusal.code(), + "message": refusal.to_string(), + }) + .to_string(), + cors_origin, + )?; + return Ok(false); + } + } let reply = { let mut guard = state.lock().unwrap_or_else(|p| p.into_inner()); let before = guard.version; @@ -464,7 +494,7 @@ pub(super) fn dispatch( /// disconnects (write error) or the hub is dropped. pub(super) fn serve_sse( stream: &mut S, - rx: Receiver, + slot: &SseSlot, current: SseTick, cors_origin: Option<&str>, ) -> Result<()> { @@ -483,20 +513,15 @@ pub(super) fn serve_sse( .map_err(|e| WebCanvasError::Transport(format!("sse headers: {e}")))?; write_sse_event(stream, current)?; loop { - match rx.recv_timeout(SSE_HEARTBEAT) { - Ok(mut tick) => { - // Coalesce any further queued bumps — only the latest tick - // matters (the client re-reads whatever the counters point at), - // so a burst of mutations collapses to a single event and the - // channel can't accumulate unboundedly behind a slow client. - while let Ok(next) = rx.try_recv() { - tick = next; - } - write_sse_event(stream, tick)?; - } - Err(RecvTimeoutError::Timeout) => { + // The slot holds only the newest tick, so coalescing is structural: + // a burst of mutations behind a slow client collapses to one event + // rather than queueing one entry per mutation. + match slot.take_latest(SSE_HEARTBEAT) { + Some(tick) => write_sse_event(stream, tick)?, + None => { // SSE comment heartbeat — no-op for the client, but a failed - // write here is how we notice it disconnected. + // write here is how we notice it disconnected, which is also + // what ends this loop and drops the slot. stream .write_all(b": ping\n\n") .map_err(|e| WebCanvasError::Transport(format!("sse heartbeat: {e}")))?; @@ -504,7 +529,6 @@ pub(super) fn serve_sse( .flush() .map_err(|e| WebCanvasError::Transport(format!("sse flush: {e}")))?; } - Err(RecvTimeoutError::Disconnected) => return Ok(()), } } } diff --git a/crates/op-host-services/src/web_canvas_server/hub_verifier.rs b/crates/op-host-services/src/web_canvas_server/hub_verifier.rs index 2d853f8a4..2d5a19465 100644 --- a/crates/op-host-services/src/web_canvas_server/hub_verifier.rs +++ b/crates/op-host-services/src/web_canvas_server/hub_verifier.rs @@ -165,13 +165,14 @@ mod tests { }); assert!(both.scopes.can_write()); - // A token that names no mcp scope at all predates scoping and keeps - // full authority; see `McpScopes::from_scope_list`. + // A token that names no mcp scope may do nothing — fail-closed. See + // `McpScopes::from_scope_list`: op-hub must issue explicit scopes. let unscoped = identity_from_token(HubToken { scopes: vec!["billing:read".into()], ..token() }); - assert!(unscoped.scopes.can_write()); + assert!(!unscoped.scopes.can_write()); + assert!(!unscoped.scopes.can_read()); } #[test] diff --git a/crates/op-host-services/src/web_canvas_server/online_mcp_tests.rs b/crates/op-host-services/src/web_canvas_server/online_mcp_tests.rs index 3d42f8eb6..57127e2c4 100644 --- a/crates/op-host-services/src/web_canvas_server/online_mcp_tests.rs +++ b/crates/op-host-services/src/web_canvas_server/online_mcp_tests.rs @@ -216,3 +216,94 @@ fn the_local_daemon_still_lists_and_calls_the_whole_catalog() { ); } } + +// --------------------------------------------------------------------------- +// H7: scopes apply to REST, not just to /mcp. +// --------------------------------------------------------------------------- + +#[test] +fn a_read_scope_token_cannot_replace_the_document_over_rest() { + // The bypass: the same token is refused `add_page` on /mcp but could + // replace the entire document here — strictly more damage. + let response = serve( + ®istry(), + &scoped_verifier(), + Request::json("POST", "/api/mcp/document", SYNC_BODY).with_bearer("tokR"), + ); + assert_eq!( + status_line(&response), + "HTTP/1.1 403 Forbidden", + "{response}" + ); + assert_eq!(body(&response)["error"], "scope-insufficient"); +} + +#[test] +fn a_read_scope_token_may_still_read_over_rest() { + let response = serve( + ®istry(), + &scoped_verifier(), + Request::new("GET", "/api/mcp/document").with_bearer("tokR"), + ); + assert_eq!(status_line(&response), "HTTP/1.1 200 OK", "{response}"); +} + +#[test] +fn a_token_with_no_scopes_is_refused_every_rest_route_but_the_probe() { + // Fail-closed: op-hub issues no tokens yet, so an unscoped token is inert. + let verifier = StaticVerifier::parse("tokN=userN:none"); + let registry = registry(); + for (method, path) in [("GET", "/api/mcp/document"), ("GET", "/api/mcp/version")] { + let response = serve( + ®istry, + &verifier, + Request::new(method, path).with_bearer("tokN"), + ); + assert_eq!( + status_line(&response), + "HTTP/1.1 403 Forbidden", + "{method} {path}: {response}" + ); + } + let push = serve( + ®istry, + &verifier, + Request::json("POST", "/api/mcp/document", SYNC_BODY).with_bearer("tokN"), + ); + assert_eq!(status_line(&push), "HTTP/1.1 403 Forbidden", "{push}"); + + // The health probe stays reachable so a client can discover the daemon. + let probe = serve( + ®istry, + &verifier, + Request::new("GET", "/api/mcp/server").with_bearer("tokN"), + ); + assert_eq!(status_line(&probe), "HTTP/1.1 200 OK", "{probe}"); +} + +#[test] +fn a_full_scope_token_is_unaffected_over_rest() { + let registry = registry(); + let response = serve( + ®istry, + &scoped_verifier(), + Request::json("POST", "/api/mcp/document", SYNC_BODY).with_bearer("tokA"), + ); + assert_eq!(status_line(&response), "HTTP/1.1 200 OK", "{response}"); +} + +#[test] +fn a_browser_session_is_not_scope_limited_over_rest() { + // A session IS the account; scopes narrow a token below it, not the + // account below itself. + let registry = registry(); + let verifier = StaticVerifier::parse("sessA=userA"); + let response = serve( + ®istry, + &verifier, + Request::json("POST", "/api/mcp/document", SYNC_BODY) + .with_session("sessA") + .with_origin(PUBLIC_ORIGIN), + ); + assert_eq!(status_line(&response), "HTTP/1.1 200 OK", "{response}"); +} diff --git a/crates/op-host-services/src/web_canvas_server/online_policy.rs b/crates/op-host-services/src/web_canvas_server/online_policy.rs index 57f403a2c..667021103 100644 --- a/crates/op-host-services/src/web_canvas_server/online_policy.rs +++ b/crates/op-host-services/src/web_canvas_server/online_policy.rs @@ -199,6 +199,30 @@ pub(super) fn cookie_write_origin_allowed(allow: &[String], origin: Option<&str> /// paints nothing. pub(super) const EMPTY_INDICATOR_RELAY: &str = r#"{"epoch":0,"active":false,"cursorAgent":null,"nodes":[],"frames":[],"previews":[],"reveals":[]}"#; +/// REST paths an API token may reach without any scope. +/// +/// Only the deployment health probe: a client has to be able to discover the +/// daemon exists before it can be told its scopes are insufficient. +const SCOPE_EXEMPT_READS: &[&str] = &["/api/mcp/server"]; + +/// Which scope a REST request needs, if any. +/// +/// Method-based: `GET`/`HEAD`/`OPTIONS` read, everything else writes. Deriving +/// it from the method rather than a per-route table means a route added later +/// is covered by default, and covered in the strict direction. +pub(super) fn rest_scope_required( + method: &str, + path: &str, +) -> Option { + if SCOPE_EXEMPT_READS.contains(&path) { + return None; + } + match method { + "GET" | "HEAD" | "OPTIONS" => Some(super::tool_scopes::RestScope::Read), + _ => Some(super::tool_scopes::RestScope::Write), + } +} + /// A route the online deployment refuses outright. /// /// Typed rather than a bare status so the refusal reads the same in a REST diff --git a/crates/op-host-services/src/web_canvas_server/online_run_loop.rs b/crates/op-host-services/src/web_canvas_server/online_run_loop.rs index 4e59b0c66..9d75587ab 100644 --- a/crates/op-host-services/src/web_canvas_server/online_run_loop.rs +++ b/crates/op-host-services/src/web_canvas_server/online_run_loop.rs @@ -123,6 +123,11 @@ pub fn run_online_web_canvas(options: ServeWebOptions) -> Result<()> { )); let conn_count = Arc::new(AtomicUsize::new(0)); let shutdown = Arc::new(AtomicBool::new(false)); + // A container stop is a SIGTERM, and without a handler it kills the + // process where it stands — losing every resident tenant that had not + // happened to be evicted. The handler only raises the flag the accept + // loop already observes, so the existing exit path (which flushes) runs. + install_shutdown_signals(&shutdown, local_addr); spawn_sweeper(®istry, &shutdown); for stream in listener.incoming() { @@ -204,6 +209,66 @@ pub(super) fn check_persistence_configured( ))) } +/// Raised by the signal handler. A `static` because that is all a handler may +/// safely touch: it runs on an arbitrary thread with almost nothing allowed, +/// so it sets one atomic and returns, and the watcher thread below does the +/// rest. +static SIGNAL_RECEIVED: AtomicBool = AtomicBool::new(false); + +/// Route SIGTERM and SIGINT into the daemon's existing shutdown path. +/// +/// Only installed by the online loop. The local and managed daemons keep the +/// lifecycle they have always had — a token-authed shutdown request, or +/// stdin EOF under a supervisor — and adding a handler there would change +/// what Ctrl-C means for an interactive operator. +fn install_shutdown_signals(shutdown: &Arc, local_addr: std::net::SocketAddr) { + #[cfg(unix)] + { + // SAFETY: `handle_shutdown_signal` is async-signal-safe — it performs + // exactly one relaxed atomic store and returns. It allocates nothing, + // takes no lock, and calls nothing re-entrant. + unsafe { + let handler = handle_shutdown_signal as *const () as libc::sighandler_t; + libc::signal(libc::SIGTERM, handler); + libc::signal(libc::SIGINT, handler); + } + let shutdown = Arc::clone(shutdown); + // The handler cannot wake a blocked `accept`, so a watcher does it: + // it raises the real flag and pokes the listener, exactly as the + // token-authed shutdown path does. + let spawned = thread::Builder::new() + .name("op-serve-web-online-signal".into()) + .spawn(move || loop { + if SIGNAL_RECEIVED.load(Ordering::Acquire) { + if !shutdown.swap(true, Ordering::AcqRel) { + let _ = std::net::TcpStream::connect(local_addr); + } + return; + } + if shutdown.load(Ordering::Acquire) { + return; + } + std::thread::sleep(std::time::Duration::from_millis(200)); + }); + if spawned.is_err() { + eprintln!( + "openpencil --serve-web --online: could not start the signal watcher; a \ + container stop will not flush tenants" + ); + } + } + #[cfg(not(unix))] + { + let _ = (shutdown, local_addr); + } +} + +/// The signal handler itself. See the safety note at its installation. +#[cfg(unix)] +extern "C" fn handle_shutdown_signal(_signal: libc::c_int) { + SIGNAL_RECEIVED.store(true, Ordering::Relaxed); +} + /// Whether the operator accepted a deployment that persists nothing. fn ephemeral_opt_in() -> bool { std::env::var(EPHEMERAL_ENV) @@ -452,6 +517,7 @@ pub(super) fn serve_one_online( // The public tool profile, narrowed further by whatever scopes // this particular credential carries. mcp_profile: crate::mcp_serve::tool_profile::McpAccessProfile::online(identity.scopes), + rest_identity: Some((identity.via, identity.scopes)), }, ) } diff --git a/crates/op-host-services/src/web_canvas_server/online_share_tests.rs b/crates/op-host-services/src/web_canvas_server/online_share_tests.rs index d677f27dd..e613caf0b 100644 --- a/crates/op-host-services/src/web_canvas_server/online_share_tests.rs +++ b/crates/op-host-services/src/web_canvas_server/online_share_tests.rs @@ -499,3 +499,165 @@ fn a_deployment_that_evicts_but_persists_nothing_refuses_to_start() { // Both is fine — the data directory simply wins. assert!(check_persistence_configured(true, true, 1800).is_ok()); } + +// --------------------------------------------------------------------------- +// H5: an unauthorised `?tenant=` must not cost a tenant slot. +// --------------------------------------------------------------------------- + +#[test] +fn unauthorised_tenant_requests_never_materialise_a_tenant() { + // The exhaustion this closes: `?tenant=` names an arbitrary account, and + // creating the tenant to discover the caller is not on its list means + // every refused request still spends the daemon's tenant budget. + let registry = registry(); + let verifier = verifier(); + for index in 0..64 { + let owner: &'static str = Box::leak(format!("victim-{index}").into_boxed_str()); + let response = serve( + ®istry, + &verifier, + Request { + tenant: Some(owner), + ..Request::new("GET", "/api/mcp/document") + } + .with_bearer("tokB"), + ); + assert_eq!(status_line(&response), "HTTP/1.1 403 Forbidden", "{owner}"); + } + assert_eq!( + registry.tenant_count(), + 0, + "a refused visitor must not leave a tenant behind" + ); +} + +#[test] +fn an_authorised_visitor_still_materialises_an_offline_owners_tenant() { + // The other half: admission-before-materialisation must not break the + // case sharing exists for — opening a document whose owner is offline. + let temp = PersistentRegistry::new("offline-owner"); + let verifier = verifier(); + share( + &temp.registry, + "tokA", + op_editor_core::share_routes::GRANT, + "userB", + ); + serve( + &temp.registry, + &verifier, + Request::json("POST", "/api/mcp/document", SYNC_BODY).with_bearer("tokA"), + ); + assert_eq!(temp.registry.evict_idle(now_unix() + 3600), 1); + assert_eq!(temp.registry.tenant_count(), 0); + + // The owner is gone from memory; the visitor is admitted from the + // persisted list and the tenant is restored for them. + let visited = serve( + &temp.registry, + &verifier, + as_tenant( + Request::new("GET", "/api/mcp/document").with_bearer("tokB"), + "userA", + ), + ); + assert_eq!(status_line(&visited), "HTTP/1.1 200 OK", "{visited}"); + assert!(visited.contains("Tenant Rect"), "{visited}"); +} + +#[test] +fn a_non_resident_tenant_admits_nobody_when_nothing_was_persisted() { + // Fail-closed: with no store the share was never durable, so a visitor + // cannot be admitted to a tenant that is not in memory. + let registry = registry(); + let response = serve( + ®istry, + &verifier(), + as_tenant( + Request::new("GET", "/api/mcp/version").with_bearer("tokB"), + "userA", + ), + ); + assert_eq!( + status_line(&response), + "HTTP/1.1 403 Forbidden", + "{response}" + ); +} + +// --------------------------------------------------------------------------- +// H6: concurrent access-list edits. +// --------------------------------------------------------------------------- + +#[test] +fn concurrent_grants_all_survive() { + // Snapshot-then-write let two grants each read the list before the + // other's insert and write back a version missing it, silently dropping + // one. The edit and its write now happen under one lock. + let temp = std::sync::Arc::new(PersistentRegistry::new("concurrent-grants")); + let identity = crate::web_canvas_server::tenant_auth::ResolvedIdentity { + user_id: "userA".into(), + username: "userA".into(), + display_name: "userA".into(), + via: crate::web_canvas_server::tenant_auth::IdentityVia::ApiToken, + scopes: crate::mcp_serve::tool_profile::McpScopes::FULL, + }; + let lease = temp.registry.lease_for(&identity).expect("lease"); + let tenant = std::sync::Arc::new(()); + let _ = tenant; + + std::thread::scope(|scope| { + for index in 0..16 { + let temp = std::sync::Arc::clone(&temp); + let lease = &lease; + scope.spawn(move || { + let change = + crate::web_canvas_server::tenant::AclChange::Grant(format!("guest-{index}")); + temp.registry + .update_acl(lease.owner_id(), lease.tenant(), change) + .expect("persisted"); + }); + } + }); + + let shared = lease.tenant().shared_with(); + assert_eq!(shared.len(), 16, "every grant must survive: {shared:?}"); + // …and the persisted list agrees with memory. + assert_eq!(temp.registry.store().load_acl("userA"), shared); +} + +#[test] +fn a_share_that_cannot_be_persisted_is_reported_and_rolled_back() { + // Reporting 200 here told the user a share had succeeded that would + // vanish on the next restart. + let temp = PersistentRegistry::new("unpersistable-share"); + std::fs::remove_dir_all(&temp.root).expect("clear root"); + std::fs::write(&temp.root, b"not a directory").expect("block the root"); + + let response = share( + &temp.registry, + "tokA", + op_editor_core::share_routes::GRANT, + "userB", + ); + assert_eq!( + status_line(&response), + "HTTP/1.1 500 Internal Server Error", + "{response}" + ); + assert_eq!(body(&response)["error"], "share-not-persisted"); + + // Rolled back, so memory and disk agree and a retry starts from a known + // state — the visitor is NOT quietly admitted in the meantime. + let visitor = serve( + &temp.registry, + &verifier(), + as_tenant( + Request::new("GET", "/api/mcp/version").with_bearer("tokB"), + "userA", + ), + ); + assert_eq!(status_line(&visitor), "HTTP/1.1 403 Forbidden", "{visitor}"); + + let _ = std::fs::remove_file(&temp.root); +} diff --git a/crates/op-host-services/src/web_canvas_server/share_routes.rs b/crates/op-host-services/src/web_canvas_server/share_routes.rs index a1cddf608..b42ebfa02 100644 --- a/crates/op-host-services/src/web_canvas_server/share_routes.rs +++ b/crates/op-host-services/src/web_canvas_server/share_routes.rs @@ -16,7 +16,7 @@ use op_editor_core::share_routes; -use super::tenant::{TenantLease, TenantRegistry}; +use super::tenant::{AclChange, TenantLease, TenantRegistry}; use super::tenant_auth::ResolvedIdentity; use super::WebReply; @@ -107,25 +107,38 @@ fn mutate( Ok(account) => account, Err(error) => return error_reply(error), }; - let changed = if granting { - lease.tenant().grant(&account) + let change = if granting { + AclChange::Grant(account) } else { - lease.tenant().revoke(&account) + AclChange::Revoke(account) }; - if changed { - // Persisted immediately rather than at eviction: a share the user was - // told had succeeded must survive a restart, and the document it - // applies to may not be written for another half hour. - registry.persist_acl(lease.owner_id(), lease.tenant()); - } - WebReply { - status: "200 OK", - body: serde_json::json!({ - "ok": true, - "changed": changed, - "sharedWith": lease.tenant().shared_with().into_iter().collect::>(), - }) - .to_string(), + // The edit and its write are one serialised operation. Persisted + // immediately rather than at eviction: a share the user was told had + // succeeded must survive a restart, and the document it applies to may not + // be written for another half hour. + match registry.update_acl(lease.owner_id(), lease.tenant(), change) { + Ok(update) => WebReply { + status: "200 OK", + body: serde_json::json!({ + "ok": true, + "changed": update.changed, + "sharedWith": update.shared_with.into_iter().collect::>(), + }) + .to_string(), + }, + // The change has been rolled back, so memory and disk agree and a + // retry starts from a known state. Reporting 200 here — as the + // previous code did — told the user a share had succeeded that would + // vanish on the next restart. + Err(error) => WebReply { + status: "500 Internal Server Error", + body: serde_json::json!({ + "ok": false, + "error": "share-not-persisted", + "message": error.to_string(), + }) + .to_string(), + }, } } diff --git a/crates/op-host-services/src/web_canvas_server/sse_hub.rs b/crates/op-host-services/src/web_canvas_server/sse_hub.rs new file mode 100644 index 000000000..ec63500d5 --- /dev/null +++ b/crates/op-host-services/src/web_canvas_server/sse_hub.rs @@ -0,0 +1,110 @@ +//! Latest-value fan-out for the SSE update stream. +//! +//! Split out of the `web_canvas_server` spine at the 800-line cap. The whole +//! design is one decision — a subscriber gets a slot, not a queue — and the +//! reasoning for it lives on `SseSlot`. + +use std::sync::{Arc, Condvar, Mutex, Weak}; +use std::time::Duration; + +use super::SseTick; + +/// One subscriber's latest-value slot. +/// +/// Deliberately NOT a queue. The client re-reads whatever the counters point +/// at, so only the newest tick has any meaning — an older one is not +/// information the client lost, it is information the newer one already +/// contains. A queue therefore buys nothing and costs the one thing that +/// matters: a subscriber that stops reading (a paused tab, a stalled socket, +/// a laptop that slept) accumulates one entry per mutation, unbounded, in a +/// process shared with every other account. +/// +/// So the slot holds exactly one tick and a publisher overwrites it. A burst +/// of a thousand mutations against a subscriber that never wakes leaves one +/// tick behind, not a thousand. +pub(crate) struct SseSlot { + latest: Mutex>, + ready: Condvar, +} + +impl SseSlot { + fn new() -> Self { + Self { + latest: Mutex::new(None), + ready: Condvar::new(), + } + } + + /// Overwrite the pending tick and wake the waiter. + fn publish(&self, tick: SseTick) { + *self.latest.lock().unwrap_or_else(|p| p.into_inner()) = Some(tick); + self.ready.notify_one(); + } + + /// Take the pending tick, waiting up to `timeout` for one. + /// + /// `None` means the timeout elapsed with nothing published — the caller's + /// cue to emit a heartbeat, which is also how a disconnected socket is + /// noticed. + pub(crate) fn take_latest(&self, timeout: Duration) -> Option { + let guard = self.latest.lock().unwrap_or_else(|p| p.into_inner()); + let (mut guard, _) = self + .ready + .wait_timeout_while(guard, timeout, |latest| latest.is_none()) + .unwrap_or_else(|p| p.into_inner()); + guard.take() + } + + /// Take the pending tick without waiting. Tests only. + #[cfg(test)] + pub(crate) fn pending(&self) -> Option { + self.latest.lock().unwrap_or_else(|p| p.into_inner()).take() + } +} + +/// Broadcast hub for SSE subscribers. Each `GET /api/mcp/events` connection +/// registers a slot; a document mutation publishes the new tick into every +/// one of them, and each SSE connection thread writes it to its socket. +/// +/// Subscribers are held weakly, so a connection that ended is pruned on the +/// next broadcast without needing the subscriber to signal anything. +#[derive(Default)] +pub struct SseHub { + subscribers: Mutex>>, +} + +impl SseHub { + /// Register a subscriber; the SSE connection thread blocks on the returned + /// slot for ticks. Dropping it unregisters. + pub(crate) fn subscribe(&self) -> Arc { + let slot = Arc::new(SseSlot::new()); + self.subscribers + .lock() + .unwrap_or_else(|p| p.into_inner()) + .push(Arc::downgrade(&slot)); + slot + } + + /// Publish a tick to all live subscribers, pruning any whose connection + /// ended (their `Arc` was dropped). + pub(crate) fn broadcast(&self, tick: SseTick) { + self.subscribers + .lock() + .unwrap_or_else(|p| p.into_inner()) + .retain(|slot| { + let Some(slot) = slot.upgrade() else { + return false; + }; + slot.publish(tick); + true + }); + } + + #[cfg(test)] + pub(crate) fn subscriber_count(&self) -> usize { + self.subscribers + .lock() + .unwrap_or_else(|p| p.into_inner()) + .len() + } +} diff --git a/crates/op-host-services/src/web_canvas_server/tenant.rs b/crates/op-host-services/src/web_canvas_server/tenant.rs index aa654cfe2..9566ee745 100644 --- a/crates/op-host-services/src/web_canvas_server/tenant.rs +++ b/crates/op-host-services/src/web_canvas_server/tenant.rs @@ -34,7 +34,23 @@ use std::sync::{Arc, Mutex}; use op_editor_core::EditorState; use super::tenant_auth::ResolvedIdentity; -use super::tenant_store::TenantStore; +use super::tenant_store::{TenantStore, TenantStoreError}; + +/// One edit to a tenant's access list. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum AclChange { + Grant(String), + Revoke(String), +} + +/// The result of an applied access-list change. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct AclUpdate { + /// Whether the list actually moved (a repeated grant does not). + pub changed: bool, + /// The list as it now stands, both in memory and on disk. + pub shared_with: BTreeSet, +} use super::{SseHub, WebCanvasState}; /// Global connection ceiling for the online daemon. @@ -224,6 +240,14 @@ impl Tenant { .remove(visitor) } + /// The access list, locked for a read-modify-write. + /// + /// `update_acl` holds this across both the edit and the disk write so two + /// concurrent grants cannot each write back a list missing the other's. + pub(super) fn shared_with_guard(&self) -> std::sync::MutexGuard<'_, BTreeSet> { + self.shared_with.lock().unwrap_or_else(|p| p.into_inner()) + } + /// A snapshot of the access list. pub fn shared_with(&self) -> BTreeSet { self.shared_with @@ -372,6 +396,19 @@ impl TenantRegistry { /// A tenant that is not resident is restored first: a visitor must be /// able to open a shared document whose owner is offline, and refusing /// until the owner next signs in would make sharing useless. + /// + /// ## Admission precedes materialisation + /// + /// Materialising first would let an unauthenticated-in-practice caller + /// spend the daemon's whole tenant budget: `?tenant=` names an arbitrary + /// account, and creating the tenant to discover the caller is not on its + /// list means every refused request still costs a resident tenant. So a + /// non-resident owner is admitted from the PERSISTED access list, which + /// reads one small file and materialises nothing. + /// + /// A resident owner is checked against the live list, which is + /// authoritative — a revoke that has not been written yet still takes + /// effect immediately. pub fn lease_for_shared( &self, owner_id: &str, @@ -380,13 +417,33 @@ impl TenantRegistry { if owner_id == visitor.user_id { return self.lease_tenant(owner_id); } + if !self.admits_visitor(owner_id, &visitor.user_id) { + return Err(TenantError::NotShared); + } let lease = self.lease_tenant(owner_id)?; + // Re-checked against the live list now that the tenant is resident: a + // revoke may have landed between the two, and the in-memory list is + // the authority. if !lease.tenant().admits(&visitor.user_id) { return Err(TenantError::NotShared); } Ok(lease) } + /// Whether `visitor` is on `owner_id`'s access list, WITHOUT materialising + /// the tenant. + /// + /// Resident tenants answer from memory; the rest answer from the persisted + /// list. A deployment with no store therefore admits nobody to a + /// non-resident tenant, which is the fail-closed direction — the share was + /// never durable in the first place. + fn admits_visitor(&self, owner_id: &str, visitor: &str) -> bool { + if let Some(tenant) = self.lock().get(owner_id) { + return tenant.admits(visitor); + } + self.store.load_acl(owner_id).contains(visitor) + } + /// Who has shared with `visitor`, across every resident tenant. /// /// Resident only, and deliberately: a full answer would mean reading every @@ -488,13 +545,49 @@ impl TenantRegistry { written } - /// Persist a tenant's access list, if persistence is on. - pub fn persist_acl(&self, user_id: &str, tenant: &Tenant) { - if !self.store.is_enabled() { - return; + /// Apply one access-list change and persist the result atomically. + /// + /// The edit and its write happen under the SAME lock, for two reasons. + /// Snapshot-then-write let two concurrent grants each read the list before + /// the other's insert and write back a version missing it — the second + /// write silently dropping the first grant. And a write that fails has to + /// be reported: the previous code logged it and answered 200, so a user + /// was told a share had succeeded that would vanish on the next restart. + /// + /// On a write failure the in-memory change is ROLLED BACK, so memory and + /// disk agree and the caller's retry starts from a known state. The + /// alternative — keep it in memory and mark it pending — would mean the + /// share works until the process restarts and then silently stops, which + /// is the harder failure to diagnose. + pub fn update_acl( + &self, + user_id: &str, + tenant: &Tenant, + change: AclChange, + ) -> Result { + let mut list = tenant.shared_with_guard(); + let changed = match &change { + AclChange::Grant(account) => list.insert(account.clone()), + AclChange::Revoke(account) => list.remove(account.as_str()), + }; + if !changed || !self.store.is_enabled() { + return Ok(AclUpdate { + changed, + shared_with: list.clone(), + }); } - if let Err(error) = self.store.save_acl_for(user_id, &tenant.shared_with()) { - eprintln!("openpencil --serve-web --online: could not persist a share list ({error})"); + match self.store.save_acl_for(user_id, &list) { + Ok(()) => Ok(AclUpdate { + changed, + shared_with: list.clone(), + }), + Err(error) => { + match &change { + AclChange::Grant(account) => list.remove(account.as_str()), + AclChange::Revoke(account) => list.insert(account.clone()), + }; + Err(error) + } } } diff --git a/crates/op-host-services/src/web_canvas_server/tenant_auth.rs b/crates/op-host-services/src/web_canvas_server/tenant_auth.rs index 4cba4f85b..a597c979d 100644 --- a/crates/op-host-services/src/web_canvas_server/tenant_auth.rs +++ b/crates/op-host-services/src/web_canvas_server/tenant_auth.rs @@ -171,12 +171,14 @@ impl StaticVerifier { ) } - /// Parse a `token=user[,token2=user2:read]` table. + /// Parse a `token=user[,token2=user2:read][,token3=user3:none]` table. /// - /// The optional `:read` suffix on the account mints a read-only - /// credential, so the scope path can be exercised without a hub. Any - /// other suffix is rejected rather than silently granting write — a typo - /// must not widen authority. + /// The optional suffix narrows the credential so the scope paths can be + /// exercised without a hub: `:read` mints a read-only token and `:none` a + /// scopeless one (what a hub token that names no `mcp:*` scope resolves + /// to). A bare `user` is full authority — the operator wrote this table by + /// hand, so that is their explicit intent. Any other suffix is rejected + /// rather than silently granting write: a typo must not widen authority. /// /// Malformed pairs are skipped rather than failing the whole table: the /// failure mode of a dropped entry is "that token does not authenticate", @@ -189,6 +191,7 @@ impl StaticVerifier { let token = token.trim(); let (user, scopes) = match account.trim().rsplit_once(':') { Some((user, "read")) => (user.trim(), McpScopes::READ_ONLY), + Some((user, "none")) => (user.trim(), McpScopes::NONE), Some((_, _)) => return None, None => (account.trim(), McpScopes::FULL), }; diff --git a/crates/op-host-services/src/web_canvas_server/tenant_store.rs b/crates/op-host-services/src/web_canvas_server/tenant_store.rs index 6dea156e9..3e61536bf 100644 --- a/crates/op-host-services/src/web_canvas_server/tenant_store.rs +++ b/crates/op-host-services/src/web_canvas_server/tenant_store.rs @@ -230,7 +230,10 @@ pub fn dir_name(user_id: &str) -> String { /// Write `bytes` to `path` via a same-directory temp file plus rename. fn atomic_write(path: &Path, bytes: &[u8]) -> Result<(), TenantStoreError> { - let temp = path.with_extension("tmp"); + // Unique per writer: a fixed `.tmp` is shared by every concurrent write to + // the same tenant, so two of them interleave into one file and the rename + // publishes a mixture of both. + let temp = path.with_extension(format!("tmp.{}.{}", std::process::id(), next_temp_id())); std::fs::write(&temp, bytes).map_err(|error| TenantStoreError::Io(error.to_string()))?; std::fs::rename(&temp, path).map_err(|error| { // Leaving the temp behind would accumulate one file per failed write. @@ -239,6 +242,12 @@ fn atomic_write(path: &Path, bytes: &[u8]) -> Result<(), TenantStoreError> { }) } +/// A per-process counter making temp file names unique. +fn next_temp_id() -> u64 { + static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed) +} + /// Move a file that would not parse out of the way, preserving it. /// /// Best effort: if the rename fails there is nothing useful left to do, and diff --git a/crates/op-host-services/src/web_canvas_server/tenant_tests.rs b/crates/op-host-services/src/web_canvas_server/tenant_tests.rs index d73e22091..2b929db9e 100644 --- a/crates/op-host-services/src/web_canvas_server/tenant_tests.rs +++ b/crates/op-host-services/src/web_canvas_server/tenant_tests.rs @@ -71,9 +71,9 @@ fn every_tenant_owns_its_own_broadcast_hub() { collab_seq: 0, }); - assert_eq!(a_sub.try_recv().expect("A hears its own bump").version, 5); + assert_eq!(a_sub.pending().expect("A hears its own bump").version, 5); assert!( - b_sub.try_recv().is_err(), + b_sub.pending().is_none(), "a tenant's version bump must not reach another tenant's subscribers" ); } diff --git a/crates/op-host-services/src/web_canvas_server/tool_scopes.rs b/crates/op-host-services/src/web_canvas_server/tool_scopes.rs new file mode 100644 index 000000000..5271d6c5a --- /dev/null +++ b/crates/op-host-services/src/web_canvas_server/tool_scopes.rs @@ -0,0 +1,85 @@ +//! Scope enforcement for the REST tier. +//! +//! The MCP dispatch has enforced scopes since M3, but `/api/*` did not — so a +//! read-only API token could simply `POST /api/mcp/document` and replace the +//! whole document, which is strictly more damage than any single tool call +//! could do. Same credential, same deployment, two different answers. +//! +//! This closes that: an API-token identity is held to the same scopes on REST +//! that it is on `/mcp`. A browser session is not — it IS the account, and +//! scopes exist to narrow a token below the account's own authority. + +use crate::mcp_serve::tool_profile::{McpScopes, MCP_READ_SCOPE, MCP_WRITE_SCOPE}; + +use super::tenant_auth::IdentityVia; + +/// Which scope a REST request needs. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RestScope { + Read, + Write, +} + +impl RestScope { + const fn name(self) -> &'static str { + match self { + Self::Read => MCP_READ_SCOPE, + Self::Write => MCP_WRITE_SCOPE, + } + } + + const fn allowed_by(self, scopes: McpScopes) -> bool { + match self { + Self::Read => scopes.can_read(), + Self::Write => scopes.can_write(), + } + } +} + +/// Why a REST request was refused on scope grounds. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct RestScopeRefusal { + required: RestScope, +} + +impl RestScopeRefusal { + pub const fn code(self) -> &'static str { + "scope-insufficient" + } + + pub const fn http_status(self) -> &'static str { + "403 Forbidden" + } +} + +impl std::fmt::Display for RestScopeRefusal { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "this credential lacks the '{}' scope", + self.required.name() + ) + } +} + +impl std::error::Error for RestScopeRefusal {} + +/// Decide whether a REST request may proceed. +/// +/// `None` when it may. A session-cookie identity is never refused here. +pub(super) fn check_rest_scope( + via: IdentityVia, + scopes: McpScopes, + method: &str, + path: &str, +) -> Option { + if via == IdentityVia::SessionCookie { + return None; + } + let required = super::online_policy::rest_scope_required(method, path)?; + (!required.allowed_by(scopes)).then_some(RestScopeRefusal { required }) +} + +#[cfg(test)] +#[path = "tool_scopes_tests.rs"] +mod tests; diff --git a/crates/op-host-services/src/web_canvas_server/tool_scopes_tests.rs b/crates/op-host-services/src/web_canvas_server/tool_scopes_tests.rs new file mode 100644 index 000000000..24b371d1c --- /dev/null +++ b/crates/op-host-services/src/web_canvas_server/tool_scopes_tests.rs @@ -0,0 +1,93 @@ +//! Tests for REST scope enforcement. + +use super::*; + +const READ_ONLY: McpScopes = McpScopes::READ_ONLY; +const FULL: McpScopes = McpScopes::FULL; +const NONE: McpScopes = McpScopes::NONE; + +fn refused(scopes: McpScopes, method: &str, path: &str) -> bool { + check_rest_scope(IdentityVia::ApiToken, scopes, method, path).is_some() +} + +#[test] +fn a_read_only_token_cannot_write_the_document_over_rest() { + // The bypass this closes: the same credential is refused `add_page` on + // /mcp but could replace the entire document here. + assert!(refused(READ_ONLY, "POST", "/api/mcp/document")); + assert!(refused(READ_ONLY, "POST", "/api/mcp/selection")); + assert!(refused(READ_ONLY, "POST", "/api/settings/credentials")); + assert!(refused(READ_ONLY, "DELETE", "/api/mcp/document")); + assert!(refused(READ_ONLY, "PUT", "/api/mcp/document")); +} + +#[test] +fn a_read_only_token_may_still_read() { + assert!(!refused(READ_ONLY, "GET", "/api/mcp/document")); + assert!(!refused(READ_ONLY, "GET", "/api/mcp/version")); + assert!(!refused(READ_ONLY, "GET", "/api/mcp/events")); +} + +#[test] +fn a_full_scope_token_is_unrestricted() { + for (method, path) in [ + ("GET", "/api/mcp/document"), + ("POST", "/api/mcp/document"), + ("DELETE", "/api/share/grant"), + ] { + assert!(!refused(FULL, method, path), "{method} {path}"); + } +} + +#[test] +fn a_token_with_no_scopes_is_refused_everything_but_the_health_probe() { + // Fail-closed: op-hub issues no tokens yet, so tightening this costs + // nothing and removes a default that would be hard to tighten later. + assert!(refused(NONE, "GET", "/api/mcp/document")); + assert!(refused(NONE, "POST", "/api/mcp/document")); + // …except the probe, so a client can discover the daemon and be told why. + assert!(!refused(NONE, "GET", "/api/mcp/server")); +} + +#[test] +fn a_browser_session_is_never_scope_limited() { + // A session IS the account; scopes exist to narrow a token below it. + for scopes in [NONE, READ_ONLY, FULL] { + for (method, path) in [ + ("GET", "/api/mcp/document"), + ("POST", "/api/mcp/document"), + ("POST", "/api/share/grant"), + ] { + assert!( + check_rest_scope(IdentityVia::SessionCookie, scopes, method, path).is_none(), + "{method} {path}" + ); + } + } +} + +#[test] +fn an_unknown_method_needs_write() { + // A route or verb added later is covered by default, in the strict + // direction. + assert!(refused(READ_ONLY, "PATCH", "/api/anything/new")); + assert!(refused(READ_ONLY, "PROPFIND", "/api/anything/new")); +} + +#[test] +fn a_refusal_is_a_typed_403_naming_the_missing_scope() { + let refusal = check_rest_scope( + IdentityVia::ApiToken, + READ_ONLY, + "POST", + "/api/mcp/document", + ) + .expect("refused"); + assert_eq!(refusal.http_status(), "403 Forbidden"); + assert_eq!(refusal.code(), "scope-insufficient"); + assert!(refusal.to_string().contains("mcp:write"), "{refusal}"); + + let read = + check_rest_scope(IdentityVia::ApiToken, NONE, "GET", "/api/mcp/document").expect("refused"); + assert!(read.to_string().contains("mcp:read"), "{read}"); +} diff --git a/crates/op-host-services/src/web_canvas_server_conn_tests.rs b/crates/op-host-services/src/web_canvas_server_conn_tests.rs index ed4f8a771..891132981 100644 --- a/crates/op-host-services/src/web_canvas_server_conn_tests.rs +++ b/crates/op-host-services/src/web_canvas_server_conn_tests.rs @@ -88,8 +88,8 @@ fn sse_hub_broadcasts_version_to_all_subscribers() { let a = hub.subscribe(); let b = hub.subscribe(); hub.broadcast(tick(5, 0)); - assert_eq!(a.recv().unwrap(), tick(5, 0)); - assert_eq!(b.recv().unwrap(), tick(5, 0)); + assert_eq!(a.pending().expect("published"), tick(5, 0)); + assert_eq!(b.pending().expect("published"), tick(5, 0)); } #[test] @@ -100,7 +100,7 @@ fn sse_hub_prunes_disconnected_subscribers() { assert_eq!(hub.subscriber_count(), 2); hub.broadcast(tick(1, 0)); // prunes the dropped one assert_eq!(hub.subscriber_count(), 1); - assert_eq!(live.recv().unwrap(), tick(1, 0)); + assert_eq!(live.pending().expect("published"), tick(1, 0)); } #[test] @@ -128,16 +128,69 @@ fn sse_payload_stays_a_superset_of_the_original_version_frame() { } #[test] -fn serve_sse_emits_initial_then_each_version_until_hub_drops() { - let (tx, rx) = mpsc::channel(); - tx.send(tick(9, 0)).expect("send"); // one bump, then the sender drops → Disconnected - drop(tx); - let mut stream = mock_stream(""); - serve_sse(&mut stream, rx, tick(7, 0), Some("*")).expect("serve_sse"); - let out = String::from_utf8_lossy(&stream.output); +fn serve_sse_emits_the_initial_tick_then_each_published_one() { + // The stream ends when a socket write fails, which is how a disconnected + // client is detected — so this writes to a stream that fails on the + // second event. + let hub = SseHub::default(); + let slot = hub.subscribe(); + hub.broadcast(tick(9, 0)); + let mut stream = FailingStream { + written: Vec::new(), + writes_before_failure: 3, + }; + let _ = serve_sse(&mut stream, &slot, tick(7, 0), Some("*")); + let out = String::from_utf8_lossy(&stream.written); assert!(out.contains("text/event-stream"), "{out}"); assert!(out.contains(r#"data: {"version":7,"#), "{out}"); // initial sync - assert!(out.contains(r#"data: {"version":9,"#), "{out}"); // broadcast bump + assert!(out.contains(r#"data: {"version":9,"#), "{out}"); // published bump +} + +/// A stream that stops accepting writes, standing in for a client that went +/// away — which is the only thing that ends an SSE loop. +struct FailingStream { + written: Vec, + writes_before_failure: usize, +} + +impl std::io::Write for FailingStream { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + if self.writes_before_failure == 0 { + return Err(std::io::Error::other("client went away")); + } + self.writes_before_failure -= 1; + self.written.extend_from_slice(buf); + Ok(buf.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +#[test] +fn a_subscriber_that_never_reads_keeps_exactly_one_pending_tick() { + // The unbounded-queue bug: a paused tab accumulated one entry per + // mutation, in a process shared with every other account. + let hub = SseHub::default(); + let slot = hub.subscribe(); + for version in 0..1000 { + hub.broadcast(tick(version, 0)); + } + // Only the newest survives — an older tick is not information the client + // lost, it is information the newest one already contains. + assert_eq!(slot.pending(), Some(tick(999, 0))); + assert_eq!(slot.pending(), None, "taking it leaves the slot empty"); +} + +#[test] +fn a_dropped_subscriber_is_pruned_without_signalling_anything() { + let hub = SseHub::default(); + let live = hub.subscribe(); + drop(hub.subscribe()); + assert_eq!(hub.subscriber_count(), 2); + hub.broadcast(tick(1, 0)); + assert_eq!(hub.subscriber_count(), 1); + assert_eq!(live.pending(), Some(tick(1, 0))); } #[test] @@ -153,7 +206,7 @@ fn serve_one_post_document_broadcasts_new_version_to_sse() { let mut stream = mock_stream(&request); serve_one(&mut stream, &state, &hub).expect("serve_one"); // The whole-doc sync bumped the version to 1 and broadcast it. - assert_eq!(sub.recv().unwrap().version, 1); + assert_eq!(sub.pending().expect("published").version, 1); } #[test] diff --git a/crates/op-host-services/src/web_canvas_server_error.rs b/crates/op-host-services/src/web_canvas_server_error.rs index 5dde87adb..74c2389c3 100644 --- a/crates/op-host-services/src/web_canvas_server_error.rs +++ b/crates/op-host-services/src/web_canvas_server_error.rs @@ -59,6 +59,11 @@ pub enum WebCanvasError { /// not a daemon fault — the document is healthy and the request was /// well-formed; it simply cannot be sequenced right now. Collab(crate::web_canvas_server::DaemonMutationRefusal), + /// A live session opened a capture for a whole-document push and then + /// declined to commit it. Distinct from `Collab`: the write was not + /// refused up front, it was discarded, so the browser's copy is now + /// definitively behind and must be refetched rather than retried. + IngestRejected(crate::web_canvas_server::IngestOutcome), } impl WebCanvasError { @@ -78,6 +83,7 @@ impl WebCanvasError { | WebCanvasError::Io(_) => "400 Bad Request", WebCanvasError::Config(_) | WebCanvasError::Transport(_) => "500 Internal Server Error", WebCanvasError::Collab(refusal) => refusal.http_status(), + WebCanvasError::IngestRejected(_) => "409 Conflict", } } @@ -89,6 +95,7 @@ impl WebCanvasError { pub fn error_code(&self) -> Option<&'static str> { match self { WebCanvasError::Collab(refusal) => Some(refusal.code()), + WebCanvasError::IngestRejected(outcome) => outcome.error_code(), _ => None, } } @@ -104,6 +111,9 @@ impl fmt::Display for WebCanvasError { | WebCanvasError::Config(m) | WebCanvasError::Transport(m) => f.write_str(m), WebCanvasError::Collab(refusal) => write!(f, "{refusal}"), + WebCanvasError::IngestRejected(_) => { + f.write_str("the live session discarded this document push; refetch and reapply") + } } } } diff --git a/crates/op-host-services/src/web_chat_standard_tests.rs b/crates/op-host-services/src/web_chat_standard_tests.rs index 3f5863a46..8c7b135ea 100644 --- a/crates/op-host-services/src/web_chat_standard_tests.rs +++ b/crates/op-host-services/src/web_chat_standard_tests.rs @@ -305,7 +305,7 @@ fn design_doc_sink_applies_and_bumps_version() { })); assert_eq!(state.lock().unwrap().version, 1); - assert_eq!(sub.recv().unwrap().version, 1); + assert_eq!(sub.pending().expect("published").version, 1); assert_eq!(sink.state().active_children().len(), 1); }