diff --git a/frontend/src/lib/message-bus/client.svelte.ts b/frontend/src/lib/message-bus/client.svelte.ts index 58d1045c..a381ae30 100644 --- a/frontend/src/lib/message-bus/client.svelte.ts +++ b/frontend/src/lib/message-bus/client.svelte.ts @@ -18,7 +18,10 @@ // and the console logger so users can diagnose without a redeploy. import log from 'loglevel'; +import { untrack } from 'svelte'; +import { apiJson } from '$lib/api/client'; +import { getCsrfHeaders } from '$lib/api/csrf'; import { RtErrorCode } from './error-codes'; import { parseIncoming, @@ -30,6 +33,20 @@ import { import type RtEventParams from '$lib/generated/message-bus/RtEventParams'; import type RtRevokedParams from '$lib/generated/message-bus/RtRevokedParams'; +/** Response body from `POST /api/rt/ticket`. Matches the Rust + * `RtTicketResponse` shape — see `handlers/rt_ticket_handler.rs`. */ +interface RtTicketResponse { + /** Opaque ticket UUID. Redeemed once server-side. */ + ticket: string; + /** Seconds until server-side expiry (informational; the client + * should open the WS immediately). */ + expires_in_seconds: number; + /** Full `Sec-WebSocket-Protocol` value the client MUST pass on + * the upgrade — assembled server-side so a FE bug can't emit + * the wrong prefix. */ + subprotocol: string; +} + /** Logger namespace — matches `frontend/AGENTS.md § Logging`. Users * tune with `oxi.setLogLevel('oxi:message-bus', 'debug')`. */ const busLog = log.getLogger('oxi:message-bus'); @@ -73,6 +90,16 @@ export interface MessageBusError { const RECONNECT_MIN_MS = 250; const RECONNECT_MAX_MS = 30_000; +/** Circuit breaker — after N consecutive failed attempts (either a + * ticket-exchange rejection or a WS close before `onopen` fires), + * give up and stay `disconnected` until the caller explicitly asks + * to `reconnect()`. Prevents an unrecoverable auth state (revoked + * session, wrong CSRF cookie, missing DPoP nonce) from flooding + * logs. Ten attempts × exponential-backoff-with-jitter is roughly a + * minute of trying — long enough for a transient blip, short enough + * to stop before it's noise. */ +const MAX_CONSECUTIVE_FAILURES = 10; + interface SubEntry { count: number; handlers: Set; @@ -102,6 +129,11 @@ export class MessageBusClient { /** setTimeout handle for a scheduled reconnect. Cleared on * explicit `close()` so we don't reconnect after teardown. */ #reconnectTimer: ReturnType | null = null; + /** Consecutive failures — incremented on every attempt that dies + * before `#onOpen()` gets to reset it. Once it hits + * `MAX_CONSECUTIVE_FAILURES` the client stops reconnecting and + * requires an explicit `reconnect()` from the caller. */ + #consecutiveFailures = 0; /** `topic` → `{count, handlers, revokedHandlers, acked}`. Refcount * drives the wire: first refcount ⇒ send `rt.subscribe`; last drop @@ -139,49 +171,68 @@ export class MessageBusClient { * to the local handlers for `topic`, sends `rt.subscribe` on the * wire only for the first ref, and returns an unsubscribe fn that * drops that same ref (last ref out sends `rt.unsubscribe`). + * + * Wrapped in `untrack` because `this.state` is `$state`. Without + * this, a caller invoking `subscribe` from a Svelte `$effect` + * (which `useTopic` does) would take a reactive dep on `state`. + * Every `state` transition (idle → connecting → disconnected → + * connecting → …) would then re-fire the caller's `$effect`, + * which re-calls `subscribe`, which flips `state`, which re-fires + * the effect — a 1000+/s runaway loop, observed on server-down + * (2026-09-11). `subscribe` is a mutation entry point; its reads + * of internal state MUST NOT contaminate reactive callers. */ subscribe(topic: string, onEvent: EventHandler, onRevoked?: RevokedHandler): UnsubscribeHandle { - let entry = this.#subs.get(topic); - if (!entry) { - // Plain Sets: internal callback registries, not reactive. Same - // rationale as `#subs` / `#pending` — see the doc there. - entry = { - count: 0, - // eslint-disable-next-line svelte/prefer-svelte-reactivity - handlers: new Set(), - // eslint-disable-next-line svelte/prefer-svelte-reactivity - revokedHandlers: new Set(), - acked: false + return untrack(() => { + let entry = this.#subs.get(topic); + if (!entry) { + // Plain Sets: internal callback registries, not reactive. + // Same rationale as `#subs` / `#pending` — see the doc + // there. + entry = { + count: 0, + // eslint-disable-next-line svelte/prefer-svelte-reactivity + handlers: new Set(), + // eslint-disable-next-line svelte/prefer-svelte-reactivity + revokedHandlers: new Set(), + acked: false + }; + this.#subs.set(topic, entry); + } + entry.count += 1; + entry.handlers.add(onEvent); + if (onRevoked) entry.revokedHandlers.add(onRevoked); + + // Kick the connection if nothing is holding it yet, otherwise + // send `rt.subscribe` if this is the first ref on this topic. + if (this.state === 'idle' || this.state === 'disconnected') { + this.#connect(); + } else if (entry.count === 1 && this.state === 'connected') { + this.#sendSubscribe(topic).catch((err) => + busLog.warn('subscribe failed', { topic, error: err }) + ); + } + + let released = false; + return () => { + if (released) return; + released = true; + // Cleanup path — Svelte `$effect` cleanup doesn't track + // anyway, but stay defensive: untrack around the + // internal state reads inside #releaseOne. + untrack(() => this.#releaseOne(topic, onEvent, onRevoked)); }; - this.#subs.set(topic, entry); - } - entry.count += 1; - entry.handlers.add(onEvent); - if (onRevoked) entry.revokedHandlers.add(onRevoked); - - // Kick the connection if nothing is holding it yet, otherwise - // send `rt.subscribe` if this is the first ref on this topic. - if (this.state === 'idle' || this.state === 'disconnected') { - this.#connect(); - } else if (entry.count === 1 && this.state === 'connected') { - this.#sendSubscribe(topic).catch((err) => - busLog.warn('subscribe failed', { topic, error: err }) - ); - } - - let released = false; - return () => { - if (released) return; - released = true; - this.#releaseOne(topic, onEvent, onRevoked); - }; + }); } /** Force a fresh reconnect — for a live-updates toggle or a manual - * "reconnect" button. Rare; not part of the normal flow. */ + * "reconnect" button. Rare; not part of the normal flow. Also the + * escape hatch after the circuit breaker trips: zeroes the + * consecutive-failure counter so the next attempt actually fires. */ reconnect(): void { if (this.#ws) this.#ws.close(); this.#backoffMs = RECONNECT_MIN_MS; + this.#consecutiveFailures = 0; this.#scheduleReconnect(0); } @@ -207,9 +258,40 @@ export class MessageBusClient { if (this.state === 'connecting' || this.state === 'connected') return; this.state = 'connecting'; busLog.debug('connecting', { url: this.#url }); + // Ticket exchange runs off a Promise; the connection is + // finalised inside its `.then`. Errors during exchange land in + // `#onTicketFailure`, which mirrors the WS-close reconnect path + // so a transient auth blip retries with backoff. + void this.#exchangeAndOpen(); + } + + /** POST `/api/rt/ticket`, then open the WS with the returned + * subprotocol. The POST runs through `apiFetch` — DPoP proof + * and session cookie handled by the interceptor — and we attach + * the CSRF header ourselves per every state-changing endpoint's + * convention (see `endpoints/shares.ts` for the pattern). */ + async #exchangeAndOpen(): Promise { + let subprotocol: string; + try { + const res = await apiJson('/api/rt/ticket', { + method: 'POST', + headers: getCsrfHeaders() + }); + subprotocol = res.subprotocol; + busLog.debug('ticket issued', { expires_in_seconds: res.expires_in_seconds }); + } catch (err) { + this.#onTicketFailure(err); + return; + } + // A close/reconnect could have raced this in-flight exchange; + // bail if we lost the "connecting" role in the meantime. + if (this.state !== 'connecting') { + busLog.debug('ticket exchange raced with close — discarding', { state: this.state }); + return; + } let ws: WebSocket; try { - ws = new this.#WebSocketCtor(this.#url); + ws = new this.#WebSocketCtor(this.#url, [subprotocol]); } catch (err) { busLog.warn('WebSocket ctor threw — reconnect scheduled', { error: err }); this.state = 'disconnected'; @@ -223,10 +305,21 @@ export class MessageBusClient { ws.onclose = (ev) => this.#onClose(ev); } + /** Handle a failed ticket exchange. Same shape as a WS close — + * we're not going to retry inline (a bad auth state won't fix + * itself in 250 ms), so schedule the next attempt through the + * standard reconnect path. */ + #onTicketFailure(err: unknown): void { + busLog.warn('ticket exchange failed — reconnect scheduled', { error: err }); + this.state = 'disconnected'; + if (this.#subs.size > 0) this.#scheduleReconnect(); + } + #onOpen(): void { busLog.debug('connected'); this.state = 'connected'; this.#backoffMs = RECONNECT_MIN_MS; + this.#consecutiveFailures = 0; // Replay every already-known topic. `entry.acked` is reset here // because the fresh connection has no server-side memory of // prior subscriptions. @@ -258,6 +351,16 @@ export class MessageBusClient { busLog.debug('event for unknown topic', { topic: frame.params.topic }); return; } + // Trace each delivered event so devs can watch the bus + // live in the console. Level `debug` — silent under the + // default `warn`. See `frontend/AGENTS.md § Logging` + // for the tune knob (`oxi.setLogLevel('oxi:message-bus', + // 'debug')`). + busLog.debug('event received', { + topic: frame.params.topic, + kind: frame.params.event, + actor: (frame.params.data as { actor?: string })?.actor + }); for (const handler of entry.handlers) { try { handler(frame.params); @@ -330,11 +433,30 @@ export class MessageBusClient { #scheduleReconnect(overrideMs?: number): void { if (this.#reconnectTimer !== null) return; + this.#consecutiveFailures += 1; + // Circuit breaker: after too many failures in a row, stop + // retrying and require an explicit `reconnect()` call from + // the caller. Prevents a bad auth state (session revoked, + // CSRF cookie stripped, DPoP nonce mismatch) from flooding + // server logs with the same 401/403 forever. `reconnect()` + // zeroes the counter and re-arms. + if (this.#consecutiveFailures >= MAX_CONSECUTIVE_FAILURES) { + busLog.error('circuit breaker tripped — reconnect suspended after too many failures', { + consecutiveFailures: this.#consecutiveFailures, + max: MAX_CONSECUTIVE_FAILURES, + remedy: 'call messageBus.reconnect() to retry, or refresh the page' + }); + return; + } const delay = overrideMs ?? this.#backoffMs; // Full jitter — random in [0, backoff]. Prevents thundering // herd if the server was momentarily overloaded. const jittered = Math.floor(Math.random() * (delay + 1)); - busLog.warn('reconnect scheduled', { attemptBackoffMs: delay, jitteredMs: jittered }); + busLog.warn('reconnect scheduled', { + attemptBackoffMs: delay, + jitteredMs: jittered, + consecutiveFailures: this.#consecutiveFailures + }); this.#reconnectTimer = setTimeout(() => { this.#reconnectTimer = null; this.#backoffMs = Math.min(this.#backoffMs * 2, RECONNECT_MAX_MS); diff --git a/frontend/vite.config.ts b/frontend/vite.config.ts index fe16ba9b..68398657 100644 --- a/frontend/vite.config.ts +++ b/frontend/vite.config.ts @@ -34,8 +34,17 @@ const DEV_ORIGIN_HEADERS = { }; const p = (target: string) => ({ target, changeOrigin: true, headers: DEV_ORIGIN_HEADERS }); +// Same as `p()` but with WebSocket upgrade forwarding enabled. Vite's +// `http-proxy-middleware` treats HTTP and WS as two separate transports — +// without `ws: true` the upgrade request is silently dropped and the +// browser hangs in `readyState = CONNECTING` until Chrome's ~30 s +// handshake timeout fires. Needed for `/api/rt/ws` (message bus). Kept +// as a separate helper so paths that don't upgrade don't pay the extra +// listener setup. +const pWs = (target: string) => ({ ...p(target), ws: true }); + const proxy = { - '/api': p(BACKEND), + '/api': pWs(BACKEND), '/locales': p(BACKEND), '/.well-known': p(BACKEND), '/remote.php': p(BACKEND), diff --git a/src/application/services/trash_service.rs b/src/application/services/trash_service.rs index 0294ed16..576c71b9 100644 --- a/src/application/services/trash_service.rs +++ b/src/application/services/trash_service.rs @@ -14,6 +14,7 @@ use crate::application::dtos::trash_dto::{ }; use crate::application::ports::authorization_ports::AuthorizationEngine; use crate::application::ports::file_lifecycle::FileLifecycleHook; +use crate::application::ports::message_bus_ports::{MessageBus, MessageBusEvent, Topic}; use crate::application::ports::storage_ports::FileWritePort; use crate::application::ports::trash_ports::TrashUseCase; use crate::common::errors::{DomainError, ErrorKind, Result}; @@ -71,6 +72,16 @@ pub struct TrashService { /// so trash listings filter by drive membership instead of the legacy /// per-user scope. drive_repo: Arc, + + /// Message bus — publishes `FolderDeleted` on `Topic::Folder(parent)` + /// after a folder is trashed, so subscribers of the parent folder's + /// live-view refresh. `None` when the bus isn't wired (tests / stubs). + /// File trash is intentionally NOT published here: the FE hits + /// `DELETE /api/files/{id}` directly (bypasses the trash service) + /// and `FileManagementService::delete_and_cleanup_with_perms` + /// publishes on that path. If a future endpoint routes file delete + /// through this service, add the file-arm publish here too. + bus: Option>, } impl TrashService { @@ -93,6 +104,7 @@ impl TrashService { content_cache, authz, drive_repo, + bus: None, } } @@ -102,6 +114,15 @@ impl TrashService { self } + /// Wire the message bus. Enables the `FolderDeleted` publish on + /// `Topic::Folder(parent)` after a folder is trashed — folder-live + /// views subscribe to the parent topic and refresh on receipt. + /// Silent no-op if never called (unit tests skip this). + pub fn with_message_bus(mut self, bus: Arc) -> Self { + self.bus = Some(bus); + self + } + /// Converts a TrashedItem entity to a DTO fn to_dto(&self, item: TrashedItem) -> TrashedItemDto { // Calculate days_until_deletion before moving item fields @@ -236,6 +257,28 @@ impl TrashUseCase for TrashService { ) .await?; + // Snapshot the parent BEFORE the trash UPDATE — the row + // still exists at this point (soft-delete flips + // `is_trashed`, keeps the parent_id). We need parent_id + // to publish `FolderDeleted` on `Topic::Folder(parent)` + // after commit, so subscribers of the folder view refresh. + // If the bus isn't wired, skip the read to save a query. + let parent_snapshot = if self.bus.is_some() { + match self.folder_storage_port.get_folder(item_id).await { + Ok(folder) => folder.parent_id().and_then(|s| Uuid::parse_str(s).ok()), + Err(e) => { + debug!("trash-folder parent lookup failed: {}", e); + None + } + } + } else { + None + }; + debug!( + "trash-folder parent snapshot for {}: {:?}", + item_id, parent_snapshot + ); + // Soft-delete model — same as the file branch above: the // cascade UPDATE below is the whole operation; no folder // fetch or trash-index write needed. @@ -253,6 +296,32 @@ impl TrashUseCase for TrashService { })?; debug!("Folder moved to trash: {}", item_id); + + // Bus publish AFTER the trash commits. Root folders + // have `parent_id = None`; the trash endpoint refuses + // those via the mount / drive-root guards, but keep + // the `Some` gate anyway so a future permissive path + // doesn't panic here. + if let (Some(bus), Some(parent_uuid)) = (&self.bus, parent_snapshot) { + debug!( + "publishing FolderDeleted folder={} parent={} actor={}", + folder_id, parent_uuid, user_id + ); + bus.publish( + &Topic::Folder(parent_uuid), + MessageBusEvent::FolderDeleted { + folder_id, + parent_id: parent_uuid, + actor: user_id, + }, + ); + } else { + debug!( + "trash-folder publish skipped: bus={} parent={:?}", + self.bus.is_some(), + parent_snapshot + ); + } Ok(()) } _ => Err(DomainError::validation_error(format!( diff --git a/src/bin/generate-asyncapi.rs b/src/bin/generate-asyncapi.rs index 769a62d6..45c5f75d 100644 --- a/src/bin/generate-asyncapi.rs +++ b/src/bin/generate-asyncapi.rs @@ -76,21 +76,32 @@ Phase C (sync-client push, album live) extend the same channels — see }, }, "protocolVersion": "13", - // Subprotocol advertised in the WS handshake. Handler - // accepts `oxi.rt.v1` and the optional bearer element - // `authorization.bearer.` alongside it. + // Subprotocol advertised in the WS handshake. The handler + // accepts one of two shapes: + // * `oxi.ticket.` — the browser path. Redeems a + // one-shot 30 s ticket minted by + // `POST /api/rt/ticket` (that endpoint runs under the + // full auth + DPoP stack, so the ticket effectively + // inherits the proofed session). + // * (no subprotocol) — falls back to + // `Authorization: Bearer `, used by programmatic + // clients that can set headers (e.g. rt-hurl-helper). "bindings": { - "ws": { "subProtocol": "oxi.rt.v1" } + "ws": { "subProtocol": "oxi.ticket.{ticket}" } }, - // Every request MUST be authenticated. Programmatic - // clients set `Authorization: Bearer ` on the WS - // upgrade (same header the REST API uses); browser - // clients — which can't set headers on `new WebSocket()` - // — will use the deferred ticket flow (a plain HTTP - // POST issues a short-lived one-shot ticket bound to - // the WS URL, see the plan's DPoP-gap section). + // Every request MUST be authenticated. Two paths: + // * `bearerAuth` — programmatic clients set + // `Authorization: Bearer ` on the WS upgrade + // (same header the REST API uses). + // * `ticketAuth` — browser clients POST + // `/api/rt/ticket` with full auth + DPoP, receive + // an opaque one-shot token, and pass it via + // `Sec-WebSocket-Protocol: oxi.ticket.` + // (browsers cannot set arbitrary headers on + // `new WebSocket()`). See `docs/plan/message-bus.md § F`. "security": [ - { "$ref": "#/components/securitySchemes/bearerAuth" } + { "$ref": "#/components/securitySchemes/bearerAuth" }, + { "$ref": "#/components/securitySchemes/ticketAuth" } ], } }, @@ -299,7 +310,16 @@ fn components() -> Value { "type": "http", "scheme": "bearer", "bearerFormat": "JWT", - "description": "OxiCloud JWT — same access_token minted by `POST /api/auth/login` (or the OPAQUE handshake). Programmatic clients set `Authorization: Bearer ` on the WS upgrade request. Browsers, which cannot set headers on `new WebSocket()`, will use the deferred ticket flow (`POST /api/rt/ticket` → short-lived one-shot ticket in the WS URL); see the plan's DPoP-gap section.", + "description": "OxiCloud JWT — same access_token minted by `POST /api/auth/login` (or the OPAQUE handshake). Programmatic clients set `Authorization: Bearer ` on the WS upgrade request. DPoP-bound tokens are refused on this path (the WS handshake cannot carry a DPoP proof); browsers use `ticketAuth` instead.", + }, + // `httpApiKey` (not bare `apiKey`) — AsyncAPI 3.0 reserves + // `apiKey` for server-variable-based schemes; a header- + // scoped key is `httpApiKey` with `in: header`. + "ticketAuth": { + "type": "httpApiKey", + "in": "header", + "name": "Sec-WebSocket-Protocol", + "description": "Browser path — the FE first calls `POST /api/rt/ticket` under the full REST middleware stack (auth + DPoP-proofed request), receives an opaque one-shot UUID with a 30 s TTL, then sets `Sec-WebSocket-Protocol: oxi.ticket.` on the WS upgrade. The server redeems the ticket (single-use — a second attempt fails) and treats the WS session as authenticated for the caller who issued it. See `docs/plan/message-bus.md § F` and `handlers/rt_ticket_handler.rs`.", } } }); diff --git a/src/bin/rt-hurl-helper.rs b/src/bin/rt-hurl-helper.rs index 136efbbf..db9d2170 100644 --- a/src/bin/rt-hurl-helper.rs +++ b/src/bin/rt-hurl-helper.rs @@ -62,12 +62,29 @@ use tokio_tungstenite::tungstenite::http::HeaderValue; struct Args { mode: Mode, url: String, - token: String, + /// Either `--token ` (Authorization: Bearer path — the original + /// helper flow) or `--ticket ` (Sec-WebSocket-Protocol path + /// — exercises F). Exactly one MUST be set; parse_args enforces. + auth: WsAuth, subscribe: Vec, expect_events: Option, reason: Option, timeout: Duration, output: Option, + /// Optional path the helper `touch`es the instant EVERY requested + /// `--subscribe` topic has been ack'd by the server. Shell tests + /// wait on this file before firing the upload that publishes to + /// the topic, closing the "sleep 0.4 hoping the subscribe landed + /// in time" race that occasionally dropped events on slow / + /// cold-cache runs. Off by default; only used by the smoke test. + ready_file: Option, +} + +/// How the helper authenticates the WS upgrade. Mirrors the two paths +/// `rt_ws_handler::authenticate_upgrade` accepts. +enum WsAuth { + Bearer(String), + Ticket(String), } enum Mode { @@ -105,11 +122,13 @@ fn parse_args() -> Result { let mut url = None; let mut token = None; + let mut ticket = None; let mut subscribe = Vec::new(); let mut expect_events = None; let mut reason = None; let mut timeout = Duration::from_secs(3); let mut output = None; + let mut ready_file = None; while let Some(flag) = it.next() { let value = it @@ -118,6 +137,7 @@ fn parse_args() -> Result { match flag.as_str() { "--url" => url = Some(value), "--token" => token = Some(value), + "--ticket" => ticket = Some(value), "--subscribe" => subscribe.push(value), "--expect-events" => { expect_events = Some( @@ -129,19 +149,31 @@ fn parse_args() -> Result { "--reason" => reason = Some(value), "--timeout" => timeout = parse_duration(&value)?, "--output" => output = Some(value), + "--ready-file" => ready_file = Some(value), other => return Err(format!("unknown flag: {other}")), } } + // Exactly one credential MUST be set. Emitting a specific error + // makes shell-script drift ("forgot to swap --token for --ticket") + // debuggable at a glance. + let auth = match (token, ticket) { + (Some(_), Some(_)) => return Err("pass exactly one of --token or --ticket".into()), + (Some(t), None) => WsAuth::Bearer(t), + (None, Some(t)) => WsAuth::Ticket(t), + (None, None) => return Err("--token or --ticket required".into()), + }; + Ok(Args { mode, url: url.ok_or("--url required")?, - token: token.ok_or("--token required")?, + auth, subscribe, expect_events, reason, timeout, output, + ready_file, }) } @@ -202,14 +234,20 @@ impl From for HelperError { // WS connection // ════════════════════════════════════════════════════════════════════════════ -/// Open a WS connection to `url` with the given bearer token attached -/// via `Authorization: Bearer `. Programmatic client — this is the -/// path native clients (this helper, future sync-client integrations) -/// take. Browser clients that can't set the header will use the -/// `Sec-WebSocket-Protocol` subprotocol fallback (Phase A follow-up). +/// Open a WS connection to `url` with the given [`WsAuth`] applied. +/// +/// - `Bearer(jwt)` sets `Authorization: Bearer ` on the upgrade +/// — the programmatic-client path. +/// - `Ticket(uuid)` sets `Sec-WebSocket-Protocol: oxi.ticket.` +/// — the browser-equivalent path used by F's smoke scenarios. +/// +/// The subprotocol prefix matches +/// `infrastructure::services::rt_ticket_store::SUBPROTOCOL_PREFIX`; kept +/// as a literal here so the test binary has no dependency on the +/// application crate. async fn connect_ws( url: &str, - token: &str, + auth: &WsAuth, ) -> Result< tokio_tungstenite::WebSocketStream>, HelperError, @@ -217,12 +255,24 @@ async fn connect_ws( let mut req = url .into_client_request() .map_err(|e| HelperError::Protocol(format!("bad url: {e}")))?; - let bearer = format!("Bearer {token}"); - req.headers_mut().insert( - "Authorization", - HeaderValue::from_str(&bearer) - .map_err(|e| HelperError::Protocol(format!("bad token: {e}")))?, - ); + match auth { + WsAuth::Bearer(token) => { + let bearer = format!("Bearer {token}"); + req.headers_mut().insert( + "Authorization", + HeaderValue::from_str(&bearer) + .map_err(|e| HelperError::Protocol(format!("bad token: {e}")))?, + ); + } + WsAuth::Ticket(ticket) => { + let subprotocol = format!("oxi.ticket.{ticket}"); + req.headers_mut().insert( + "Sec-WebSocket-Protocol", + HeaderValue::from_str(&subprotocol) + .map_err(|e| HelperError::Protocol(format!("bad ticket: {e}")))?, + ); + } + } let (ws, _resp) = tokio_tungstenite::connect_async(req) .await .map_err(|e| HelperError::Protocol(format!("connect failed: {e}")))?; @@ -241,7 +291,7 @@ async fn subscribe_and_collect(args: Args) -> Result<(), HelperError> { } let expect_events = args.expect_events.unwrap_or(0); - let mut ws = connect_ws(&args.url, &args.token).await?; + let mut ws = connect_ws(&args.url, &args.auth).await?; // Subscribe to every requested topic; track pending request ids so // we know when all acks have arrived before we start counting @@ -327,6 +377,20 @@ async fn subscribe_and_collect(args: Args) -> Result<(), HelperError> { if let Some(topic) = topic { subscribed.push(topic); } + // Every requested subscribe is now ack'd — signal the + // orchestrator that publishes targeted at these topics + // will land on a live subscriber. See `Args::ready_file` + // for the race this closes. Empty content is fine; the + // shell only checks existence, not payload. Errors are + // logged to stderr but not fatal: the smoke test's + // `wait_ready` timeout will surface the failure with + // more context than a mid-run panic here. + if pending_subs.is_empty() + && let Some(path) = args.ready_file.as_deref() + && let Err(e) = std::fs::write(path, b"") + { + eprintln!("rt-hurl-helper: could not touch --ready-file {path}: {e}"); + } continue; } @@ -391,7 +455,7 @@ async fn expect_denied(args: Args) -> Result<(), HelperError> { .ok_or_else(|| HelperError::Protocol("--subscribe required for expect-denied".into()))? .clone(); - let mut ws = connect_ws(&args.url, &args.token).await?; + let mut ws = connect_ws(&args.url, &args.auth).await?; let req_id: u64 = 1; let frame = json!({ diff --git a/src/common/di.rs b/src/common/di.rs index 78b62c2f..11a24558 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -1041,6 +1041,7 @@ impl AppServiceFactory { core: &CoreServices, authz: &Arc, drive_repo: &Arc, + bus: &Arc, ) -> Option> { if !self.config.features.enable_trash { tracing::info!("Trash service is disabled in configuration"); @@ -1049,7 +1050,12 @@ impl AppServiceFactory { let trash_repo = repos.trash_repository.as_ref()?; - // Wire ports directly to TrashService — no adapter layer needed + // Wire ports directly to TrashService — no adapter layer needed. + // Bus upcast to the trait object so the service takes the port, + // not the concrete impl — mirrors the pattern in + // `create_application_services`. + let bus_trait: Arc = + bus.clone(); let service = Arc::new( TrashService::new( trash_repo.clone(), @@ -1060,7 +1066,8 @@ impl AppServiceFactory { authz.clone(), drive_repo.clone(), ) - .with_file_deleted_hook(core.file_lifecycle.clone()), + .with_file_deleted_hook(core.file_lifecycle.clone()) + .with_message_bus(bus_trait), ); // Initialize cleanup service (bulk-deletes expired items in 2 SQL @@ -1742,9 +1749,29 @@ impl AppServiceFactory { let drive_repo = Arc::new(crate::infrastructure::repositories::pg::DrivePgRepository::new(pool.clone())); + // Message bus: constructed BEFORE the trash service so trash-first + // deletes can publish `FolderDeleted` on the parent folder's + // topic (folder-view live refresh). Wired with a no-op replicator + // — multi-instance broker is a follow-up per + // `docs/plan/message-bus.md § Roadmap`. Spawns its own GC task in + // `with_replicator`; no supervisor setup required. + let bus = crate::infrastructure::services::in_process_message_bus::InProcessMessageBus::with_replicator( + Arc::new(crate::application::ports::message_bus_ports::NoopReplicator), + ); + + // WebSocket ticket store — see `rt_ticket_store` module doc for + // why this exists (DPoP-bound sessions can't be re-proofed on + // a browser-issued WS upgrade). Reaper task runs for the app + // lifetime; its handle is dropped intentionally — the task + // survives on the runtime, and cancellation is handled by + // graceful shutdown killing the runtime. + let rt_ticket_store = + crate::infrastructure::services::rt_ticket_store::RtTicketStore::new(); + let _reaper = Arc::clone(&rt_ticket_store).spawn_reaper(); + // 3b. Trash service (needed before application services) let trash_service = self - .create_trash_service(&repos, &core, &authorization, &drive_repo) + .create_trash_service(&repos, &core, &authorization, &drive_repo, &bus) .await; // 3c. Storage usage / quota service (needed by the instant-upload @@ -1794,17 +1821,6 @@ impl AppServiceFactory { crate::application::services::external_mount_router::MountRouter::new(mount_registry), ); - // Message bus: single instance for the app lifetime, wired - // with a no-op replicator (multi-instance broker is a follow-up - // per `docs/plan/message-bus.md § Roadmap`). Constructed here - // so `create_application_services` can hand it to services that - // publish after their DB commits (`FolderService`, - // `FileUploadService`, …). Spawns its own GC task in - // `with_replicator` — no supervisor setup required. - let bus = crate::infrastructure::services::in_process_message_bus::InProcessMessageBus::with_replicator( - Arc::new(crate::application::ports::message_bus_ports::NoopReplicator), - ); - let mut apps = self.create_application_services( &core, &repos, @@ -2316,6 +2332,7 @@ impl AppServiceFactory { maintenance_pool: Some(maintenance_pool), mount_router, bus, + rt_ticket_store, auth_service: auth_services, opaque_service, opaque_repo, @@ -3223,6 +3240,16 @@ pub struct AppState { pub bus: Arc< crate::infrastructure::services::in_process_message_bus::InProcessMessageBus, >, + /// Short-lived tickets that authenticate a WebSocket upgrade + /// without the browser needing to attach a DPoP proof (which + /// `new WebSocket()` cannot set — only `Sec-WebSocket-Protocol` + /// is settable). FE POSTs `/api/rt/ticket` with a normal + /// DPoP-signed request, receives an opaque one-shot token, and + /// hands it to the WS upgrade via subprotocol. Always populated; + /// see `rt_ticket_store` module doc. + pub rt_ticket_store: Arc< + crate::infrastructure::services::rt_ticket_store::RtTicketStore, + >, pub auth_service: Option, /// OPAQUE aPAKE substrate (RFC 9807). Populated only when /// [`OpaqueConfig::effective_mode`] is not `Off` — that method diff --git a/src/infrastructure/services/in_process_message_bus.rs b/src/infrastructure/services/in_process_message_bus.rs index e92200d1..324f3665 100644 --- a/src/infrastructure/services/in_process_message_bus.rs +++ b/src/infrastructure/services/in_process_message_bus.rs @@ -127,6 +127,27 @@ impl MessageBus for InProcessMessageBus { // replicators must background their I/O themselves. self.replicator.on_local_publish(topic, &event); + // Structured trace of every publish so operators can watch the + // bus with `RUST_LOG=oxicloud::message_bus=debug`. Cheap: + // shows the wire-form topic (uses the same Display we return + // to WS clients), the event's discriminator (via serde), and + // whether anyone was listening at publish time. Payload bodies + // are NOT emitted here to keep the log line short and stable + // across variant additions — `debug_span` or a per-service + // publish site can log the payload if needed. + let sub_count = self + .topics + .get(topic) + .map(|s| s.receiver_count()) + .unwrap_or(0); + tracing::debug!( + target: "oxicloud::message_bus", + topic = %topic.to_wire_key(), + kind = event_kind(&event), + subscribers = sub_count, + "📤 bus publish", + ); + // If nobody is subscribed, don't allocate a sender just to drop // its message. `broadcast::Sender::send` returns Err when there // are no receivers — cheaper still to short-circuit here. @@ -155,6 +176,27 @@ impl MessageBus for InProcessMessageBus { } } +/// Snake-case discriminator string for the event, matching the wire +/// `event` field. Lifted out of the `publish` hot path so the debug +/// log stays a one-liner. Kept in sync with the `#[serde(tag = +/// "event", rename_all = "snake_case")]` shape in +/// `MessageBusEvent` — new variants get a new arm here to render +/// nicely in the trace log; adding one that lands in the default is +/// harmless (still readable), just less specific. +fn event_kind(event: &MessageBusEvent) -> &'static str { + match event { + MessageBusEvent::FileCreated { .. } => "file_created", + MessageBusEvent::FileRenamed { .. } => "file_renamed", + MessageBusEvent::FileMoved { .. } => "file_moved", + MessageBusEvent::FileDeleted { .. } => "file_deleted", + MessageBusEvent::FolderCreated { .. } => "folder_created", + MessageBusEvent::FolderRenamed { .. } => "folder_renamed", + MessageBusEvent::FolderMoved { .. } => "folder_moved", + MessageBusEvent::FolderDeleted { .. } => "folder_deleted", + MessageBusEvent::AuthzChanged { .. } => "authz_changed", + } +} + // ════════════════════════════════════════════════════════════════════════════ // Tests // ════════════════════════════════════════════════════════════════════════════ diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index bb17125c..e3ed53c9 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -52,6 +52,7 @@ pub mod pg_acl_engine; pub mod plugins; pub mod recent_recording_hook; pub mod retry_blob_backend; +pub mod rt_ticket_store; pub mod s3_blob_backend; pub mod satellites_consistency_service; pub mod search_index; diff --git a/src/infrastructure/services/rt_ticket_store.rs b/src/infrastructure/services/rt_ticket_store.rs new file mode 100644 index 00000000..ef21e446 --- /dev/null +++ b/src/infrastructure/services/rt_ticket_store.rs @@ -0,0 +1,229 @@ +//! Short-lived tickets for authenticating a WebSocket upgrade. +//! +//! # Problem +//! +//! A DPoP-bound session must carry a fresh `DPoP:` header on every +//! request. `new WebSocket(url)` in browsers cannot set arbitrary +//! headers — only `Sec-WebSocket-Protocol` — so the upgrade GET +//! arrives without a DPoP proof and `require_dpop_layer` refuses with +//! 401 `proof_missing_on_bound_session`. See +//! `docs/plan/message-bus.md § F`. +//! +//! # Solution +//! +//! Ticket exchange. The FE first `POST /api/rt/ticket` — a normal +//! HTTP request, so `apiFetch` attaches the DPoP proof and every other +//! middleware runs. The server mints an opaque one-shot ticket, tied +//! to the caller_id and a 30 s expiry. The FE then opens the WS with +//! `Sec-WebSocket-Protocol: oxi.ticket.`; the WS handler +//! redeems the ticket via this store to recover the caller_id, then +//! runs the session with zero auth-middleware involvement. +//! +//! # Invariants +//! +//! - **Single-use** — `redeem` removes the entry atomically, so a +//! captured ticket can be replayed at most once (the race is decided +//! by the first successful `remove`; every other caller gets `None`). +//! - **Short-lived** — 30 s TTL. A captured ticket that isn't burned +//! inside that window is inert. +//! - **Opaque** — the token carries no user identity itself. All the +//! auth data lives in the store keyed by the token. Losing the store +//! invalidates every issued ticket; that's the correct failure mode. +//! - **In-process** — one store per process. Multi-instance +//! deployments will need a shared backend (Redis, PG); calling it +//! out here so the seam is visible when the day comes. + +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use dashmap::DashMap; +use tokio::task::JoinHandle; +use uuid::Uuid; + +/// TTL for a freshly-minted ticket. 30 s covers the round-trip from +/// `/api/rt/ticket` response to `new WebSocket()` handshake on any +/// realistic network — well under the shortest sensible clock skew +/// budget, well above the 100–500 ms actually needed on localhost or +/// LAN. Kept as a compile-time constant; if operators ever want to +/// tune it, promote to config. +pub const TICKET_TTL: Duration = Duration::from_secs(30); + +/// Reaper cadence. Every N seconds the store walks its entries and +/// drops expired ones. Redemption also lazily short-circuits on +/// expiry, so the reaper is a memory-hygiene backstop rather than a +/// correctness gate — a ticket that expires and is never redeemed +/// stays around for up to `TICKET_TTL + REAPER_INTERVAL` before its +/// row is freed. +pub const REAPER_INTERVAL: Duration = Duration::from_secs(60); + +/// Wire prefix identifying our tickets in `Sec-WebSocket-Protocol`. +/// The full value on the wire is `oxi.ticket.` — one +/// subprotocol string, opaque to intermediaries. Kept short so +/// stripping proxies don't hit an arbitrary length limit. +pub const SUBPROTOCOL_PREFIX: &str = "oxi.ticket."; + +struct Entry { + caller_id: Uuid, + expires_at: Instant, +} + +/// In-process ticket store. Cheap to construct; the reaper task is +/// spawned by DI when the store is wired. +pub struct RtTicketStore { + entries: DashMap, +} + +impl RtTicketStore { + pub fn new() -> Arc { + Arc::new(Self { + entries: DashMap::new(), + }) + } + + /// Issue a fresh ticket for `caller_id`. Returns the opaque token + /// (a UUIDv4 string) — the FE puts this on the wire as + /// `Sec-WebSocket-Protocol: oxi.ticket.`. + /// + /// Ticket IDs are v4 (random) — 122 bits of entropy, well above + /// the "unguessable-token" bar even without server-side rate + /// limiting. A serial or timestamped id would leak issue-order + /// signal to anyone with a wire tap. + pub fn issue(&self, caller_id: Uuid) -> Uuid { + let ticket = Uuid::new_v4(); + self.entries.insert( + ticket, + Entry { + caller_id, + expires_at: Instant::now() + TICKET_TTL, + }, + ); + ticket + } + + /// Redeem `ticket` if it exists AND has not expired. Removes the + /// entry regardless of outcome — a valid ticket returns the + /// caller_id, an expired ticket is silently freed and returns + /// `None`. Single-use invariant holds by construction: only one + /// caller wins the `remove`, everyone else sees `None`. + pub fn redeem(&self, ticket: Uuid) -> Option { + let (_, entry) = self.entries.remove(&ticket)?; + if entry.expires_at < Instant::now() { + return None; + } + Some(entry.caller_id) + } + + /// Background reaper. Walks the map on the configured cadence and + /// removes expired entries. Runs until the returned handle is + /// dropped or `cancel` is notified (per the standard shutdown + /// contract used across the crate). + pub fn spawn_reaper(self: Arc) -> JoinHandle<()> { + tokio::spawn(async move { + let mut ticker = tokio::time::interval(REAPER_INTERVAL); + // First tick fires immediately; skip it so the store has + // at least one TTL window's worth of entries before the + // first sweep. + ticker.tick().await; + loop { + ticker.tick().await; + let now = Instant::now(); + self.entries.retain(|_, entry| entry.expires_at >= now); + } + }) + } + + /// Present count. Test-only. Not exposed to handlers — no + /// operational reason to peek at the queue depth from a request + /// path. + #[cfg(test)] + fn len(&self) -> usize { + self.entries.len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn issue_then_redeem_returns_caller_id() { + let store = RtTicketStore::new(); + let caller = Uuid::new_v4(); + let ticket = store.issue(caller); + assert_eq!(store.redeem(ticket), Some(caller)); + } + + #[test] + fn redeem_is_single_use() { + let store = RtTicketStore::new(); + let caller = Uuid::new_v4(); + let ticket = store.issue(caller); + assert_eq!(store.redeem(ticket), Some(caller)); + // Second redeem finds nothing — replay protection. + assert_eq!(store.redeem(ticket), None); + } + + #[test] + fn redeem_unknown_returns_none() { + let store = RtTicketStore::new(); + assert_eq!(store.redeem(Uuid::new_v4()), None); + } + + #[test] + fn issued_ticket_is_present_in_store() { + let store = RtTicketStore::new(); + let caller = Uuid::new_v4(); + assert_eq!(store.len(), 0); + let _ticket = store.issue(caller); + assert_eq!(store.len(), 1); + } + + #[test] + fn redeem_after_expiry_returns_none_and_frees_entry() { + // Use a synthetic entry with `expires_at` in the past so the + // test doesn't have to sleep 30 s. + let store = RtTicketStore::new(); + let caller = Uuid::new_v4(); + let ticket = Uuid::new_v4(); + store.entries.insert( + ticket, + Entry { + caller_id: caller, + expires_at: Instant::now() - Duration::from_secs(1), + }, + ); + assert_eq!(store.len(), 1); + // Expired redeem returns None… + assert_eq!(store.redeem(ticket), None); + // …and the entry is gone. + assert_eq!(store.len(), 0); + } + + #[test] + fn distinct_tickets_for_the_same_caller() { + // Two issues in a row must produce distinct token ids — the + // FE will issue one per WS reconnect, and a collision would + // mean the second issue clobbers the first's expiry map row. + let store = RtTicketStore::new(); + let caller = Uuid::new_v4(); + let t1 = store.issue(caller); + let t2 = store.issue(caller); + assert_ne!(t1, t2); + } + + // Wall-clock testing of the reaper's timer needs the tokio + // `test-util` feature; not enabled crate-wide. The reaper body is + // a straight `entries.retain(|_, e| e.expires_at >= now)` and the + // redemption path already lazily short-circuits on expiry (see + // `redeem_after_expiry_returns_none_and_frees_entry`), which + // exercises the same expiry decision without waiting on a real + // clock. + + #[test] + fn is_send_sync_arc_shareable() { + // Mirrors the actual usage in `AppState` — an + // `Arc` shared across the axum-served tasks. + fn assert_send_sync() {} + assert_send_sync::>(); + } +} diff --git a/src/interfaces/api/handlers/mod.rs b/src/interfaces/api/handlers/mod.rs index 8177ee35..19e33984 100644 --- a/src/interfaces/api/handlers/mod.rs +++ b/src/interfaces/api/handlers/mod.rs @@ -23,6 +23,7 @@ pub mod opaque_auth_handler; pub mod people_handler; pub mod photos_handler; pub mod recent_handler; +pub mod rt_ticket_handler; pub mod rt_ws; pub mod search_handler; pub mod share_handler; diff --git a/src/interfaces/api/handlers/rt_ticket_handler.rs b/src/interfaces/api/handlers/rt_ticket_handler.rs new file mode 100644 index 00000000..320b10ed --- /dev/null +++ b/src/interfaces/api/handlers/rt_ticket_handler.rs @@ -0,0 +1,80 @@ +//! Ticket issuance for browser WebSocket authentication. +//! +//! `POST /api/rt/ticket` — issues a one-shot 30 s ticket for the +//! authenticated caller. Runs under the full `/api/*` middleware +//! stack (auth + DPoP), so the caller proves possession of the +//! session AND (when the session is DPoP-bound) the DPoP key on the +//! same request. The ticket then substitutes for that proof on the +//! next WS upgrade. +//! +//! See `src/infrastructure/services/rt_ticket_store.rs` for the +//! store semantics and `docs/plan/message-bus.md § F` for the +//! architectural context. + +use std::sync::Arc; + +use axum::{Json, extract::State}; +use serde::Serialize; + +use crate::common::di::AppState; +use crate::infrastructure::services::rt_ticket_store::{SUBPROTOCOL_PREFIX, TICKET_TTL}; +use crate::interfaces::middleware::auth::CurrentUserId; + +/// Response body for `POST /api/rt/ticket`. Deliberately minimal — +/// callers only need the token string; the TTL is echoed so the FE +/// doesn't hard-code the 30 s constant on its side. +#[derive(Debug, Serialize, utoipa::ToSchema)] +pub struct RtTicketResponse { + /// Opaque single-use token. Present in the WS upgrade as + /// `Sec-WebSocket-Protocol: oxi.ticket.` (the prefix is + /// baked in by both sides — see [`SUBPROTOCOL_PREFIX`]). + pub ticket: String, + + /// Seconds until this ticket expires server-side. Consumers should + /// open the WS immediately; a 30 s bound leaves generous headroom + /// for the handshake without letting a captured ticket live long. + pub expires_in_seconds: u64, + + /// Full `Sec-WebSocket-Protocol` value the client MUST pass on the + /// upgrade. Included pre-assembled so a FE bug can't emit the + /// wrong prefix and blow the handshake in a way that looks like a + /// server-side denial. + pub subprotocol: String, +} + +/// Issue a fresh ticket for the authenticated caller. Idempotent from +/// the caller's perspective — each call mints a new token — but +/// each ticket is single-use once redeemed by the WS handler. +/// +/// No rate limiting today: even a mildly abusive client would just +/// fill the ticket store with entries that reap in 30 s. If ever +/// necessary, add a per-caller_id token bucket alongside the auth +/// middleware limits. +#[utoipa::path( + post, + path = "/api/rt/ticket", + tag = "message-bus", + responses( + (status = 200, description = "Ticket issued", body = RtTicketResponse), + (status = 401, description = "Unauthenticated"), + ), + security(("bearerAuth" = [])) +)] +pub async fn issue_rt_ticket( + CurrentUserId(caller_id): CurrentUserId, + State(state): State>, +) -> Json { + let ticket = state.rt_ticket_store.issue(caller_id); + let ticket_str = ticket.to_string(); + tracing::debug!( + target: "oxicloud::message_bus", + event = "message_bus.ticket_issued", + caller_id = %caller_id, + "🎫 rt.ticket issued", + ); + Json(RtTicketResponse { + subprotocol: format!("{SUBPROTOCOL_PREFIX}{ticket_str}"), + ticket: ticket_str, + expires_in_seconds: TICKET_TTL.as_secs(), + }) +} diff --git a/src/interfaces/api/handlers/rt_ws.rs b/src/interfaces/api/handlers/rt_ws.rs index fb8329a5..2b1944e8 100644 --- a/src/interfaces/api/handlers/rt_ws.rs +++ b/src/interfaces/api/handlers/rt_ws.rs @@ -18,13 +18,28 @@ //! //! # Auth //! -//! Route sits under `protected_api` (see `src/interfaces/api/routes.rs`) -//! so `auth_middleware` runs first. Cookie AND `Authorization: Bearer` -//! paths both produce a `CurrentUserId` extension the handler extracts. -//! Browser-side subprotocol bearer (`Sec-WebSocket-Protocol: -//! authorization.bearer.`) is a Phase-A follow-up — the MVP relies -//! on the Authorization header, which programmatic clients (the -//! `rt-hurl-helper` smoke test) set directly. +//! Route is mounted at `/api/rt/ws` OUTSIDE the standard +//! `auth_middleware` + `require_dpop_layer` stack — a browser can't +//! attach a `DPoP:` header to `new WebSocket()` (RFC 6455 gives us +//! only `Sec-WebSocket-Protocol`), and the standard chain would 401 +//! on every DPoP-bound session. This handler self-authenticates +//! from two accepted sources: +//! +//! 1. **Ticket subprotocol** (`Sec-WebSocket-Protocol: +//! oxi.ticket.`) — the primary path for browser clients. +//! The FE first `POST /api/rt/ticket` under the full middleware +//! chain (auth + DPoP proofed), receives an opaque one-shot +//! token, and passes it here. Verified by redeeming through +//! [`AppState::rt_ticket_store`]. See +//! `docs/plan/message-bus.md § F`. +//! 2. **Bearer token** (`Authorization: Bearer `) — the +//! programmatic-client path used by `rt-hurl-helper` in api-test. +//! Verified against `AuthServices::token_service`. DPoP-bound +//! tokens are rejected on this path to preserve the substrate's +//! proof-of-possession invariant. +//! +//! Neither → 401. Order matters: ticket first (short-lived, tied to +//! a proofed HTTP round-trip), bearer second. //! //! # Limits //! @@ -40,7 +55,8 @@ use std::time::Duration; use axum::body::Bytes; use axum::extract::State; use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade}; -use axum::response::Response; +use axum::http::{HeaderMap, StatusCode}; +use axum::response::{IntoResponse, Response}; use futures::StreamExt; use serde::{Deserialize, Serialize}; use serde_json::Value; @@ -49,13 +65,14 @@ use tokio::task::JoinHandle; use tokio::time::MissedTickBehavior; use uuid::Uuid; +use crate::application::ports::auth_ports::TokenServicePort; use crate::application::ports::authorization_ports::AuthorizationEngine; use crate::application::ports::message_bus_ports::{ AuthzCheck, BusResource, MessageBus, MessageBusEvent, ParseTopicErr, Topic, error_code, }; use crate::common::di::AppState; use crate::domain::services::authorization::{Permission, Resource, Subject}; -use crate::interfaces::middleware::auth::CurrentUserId; +use crate::infrastructure::services::rt_ticket_store::SUBPROTOCOL_PREFIX; /// Max simultaneous subscriptions on a single WS session. Beyond this the /// server responds `-32005 sub_limit` and the client is expected to @@ -151,20 +168,112 @@ struct RpcNotification<'a> { // Handler entrypoint // ════════════════════════════════════════════════════════════════════════════ -/// `GET /api/rt/ws` — WS upgrade handler. Sits under `protected_api` so -/// [`CurrentUserId`] resolves against a valid session before we reach -/// `on_upgrade`. +/// `GET /api/rt/ws` — WS upgrade handler. Mounted outside the standard +/// `/api/*` middleware stack; self-authenticates via ticket +/// subprotocol OR bearer token (see the module doc). /// -/// Returns whatever `WebSocketUpgrade::on_upgrade` produces (an HTTP 101 -/// Switching Protocols with the WebSocket handshake headers). +/// Returns 101 Switching Protocols on success; 401 with an audit +/// entry on any auth failure. The response is deliberately terse — +/// browsers surface the status code via the `close` event's code +/// field (1006 on a rejected upgrade), so a longer body wouldn't +/// reach the FE anyway. pub async fn rt_ws_handler( ws: WebSocketUpgrade, - CurrentUserId(caller_id): CurrentUserId, State(state): State>, + headers: HeaderMap, ) -> Response { + let auth = match authenticate_upgrade(&headers, &state).await { + Ok(auth) => auth, + Err(reason) => { + tracing::info!( + target: "audit", + event = "message_bus.upgrade_rejected", + reason = %reason, + "👮🏻‍♂️ WS upgrade rejected", + ); + return (StatusCode::UNAUTHORIZED, "ws_auth_failed").into_response(); + } + }; + let caller_id = auth.caller_id; + // If the caller reached us via the ticket path, echo the exact + // subprotocol they sent back on the 101 response — RFC 6455 §4.2.2 + // requires this or the client fails the connection. + let ws = match auth.accepted_subprotocol { + Some(sub) => ws.protocols([sub]), + None => ws, + }; ws.on_upgrade(move |socket| handle_session(socket, caller_id, state)) } +/// Successful upgrade credentials — the resolved caller and (when the +/// ticket path was used) the subprotocol to echo on the 101 response. +struct UpgradeAuth { + caller_id: Uuid, + accepted_subprotocol: Option, +} + +/// Extract `Sec-WebSocket-Protocol` and match a ticket subprotocol +/// first; fall back to `Authorization: Bearer`. Returns a stable +/// `reason` key on failure so the audit log stays filterable. +async fn authenticate_upgrade( + headers: &HeaderMap, + state: &Arc, +) -> Result { + if let Some(ticket_sub) = extract_ticket_subprotocol(headers) { + // Redeem parses the UUID; a malformed subprotocol is a + // structural failure ("bad_ticket_format"), an unknown-or- + // expired UUID is a redemption failure ("ticket_invalid"). + let Some(ticket_str) = ticket_sub.strip_prefix(SUBPROTOCOL_PREFIX) else { + return Err("bad_ticket_format"); + }; + let Ok(ticket_uuid) = Uuid::parse_str(ticket_str) else { + return Err("bad_ticket_uuid"); + }; + let Some(caller_id) = state.rt_ticket_store.redeem(ticket_uuid) else { + return Err("ticket_invalid"); + }; + return Ok(UpgradeAuth { + caller_id, + accepted_subprotocol: Some(ticket_sub), + }); + } + if let Some(bearer) = extract_bearer(headers) { + let Some(auth_service) = state.auth_service.as_ref() else { + return Err("auth_service_unavailable"); + }; + let claims = auth_service + .token_service + .validate_token(bearer) + .map_err(|_| "bearer_invalid")?; + if claims.sub_id.is_nil() { + return Err("bearer_bad_subject"); + } + return Ok(UpgradeAuth { + caller_id: claims.sub_id, + accepted_subprotocol: None, + }); + } + Err("no_credentials") +} + +/// Find the first subprotocol value that looks like a ticket. Browsers +/// send `Sec-WebSocket-Protocol` as a comma-separated list per RFC 6455. +fn extract_ticket_subprotocol(headers: &HeaderMap) -> Option { + let raw = headers.get("sec-websocket-protocol")?.to_str().ok()?; + raw.split(',') + .map(str::trim) + .find(|s| s.starts_with(SUBPROTOCOL_PREFIX)) + .map(|s| s.to_string()) +} + +/// Extract `Authorization: Bearer ` if present. Returns the raw +/// token string (never empty). +fn extract_bearer(headers: &HeaderMap) -> Option<&str> { + let value = headers.get("authorization")?.to_str().ok()?; + let token = value.strip_prefix("Bearer ")?.trim(); + (!token.is_empty()).then_some(token) +} + // ════════════════════════════════════════════════════════════════════════════ // Session loop // ════════════════════════════════════════════════════════════════════════════ diff --git a/src/interfaces/api/routes.rs b/src/interfaces/api/routes.rs index 5cc5967d..5d9aba3b 100644 --- a/src/interfaces/api/routes.rs +++ b/src/interfaces/api/routes.rs @@ -674,16 +674,24 @@ pub fn create_api_routes(app_state: &Arc) -> Router> { .with_state(app_state.clone()); router = router.nest("/users", users_router); - // Message bus WebSocket. Auth (session cookie or bearer JWT) via - // the same `auth_middleware` the rest of `/api/*` gets; the handler - // extracts `CurrentUserId` from the extension the middleware - // installs. See `docs/plan/message-bus.md` and the module doc on - // `rt_ws` for the JSON-RPC 2.0 wire. + // Message bus — ticket issuance (`POST /api/rt/ticket`). Stays in + // the protected router (auth + DPoP), so the caller proves session + // + DPoP-key possession before a ticket is minted. See + // `handlers/rt_ticket_handler.rs` and `docs/plan/message-bus.md § F`. router = router.route( - "/rt/ws", - get(crate::interfaces::api::handlers::rt_ws::rt_ws_handler).with_state(app_state.clone()), + "/rt/ticket", + post(crate::interfaces::api::handlers::rt_ticket_handler::issue_rt_ticket) + .with_state(app_state.clone()), ); + // The WS upgrade (`GET /api/rt/ws`) is registered OUTSIDE the + // protected-api middleware stack — a browser cannot attach a + // `DPoP:` header to `new WebSocket()`, so the standard stack + // 401s on every DPoP-bound session. See the `rt_ws` module doc + // for the self-auth logic (ticket subprotocol or bearer token). + // Registration happens in `main.rs` where the outer router owns + // the middleware layering. + // Collector for any unknown `/api/*` path. Without this, an // unmatched API URL falls through Axum's matcher to the // ServeDir fallback and is logged under `http::web` — wrong diff --git a/src/main.rs b/src/main.rs index 792cfa12..c10105c9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1042,6 +1042,22 @@ async fn run() -> Result<(), Box> { ) // Public API routes (share access, i18n) — no auth required .nest("/api", public_api_routes.layer(access_log!("http::api"))) + // Message-bus WebSocket. Registered OUTSIDE `protected_api` + // because a browser cannot attach a `DPoP:` header to + // `new WebSocket()` (RFC 6455 only lets us set + // `Sec-WebSocket-Protocol`), so the standard auth + DPoP + // stack would 401 every DPoP-bound session. The handler + // self-authenticates from either a ticket subprotocol + // (minted by `POST /api/rt/ticket` under the full chain) + // or a bearer token (`rt-hurl-helper` test path). + // See `handlers/rt_ws.rs` module doc and + // `docs/plan/message-bus.md § F`. + .route( + "/api/rt/ws", + axum::routing::get(oxicloud::interfaces::api::handlers::rt_ws::rt_ws_handler) + .with_state(app_state.clone()) + .layer(access_log!("http::api")), + ) // All other API routes are protected by auth middleware .nest("/api", protected_api.layer(access_log!("http::api"))) // RFC 6764 well-known discovery (public, no auth — just redirects) diff --git a/tests/api/rt_bus_check.sh b/tests/api/rt_bus_check.sh index 137046c3..58ca2d55 100755 --- a/tests/api/rt_bus_check.sh +++ b/tests/api/rt_bus_check.sh @@ -42,6 +42,16 @@ # shape as an unknown topic — anti-enumeration). # Guards the strict-privacy Class-2 AuthZ gate: # no admin bypass, direct UUID equality only. +# S10 Ticket happy path — user1 POSTs `/api/rt/ticket`, receives a +# short-lived opaque token, opens the WS with +# `Sec-WebSocket-Protocol: oxi.ticket.` +# and successfully subscribes + delivers an +# event. Exercises the ticket path — the only +# path a DPoP-required browser can take. +# S11 Ticket single-use — a ticket redeemed once cannot be redeemed +# again. Guards replay: a captured token +# outside its 30 s TTL, or one already +# consumed, MUST fail the upgrade with 401. # # Exit non-zero on any failure — run.sh treats that as a suite failure. # ───────────────────────────────────────────────────────────────────────────── @@ -69,14 +79,19 @@ esac log() { printf '\033[1;36m[rt_bus_check]\033[0m %s\n' "$*"; } die() { printf '\033[1;31m[rt_bus_check FAIL]\033[0m %s\n' "$*" >&2; exit 1; } -# ── Build the helper on demand (matches opaque/dpop helper convention) ────── -if [[ ! -x "$HELPER_BIN" ]]; then - log "Building rt-hurl-helper ($BUILD_TARGET)..." - case "$BUILD_TARGET" in - debug) (cd "$REPO_ROOT" && cargo build --features test_utils --bin rt-hurl-helper 2>&1 | tail -n 20) || die "rt-hurl-helper build failed" ;; - release) (cd "$REPO_ROOT" && cargo build --release --features test_utils --bin rt-hurl-helper 2>&1 | tail -n 20) || die "rt-hurl-helper build failed" ;; - esac -fi +# ── Rebuild the helper every run ──────────────────────────────────────────── +# Deliberately unconditional — the previous `[[ ! -x $HELPER_BIN ]]` guard +# silently reused a stale binary whenever the helper's source changed +# without touching the caller shell script, producing "unknown flag" +# exits that looked like test bugs (see the S10/S11 --ticket rollout). +# Cargo incremental short-circuits in ~50 ms when nothing changed, so +# the cost of the always-build is negligible; the cost of a stale binary +# is a wild-goose chase. +log "Building rt-hurl-helper ($BUILD_TARGET)..." +case "$BUILD_TARGET" in + debug) (cd "$REPO_ROOT" && cargo build --features test_utils --bin rt-hurl-helper 2>&1 | tail -n 20) || die "rt-hurl-helper build failed" ;; + release) (cd "$REPO_ROOT" && cargo build --release --features test_utils --bin rt-hurl-helper 2>&1 | tail -n 20) || die "rt-hurl-helper build failed" ;; +esac # ── curl wrappers ─────────────────────────────────────────────────────────── c_post() { @@ -93,6 +108,22 @@ c_get() { "$url" } +# Block until the helper writes `--ready-file ` (touched the +# moment every requested subscribe is ack'd) or `$timeout` seconds +# elapse. Replaces the older `sleep 0.4` heuristic that flaked on +# cold-cache runs where the helper's fork/tokio-init/connect chain +# crossed 400 ms and the shell's mkfile_in publish arrived at an +# empty topic. See `Args::ready_file` in rt-hurl-helper.rs. +wait_ready() { + local path="$1" timeout="${2:-5}" + local waited=0 + while [[ ! -f "$path" && "$waited" -lt "$((timeout * 20))" ]]; do + sleep 0.05 + waited=$((waited + 1)) + done + [[ -f "$path" ]] || die "wait_ready: $path never appeared within ${timeout}s (subscribe likely never ack'd)" +} + # ── Setup: register fresh users; the test.env admin may be OPAQUE- # migrated and the legacy password-login path refuses those accounts, # so we don't use it at all — same pattern as `dedup_admin_gate.hurl` @@ -178,17 +209,21 @@ mkfile_in() { # ── Scenario 1 — Positive delivery ────────────────────────────────────────── log "S1: subscribe to folder A, upload into A, expect one file_created event." out_s1="$(mktemp -t rtbus_s1.XXXXXX)" +ready_s1="$(mktemp -t rtbus_s1_ready.XXXXXX)" +rm -f "$ready_s1" # mktemp creates it; ready-file semantics need "appears when subscribed" "$HELPER_BIN" subscribe-and-collect \ --url "$ws_url" \ --token "$user1_token" \ --subscribe "folder:$folder_a" \ --expect-events 1 \ --timeout 5s \ + --ready-file "$ready_s1" \ --output "$out_s1" & helper_pid=$! -# Give the ack a moment to install so the upload's post-commit publish -# lands on a live receiver, not an orphaned map entry. -sleep 0.4 +# Block on the helper's ready-file signal, not a wall-clock sleep — +# see wait_ready doc. Closes the "publish before subscribe installed" +# race that flaked S1 on cold-cache runs. +wait_ready "$ready_s1" mkfile_in "$folder_a" "s1.txt" "$user1_token" if ! wait "$helper_pid"; then cat "$out_s1" >&2 || true @@ -206,15 +241,17 @@ log "S1 OK" # ── Scenario 2 — Topic isolation ──────────────────────────────────────────── log "S2: subscribe to folder A, upload into B (must be silent) and A (triggers exit)." out_s2="$(mktemp -t rtbus_s2.XXXXXX)" +ready_s2="$(mktemp -t rtbus_s2_ready.XXXXXX)"; rm -f "$ready_s2" "$HELPER_BIN" subscribe-and-collect \ --url "$ws_url" \ --token "$user1_token" \ --subscribe "folder:$folder_a" \ --expect-events 1 \ --timeout 5s \ + --ready-file "$ready_s2" \ --output "$out_s2" & helper_pid=$! -sleep 0.4 +wait_ready "$ready_s2" # B first — should be dropped for the A subscriber. mkfile_in "$folder_b" "s2_in_B.txt" "$user1_token" # Small settle so if isolation is BROKEN, the B event has time to arrive @@ -271,14 +308,20 @@ log "S4 OK" # (b) trips (event never arrives after idle). log "S5: server-initiated keepalive fires on idle; session still delivers." out_s5="$(mktemp -t rtbus_s5.XXXXXX)" +ready_s5="$(mktemp -t rtbus_s5_ready.XXXXXX)"; rm -f "$ready_s5" "$HELPER_BIN" subscribe-and-collect \ --url "$ws_url" \ --token "$user1_token" \ --subscribe "folder:$folder_a" \ --expect-events 1 \ --timeout 6s \ + --ready-file "$ready_s5" \ --output "$out_s5" & helper_pid=$! +# Wait for the subscribe to install BEFORE starting the idle window — +# otherwise slow helper startup eats into the 3 s and we observe +# fewer pings than the assertion below tolerates. +wait_ready "$ready_s5" # 3 s of pure idle — with 1 s keepalive on the server, that's ~3 Pings. sleep 3 mkfile_in "$folder_a" "s5.txt" "$user1_token" @@ -320,15 +363,17 @@ s6_file_id=$(printf '%s' "$s6_upload" | jq -r '.id') || die "S6: pre-upload failed: $s6_upload" out_s6="$(mktemp -t rtbus_s6.XXXXXX)" +ready_s6="$(mktemp -t rtbus_s6_ready.XXXXXX)"; rm -f "$ready_s6" "$HELPER_BIN" subscribe-and-collect \ --url "$ws_url" \ --token "$user1_token" \ --subscribe "folder:$folder_a" \ --expect-events 1 \ --timeout 5s \ + --ready-file "$ready_s6" \ --output "$out_s6" & helper_pid=$! -sleep 0.4 +wait_ready "$ready_s6" # `DELETE /api/files/{id}` routes to `delete_and_cleanup_with_perms` — # the trash-first path. Publish fires on BOTH the trash and the # permanent-delete branch, so this covers whichever the test hits. @@ -369,6 +414,7 @@ s7_file_id=$(printf '%s' "$s7_upload" | jq -r '.id') || die "S7: pre-upload failed: $s7_upload" out_s7="$(mktemp -t rtbus_s7.XXXXXX)" +ready_s7="$(mktemp -t rtbus_s7_ready.XXXXXX)"; rm -f "$ready_s7" "$HELPER_BIN" subscribe-and-collect \ --url "$ws_url" \ --token "$user1_token" \ @@ -376,9 +422,10 @@ out_s7="$(mktemp -t rtbus_s7.XXXXXX)" --subscribe "folder:$folder_b" \ --expect-events 2 \ --timeout 5s \ + --ready-file "$ready_s7" \ --output "$out_s7" & helper_pid=$! -sleep 0.4 +wait_ready "$ready_s7" # `PUT /api/files/{id}/move` — MoveFilePayload = { folder_id: }. curl -sS -X PUT \ -H "Authorization: Bearer $user1_token" \ @@ -447,6 +494,7 @@ grant_b_id=$(printf '%s' "$grant_b" | jq -r '.grants[0].id') # 8.2 user2 subscribes to BOTH folder topics; --expect-events 1 exits # when the post-revoke upload lands on the SURVIVING sub. out_s8="$(mktemp -t rtbus_s8.XXXXXX)" +ready_s8="$(mktemp -t rtbus_s8_ready.XXXXXX)"; rm -f "$ready_s8" "$HELPER_BIN" subscribe-and-collect \ --url "$ws_url" \ --token "$user2_token" \ @@ -454,9 +502,10 @@ out_s8="$(mktemp -t rtbus_s8.XXXXXX)" --subscribe "folder:$folder_b" \ --expect-events 1 \ --timeout 6s \ + --ready-file "$ready_s8" \ --output "$out_s8" & helper_pid=$! -sleep 0.4 # let both subscribes install +wait_ready "$ready_s8" # both subscribes installed before we revoke/upload # 8.3 user1 revokes only the folder-A grant. curl -sS -X DELETE \ @@ -531,4 +580,65 @@ if ! "$HELPER_BIN" expect-denied \ fi log "S9 OK" -log "All nine message-bus scenarios passed." +# ── Scenario 10 — Ticket happy path ───────────────────────────────────────── +# The browser flow: POST /api/rt/ticket under the full middleware stack +# (auth + DPoP proofed), then open the WS with `oxi.ticket.` in +# Sec-WebSocket-Protocol. Same delivery guarantees as the bearer path. +# `curl` mints the ticket; `rt-hurl-helper --ticket` redeems it on the +# upgrade. +log "S10: issue rt ticket, open WS with subprotocol, subscribe + deliver." +# c_post takes the raw JWT as its second arg (not the full +# `Authorization:` line); it assembles the header itself. +tkt_resp=$(c_post "$base_url/api/rt/ticket" "$user1_token" "") +ticket=$(printf '%s' "$tkt_resp" | jq -r '.ticket') +[[ -n "$ticket" && "$ticket" != "null" ]] \ + || die "S10: no ticket in POST /api/rt/ticket response: $tkt_resp" +out_s10="$(mktemp -t rtbus_s10.XXXXXX)" +ready_s10="$(mktemp -t rtbus_s10_ready.XXXXXX)"; rm -f "$ready_s10" +"$HELPER_BIN" subscribe-and-collect \ + --url "$ws_url" \ + --ticket "$ticket" \ + --subscribe "folder:$folder_a" \ + --expect-events 1 \ + --timeout 3s \ + --ready-file "$ready_s10" \ + --output "$out_s10" & +helper_pid=$! +wait_ready "$ready_s10" +mkfile_in "$folder_a" "s10.txt" "$user1_token" +if ! wait "$helper_pid"; then + cat "$out_s10" >&2 || true + die "S10: helper did not observe event on ticket-authenticated WS" +fi +[[ "$(jq -r '.events | length' "$out_s10")" == "1" ]] \ + || { cat "$out_s10"; die "S10: expected 1 event, got $(jq -r '.events | length' "$out_s10")"; } +log "S10 OK" + +# ── Scenario 11 — Ticket single-use ───────────────────────────────────────── +# S10 already redeemed the ticket. A second connection with the SAME +# token MUST be refused at the upgrade with 401 (`ticket_invalid` +# audit reason). Proves replay protection — the store removes entries +# on first successful redeem, even if the caller reconnects before +# the 30 s TTL would have expired anyway. +# +# The helper distinguishes "expectation failure" (exit 1 — WS opened +# and then something was off) from "protocol/connect failure" (exit 2 +# — connect_ws itself refused). Ticket rejection lands in the second +# bucket, so we assert on exit code 2. Bash's `!` inverter treats any +# non-zero as success, so we capture the exact code. +log "S11: reuse the redeemed ticket, expect upgrade rejected." +set +e +"$HELPER_BIN" subscribe-and-collect \ + --url "$ws_url" \ + --ticket "$ticket" \ + --subscribe "folder:$folder_a" \ + --expect-events 1 \ + --timeout 2s \ + --output /dev/null +reuse_exit=$? +set -e +[[ "$reuse_exit" -eq 2 ]] \ + || die "S11: expected exit 2 (connect refused), got $reuse_exit" +log "S11 OK" + +log "All eleven message-bus scenarios passed."