Feat/ws grants bug 215

aimdb-core/CHANGELOG.md

@@ -7,6 +7,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed (breaking, API)
+
+- **`ConnectorConfig` gains `record_index: Option<usize>`**, the id of the
+  record an outbound publish comes from — its registration index, the same
+  `record_id` `AimDb::list_records` reports. A topic alone cannot identify the
+  record, since several records may publish on one topic (or on topics a
+  `TopicProvider` picks per value), so a connector enforcing per-record
+  authorization needs the index at publish time; the WebSocket connector uses it
+  to gate delivery against each client's granted records. The struct is not
+  `#[non_exhaustive]`, so code building a `ConnectorConfig` with a struct
+  literal must add the field (or use `..Default::default()`); it defaults to
+  `None`.
+
+- **`AimDb::collect_outbound_routes` stamps each route with its record index**,
+  appending `("record_index", "<id>")` to `OutboundRoute::config`, and
+  `ConnectorConfig::from_query` parses that pair into `record_index` rather than
+  passing it on in `protocol_options`. `record_index` is therefore a reserved
+  config key: a value set through `with_config("record_index", …)` is
+  overridden by the stamped one, which is appended last.
+
 ## [2.0.0] - 2026-09-18
 
 ### Added

aimdb-core/src/builder.rs

@@ -64,6 +64,9 @@ struct RecordEntry {
 pub struct AimDbInner {
     /// Record entries, indexed by `RecordId`. Order matches registration
     /// order. Immutable after build().
+    ///
+    /// Per-client authorization is based on index of RecordEntry in `storage`
+    /// Never reorder, remove, or `swap_remove` entries as these could break current logic.
     storages: Vec<RecordEntry>,
 
     /// Name → RecordId lookup (control plane)
@@ -1222,7 +1225,14 @@ impl AimDb {
     pub fn collect_outbound_routes(&self, scheme: &str) -> Vec<OutboundRoute> {
         let mut routes = Vec::new();
 
-        for entry in &self.inner.storages {
+        for (i, entry) in self.inner.storages.iter().enumerate() {
+            // i and RecordId must match
+            debug_assert_eq!(
+                self.inner.by_key.get(&entry.key).map(|id| id.index()),
+                Some(i),
+                "record storage order diverges from RecordId for key {}",
+                entry.key.as_str()
+            );
             let outbound_links = entry.record.outbound_connectors();
 
             for link in outbound_links {
@@ -1231,11 +1241,15 @@ impl AimDb {
                     continue;
                 }
 
+                // config must carry the record index
+                let mut config = link.config.clone();
+                config.push(("record_index".to_string(), i.to_string()));
+
                 // Create the fused source using the stored factory
                 routes.push(OutboundRoute {
                     topic: link.url.resource_id().to_string(),
                     source: link.create_source(self),
-                    config: link.config.clone(),
+                    config,
                 });
             }
         }

aimdb-core/src/transport.rs

@@ -33,13 +33,19 @@ pub struct ConnectorConfig {
     /// Protocol-specific options as key-value pairs
     /// Allows custom configuration without polluting the base struct
     pub protocol_options: Vec<(String, String)>,
+
+    /// The index of the record key that is setup together with the outbound route.
+    /// The record key is unique, and its order in `AimDb::inner.storage` is immutable
+    /// so this `record_index` could be used for `O(1)` lookup.
+    pub record_index: Option<usize>,
 }
 
 impl Default for ConnectorConfig {
     fn default() -> Self {
         Self {
             timeout_ms: Some(5000),
             protocol_options: Vec::new(),
+            record_index: None,
         }
     }
 }
@@ -51,8 +57,9 @@ impl ConnectorConfig {
     /// per-route configuration through to [`Connector::publish`] without changing
     /// the `publish` signature.
     ///
-    /// Only the protocol-agnostic `timeout_ms` is lifted into the typed field;
-    /// every other key is passed through verbatim in
+    /// Only the protocol-agnostic `timeout_ms` and `record_index` (stamped by
+    /// `AimDb::collect_outbound_routes`; the last occurrence wins) are lifted
+    /// into typed fields; every other key is passed through verbatim in
     /// [`protocol_options`](ConnectorConfig::protocol_options) for the
     /// connector to interpret with its own defaults.
     pub fn from_query(query: &[(String, String)]) -> ConnectorConfig {
@@ -64,6 +71,11 @@ impl ConnectorConfig {
                         cfg.timeout_ms = Some(n);
                     }
                 }
+                "record_index" => {
+                    if let Ok(i) = v.parse::<usize>() {
+                        cfg.record_index = Some(i);
+                    }
+                }
                 _ => cfg.protocol_options.push((k.clone(), v.clone())),
             }
         }

aimdb-websocket-connector/CHANGELOG.md

@@ -7,6 +7,118 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed (breaking, API)
+
+- **Read grants are record-key patterns, not topic patterns.**
+  `Permissions::subscribe_patterns` is renamed to `read_patterns` and
+  `Permissions::can_subscribe(topic)` is renamed `can_read(key)`. The rename
+  carries a sublte change: the read patterns are now matched against **record
+  keys** — the string passed to `configure::<T>(key, ...)` — rather than against
+  the ws topic registered at `link_to("ws://...")`. `write_patterns` and `can_write` are
+  unchanged and remain **topic**-based. An operator migrating a config
+  replaces each topic grant with the key(s) of the records publishing there;
+  where the two namespaces coincide (`link_to("ws://cfg")` on record `cfg`) no
+  change is needed.
+
+- **`AuthHandler::authorize_subscribe`, `authorize_query` and `authorize_list`
+  are removed.** Authorization is resolved once, at the HTTP upgrade, into a
+  per-client record bitset; the three read paths consult that bitset instead of
+  calling back into the handler. A handler that overrode any of them must move
+  the logic into `authenticate`, which now returns the grants that become the
+  bitset. `authorize_write` is unaffected.
+
+  **Read grants are now fixed for the lifetime of a connection.** Previously a
+  handler overriding these hooks could consult external or mutable state (an ACL
+  service, a revocation list) on every `sub`, `record.query` and `record.list`,
+  and so refuse *new* read requests on an open connection once a grant changed.
+  That per-request re-check is gone: nothing on the read path calls back into
+  the handler after `authenticate`. With the default handler nothing changes —
+  grants were already fixed at the upgrade. Live subscriptions were never
+  re-checked per event, before or after this change, so revoking a client's
+  reads still means closing its connection and letting it re-authenticate.
+
+- **`AuthHandler::authorize_query_record` added**, a `record.query` post-filter
+  receiving the handler's rows and returning the subset the client may read.
+  It defaults to `Permissions::can_read` on each row's key, so implementors need
+  it only to narrow further. This is deliberately pattern-based rather than
+  bitset-based: history outlives configuration, so a query may legitimately name
+  a record the running server no longer registers.
+
+- **`ClientInfo` carries `record_perms: Arc<RecordsBits>`**, the resolved bitset,
+  alongside the `Permissions` it was built from. `RecordsBits` is public
+  (`new`, `set`, `is_allowed`, `len`, `is_empty`,
+  `resolve_permissions`); one bit per registered record, indexed by record id —
+  registration order in the builder, which is also the index
+  `AimDb::list_records` reports.
+
+- **`ClientManager::subscribe` and `broadcast` signatures changed.**
+  `subscribe(pattern)` becomes `subscribe(pattern, Arc<RecordsBits>)` — a
+  subscription now carries the grants it delivers under — and
+  `broadcast(topic, payload)` becomes `broadcast(topic, record_index, payload)`,
+  since the topic alone no longer identifies which record a message came from.
+
+- **`SnapshotProvider::snapshots` returns `Vec<(usize, String, Vec<u8>)>`**
+  instead of `Vec<(String, Vec<u8>)>`; the added `usize` is the record id the
+  cached value belongs to. Two records publishing on one topic now yield two
+  entries rather than one overwriting the other.
+
+- **`ConnectorConfig` carry an additional attr `record_index: Option<usize>`
+  to `WsBusSink`**, so record index could join topic at outbound routes.
+
+### Changed (breaking, wire)
+
+- **A late-join burst may carry several `snap` frames for the same topic** — one
+  per granted record publishing there — where previously a topic produced at
+  most one. Each rides its own `seq`, and `last` still closes the burst, so a
+  client tracking sequence numbers is unaffected; a client indexing snapshots by
+  topic must expect collisions.
+
+- **`record.query`'s `total` is the number of rows returned**, after
+  authorization, rather than the handler's own match count. The trait doc for
+  `QueryHandler::handle_query` is updated to say so: a client is never told how
+  many rows it was not allowed to see. The built-in persistence handler already
+  returned `records.len()`, so only custom handlers change behaviour.
+
+  However, there is a limit to this design: in case `total` < `limit`, it does not mean
+  that there is no more record, but records returned from `QueryHandlerFn` are filtered by grants.
+  A better design is to resolve search name against grants then dedicate the filtering
+  to `QueryHandlerFn` instead of let it stay in `WsSession`.
+
+- **`record.query` distinguishes "not permitted" from "nothing matched".**
+  `denied` is returned only when the client holds no read grants at all;
+  a client with grants that match no stored rows gets `{"records": [], "total": 0}`.
+  Querying a record the server no longer registers is allowed if the grant
+  covers its key, so history survives a record's retirement from the config.
+
+- **Subscribing to a topic no records publish on now succeeds and stays silent**
+  rather than being refused. Topics are not part of the authorization model any
+  more, so the server cannot tell an unregistered topic from one whose records
+  the client may not read. Only a client holding no read grants at all is
+  refused with `denied`; a client whose grants match no registered record
+  subscribes and stays silent, as does any client on a server with no records.
+
+- **`record.list` returns only the records the client may read.** Previously the
+  full database was enumerable by any authenticated client.
+
+### Fixed
+
+- **Late-join snapshots no longer lose a record when two share a topic.** The
+  snapshot cache was keyed by topic, so the most recent publisher overwrote the
+  other's cached value and a late-joining client received only one of them —
+  and, once grants became per record, sometimes neither. It is now keyed by
+  `(record id, topic)`.
+
+### Security
+
+- **A read grant no longer leaks records it does not name.** Grants lived in
+  topic space while records are keyed independently, so a client granted one
+  record's topic received every record publishing there — including records it
+  held no grant for, over `event` frames, late-join snapshots, `record.list` and
+  `record.query` alike. Records whose topic comes from a `TopicProvider` made
+  this unbounded, since the topic is chosen per value at runtime. Grants are now
+  resolved against record keys at the upgrade and enforced at every delivery
+  point, so a record's data reaches only clients granted that record.
+
 ## [0.3.0] - 2026-09-18
 
 ### Added

aimdb-websocket-connector/src/lib.rs

@@ -125,7 +125,7 @@ pub type WebSocketConnector = server::builder::WebSocketConnectorBuilder;
 
 #[cfg(feature = "server")]
 pub use server::auth::{
-    AuthError, AuthHandler, AuthRequest, ClientId, ClientInfo, NoAuth, Permissions,
+    AuthError, AuthHandler, AuthRequest, ClientId, ClientInfo, NoAuth, Permissions, RecordsBits,
 };
 #[cfg(feature = "server")]
 pub use server::client_manager::ClientManager;