api: a stream opened with a token ends with it

A token was checked when a request came, and a stream answered to it (the dashboard's
events, a subscription's links and messages, the store's streams) then went on for as
long as the client kept it open: past the token's expiry, past its user's tokens being
revoked, its user disabled or removed, its session ended.

Now a middleware inside `login::bearer` (`login::streams_end`) ends a streaming answer
(`text/event-stream`, `application/x-ndjson`) to a request with a token at the token's
deadline, its expiry, or as soon as the document no longer takes it
(`token::still_taken`, the checks `verify` makes of the document, now its own
function): looked at when a change touches protocol data (the root, a switch, among
it), and when a session ends on the node itself in a swarm without a registry
(`AppState::session_ended`). No polling: a timer for the expiry, and the changes as
they come. The client reconnects with the token it has then; a refreshed token is a
new one, so a stream lasts a token's lifetime at most (12 hours by default).

`futures` is a dependency of every build now (for `take_until`), not only of libp2p's
and Tor's.

7 files changed, 126 insertions, 15 deletions

diff --git a/Cargo.toml b/Cargo.toml
index e6cd6c6..920cbfe 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -13,7 +13,7 @@ dashboard = []
 # supervises or one running elsewhere.
 mycelium = ["dep:reqwest", "dep:rustls", "dep:libc"]
 # The libp2p transport (`--libp2p-enable`).
-libp2p = ["dep:libp2p", "dep:libp2p-stream", "dep:futures"]
+libp2p = ["dep:libp2p", "dep:libp2p-stream"]
 # The Tor transport (`--tor-enable`): onion services, with arti.
 tor = [
     "dep:arti-client",
@@ -22,7 +22,6 @@ tor = [
     "dep:tor-proto",
     "dep:tor-rtcompat",
     "dep:safelog",
-    "dep:futures",
 ]
 
 [dependencies]
@@ -48,7 +47,8 @@ clap = "4.6"
 dashmap = "6.2"
 # Deflate (pure Rust), for large sync messages; Automerge uses it already.
 flate2 = "1.1"
-futures = { version = "0.3", optional = true }
+# Streams: an API stream ended with its token (`api::login::streams_end`), libp2p, tor.
+futures = "0.3"
 libp2p = { version = "0.57", optional = true, features = ["tokio", "tcp", "quic", "dns", "noise", "yamux", "ed25519", "identify", "ping", "macros", "relay", "dcutr", "upnp", "autonat", "mdns", "kad"] }
 libp2p-stream = { version = "0.5.0-alpha", optional = true }
 ed25519-dalek = "3"
diff --git a/docs/access.md b/docs/access.md
index 988d619..1385950 100644
--- a/docs/access.md
+++ b/docs/access.md
@@ -128,7 +128,10 @@ curl -H "Authorization: Bearer ct1.…" http://other-node:3000/store
   "log out"): `_continuum/sessions/<name>/<session>`, until when the session would have
   lasted, written by the user or one who manages users; a remake's snapshot leaves it
   out once that has passed. Without a registry, the node ending it refuses the
-  session's tokens itself, until it restarts; others take them until they expire. A token of a node banned or forgotten by the swarm isn't
+  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
   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 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