parse_rust_rest/
acl.rs

1//! ACL enforcement: the boundary between the `ACL` field a client sees and the `_rperm`/`_wperm`
2//! columns storage holds.
3//!
4//! **The rule that must not be got wrong: absent permission columns mean public.**
5//! `addReadACL` emits `_rperm: {$in: [null, '*', ...acl]}` and `null` in a Mongo `$in` matches a
6//! document where the field is *missing*, which is how a row saved without an ACL stays readable.
7//! Omitting the null silently hides every such row, and there is no error to notice.
8
9use parse_rust_core::{Acl, ErrorCode, ParseError, ParseMap, ParseValue, Permissions, Principal};
10use parse_rust_storage::{Comparison, Constraint};
11
12/// Who a request is acting as, for ACL purposes.
13///
14/// An enum rather than an `Option<String>` so that "no ACL constraint at all" cannot be reached
15/// by forgetting to set a field. `acl === undefined` as a master sentinel is the upstream shape
16/// this deliberately does not copy.
17#[derive(Debug, Clone, PartialEq, Eq)]
18pub enum AclScope {
19    /// Master or maintenance: no ACL constraint is applied at all.
20    Unrestricted,
21    /// A caller acting as nobody in particular.
22    Anonymous,
23    /// A logged-in user, with the transitive closure of their roles.
24    ///
25    /// **Roles are bare names here, with no `role:` prefix.** The prefix is added by
26    /// [`AclScope::acl_group`] and by `AclScope::principals`, so there is exactly one place
27    /// that knows the wire spelling. Construct through [`AclScope::user`] rather than by
28    /// literal, so that a `role:`-prefixed objectId cannot reach `object_id`.
29    User {
30        object_id: String,
31        roles: Vec<String>,
32    },
33}
34
35impl AclScope {
36    /// Build a user scope, refusing a `role:`-prefixed objectId.
37    ///
38    /// A user whose objectId began with `role:` would be granted that role by every ACL and CLP
39    /// check, because the entity namespace is one flat string space on the wire. Upstream guards
40    /// it at two session-resolution sites with the same code and message (`Auth.js:195`, `:237`);
41    /// here the guard is at the one place a scope can be built.
42    pub fn user(object_id: impl Into<String>, roles: Vec<String>) -> Result<Self, ParseError> {
43        let object_id = object_id.into();
44        if object_id.starts_with("role:") {
45            return Err(ParseError::new(
46                ErrorCode::InternalServerError,
47                "Invalid object ID.",
48            ));
49        }
50        Ok(AclScope::User { object_id, roles })
51    }
52
53    pub fn is_master(&self) -> bool {
54        matches!(self, AclScope::Unrestricted)
55    }
56
57    pub fn user_id(&self) -> Option<&str> {
58        match self {
59            AclScope::User { object_id, .. } => Some(object_id),
60            _ => None,
61        }
62    }
63
64    /// Does the caller hold this role? The name is bare, with no `role:` prefix.
65    pub fn has_role(&self, name: &str) -> bool {
66        match self {
67            AclScope::User { roles, .. } => roles.iter().any(|r| r == name),
68            _ => false,
69        }
70    }
71
72    /// Upstream's `aclGroup`: `['*']`, then every role as `role:<name>`, then the user's objectId
73    /// (`RestWrite.js:184`, `RestQuery.js:427`, both `['*'].concat(roles, [user.id])`).
74    ///
75    /// Master is the empty list, because upstream never reaches a caller that consumes an
76    /// `aclGroup` without first branching on `isMaster`.
77    ///
78    /// Order matters twice over. `addPointerPermissions` extracts the single user id by filtering
79    /// out `role:` and `*` (`DatabaseController.js:1745-1747`), and the compiled `$in` array is
80    /// snapshot-compared.
81    pub fn acl_group(&self) -> Vec<String> {
82        match self {
83            AclScope::Unrestricted => Vec::new(),
84            AclScope::Anonymous => vec!["*".to_string()],
85            AclScope::User { object_id, roles } => {
86                let mut out = Vec::with_capacity(roles.len() + 2);
87                out.push("*".to_string());
88                out.extend(roles.iter().map(|r| format!("role:{r}")));
89                // Defensive, and unreachable through `AclScope::user`. A `role:`-prefixed
90                // objectId that arrived by literal construction is dropped rather than emitted,
91                // which costs the caller access to their own rows and grants nothing.
92                if !object_id.starts_with("role:") {
93                    out.push(object_id.clone());
94                }
95                out
96            }
97        }
98    }
99
100    /// The principals this caller matches in an `_rperm`/`_wperm` lookup.
101    ///
102    /// `null` first, then optionally a literal `'*'`, then the `aclGroup`
103    /// (`DatabaseController.js:81`, `:88`).
104    fn principals(&self, seed_public: bool) -> Vec<ParseValue> {
105        // `null` matches a row with no permission column, i.e. a public row.
106        let mut out = vec![ParseValue::Null];
107        if seed_public {
108            out.push(ParseValue::String("*".to_string()));
109        }
110        out.extend(self.acl_group().into_iter().map(ParseValue::String));
111        out
112    }
113
114    /// The constraint to add to a read.
115    ///
116    /// `None` for [`AclScope::Unrestricted`], which is the only case where no constraint is
117    /// applied. Returning `Option` makes the master case explicit at every call site instead of
118    /// being the absence of a step.
119    ///
120    /// UPSTREAM-QUIRK: the emitted list carries `'*'` twice, once seeded by `addReadACL`
121    /// (`DatabaseController.js:88`) and once already present in the `aclGroup`
122    /// (`RestQuery.js:427`). A duplicate in an `$in` changes nothing, and removing it would make
123    /// the compiled query differ from upstream's for no gain.
124    pub fn read_constraint(&self) -> Option<Constraint> {
125        match self {
126            AclScope::Unrestricted => None,
127            _ => Some(Constraint {
128                field: "_rperm".to_string(),
129                comparison: Comparison::In(self.principals(true)),
130            }),
131        }
132    }
133
134    /// The constraint to add to a write.
135    ///
136    /// Note the asymmetry with reads: `addWriteACL` omits `'*'` from the injected list, because
137    /// `getUserAndRoleACL` already seeds it for every non-master caller. Reproduced rather than
138    /// unified, since the two functions are not symmetric upstream and a caller path that builds
139    /// its own list would behave differently.
140    pub fn write_constraint(&self) -> Option<Constraint> {
141        match self {
142            AclScope::Unrestricted => None,
143            _ => Some(Constraint {
144                field: "_wperm".to_string(),
145                comparison: Comparison::In(self.principals(false)),
146            }),
147        }
148    }
149}
150
151/// Split an `ACL` field out of a row into the two storage columns.
152///
153/// Returns the row with `ACL` removed and the columns added. A row with no `ACL` gets no columns,
154/// which is what makes it public.
155///
156/// **The test upstream applies is falsiness, not "is it an object"** (`DatabaseController.js:94-96`,
157/// literally `if (!ACL) return result`). Everything truthy falls through to a `for...in` that reads
158/// `.read` and `.write` off each entry, so a string, a number or an array yields no principals but
159/// **still writes both columns as empty arrays**, which is a master-only row. Skipping the columns
160/// instead writes a row with no `_rperm`/`_wperm` at all, and an absent column is public.
161///
162/// Getting this wrong is not a cosmetic divergence. Nothing type-checks `ACL` on either side, by
163/// design (`SchemaController.js:1312-1315`), so `{"ACL":"x"}` reaches here from any client. The
164/// consequential class is `_Role`: its required-column check tests presence and truthiness only, so
165/// a non-object `ACL` would satisfy it and then produce a world-writable role that any caller can
166/// add itself to.
167///
168/// The update path applies the same test, in `lower_acl_into_update`. It did not until a review:
169/// it tested for `null` alone, so `false`, `0` and `""` fell through and cleared both columns on a
170/// row that already had permissions. Both paths now branch on truthiness, and the tests on each
171/// side loop over the falsy values rather than checking one, because checking one is what let the
172/// other three through.
173pub fn lower_acl(mut row: ParseMap) -> ParseMap {
174    let Some(acl_value) = row.shift_remove("ACL") else {
175        return row;
176    };
177    if !parse_rust_core::is_js_truthy(&acl_value) {
178        return row;
179    }
180    // `None` here is a truthy non-object, which upstream's loop walks and takes nothing from.
181    let acl = acl_from_value(&acl_value).unwrap_or_default();
182    let (rperm, wperm) = acl.to_perms();
183    row.insert(
184        "_rperm".to_string(),
185        ParseValue::Array(rperm.into_iter().map(ParseValue::String).collect()),
186    );
187    row.insert(
188        "_wperm".to_string(),
189        ParseValue::Array(wperm.into_iter().map(ParseValue::String).collect()),
190    );
191    row
192}
193
194/// Rebuild the `ACL` field from the two storage columns, then drop them.
195///
196/// Reproduces `untransformObjectACL` exactly, including that both columns absent produce **no
197/// `ACL` key at all** rather than `null` or `{}`.
198pub fn raise_acl(mut row: ParseMap) -> ParseMap {
199    let rperm = take_string_array(&mut row, "_rperm");
200    let wperm = take_string_array(&mut row, "_wperm");
201
202    let Some(acl) = Acl::from_perms(rperm.as_deref(), wperm.as_deref()) else {
203        return row;
204    };
205
206    let mut map = ParseMap::new();
207    for (principal, perms) in acl.iter() {
208        if perms.is_empty() {
209            continue;
210        }
211        let mut entry = ParseMap::new();
212        // Only true flags are emitted. UPSTREAM-QUIRK, see `parse_rust_core::acl`.
213        if perms.read {
214            entry.insert("read".to_string(), ParseValue::Bool(true));
215        }
216        if perms.write {
217            entry.insert("write".to_string(), ParseValue::Bool(true));
218        }
219        map.insert(principal.as_key(), ParseValue::Object(entry));
220    }
221    row.insert("ACL".to_string(), ParseValue::Object(map));
222    row
223}
224
225fn take_string_array(row: &mut ParseMap, key: &str) -> Option<Vec<String>> {
226    match row.shift_remove(key) {
227        Some(ParseValue::Array(items)) => Some(
228            items
229                .into_iter()
230                .filter_map(|v| match v {
231                    ParseValue::String(s) => Some(s),
232                    _ => None,
233                })
234                .collect(),
235        ),
236        _ => None,
237    }
238}
239
240/// Read a client-supplied `ACL` value.
241fn acl_from_value(value: &ParseValue) -> Option<Acl> {
242    let ParseValue::Object(map) = value else {
243        return None;
244    };
245    let mut acl = Acl::new();
246    for (key, entry) in map {
247        let ParseValue::Object(flags) = entry else {
248            continue;
249        };
250        let flag = |name: &str| matches!(flags.get(name), Some(ParseValue::Bool(true)));
251        acl.set(
252            Principal::parse(key),
253            Permissions {
254                read: flag("read"),
255                write: flag("write"),
256            },
257        );
258    }
259    Some(acl)
260}
261
262#[cfg(test)]
263mod tests {
264    use super::*;
265
266    fn row(pairs: Vec<(&str, ParseValue)>) -> ParseMap {
267        let mut m = ParseMap::new();
268        for (k, v) in pairs {
269            m.insert(k.to_string(), v);
270        }
271        m
272    }
273
274    /// The single most important assertion in this module.
275    #[test]
276    fn the_read_constraint_includes_null_so_public_rows_stay_visible() {
277        let c = AclScope::Anonymous
278            .read_constraint()
279            .expect("anonymous is constrained");
280        assert_eq!(c.field, "_rperm");
281        match c.comparison {
282            Comparison::In(values) => {
283                assert!(
284                    values.iter().any(|v| matches!(v, ParseValue::Null)),
285                    "null must be in the list, or every row saved without an ACL becomes invisible"
286                );
287                assert!(values
288                    .iter()
289                    .any(|v| matches!(v, ParseValue::String(s) if s == "*")));
290            }
291            other => panic!("expected In, got {other:?}"),
292        }
293    }
294
295    /// The falsy-versus-not-an-object distinction, which is the whole of `lower_acl`'s contract.
296    ///
297    /// A truthy non-object must produce two empty arrays, which is a master-only row. Producing no
298    /// columns instead is a public row, and on `_Role` that is a world-writable role any caller can
299    /// add itself to, because the required-column check tests truthiness and stops there.
300    #[test]
301    fn a_truthy_non_object_acl_writes_empty_columns_rather_than_none() {
302        for value in [
303            ParseValue::String("x".into()),
304            ParseValue::Number(1.0),
305            ParseValue::Array(vec![]),
306            ParseValue::Array(vec![ParseValue::String("*".into())]),
307            ParseValue::Bool(true),
308            ParseValue::Object(ParseMap::new()),
309        ] {
310            let lowered = lower_acl(row(vec![("ACL", value.clone())]));
311            for column in ["_rperm", "_wperm"] {
312                assert!(
313                    matches!(lowered.get(column), Some(ParseValue::Array(a)) if a.is_empty()),
314                    "a truthy ACL must write an empty {column}, got {:?} for {value:?}",
315                    lowered.get(column)
316                );
317            }
318            assert!(!lowered.contains_key("ACL"));
319        }
320    }
321
322    /// The other half. Falsy means no columns, which is a public row, and that is upstream's
323    /// `if (!ACL) return result`.
324    #[test]
325    fn a_falsy_or_absent_acl_writes_no_columns() {
326        for value in [
327            ParseValue::Null,
328            ParseValue::Bool(false),
329            ParseValue::Number(0.0),
330            ParseValue::String(String::new()),
331        ] {
332            let lowered = lower_acl(row(vec![("ACL", value.clone())]));
333            assert!(!lowered.contains_key("_rperm"), "falsy ACL: {value:?}");
334            assert!(!lowered.contains_key("_wperm"), "falsy ACL: {value:?}");
335        }
336        let untouched = lower_acl(row(vec![("title", ParseValue::String("x".into()))]));
337        assert!(!untouched.contains_key("_rperm"));
338        assert!(!untouched.contains_key("_wperm"));
339    }
340
341    #[test]
342    fn master_applies_no_constraint_at_all() {
343        assert!(AclScope::Unrestricted.read_constraint().is_none());
344        assert!(AclScope::Unrestricted.write_constraint().is_none());
345    }
346
347    #[test]
348    fn a_user_scope_carries_its_object_id() {
349        let c = AclScope::user("u1", vec![])
350            .expect("plain id")
351            .read_constraint()
352            .expect("constrained");
353        match c.comparison {
354            Comparison::In(values) => assert!(values
355                .iter()
356                .any(|v| matches!(v, ParseValue::String(s) if s == "u1"))),
357            other => panic!("expected In, got {other:?}"),
358        }
359    }
360
361    /// 0.1.0 emitted no `role:` entry at all, so a `role:Admins` entry in `_rperm` matched
362    /// nobody and every role-protected row was invisible to its own members.
363    #[test]
364    fn a_role_entry_now_matches() {
365        let scope = AclScope::user("u1", vec!["Admins".into(), "Editors".into()]).expect("scope");
366        let c = scope.read_constraint().expect("constrained");
367        match c.comparison {
368            Comparison::In(values) => {
369                let strings: Vec<&str> = values
370                    .iter()
371                    .filter_map(|v| match v {
372                        ParseValue::String(s) => Some(s.as_str()),
373                        _ => None,
374                    })
375                    .collect();
376                assert!(strings.contains(&"role:Admins"), "{strings:?}");
377                assert!(strings.contains(&"role:Editors"), "{strings:?}");
378                assert!(strings.contains(&"u1"));
379            }
380            other => panic!("expected In, got {other:?}"),
381        }
382    }
383
384    /// Upstream order: `['*']`, then roles, then the user id. `addPointerPermissions` recovers
385    /// the single user id by filtering the first two out, so the shape is load bearing.
386    #[test]
387    fn the_acl_group_is_star_then_roles_then_the_user() {
388        let scope = AclScope::user("u1", vec!["A".into()]).expect("scope");
389        assert_eq!(scope.acl_group(), vec!["*", "role:A", "u1"]);
390        assert_eq!(AclScope::Anonymous.acl_group(), vec!["*"]);
391        assert!(AclScope::Unrestricted.acl_group().is_empty());
392    }
393
394    #[test]
395    fn accessors_answer_for_every_variant() {
396        let scope = AclScope::user("u1", vec!["A".into()]).expect("scope");
397        assert!(scope.has_role("A"));
398        assert!(!scope.has_role("role:A"), "roles are stored bare");
399        assert!(!scope.has_role("B"));
400        assert_eq!(scope.user_id(), Some("u1"));
401        assert!(!scope.is_master());
402        assert!(AclScope::Unrestricted.is_master());
403        assert_eq!(AclScope::Anonymous.user_id(), None);
404        assert!(!AclScope::Anonymous.has_role("A"));
405    }
406
407    /// The `role:` objectId collision, guarded at the one place a scope can be built.
408    #[test]
409    fn a_role_prefixed_object_id_is_refused() {
410        let e = AclScope::user("role:Admins", vec![]).unwrap_err();
411        assert_eq!(e.code, parse_rust_core::ErrorCode::InternalServerError);
412        assert_eq!(e.message, "Invalid object ID.");
413    }
414
415    #[test]
416    fn acl_lowers_to_two_columns_and_raises_back() {
417        let mut acl_map = ParseMap::new();
418        let mut public = ParseMap::new();
419        public.insert("read".into(), ParseValue::Bool(true));
420        acl_map.insert("*".into(), ParseValue::Object(public));
421        let mut owner = ParseMap::new();
422        owner.insert("read".into(), ParseValue::Bool(true));
423        owner.insert("write".into(), ParseValue::Bool(true));
424        acl_map.insert("u1".into(), ParseValue::Object(owner));
425
426        let lowered = lower_acl(row(vec![
427            ("title", ParseValue::String("x".into())),
428            ("ACL", ParseValue::Object(acl_map)),
429        ]));
430        assert!(
431            lowered.get("ACL").is_none(),
432            "ACL must not be stored as a field"
433        );
434        assert!(matches!(lowered.get("_rperm"), Some(ParseValue::Array(a)) if a.len() == 2));
435        assert!(matches!(lowered.get("_wperm"), Some(ParseValue::Array(a)) if a.len() == 1));
436
437        let raised = raise_acl(lowered);
438        assert!(raised.get("_rperm").is_none() && raised.get("_wperm").is_none());
439        let ParseValue::Object(acl) = raised.get("ACL").expect("ACL restored") else {
440            panic!("ACL should be an object");
441        };
442        assert!(acl.contains_key("*") && acl.contains_key("u1"));
443    }
444
445    /// UPSTREAM-QUIRK, reproduced end to end.
446    #[test]
447    fn a_false_flag_disappears_on_the_round_trip() {
448        let mut entry = ParseMap::new();
449        entry.insert("read".into(), ParseValue::Bool(true));
450        entry.insert("write".into(), ParseValue::Bool(false));
451        let mut acl_map = ParseMap::new();
452        acl_map.insert("*".into(), ParseValue::Object(entry));
453
454        let raised = raise_acl(lower_acl(row(vec![("ACL", ParseValue::Object(acl_map))])));
455        let ParseValue::Object(acl) = raised.get("ACL").expect("ACL") else {
456            panic!()
457        };
458        let ParseValue::Object(star) = acl.get("*").expect("*") else {
459            panic!()
460        };
461        assert!(star.contains_key("read"));
462        assert!(
463            !star.contains_key("write"),
464            "the false key is dropped, matching untransformObjectACL"
465        );
466    }
467
468    #[test]
469    fn a_row_with_no_acl_gets_no_columns_and_no_acl_key_back() {
470        let lowered = lower_acl(row(vec![("title", ParseValue::String("x".into()))]));
471        assert!(lowered.get("_rperm").is_none());
472        let raised = raise_acl(lowered);
473        assert!(
474            raised.get("ACL").is_none(),
475            "absent columns produce no ACL key at all, not null and not an empty object"
476        );
477    }
478}
parse_rust_rest/acl.rs

parse_rust_rest/
acl.rs
