diff --git a/v2/Cargo.lock b/v2/Cargo.lock index bfb70e58..fc1218c9 100644 --- a/v2/Cargo.lock +++ b/v2/Cargo.lock @@ -7637,6 +7637,7 @@ version = "0.1.0" dependencies = [ "base64 0.21.7", "jsonwebtoken", + "libc", "p256", "rand 0.8.5", "reqwest 0.12.28", diff --git a/v2/crates/ruview-auth/Cargo.toml b/v2/crates/ruview-auth/Cargo.toml index 6aadf138..2b2c5931 100644 --- a/v2/crates/ruview-auth/Cargo.toml +++ b/v2/crates/ruview-auth/Cargo.toml @@ -34,13 +34,15 @@ rand = { version = "0.8", optional = true } sha2 = { workspace = true, optional = true } base64 = { version = "0.21", optional = true } url = { version = "2", optional = true } +# Advisory cross-process file lock around the refresh critical section (Unix). +libc = { version = "0.2", optional = true } [features] default = ["ureq-transport"] ureq-transport = ["dep:ureq"] # Interactive OAuth login: PKCE, loopback callback, OOB paste fallback, # credential storage, single-flight refresh. Opt in from a CLI or desktop app. -login = ["dep:reqwest", "dep:tokio", "dep:rand", "dep:sha2", "dep:base64", "dep:url"] +login = ["dep:reqwest", "dep:tokio", "dep:rand", "dep:sha2", "dep:base64", "dep:url", "dep:libc"] [dev-dependencies] # Test-only: sign real ES256 tokens so the negative matrix exercises the same diff --git a/v2/crates/ruview-auth/src/login/store.rs b/v2/crates/ruview-auth/src/login/store.rs index ac9d2b62..17a058a6 100644 --- a/v2/crates/ruview-auth/src/login/store.rs +++ b/v2/crates/ruview-auth/src/login/store.rs @@ -17,6 +17,21 @@ //! await**, re-checks expiry after acquiring it (the task that waited may find //! the work already done), persists the rotated token **before** returning, and //! never retries. +//! +//! ## The in-process mutex is not enough +//! +//! Every CLI invocation is a NEW process with its own `Session` and its own +//! mutex, all sharing one credential file. Two `wifi-densepose` commands run +//! close together inside the refresh window would each load the same refresh +//! token and each present it — and the second is replay, so the user is logged +//! out for running two commands at once. +//! +//! So the critical section is also guarded by an advisory **file lock**, taken +//! NON-BLOCKING. If another process holds it, that process is already +//! refreshing: we wait briefly and re-read the file rather than queue up to do +//! the same work with a token that is about to be spent. Blocking on the lock +//! would also park the async executor — the same mistake this crate had to fix +//! in `jwks.rs`. use std::path::{Path, PathBuf}; use std::sync::Arc; @@ -26,6 +41,13 @@ use tokio::sync::Mutex; use super::client::{self, OAuthError, TokenResponse}; +/// How many times to re-read the credential file while another process holds +/// the refresh lock, before giving up on it and refreshing ourselves. +const RELOAD_ATTEMPTS: usize = 20; +/// Gap between those re-reads. 20 x 150ms = 3s, comfortably longer than a +/// healthy token exchange and shorter than a user notices. +const RELOAD_INTERVAL: std::time::Duration = std::time::Duration::from_millis(150); + /// Refresh this many seconds before `exp`. Matches the figure meta-proxy and /// musica independently arrived at against the same 15-minute token. const REFRESH_SKEW_SECS: i64 = 60; @@ -55,7 +77,10 @@ pub enum StoreError { } /// The persisted session. Deliberately small: this file holds live credentials. -#[derive(Debug, Clone, Serialize, Deserialize)] +/// +/// `Debug` is hand-written and REDACTING — a derived impl prints both tokens in +/// full, and this type is the obvious thing to log when a session misbehaves. +#[derive(Clone, Serialize, Deserialize)] pub struct StoredCredentials { pub schema_version: u8, pub access_token: String, @@ -67,6 +92,20 @@ pub struct StoredCredentials { pub issuer: String, } +impl std::fmt::Debug for StoredCredentials { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("StoredCredentials") + .field("schema_version", &self.schema_version) + .field("access_token", &"") + .field("refresh_token", &self.refresh_token.as_ref().map(|_| "")) + .field("expires_at", &self.expires_at) + .field("scope", &self.scope) + .field("account_email", &self.account_email) + .field("issuer", &self.issuer) + .finish() + } +} + impl StoredCredentials { pub const SCHEMA_VERSION: u8 = 1; @@ -253,6 +292,55 @@ pub fn clear(path: &Path) -> Result { } } +/// An advisory, cross-process exclusive lock on the credential file. +/// +/// Unix only. On other platforms this is a no-op and the cross-process race +/// remains — stated rather than silently pretended, since a lock that does +/// nothing while claiming to protect is worse than none. +struct FileLock { + #[cfg(unix)] + file: std::fs::File, +} + +impl FileLock { + /// `None` if another process holds it. Never blocks. + #[cfg(unix)] + fn try_acquire(credentials_path: &Path) -> Option { + use std::os::unix::io::AsRawFd; + let path = credentials_path.with_extension("lock"); + if let Some(dir) = path.parent() { + let _ = std::fs::create_dir_all(dir); + } + let file = std::fs::OpenOptions::new() + .create(true) + .truncate(false) + .write(true) + .open(&path) + .ok()?; + // LOCK_EX | LOCK_NB + let rc = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) }; + if rc == 0 { + Some(Self { file }) + } else { + None + } + } + + #[cfg(not(unix))] + fn try_acquire(_credentials_path: &Path) -> Option { + Some(Self {}) + } +} + +#[cfg(unix)] +impl Drop for FileLock { + fn drop(&mut self) { + use std::os::unix::io::AsRawFd; + // Released on close anyway; explicit so the intent is legible. + unsafe { libc::flock(self.file.as_raw_fd(), libc::LOCK_UN) }; + } +} + /// A live session that refreshes itself, safely, at most once at a time. #[derive(Clone)] pub struct Session { @@ -304,6 +392,32 @@ impl Session { return Ok(guard.access_token.clone()); } + // Cross-process guard. Non-blocking on purpose: a busy lock means + // another process is mid-refresh, so the useful move is to wait for its + // result rather than race it with a token it is about to spend. + let _file_lock: Option = match FileLock::try_acquire(&self.path) { + Some(lock) => Some(lock), + None => { + for _ in 0..RELOAD_ATTEMPTS { + tokio::time::sleep(RELOAD_INTERVAL).await; + if let Ok(fresh) = load(&self.path) { + if !fresh.needs_refresh() { + let token = fresh.access_token.clone(); + *guard = fresh; + return Ok(token); + } + } + } + // The other process died or is wedged. Fall through and refresh + // ourselves — the lock is advisory, not a correctness barrier. + tracing::warn!( + "another process held the credential lock without completing a refresh; \ + proceeding" + ); + None + } + }; + let Some(refresh_token) = guard.refresh_token.clone() else { return Err(StoreError::NoRefreshToken); }; @@ -370,6 +484,52 @@ mod tests { assert!(creds(Some(now_unix() - 1)).needs_refresh()); } + #[cfg(unix)] + #[test] + fn the_credential_lock_is_exclusive_and_non_blocking() { + // Guards the cross-process race: two CLI invocations inside the refresh + // window used to be able to present the same rotating refresh token, + // which identity treats as replay and answers by revoking the session. + let dir = std::env::temp_dir().join(format!("ruview-auth-lock-{}", std::process::id())); + let path = dir.join("credentials.json"); + std::fs::create_dir_all(&dir).unwrap(); + + let first = FileLock::try_acquire(&path).expect("first acquire succeeds"); + assert!( + FileLock::try_acquire(&path).is_none(), + "a second holder must be refused, and refused WITHOUT blocking" + ); + drop(first); + assert!( + FileLock::try_acquire(&path).is_some(), + "the lock must be released on drop" + ); + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn redacted_debug_never_prints_token_material() { + // A derived Debug prints both tokens in full, and this is the obvious + // type to log when a session misbehaves. + // Distinctive values — an earlier version of this test used "at"/"rt", + // which collide with `expires_at` and produce a false failure. + let mut c = creds(Some(1)); + c.access_token = "SECRET-ACCESS-VALUE".into(); + c.refresh_token = Some("SECRET-REFRESH-VALUE".into()); + let rendered = format!("{c:?}"); + assert!( + !rendered.contains("SECRET-ACCESS-VALUE"), + "access token leaked: {rendered}" + ); + assert!( + !rendered.contains("SECRET-REFRESH-VALUE"), + "refresh token leaked: {rendered}" + ); + assert!(rendered.contains("")); + // Non-secret fields stay visible or the type is useless for debugging. + assert!(rendered.contains("https://auth.test")); + } + #[test] fn save_then_load_round_trips() { let dir = std::env::temp_dir().join(format!("ruview-auth-test-{}", std::process::id())); diff --git a/v2/crates/wifi-densepose-sensing-server/src/bearer_auth.rs b/v2/crates/wifi-densepose-sensing-server/src/bearer_auth.rs index 0cea8bad..b86b7f39 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/bearer_auth.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/bearer_auth.rs @@ -190,7 +190,13 @@ pub enum OAuthConfigError { } /// Cheap, cloneable handle to the configured credentials. -#[derive(Debug, Clone, Default)] +/// +/// `Debug` is hand-written and REDACTING: this holds the raw +/// `RUVIEW_API_TOKEN`. A derived impl would print it in full the first time +/// anyone writes `tracing::debug!(?auth, ...)`. `OAuthState` and `TicketStore` +/// already redact for the same reason; the type actually holding the secret +/// should not be the one that does not. +#[derive(Clone, Default)] pub struct AuthState { /// The expected static bearer token, if any. token: Option>, @@ -203,6 +209,17 @@ pub struct AuthState { legacy_ws: bool, } +impl std::fmt::Debug for AuthState { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("AuthState") + .field("static_token", &self.token.as_ref().map(|_| "")) + .field("oauth", &self.oauth) + .field("tickets", &self.tickets) + .field("legacy_ws", &self.legacy_ws) + .finish() + } +} + impl AuthState { /// Build an [`AuthState`] from an explicit string. Empty ⇒ disabled. pub fn from_token(t: impl Into) -> Self { @@ -909,7 +926,11 @@ mod oauth_tests { "exp": now + 900, "setup": false, "workload": false, - "iss": ISSUER, + // NO `iss`. Real Cognitum tokens carry none — mirroring production + // here matters even though the verifier ignores the claim either + // way: a fixture that invents a claim reality lacks is exactly what + // hid the original `iss` bug for a day. See ruview-auth's + // `a_token_with_no_iss_claim_is_accepted_because_cognitum_issues_none`. }); let mut header = Header::new(jsonwebtoken::Algorithm::ES256); header.kid = Some(KID.to_string()); diff --git a/v2/crates/wifi-densepose-sensing-server/src/ws_ticket.rs b/v2/crates/wifi-densepose-sensing-server/src/ws_ticket.rs index d8584963..8dc00c7a 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/ws_ticket.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/ws_ticket.rs @@ -45,11 +45,21 @@ use rand::RngCore; /// someone who found the URL later. pub const TICKET_TTL: Duration = Duration::from_secs(30); -/// Cap on outstanding tickets, so a caller with a valid credential cannot grow -/// the map without bound by requesting tickets in a loop. Well above any real -/// page's needs. +/// Global cap on outstanding tickets. const MAX_OUTSTANDING: usize = 512; +/// Per-principal cap. +/// +/// The global cap alone is not enough: one authenticated `sensing:read` caller +/// looping on `POST /api/v1/ws-ticket` could occupy all 512 slots for 30 +/// seconds and 503 every other user — a denial of service by an ordinary, +/// lowest-privilege account. A page needs a handful of concurrent sockets, so +/// this is generous while making one caller unable to starve the rest. +/// +/// Tickets issued to the legacy static token share the `None` bucket, since +/// that credential carries no subject to attribute them to. +const MAX_PER_PRINCIPAL: usize = 16; + /// What a redeemed ticket authorizes. /// /// The scopes are captured at issue time from the authenticated request, so a @@ -106,7 +116,20 @@ impl TicketStore { if map.len() >= MAX_OUTSTANDING { tracing::warn!( outstanding = map.len(), - "refusing to issue a WebSocket ticket: too many outstanding" + "refusing to issue a WebSocket ticket: global cap reached" + ); + return None; + } + // Per-principal cap, so one caller cannot starve every other user. + let held_by_this_principal = map + .values() + .filter(|e| e.grant.subject == grant.subject) + .count(); + if held_by_this_principal >= MAX_PER_PRINCIPAL { + tracing::warn!( + subject = ?grant.subject, + held = held_by_this_principal, + "refusing to issue a WebSocket ticket: per-principal cap reached" ); return None; } @@ -254,24 +277,75 @@ mod tests { } #[test] - fn issuing_is_refused_once_too_many_are_outstanding() { + fn one_principal_cannot_starve_the_global_pool() { + // The reported DoS: an ordinary sensing:read caller looping on + // /api/v1/ws-ticket used to be able to occupy every slot and 503 + // everyone else. let store = TicketStore::new(); - for _ in 0..MAX_OUTSTANDING { - assert!(store.issue(grant()).is_some()); + let noisy = TicketGrant { + scopes: Some("sensing:read".into()), + subject: Some("noisy-user".into()), + }; + for _ in 0..MAX_PER_PRINCIPAL { + assert!(store.issue(noisy.clone()).is_some()); } assert!( - store.issue(grant()).is_none(), - "an authenticated but misbehaving caller must not grow the map without bound" + store.issue(noisy).is_none(), + "one principal must hit its own cap" + ); + // ...and a different user is entirely unaffected. + let other = TicketGrant { + scopes: Some("sensing:read".into()), + subject: Some("quiet-user".into()), + }; + assert!( + store.issue(other).is_some(), + "another principal must still be served" + ); + assert!( + store.outstanding() < MAX_OUTSTANDING, + "the global pool was never exhausted" + ); + } + + #[test] + fn issuing_is_refused_once_too_many_are_outstanding() { + let store = TicketStore::new(); + for i in 0..MAX_OUTSTANDING { + // Distinct subjects, so this exercises the GLOBAL cap and not the + // per-principal one. + let g = TicketGrant { + scopes: Some("sensing:read".into()), + subject: Some(format!("user-{i}")), + }; + assert!(store.issue(g).is_some()); + } + assert!( + store + .issue(TicketGrant { + scopes: Some("sensing:read".into()), + subject: Some("one-more".into()) + }) + .is_none(), + "the global cap must still hold" ); } #[test] fn expired_tickets_free_capacity_again() { let store = TicketStore::new(); - for _ in 0..MAX_OUTSTANDING { - store.issue(grant()); + for i in 0..MAX_OUTSTANDING { + store.issue(TicketGrant { + scopes: Some("sensing:read".into()), + subject: Some(format!("user-{i}")), + }); } - assert!(store.issue(grant()).is_none()); + assert!(store + .issue(TicketGrant { + scopes: Some("sensing:read".into()), + subject: Some("blocked".into()) + }) + .is_none()); { let mut map = store.inner.lock().unwrap(); for e in map.values_mut() {