From 2e9410cb40493208dd415dea500724929a768d82 Mon Sep 17 00:00:00 2001 From: Spriha Anand Date: Wed, 28 Oct 2026 16:46:03 +0530 Subject: [PATCH] Bind each client group to a fixed-size identity claim on the Group itself --- src/client/backend.rs | 124 +++++++++++++++++++++++++++++++++++ src/client/connection.rs | 1 + src/client/groups.rs | 138 ++++++++++++++++++++++++++++++++++++++- 3 files changed, 260 insertions(+), 3 deletions(-) diff --git a/src/client/backend.rs b/src/client/backend.rs index 98c00c3..7052552 100644 --- a/src/client/backend.rs +++ b/src/client/backend.rs @@ -37,6 +37,80 @@ pub struct Identity { pub token: Option, } +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 { + 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 { + 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); + } +} diff --git a/src/client/connection.rs b/src/client/connection.rs index 8f285fa..af61aad 100644 --- a/src/client/connection.rs +++ b/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, diff --git a/src/client/groups.rs b/src/client/groups.rs index 436673d..2e10fd8 100644 --- a/src/client/groups.rs +++ b/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, + /// 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, lmids: Vec<(String, i64)>, results: Vec, reply: oneshot::Sender, @@ -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, sockets: HashMap, desired: HashMap>, queries: HashMap, @@ -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, + claim: Option, lmids: Vec<(String, i64)>, results: Vec, ) -> 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| { + 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, 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(), );