From c1e750c5f795d4dd659be99f873ab7f98d5a2e83 Mon Sep 17 00:00:00 2001 From: tv Date: Mon, 9 Nov 2026 15:34:54 +0000 Subject: api: a stream opened by password or a proxy's header ends with its user Review 4, F11: only streams answered to a token ended with it (1273); one opened by password (basic authentication) or a proxy's header went on after its user was disabled or removed. `streams_end` now wraps those too, by the user `basic_auth` leaves on the answer: such a stream ends once the document no longer has its user, enabled (`token::user_taken`, the part of a token's check about its user, which `still_taken` uses too), looked at as the stream begins and as protocol data changes. No expiry: a password has none. --- docs/access.md | 3 +- src/access/token.rs | 20 +++++++--- src/api/login.rs | 106 +++++++++++++++++++++++++++++++++------------------- 3 files changed, 85 insertions(+), 44 deletions(-) diff --git a/docs/access.md b/docs/access.md index 6422b51..3d4aeaf 100644 --- a/docs/access.md +++ b/docs/access.md @@ -133,7 +133,8 @@ curl -H "Authorization: Bearer ct1.…" http://other-node:3000/store session's tokens itself, until it restarts; others take them until they expire. A stream opened with a token (server-sent events, the store's streams) ends with it: when it expires, or once the node holds a change that revokes it (looked at as - protocol data changes), a session's end on the node itself included. A token of a node banned or forgotten by the swarm isn't + protocol data changes), a session's end on the node itself included. One opened by + password or a proxy's header ends once its user is disabled or removed. A token of a node banned or forgotten by the swarm isn't taken either. In a swarm without a registry, a token is taken until it expires. The user a logout names is its node's word, and a logout no deciding path, even under a strict policy: any member could log any user out. It ends sessions; and a value there diff --git a/src/access/token.rs b/src/access/token.rs index a9fc270..153737e 100644 --- a/src/access/token.rs +++ b/src/access/token.rs @@ -174,11 +174,8 @@ fn issuer_counts(doc: &impl ReadDoc, claims: &Claims) -> Result<(), Refused> { /// In a swarm with a registry: its user registered, enabled, its tokens not revoked /// since it was issued, its session not ended. fn user_counts(doc: &impl ReadDoc, claims: &Claims) -> Result<(), Refused> { - if registry::exists(doc) { - let user = registry::user(doc, &claims.user).ok_or(Refused::UnknownUser)?; - if user.disabled { - return Err(Refused::Disabled); - } + user_taken(doc, &claims.user)?; + if let Some(user) = registry::user(doc, &claims.user) { if claims.iat < user.tokens_not_before { return Err(Refused::Revoked); } @@ -189,6 +186,19 @@ fn user_counts(doc: &impl ReadDoc, claims: &Claims) -> Result<(), Refused> { Ok(()) } +/// In a swarm with a registry: user `name` registered and enabled. What a stream opened +/// by password or a proxy's header is judged by as the document changes +/// (`api::login::streams_end`), as one opened with a token is by `still_taken`. +pub(crate) fn user_taken(doc: &impl ReadDoc, name: &str) -> Result<(), Refused> { + if registry::exists(doc) { + let user = registry::user(doc, name).ok_or(Refused::UnknownUser)?; + if user.disabled { + return Err(Refused::Disabled); + } + } + Ok(()) +} + /// A new session's ID (8 random bytes, hex). pub(crate) fn new_session() -> String { registry::new_id() diff --git a/src/api/login.rs b/src/api/login.rs index 83e7f25..0dc74e5 100644 --- a/src/api/login.rs +++ b/src/api/login.rs @@ -363,13 +363,15 @@ pub(crate) async fn login_key( answer(&state, &claims) } -/// A stream answered to a request with a token (server-sent events, the store's -/// streams) ends with the token: when it expires, or as soon as the document says it -/// isn't taken any more (`token::still_taken`: its user disabled or removed, its tokens -/// revoked, its session ended, its issuer banned), looked at when a change touches -/// protocol data, or when a session ends on this node. The page reconnects with the -/// token it has then, if any; a refreshed one is a new token, so a stream lasts at most -/// a token's lifetime. Answers that aren't streams pass as they are. +/// A stream answered to a request (server-sent events, the store's streams) ends with +/// what let it in: a token when it expires, or as soon as the document says it isn't +/// taken any more (`token::still_taken`: its user disabled or removed, its tokens +/// revoked, its session ended, its issuer banned); a password or a proxy's header as +/// soon as its user is disabled or removed (`token::user_taken`). Looked at as the +/// stream begins, when a change touches protocol data, and when a session ends on this +/// node. The page reconnects with what it has then, if anything; a refreshed token is a +/// new one, so a stream opened with one lasts a token's lifetime at most. Answers that +/// aren't streams pass as they are. pub(crate) async fn streams_end( State(state): State, request: Request, @@ -379,13 +381,10 @@ pub(crate) async fn streams_end( .extensions() .get::() .map(|token| Arc::clone(&token.claims)); - // Listening before the answer is made: a change revoking the token that comes while - // it is (the node applying what it was behind on) isn't missed. - let changes = token.as_ref().map(|_| state.change_events.subscribe()); + // Listening before the answer is made: a change revoking what let it in that comes + // while it is (the node applying what it was behind on) isn't missed. + let changes = state.change_events.subscribe(); let response = next.run(request).await; - let (Some(claims), Some(changes)) = (token, changes) else { - return response; - }; let streaming = response .headers() .get(axum::http::header::CONTENT_TYPE) @@ -396,26 +395,55 @@ pub(crate) async fn streams_end( if !streaming { return response; } + // By a password or a proxy's header: the user `basic_auth` leaves on the answer. + let who = match token { + Some(claims) => Credentials::Token(claims), + None => match response.extensions().get::() { + Some(AuthenticatedUser(user)) => Credentials::User(user.clone()), + None => return response, + }, + }; let (parts, body) = response.into_parts(); - let ended = token_ends(state, claims, changes); + let ended = credentials_end(state, who, changes); let body = futures::StreamExt::take_until(body.into_data_stream(), Box::pin(ended)); Response::from_parts(parts, Body::from_stream(body)) } -/// When the token of `claims` ends: it expires, or the document (or this node's ended -/// sessions) no longer takes it, looked at once at first (what changed since the -/// request was let in) and with every change of protocol data in `changes` after. -async fn token_ends( +/// What let a stream in (`streams_end`). +enum Credentials { + Token(Arc), + /// A password, or a proxy's header: the user. + User(String), +} + +/// When `who` ends: a token expires; the document (or this node's ended sessions) no +/// longer takes it, looked at once at first (what changed since the request was let in) +/// and with every change of protocol data in `changes` after. +async fn credentials_end( state: AppState, - claims: Arc, + who: Credentials, mut changes: tokio::sync::broadcast::Receiver>, ) { - if let Some(refused) = refused_now(&state, &claims).await { - debug!(user = %claims.user, reason = %refused, "Ended a stream with its token"); + let user = match &who { + Credentials::Token(claims) => claims.user.clone(), + Credentials::User(user) => user.clone(), + }; + if let Some(refused) = refused_now(&state, &who).await { + debug!(user = %user, reason = %refused, "Ended a stream with what let it in"); return; } - let left = claims.exp.saturating_sub(state.clock.now_millis().as_u64()); - let expiry = tokio::time::sleep(std::time::Duration::from_millis(left)); + let left = match &who { + Credentials::Token(claims) => { + Some(claims.exp.saturating_sub(state.clock.now_millis().as_u64())) + } + Credentials::User(_) => None, + }; + let expiry = async { + match left { + Some(left) => tokio::time::sleep(std::time::Duration::from_millis(left)).await, + None => std::future::pending().await, + } + }; tokio::pin!(expiry); loop { let session_ended = state.session_ended.notified(); @@ -431,27 +459,29 @@ async fn token_ends( Err(tokio::sync::broadcast::error::RecvError::Closed) => return, }, } - if let Some(refused) = refused_now(&state, &claims).await { - debug!(user = %claims.user, reason = %refused, "Ended a stream with its token"); + if let Some(refused) = refused_now(&state, &who).await { + debug!(user = %user, reason = %refused, "Ended a stream with what let it in"); return; } } } -/// Why the token of `claims` isn't taken any more, if it isn't: by the document, or -/// this node's ended sessions. -async fn refused_now(state: &AppState, claims: &Claims) -> Option { - { - let replica = state.doc.read().await; - token::still_taken(&replica.doc, claims).err() +/// Why `who` isn't taken any more, if it isn't: by the document, or this node's ended +/// sessions. +async fn refused_now(state: &AppState, who: &Credentials) -> Option { + let replica = state.doc.read().await; + match who { + Credentials::Token(claims) => { + token::still_taken(&replica.doc, claims).err().or_else(|| { + state + .ended_sessions + .lock_or_recover() + .contains_key(&claims.session) + .then_some(token::Refused::Ended) + }) + } + Credentials::User(user) => token::user_taken(&replica.doc, user).err(), } - .or_else(|| { - state - .ended_sessions - .lock_or_recover() - .contains_key(&claims.session) - .then_some(token::Refused::Ended) - }) } /// `POST /auth/refresh`. -- cgit v1.3.1