mapping of group_id and identifying stamp

src/client/backend.rs

@@ -37,6 +37,80 @@ pub struct Identity {
     pub token: Option<String>,
 }
 
+impl Identity {
+    /// The claim that binds a client group to its first claimant. Prefers
+    /// the JWT `sub` (stable across token rotation); then a token that is
+    /// not a JWT (it is the identity in its own right); then the cookie
+    /// pairs. Fixed-size, so a bind is a `Copy` compare and no raw
+    /// credential is kept.
+    pub fn claim(&self) -> Option<u64> {
+        if let Some(token) = self
+            .token
+            .as_deref()
+            .map(str::trim)
+            .filter(|token| !token.is_empty())
+        {
+            if is_jwt(token) {
+                // A JWT's stable claim is its `sub`. One with no `sub` has
+                // no claim of its own: its raw form rotates, so fall
+                // through to the cookies rather than bind to it.
+                if let Some(sub) = jwt_sub(token) {
+                    return Some(digest("sub", &sub));
+                }
+            } else {
+                return Some(digest("token", token));
+            }
+        }
+        let cookie = self.cookie.as_deref().map(str::trim)?;
+        let mut pairs: Vec<&str> = cookie
+            .split(';')
+            .map(str::trim)
+            .filter(|pair| !pair.is_empty())
+            .collect();
+        if pairs.is_empty() {
+            return None;
+        }
+        // Cookie order is not part of a browser's identity: the same pairs
+        // in another order are the same claim.
+        pairs.sort_unstable();
+        pairs.dedup();
+        Some(digest("cookie", &pairs.join("\n")))
+    }
+}
+
+/// Whether `token` is a JWT: the three dot-separated segments of one.
+fn is_jwt(token: &str) -> bool {
+    token.split('.').count() == 3
+}
+
+/// Extract the `sub` claim from a JWT without verifying the signature.
+/// Returns `None` when the token is blank, malformed, or has no non-blank `sub`.
+fn jwt_sub(token: &str) -> Option<String> {
+    let token = token.trim();
+    if token.is_empty() {
+        return None;
+    }
+    let payload_b64 = token.split('.').nth(1)?;
+    use base64::Engine as _;
+    let payload_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
+        .decode(payload_b64)
+        .or_else(|_| base64::engine::general_purpose::URL_SAFE.decode(payload_b64))
+        .ok()?;
+    let payload: serde_json::Value = serde_json::from_slice(&payload_bytes).ok()?;
+    let sub = payload.get("sub").and_then(|v| v.as_str())?;
+    if sub.is_empty() { None } else { Some(sub.to_owned()) }
+}
+
+/// A fixed-size digest of `kind` and `material`.
+fn digest(kind: &str, material: &str) -> u64 {
+    use rustc_hash::FxHasher;
+    use std::hash::{Hash, Hasher};
+    let mut hasher = FxHasher::default();
+    kind.hash(&mut hasher);
+    material.hash(&mut hasher);
+    hasher.finish()
+}
+
 /// The query endpoint's answer.
 #[derive(Debug)]
 pub enum TransformOutcome {
@@ -276,3 +350,53 @@ fn preview(text: &str) -> String {
     }
     format!("{}...", &text[..end])
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn make(token: Option<&str>, cookie: Option<&str>) -> Identity {
+        Identity {
+            token: token.map(str::to_owned),
+            cookie: cookie.map(str::to_owned),
+            ..Identity::default()
+        }
+    }
+
+    // Two JWTs with the same `sub` ("user-123") but different `iat`.
+    const JWT_V1: &str = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9\
+        .eyJzdWIiOiAidXNlci0xMjMiLCAiaWF0IjogMTcwMDAwMDAwMH0.fakesig";
+    const JWT_V2: &str = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9\
+        .eyJzdWIiOiAidXNlci0xMjMiLCAiaWF0IjogMTcwMDAwMzYwMH0.fakesig";
+    const JWT_OTHER: &str = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9\
+        .eyJzdWIiOiAidXNlci00NTYiLCAiaWF0IjogMTcwMDAwMDAwMH0.fakesig";
+    // JWT with no `sub`.
+    const JWT_NOSUB: &str = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9\
+        .eyJzdWIiOiBudWxsLCAiaWF0IjogMTcwMDAwMDAwMH0.fakesig";
+
+    /// A claim is the JWT `sub` when the token has one (so rotated tokens
+    /// for the same user match), an opaque token when it is not a JWT, and
+    /// the cookie pairs otherwise. Never blank.
+    #[test]
+    fn a_claim_is_the_jwt_sub_the_token_or_the_cookies() {
+        // JWT sub; two rotated tokens for the same user match.
+        assert_eq!(make(Some(JWT_V1), None).claim(), make(Some(JWT_V2), None).claim());
+        assert_ne!(make(Some(JWT_V1), None).claim(), make(Some(JWT_OTHER), None).claim());
+
+        // An opaque token is the claim in its own right.
+        let opaque = make(Some("opaque-token"), None).claim().unwrap();
+        assert_ne!(opaque, make(Some(JWT_V1), None).claim().unwrap());
+        assert_eq!(opaque, make(Some("  opaque-token  "), None).claim().unwrap());
+
+        // No token, or a JWT with no `sub`, falls through to the cookies;
+        // the same pairs in another order are the same claim.
+        let cookie = make(None, Some("sid=1; uid=2")).claim();
+        assert_eq!(cookie, make(Some(""), Some("sid=1; uid=2")).claim());
+        assert_eq!(cookie, make(Some(JWT_NOSUB), Some("sid=1; uid=2")).claim());
+        assert_eq!(cookie, make(None, Some("uid=2; sid=1")).claim());
+
+        // No token and no cookie is no claim.
+        assert_eq!(make(None, None).claim(), None);
+        assert_eq!(make(None, Some("  ;  ")).claim(), None);
+    }
+}

src/client/connection.rs

@@ -452,6 +452,7 @@ async fn handle(
             sink: out.clone(),
         },
         base_cookie: conn.params.base_cookie.clone(),
+        claim: conn.identity.claim(),
         lmids,
         results,
         reply: reply_tx,

src/client/groups.rs

@@ -72,6 +72,9 @@
 //!   every replayed poke gets the same current ids, in its last part.
 //! - **Anything else** (a cookie older than the log, or from a server
 //!   that is gone): the client is told to start over, and syncs afresh.
+//! - **Before any of that**, the identity claim: a live group answers only
+//!   the claim that claimed it ([`super::backend::Identity::claim`]); a
+//!   connection with no claim joins only unclaimed groups.
 //!
 //! The application's mutation ids arrive as writes to its clients table,
 //! which the engine side carries inside the transaction's own
@@ -191,6 +194,10 @@ pub enum Request {
         wsid: String,
         socket: Socket,
         base_cookie: Option<String>,
+        /// The connection's identity claim, as [`super::backend::Identity::claim`]
+        /// derives it: used only to bind the group to its first claimant —
+        /// never forwarded.
+        claim: Option<u64>,
         lmids: Vec<(String, i64)>,
         results: Vec<MutationResult>,
         reply: oneshot::Sender<ConnectReply>,
@@ -383,6 +390,9 @@ fn release_rows(rows: &mut Ledger, sub: SubId) -> Vec<(TableName, DataFrameKey)>
 struct Group {
     version: u64,
     generation: u64,
+    /// The claim that owns this group, bound on the first connect that
+    /// carried one. None until then: an unclaimed group answers anyone.
+    claim: Option<u64>,
     sockets: HashMap<String, Socket>,
     desired: HashMap<String, HashSet<String>>,
     queries: HashMap<String, QueryState>,
@@ -481,6 +491,7 @@ impl Group {
         Group {
             version: 0,
             generation: 0,
+            claim: None,
             sockets: HashMap::new(),
             desired: HashMap::new(),
             queries: HashMap::new(),
@@ -646,11 +657,12 @@ impl Groups {
                 wsid,
                 socket,
                 base_cookie,
+                claim,
                 lmids,
                 results,
                 reply,
             } => {
-                let outcome = self.connect(&group, wsid, socket, base_cookie, lmids, results);
+                let outcome = self.connect(&group, wsid, socket, base_cookie, claim, lmids, results);
                 let _ = reply.send(outcome);
             }
             Request::Disconnect { group, wsid } => self.disconnect(&group, &wsid),
@@ -819,15 +831,34 @@ impl Groups {
     /// whole state when it has none and the group is under way, the logged
     /// pokes after an older cookie. A client that cannot be caught up is
     /// told to start over; no other connection of the group is touched.
+    #[allow(clippy::too_many_arguments)]
     fn connect(
         &mut self,
         group_id: &str,
         wsid: String,
         socket: Socket,
         base_cookie: Option<String>,
+        claim: Option<u64>,
         lmids: Vec<(String, i64)>,
         results: Vec<MutationResult>,
     ) -> ConnectReply {
+        // A live group answers only the identity that claimed it: a
+        // stranger holding the group id is turned away before any of its
+        // state is read. A connection with no claim joins only unclaimed
+        // groups, which is what deployments without credentials (and the
+        // test benches) do everywhere.
+        if self
+            .groups
+            .get(group_id)
+            .is_some_and(|group| group.claim.is_some_and(|bound| claim != Some(bound)))
+        {
+            log_info!(
+                "connection {wsid} refused client group {group_id}: the group is bound to another identity"
+            );
+            return ConnectReply::Reset {
+                reason: "this client group belongs to another identity".to_owned(),
+            };
+        }
         let offered = match base_cookie.as_deref().map(protocol::version_of) {
             None => 0,
             Some(Some(version)) => version,
@@ -869,6 +900,11 @@ impl Groups {
         self.next_generation += 1;
         let generation = self.next_generation;
         let group = self.groups.get_mut(group_id).expect("just ensured");
+        // The group is live and every check has passed: bind it to this
+        // claim now, so a failed connect never claims a group id alone.
+        if group.claim.is_none() {
+            group.claim = claim;
+        }
         for (client, lmid) in lmids {
             let known = group.lmids.entry(client.clone()).or_insert(0);
             if lmid > *known {
@@ -1021,7 +1057,8 @@ impl Groups {
         }
     }
 
-    /// Forget a group and every subscription it had.
+    /// Forget a group and every subscription it had. Its claim goes with
+    /// it: a stale group id claims nothing afterwards.
     fn drop_group(&mut self, group_id: &str) {
         if let Some(group) = self.groups.remove(group_id) {
             self.stats.client_groups.fetch_sub(1, Ordering::Relaxed);
@@ -2027,6 +2064,7 @@ mod tests {
                 wsid.to_owned(),
                 socket,
                 cookie.map(str::to_owned),
+                None,
                 lmids,
                 results,
             ) {
@@ -2730,6 +2768,98 @@ mod tests {
         });
     }
 
+    /// A group is bound to the identity claim that claimed it: the same
+    /// claim joins again, a stranger's is turned away, and once the group
+    /// has expired the id claims nothing — a fresh identity starts fresh.
+    #[test]
+    fn a_group_answers_only_the_identity_that_claimed_it() {
+        on_local(|| {
+            let mut bench = Bench::new("262144");
+            let make_socket = |wsid: &str| {
+                let (sink, _frames) = mpsc::unbounded_channel();
+                Socket {
+                    client: format!("client-{wsid}"),
+                    sink,
+                }
+            };
+            let connect = |bench: &mut Bench, wsid: &str, claim: Option<u64>| {
+                bench.core.connect(
+                    "g",
+                    wsid.to_owned(),
+                    make_socket(wsid),
+                    None,
+                    claim,
+                    Vec::new(),
+                    Vec::new(),
+                )
+            };
+
+            assert!(matches!(
+                connect(&mut bench, "owner-1", Some(1)),
+                ConnectReply::Accepted
+            ));
+            assert!(matches!(
+                connect(&mut bench, "owner-2", Some(1)),
+                ConnectReply::Accepted
+            ), "the same identity joins its own group again");
+            assert!(
+                matches!(connect(&mut bench, "stranger", Some(2)), ConnectReply::Reset { .. }),
+                "a stranger holding the group id is turned away"
+            );
+            assert!(
+                matches!(connect(&mut bench, "anon", None), ConnectReply::Reset { .. }),
+                "an anonymous connection can not join a claimed group"
+            );
+
+            bench.core.drop_group("g");
+            assert!(matches!(
+                connect(&mut bench, "new-owner", Some(2)),
+                ConnectReply::Accepted
+            ), "an expired group id no longer claims anything");
+        });
+    }
+
+    /// A connection that claims a group id but fails validation (bad
+    /// cookie) must not leave the binding behind: the group id is still
+    /// free for the identity that is actually entitled to it.
+    #[test]
+    fn a_failed_connect_does_not_claim_a_group_id() {
+        on_local(|| {
+            let mut bench = Bench::new("262144");
+            let make_socket = |wsid: &str| {
+                let (sink, _frames) = mpsc::unbounded_channel();
+                Socket {
+                    client: format!("client-{wsid}"),
+                    sink,
+                }
+            };
+            let connect = |bench: &mut Bench, wsid: &str, claim: Option<u64>, cookie: Option<&str>| {
+                bench.core.connect(
+                    "g",
+                    wsid.to_owned(),
+                    make_socket(wsid),
+                    cookie.map(str::to_owned),
+                    claim,
+                    Vec::new(),
+                    Vec::new(),
+                )
+            };
+
+            // Mallory tries to claim the group with a bad cookie: rejected.
+            assert!(matches!(
+                connect(&mut bench, "mallory", Some(2), Some("bogus")),
+                ConnectReply::Reset { .. }
+            ), "a bad cookie is rejected");
+
+            // Alice connects with a different identity: still accepted,
+            // because the failed connect left no binding behind.
+            assert!(matches!(
+                connect(&mut bench, "alice", Some(1), None),
+                ConnectReply::Accepted
+            ), "a failed connect must not claim a group id");
+        });
+    }
+
     /// The two ways of keeping what a client missed meet without a gap or
     /// an overlap. A socket dies unnoticed at cookie `c`: the pokes sent to
     /// it meanwhile are in the log. The server then learns of it and the
@@ -3411,6 +3541,7 @@ mod tests {
                             sink,
                         },
                         None,
+                        None,
                         Vec::new(),
                         Vec::new(),
                     );
@@ -4024,7 +4155,7 @@ mod tests {
                     readiness: watch::channel(true).0,
                     backlog: Vec::new(),
                     groups: HashMap::new(),
-                    by_sub: HashMap::new(),
+                        by_sub: HashMap::new(),
                     awaiting: HashMap::new(),
                     pending: HashMap::new(),
                     lmid_changes: HashMap::new(),
@@ -4114,6 +4245,7 @@ mod tests {
                                         sink,
                                     },
                                     None,
+                                    None,
                                     Vec::new(),
                                     Vec::new(),
                                 );