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.

1 files changed, 33 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).