diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/access/token.rs | 44 | ||||
| -rw-r--r-- | src/api/login.rs | 77 | ||||
| -rw-r--r-- | src/api/mod.rs | 5 | ||||
| -rw-r--r-- | src/node.rs | 1 | ||||
| -rw-r--r-- | src/state/mod.rs | 3 |
5 files changed, 119 insertions, 11 deletions
diff --git a/src/access/token.rs b/src/access/token.rs index e7aff42..ae1546c 100644 --- a/src/access/token.rs +++ b/src/access/token.rs @@ -124,16 +124,7 @@ pub(crate) fn verify( return Err(Refused::OtherSwarm); } let key = identity::public_of(&claims.issuer).ok_or(Refused::Malformed)?; - let member = store::read_path(doc, &meta::path(&[meta::MEMBERS, &claims.issuer])) - .ok() - .flatten() - .is_some(); - if !member { - return Err(Refused::UnknownIssuer); - } - if super::bans::banned(doc, &claims.issuer) { - return Err(Refused::BannedIssuer); - } + issuer_counts(doc, &claims)?; let mut signed = LABEL.to_vec(); signed.extend_from_slice(&payload); if !identity::verify(&key, &signed, &signature) { @@ -145,6 +136,37 @@ pub(crate) fn verify( if claims.iat > now.saturating_add(SKEW_MS) { return Err(Refused::Future); } + user_counts(doc, &claims)?; + Ok(claims) +} + +/// Whether a token of `claims`, taken once (`verify`), still is by what `doc` says now: +/// its issuer a member and not banned, its user registered, enabled, its tokens not +/// revoked since and its session not ended. What a stream opened with it is judged by +/// as the document changes (`api::login::streams_end`). +pub(crate) fn still_taken(doc: &impl ReadDoc, claims: &Claims) -> Result<(), Refused> { + issuer_counts(doc, claims)?; + user_counts(doc, claims) +} + +/// Its issuer a member of the swarm, not banned. +fn issuer_counts(doc: &impl ReadDoc, claims: &Claims) -> Result<(), Refused> { + let member = store::read_path(doc, &meta::path(&[meta::MEMBERS, &claims.issuer])) + .ok() + .flatten() + .is_some(); + if !member { + return Err(Refused::UnknownIssuer); + } + if super::bans::banned(doc, &claims.issuer) { + return Err(Refused::BannedIssuer); + } + Ok(()) +} + +/// 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 { @@ -157,7 +179,7 @@ pub(crate) fn verify( return Err(Refused::Ended); } } - Ok(claims) + Ok(()) } /// A new session's ID (8 random bytes, hex). diff --git a/src/api/login.rs b/src/api/login.rs index ec0c9cf..57c1859 100644 --- a/src/api/login.rs +++ b/src/api/login.rs @@ -362,6 +362,81 @@ 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. +pub(crate) async fn streams_end( + State(state): State<AppState>, + request: Request<Body>, + next: axum::middleware::Next, +) -> Response { + let token = request + .extensions() + .get::<Token>() + .map(|token| Arc::clone(&token.claims)); + let response = next.run(request).await; + let Some(claims) = token else { + return response; + }; + let streaming = response + .headers() + .get(axum::http::header::CONTENT_TYPE) + .and_then(|kind| kind.to_str().ok()) + .is_some_and(|kind| { + kind.starts_with("text/event-stream") || kind.starts_with("application/x-ndjson") + }); + if !streaming { + return response; + } + let (parts, body) = response.into_parts(); + let ended = token_ends(state, claims); + 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. +async fn token_ends(state: AppState, claims: Arc<Claims>) { + let mut changes = state.change_events.subscribe(); + let left = claims.exp.saturating_sub(state.clock.now_millis().as_u64()); + let expiry = tokio::time::sleep(std::time::Duration::from_millis(left)); + tokio::pin!(expiry); + loop { + let session_ended = state.session_ended.notified(); + tokio::select! { + () = &mut expiry => return, + () = session_ended => {} + change = changes.recv() => match change { + // Protocol data, the root (a switch) among it. + Ok(set) if !set.paths.iter().any(|path| { + path.first().is_none_or(|first| first == crate::document::meta::ROOT) + }) => continue, + Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {} + Err(tokio::sync::broadcast::error::RecvError::Closed) => return, + }, + } + let refused = { + let replica = state.doc.read().await; + token::still_taken(&replica.doc, &claims).err() + } + .or_else(|| { + state + .ended_sessions + .lock_or_recover() + .contains_key(&claims.session) + .then_some(token::Refused::Ended) + }); + if let Some(refused) = refused { + debug!(user = %claims.user, reason = %refused, "Ended a stream with its token"); + return; + } + } +} + /// `POST /auth/refresh`. pub(crate) async fn refresh( State(state): State<AppState>, @@ -477,6 +552,8 @@ async fn end(state: &AppState, claims: &Claims, by: &str) -> Result<(), ApiError let mut ended = state.ended_sessions.lock_or_recover(); ended.retain(|_, until| *until > now); ended.insert(claims.session.clone(), until); + drop(ended); + state.session_ended.notify_waiters(); } info!(user = %claims.user, session = %claims.session, by = %by, recorded, "Ended a session"); Ok(()) diff --git a/src/api/mod.rs b/src/api/mod.rs index 6631bc9..97724ff 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -121,6 +121,11 @@ pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) -> )) // Who has a password here (`/auth/keys`: a user's last key goes only then). .layer(axum::Extension(auth_state)) + // A stream answered to a token ends with it (expired, revoked, its session ended). + .layer(middleware::from_fn_with_state( + state.clone(), + login::streams_end, + )) // Outside authentication: a token, if the request brings one, names the user. .layer(middleware::from_fn_with_state(state.clone(), login::bearer)) // Outermost: pages of other origins, and their preflights, which bring no diff --git a/src/node.rs b/src/node.rs index 309c126..6f08e69 100644 --- a/src/node.rs +++ b/src/node.rs @@ -193,6 +193,7 @@ impl Node { genesis_root: config.genesis_root, challenges: Default::default(), ended_sessions: Default::default(), + session_ended: Default::default(), token_lifetime: config.token_lifetime, session_lifetime: config.session_lifetime, challenge_lifetime: config.challenge_lifetime, diff --git a/src/state/mod.rs b/src/state/mod.rs index a2da643..2fbfd31 100644 --- a/src/state/mod.rs +++ b/src/state/mod.rs @@ -161,6 +161,9 @@ pub(crate) struct AppState { /// (`api::login::end_session`). This node refuses their tokens; others take them /// until they expire, and so does this one after a restart. pub(crate) ended_sessions: Arc<std::sync::Mutex<std::collections::BTreeMap<String, u64>>>, + /// Woken when a session ends on this node (`ended_sessions`): streams opened with + /// its tokens end (`api::login::streams_end`). + pub(crate) session_ended: Arc<tokio::sync::Notify>, pub(crate) token_lifetime: Duration, pub(crate) session_lifetime: Duration, /// How long a challenge to log in by key is valid, and how many logins by key a user |