arete_server/
row_scope.rs

1//! Row scoping: rows of a scoped entity reach only the sessions entitled to
2//! them.
3//!
4//! An entity may name one string field as its **scope field** (the stack's
5//! LiveSpec `rowScopes`, carried on the entity spec as `scope_field`). Each
6//! row's value in that field is its scope id:
7//!
8//! - an absent, `null` or empty value makes the row **public**: every session
9//!   receives it;
10//! - any other string makes it **scoped**: a session receives it only when
11//!   its verified token's `row_scopes` claim lists that id;
12//! - any other value makes it visible to nobody, so a malformed row fails
13//!   closed.
14//!
15//! The scopes a session may read come only from its verified token, never
16//! from the subscription request. A client's own `partition` and `filters`
17//! still apply, but only on top of the scope check, so they can narrow what a
18//! session receives and never widen it. Sessions without the claim, and
19//! servers without authentication, receive only public rows of scoped
20//! entities. Entities without a scope field are unaffected.
21
22use std::collections::BTreeSet;
23use std::sync::Arc;
24
25use serde_json::Value;
26use tokio::sync::{RwLock, RwLockReadGuard};
27
28use crate::shared_entity::{lookup, EntityFields};
29use crate::websocket::auth::AuthContext;
30
31/// Default cap on the scope partitions one sorted view keeps (see
32/// [`SortedViewCache::partition_by_scope`](crate::sorted_cache::SortedViewCache::partition_by_scope)).
33pub const DEFAULT_MAX_SCOPE_PARTITIONS: usize = 10_000;
34
35/// The field of a scoped entity's rows that holds their scope id.
36#[derive(Debug, Clone, PartialEq, Eq)]
37pub struct ScopeField {
38    path: Arc<[String]>,
39}
40
41impl ScopeField {
42    /// A scope field from its dotted path (`meta.scope`).
43    pub fn new(dotted_path: &str) -> Self {
44        Self {
45            path: dotted_path.split('.').map(str::to_string).collect(),
46        }
47    }
48
49    /// The field's path segments.
50    pub fn path(&self) -> &[String] {
51        &self.path
52    }
53
54    /// The scope of a whole row: an absent field is public.
55    pub fn scope_of<E: EntityFields + ?Sized>(&self, row: &E) -> RowScope {
56        match lookup(row, self.path.iter()) {
57            Some(value) => RowScope::of_value(&value),
58            None => RowScope::Public,
59        }
60    }
61
62    /// The scope a patch sets. Replacing an ancestor with a non-object
63    /// clears its nested scope; omitting a field from a partial object leaves
64    /// its old value untouched, so only that case returns `None`.
65    pub fn scope_in<E: EntityFields + ?Sized>(&self, data: &E) -> Option<RowScope> {
66        let mut path = self.path.iter();
67        let first = path.next()?;
68        let Some(mut current) = data.field(first) else {
69            return (!data.whole().is_object()).then_some(RowScope::Public);
70        };
71        for segment in path {
72            if !current.is_object() {
73                return Some(RowScope::Public);
74            }
75            current = current.get(segment)?;
76        }
77        Some(RowScope::of_value(current))
78    }
79}
80
81/// A row's scope.
82#[derive(Debug, Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
83pub enum RowScope {
84    /// No scope: every session receives the row.
85    Public,
86    /// Only sessions whose row scopes list this id receive the row.
87    Scoped(String),
88    /// The scope field holds something other than a string: nobody receives
89    /// the row.
90    Invalid,
91}
92
93impl RowScope {
94    fn of_value(value: &Value) -> Self {
95        match value {
96            Value::Null => Self::Public,
97            Value::String(scope) if scope.is_empty() => Self::Public,
98            Value::String(scope) => Self::Scoped(scope.clone()),
99            _ => Self::Invalid,
100        }
101    }
102
103    /// The partition a sorted view keeps the row in: `""` for public rows,
104    /// the scope id otherwise, and none for an invalid scope.
105    pub(crate) fn partition(&self) -> Option<&str> {
106        match self {
107            Self::Public => Some(""),
108            Self::Scoped(scope) => Some(scope),
109            Self::Invalid => None,
110        }
111    }
112}
113
114/// The row scopes a verified session may read: its `row_scopes` claim,
115/// without empty ids. Empty for a session without the claim or without
116/// authentication.
117pub fn session_row_scopes(auth: Option<&AuthContext>) -> BTreeSet<String> {
118    auth.and_then(|auth| auth.row_scopes.as_ref())
119        .map(|scopes| {
120            scopes
121                .iter()
122                .filter(|scope| !scope.is_empty())
123                .cloned()
124                .collect()
125        })
126        .unwrap_or_default()
127}
128
129/// Which rows of one scoped view one subscription may receive: the view's
130/// scope field and the session's row scopes, taken from its verified token
131/// when the subscription attached.
132///
133/// It also carries the connection's row-scope generation. A token refresh
134/// that changes the session's row scopes bumps the generation (under the
135/// write half of the connection's lock) before it resnapshots, so a
136/// subscription task still running on the old scopes can no longer send:
137/// see [`RowFilter::hold`].
138#[derive(Debug, Clone)]
139pub(crate) struct RowFilter {
140    field: ScopeField,
141    scopes: Arc<BTreeSet<String>>,
142    lock: Arc<RwLock<u64>>,
143    generation: u64,
144}
145
146impl RowFilter {
147    pub(crate) fn new(
148        field: ScopeField,
149        scopes: BTreeSet<String>,
150        lock: Arc<RwLock<u64>>,
151        generation: u64,
152    ) -> Self {
153        Self {
154            field,
155            scopes: Arc::new(scopes),
156            lock,
157            generation,
158        }
159    }
160
161    pub(crate) fn field(&self) -> &ScopeField {
162        &self.field
163    }
164
165    /// The session's row scopes.
166    pub(crate) fn scopes(&self) -> &BTreeSet<String> {
167        &self.scopes
168    }
169
170    pub(crate) fn allows_scope(&self, scope: &RowScope) -> bool {
171        match scope {
172            RowScope::Public => true,
173            RowScope::Scoped(scope) => self.scopes.contains(scope),
174            RowScope::Invalid => false,
175        }
176    }
177
178    /// Whether the session may receive this whole row.
179    pub(crate) fn allows<E: EntityFields + ?Sized>(&self, row: &E) -> bool {
180        self.allows_scope(&self.field.scope_of(row))
181    }
182
183    /// Whether frame data that carries the scope field moves the row out of
184    /// this session's reach. `false` when the data leaves the scope as it was.
185    pub(crate) fn denies_frame_data<E: EntityFields + ?Sized>(&self, data: &E) -> bool {
186        self.field
187            .scope_in(data)
188            .is_some_and(|scope| !self.allows_scope(&scope))
189    }
190
191    /// Hold the right to send on these scopes, without waiting. `None` once a
192    /// refresh changed the session's row scopes, or while one is changing
193    /// them: the subscription must stop.
194    pub(crate) fn try_hold(&self) -> Option<RwLockReadGuard<'_, u64>> {
195        self.lock
196            .try_read()
197            .ok()
198            .filter(|generation| **generation == self.generation)
199    }
200
201    /// [`Self::try_hold`], waiting out a refresh in progress.
202    pub(crate) async fn hold(&self) -> Option<RwLockReadGuard<'_, u64>> {
203        let generation = self.lock.read().await;
204        (*generation == self.generation).then_some(generation)
205    }
206}
207
208#[cfg(test)]
209mod tests {
210    use super::*;
211    use serde_json::json;
212
213    fn filter(scopes: &[&str]) -> RowFilter {
214        RowFilter::new(
215            ScopeField::new("meta.scope"),
216            scopes.iter().map(|scope| scope.to_string()).collect(),
217            Arc::new(RwLock::new(0)),
218            0,
219        )
220    }
221
222    #[test]
223    fn absent_null_and_empty_scopes_are_public() {
224        let field = ScopeField::new("meta.scope");
225        assert_eq!(field.scope_of(&json!({})), RowScope::Public);
226        assert_eq!(field.scope_of(&json!({"meta": {}})), RowScope::Public);
227        assert_eq!(
228            field.scope_of(&json!({"meta": {"scope": null}})),
229            RowScope::Public
230        );
231        assert_eq!(
232            field.scope_of(&json!({"meta": {"scope": ""}})),
233            RowScope::Public
234        );
235        assert_eq!(
236            field.scope_of(&json!({"meta": {"scope": "a"}})),
237            RowScope::Scoped("a".into())
238        );
239        assert_eq!(
240            field.scope_of(&json!({"meta": {"scope": 7}})),
241            RowScope::Invalid
242        );
243    }
244
245    #[test]
246    fn a_session_reads_public_rows_and_its_own_scopes_only() {
247        let scoped_to_a = filter(&["a"]);
248        assert!(scoped_to_a.allows(&json!({"meta": {"scope": "a"}})));
249        assert!(scoped_to_a.allows(&json!({"name": "public"})));
250        assert!(!scoped_to_a.allows(&json!({"meta": {"scope": "b"}})));
251        assert!(!scoped_to_a.allows(&json!({"meta": {"scope": ["a"]}})));
252
253        let unscoped = filter(&[]);
254        assert!(unscoped.allows(&json!({"meta": {"scope": ""}})));
255        assert!(!unscoped.allows(&json!({"meta": {"scope": "a"}})));
256    }
257
258    #[test]
259    fn frame_data_without_the_field_leaves_the_scope_alone() {
260        let scoped_to_a = filter(&["a"]);
261        assert!(!scoped_to_a.denies_frame_data(&json!({"count": 2})));
262        assert!(!scoped_to_a.denies_frame_data(&json!({"meta": {"scope": "a"}})));
263        assert!(scoped_to_a.denies_frame_data(&json!({"meta": {"scope": "b"}})));
264    }
265
266    #[test]
267    fn parent_replacements_clear_scope_but_partial_objects_keep_it() {
268        let field = ScopeField::new("meta.access.scope");
269        for patch in [
270            json!({"meta":null}),
271            json!({"meta":false}),
272            json!({"meta":[]}),
273            json!({"meta":"replaced"}),
274            json!({"meta":{"access":null}}),
275            json!({"meta":{"access":7}}),
276            Value::Null,
277        ] {
278            assert_eq!(field.scope_in(&patch), Some(RowScope::Public), "{patch}");
279            assert_eq!(
280                field.scope_in(&crate::shared_entity::SharedEntity::new(patch)),
281                Some(RowScope::Public)
282            );
283        }
284        for patch in [
285            json!({"value":1}),
286            json!({"meta":{}}),
287            json!({"meta":{"access":{}}}),
288            json!({"meta":{"other":true}}),
289        ] {
290            assert_eq!(field.scope_in(&patch), None, "{patch}");
291        }
292        assert_eq!(
293            field.scope_in(&json!({"meta":{"access":{"scope":false}}})),
294            Some(RowScope::Invalid)
295        );
296        assert_eq!(
297            field.scope_in(&json!({"meta":{"access":{"scope":"B"}}})),
298            Some(RowScope::Scoped("B".into()))
299        );
300    }
301
302    #[test]
303    fn session_row_scopes_come_from_the_claim_without_empty_ids() {
304        let claims = arete_auth::SessionClaims::builder("issuer", "subject", "audience")
305            .with_row_scopes(["a", "", "b"])
306            .build();
307        let context = AuthContext::from_claims(claims);
308        assert_eq!(
309            session_row_scopes(Some(&context)),
310            ["a".to_string(), "b".to_string()].into_iter().collect()
311        );
312        assert!(session_row_scopes(None).is_empty());
313    }
314
315    #[tokio::test]
316    async fn a_bumped_generation_stops_the_old_filter() {
317        let lock = Arc::new(RwLock::new(0));
318        let old = RowFilter::new(ScopeField::new("scope"), BTreeSet::new(), lock.clone(), 0);
319        assert!(old.try_hold().is_some());
320        *lock.write().await += 1;
321        assert!(old.try_hold().is_none());
322        assert!(old.hold().await.is_none());
323    }
324}
arete_server/row_scope.rs

arete_server/
row_scope.rs
