SemianiakaVY/openpencil
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

fix(web): close the review's high-severity online gaps

Browse source 
- SSE subscribers hold a single latest-value slot (weak-pruned) so a
  stalled reader coalesces ticks instead of growing an unbounded queue
- /api/file/save clears the same whole-document gate as open-recent:
  409 collab-active during a session instead of silently desyncing the
  runtime's hash and commit log
- document ingest returns a typed outcome; only Committed bumps the
  version, Rejected/Failed surface codes the browser's existing
  conflict recovery already handles, and the local-edit capture closes
  through an RAII guard on every path including unwind
- a ?tenant= request now passes persisted-ACL admission before any
  tenant materialises, so unauthorised probes cannot occupy capacity
- ACL edits and their durable write serialise under one mutex with
  unique temp files; a failed write rolls back and returns 500 rather
  than resurrecting revoked grants after restart
- API-token REST calls are scope-checked by method (session cookies
  exempt) and a token naming no mcp:* scope now gets nothing instead
  of everything — hub-issued tokens must carry explicit scopes
- SIGTERM/SIGINT in the online loop raise the existing shutdown flag
  so container stops run the tenant flush
This commit is contained in:
Kayshen-X 2026-08-08 17:03:47 +08:00
parent f70bf89006
commit d909578ddb
25 changed files with 1159 additions and 150 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -4137,6 +4137,7 @@ dependencies = [
"glam",
"hmac",
"jian-ops-schema",
"libc",
"memmap2",
"op-acp",
"op-ai",

6
crates/op-host-services/Cargo.toml
View file

@ -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"] }

31
crates/op-host-services/src/mcp_serve/tool_profile.rs
View file

@ -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<S: AsRef<str>>(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,

8
crates/op-host-services/src/mcp_serve/tool_profile_tests.rs
View file

@ -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"]),

102
crates/op-host-services/src/web_canvas_server.rs
View file

@ -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<Vec<mpsc::Sender<SseTick>>>,
}
impl SseHub {
/// Register a subscriber; the SSE connection thread blocks on the returned
/// receiver for ticks.
pub(crate) fn subscribe(&self) -> Receiver<SseTick> {
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<AtomicUsize>);
@ -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::*;

12
crates/op-host-services/src/web_canvas_server/collab_driver.rs
View file

@ -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]

103
crates/op-host-services/src/web_canvas_server/collab_state.rs
View file

@ -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
}
}

77
crates/op-host-services/src/web_canvas_server/collab_state_tests.rs
View file

@ -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"));
}

58
crates/op-host-services/src/web_canvas_server/connection.rs
View file

@ -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<S: Read + Write>(
hub,
mode,
mcp_profile: crate::mcp_serve::tool_profile::McpAccessProfile::UNRESTRICTED,
rest_identity: None,
},
)
}
@ -246,9 +255,9 @@ pub(super) fn dispatch<S: Read + Write>(
// 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<S: Read + Write>(
// 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<S: Read + Write>(
/// disconnects (write error) or the hub is dropped.
pub(super) fn serve_sse<S: Write>(
stream: &mut S,
rx: Receiver<SseTick>,
slot: &SseSlot,
current: SseTick,
cors_origin: Option<&str>,
) -> Result<()> {
@ -483,20 +513,15 @@ pub(super) fn serve_sse<S: Write>(
.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<S: Write>(
.flush()
.map_err(|e| WebCanvasError::Transport(format!("sse flush: {e}")))?;
}
Err(RecvTimeoutError::Disconnected) => return Ok(()),
}
}
}

7
crates/op-host-services/src/web_canvas_server/hub_verifier.rs
View file

@ -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]

91
crates/op-host-services/src/web_canvas_server/online_mcp_tests.rs
View file

@ -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(
&registry(),
&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(
&registry(),
&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(
&registry,
&verifier,
Request::new(method, path).with_bearer("tokN"),
);
assert_eq!(
status_line(&response),
"HTTP/1.1 403 Forbidden",
"{method} {path}: {response}"
);
}
let push = serve(
&registry,
&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(
&registry,
&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(
&registry,
&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(
&registry,
&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}");
}

24
crates/op-host-services/src/web_canvas_server/online_policy.rs
View file

@ -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<super::tool_scopes::RestScope> {
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

66
crates/op-host-services/src/web_canvas_server/online_run_loop.rs
View file

@ -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(&registry, &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<AtomicBool>, 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<S: Read + Write>(
// 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)),
},
)
}

162
crates/op-host-services/src/web_canvas_server/online_share_tests.rs
View file

@ -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(
&registry,
&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(
&registry,
&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);
}

49
crates/op-host-services/src/web_canvas_server/share_routes.rs
View file

@ -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::<Vec<_>>(),
})
.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::<Vec<_>>(),
})
.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(),
},
}
}

110
crates/op-host-services/src/web_canvas_server/sse_hub.rs Normal file
View file

@ -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<Option<SseTick>>,
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<SseTick> {
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<SseTick> {
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<Vec<Weak<SseSlot>>>,
}
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<SseSlot> {
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()
}
}

107
crates/op-host-services/src/web_canvas_server/tenant.rs
View file

@ -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<String>,
}
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<String>> {
self.shared_with.lock().unwrap_or_else(|p| p.into_inner())
}
/// A snapshot of the access list.
pub fn shared_with(&self) -> BTreeSet<String> {
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<AclUpdate, TenantStoreError> {
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)
}
}
}

13
crates/op-host-services/src/web_canvas_server/tenant_auth.rs
View file

@ -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),
};

11
crates/op-host-services/src/web_canvas_server/tenant_store.rs
View file

@ -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

4
crates/op-host-services/src/web_canvas_server/tenant_tests.rs
View file

@ -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"
);
}

85
crates/op-host-services/src/web_canvas_server/tool_scopes.rs Normal file
View file

@ -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<RestScopeRefusal> {
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;

93
crates/op-host-services/src/web_canvas_server/tool_scopes_tests.rs Normal file
View file

@ -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}");
}

77
crates/op-host-services/src/web_canvas_server_conn_tests.rs
View file

@ -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<u8>,
writes_before_failure: usize,
}
impl std::io::Write for FailingStream {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
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]

10
crates/op-host-services/src/web_canvas_server_error.rs
View file

@ -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")
}
}
}
}

2
crates/op-host-services/src/web_chat_standard_tests.rs
View file

@ -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);
}