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.

1 files changed, 68 insertions, 38 deletions

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<AppState>,
     request: Request<Body>,
@@ -379,13 +381,10 @@ pub(crate) async fn streams_end(
         .extensions()
         .get::<Token>()
         .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::<AuthenticatedUser>() {
+            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<Claims>),
+    /// 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<Claims>,
+    who: Credentials,
     mut changes: tokio::sync::broadcast::Receiver<Arc<crate::document::store::ChangeSet>>,
 ) {
-    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<token::Refused> {
-    {
-        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<token::Refused> {
+    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`.