parse-rust-rest 0.2.1

Parse read and write pipelines for parse-rust-server: schema validation, ACL enforcement, encoding.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
//! ACL enforcement: the boundary between the `ACL` field a client sees and the `_rperm`/`_wperm`
//! columns storage holds.
//!
//! **The rule that must not be got wrong: absent permission columns mean public.**
//! `addReadACL` emits `_rperm: {$in: [null, '*', ...acl]}` and `null` in a Mongo `$in` matches a
//! document where the field is *missing*, which is how a row saved without an ACL stays readable.
//! Omitting the null silently hides every such row, and there is no error to notice.
//!
//! **Two known differences from upstream live in [`lower_acl`], both recorded as deliberate
//! differences and both deferred rather than fixed here.**
//!
//! It reads principals from a map and nothing else, so an `ACL` that is an *array* takes the
//! truthy-non-object path below: two **empty** columns, which is a master-only row rather than a
//! column-less public one. Upstream enumerates the array, so `[{"read":true}]` grants principal
//! `"0"` there, an index being a property name. **That does change who may read the row**, in the
//! restrictive direction: a principal upstream grants is granted nothing here.
//!
//! And the columns come out in wire order, where upstream enumerates a JavaScript object and puts
//! integer-like keys first. That one grants the same rights to the same principals and is visible
//! only to a client preserving map order, or to a mixed fleet comparing stored rows.

use parse_rust_core::{Acl, ErrorCode, ParseError, ParseMap, ParseValue, Permissions, Principal};
use parse_rust_storage::{Comparison, Constraint};

/// Who a request is acting as, for ACL purposes.
///
/// An enum rather than an `Option<String>` so that "no ACL constraint at all" cannot be reached
/// by forgetting to set a field. `acl === undefined` as a master sentinel is the upstream shape
/// this deliberately does not copy.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AclScope {
    /// Master or maintenance: no ACL constraint is applied at all.
    Unrestricted,
    /// A caller acting as nobody in particular.
    Anonymous,
    /// A logged-in user, with the transitive closure of their roles.
    ///
    /// **Roles are bare names here, with no `role:` prefix.** The prefix is added by
    /// [`AclScope::acl_group`] and by `AclScope::principals`, so there is exactly one place
    /// that knows the wire spelling. Construct through [`AclScope::user`] rather than by
    /// literal, so that a `role:`-prefixed objectId cannot reach `object_id`.
    User {
        object_id: String,
        roles: Vec<String>,
    },
}

impl AclScope {
    /// Build a user scope, refusing a `role:`-prefixed objectId.
    ///
    /// A user whose objectId began with `role:` would be granted that role by every ACL and CLP
    /// check, because the entity namespace is one flat string space on the wire. Upstream guards
    /// it at two session-resolution sites with the same code and message (`Auth.js:195`, `:237`);
    /// here the guard is at the one place a scope can be built.
    pub fn user(object_id: impl Into<String>, roles: Vec<String>) -> Result<Self, ParseError> {
        let object_id = object_id.into();
        if object_id.starts_with("role:") {
            return Err(ParseError::new(
                ErrorCode::InternalServerError,
                "Invalid object ID.",
            ));
        }
        Ok(AclScope::User { object_id, roles })
    }

    pub fn is_master(&self) -> bool {
        matches!(self, AclScope::Unrestricted)
    }

    pub fn user_id(&self) -> Option<&str> {
        match self {
            AclScope::User { object_id, .. } => Some(object_id),
            _ => None,
        }
    }

    /// Does the caller hold this role? The name is bare, with no `role:` prefix.
    pub fn has_role(&self, name: &str) -> bool {
        match self {
            AclScope::User { roles, .. } => roles.iter().any(|r| r == name),
            _ => false,
        }
    }

    /// Upstream's `aclGroup`: `['*']`, then every role as `role:<name>`, then the user's objectId
    /// (`RestWrite.js:184`, `RestQuery.js:427`, both `['*'].concat(roles, [user.id])`).
    ///
    /// Master is the empty list, because upstream never reaches a caller that consumes an
    /// `aclGroup` without first branching on `isMaster`.
    ///
    /// Order matters twice over. `addPointerPermissions` extracts the single user id by filtering
    /// out `role:` and `*` (`DatabaseController.js:1745-1747`), and the compiled `$in` array is
    /// snapshot-compared.
    pub fn acl_group(&self) -> Vec<String> {
        match self {
            AclScope::Unrestricted => Vec::new(),
            AclScope::Anonymous => vec!["*".to_string()],
            AclScope::User { object_id, roles } => {
                let mut out = Vec::with_capacity(roles.len() + 2);
                out.push("*".to_string());
                out.extend(roles.iter().map(|r| format!("role:{r}")));
                // Defensive, and unreachable through `AclScope::user`. A `role:`-prefixed
                // objectId that arrived by literal construction is dropped rather than emitted,
                // which costs the caller access to their own rows and grants nothing.
                if !object_id.starts_with("role:") {
                    out.push(object_id.clone());
                }
                out
            }
        }
    }

    /// The principals this caller matches in an `_rperm`/`_wperm` lookup.
    ///
    /// `null` first, then optionally a literal `'*'`, then the `aclGroup`
    /// (`DatabaseController.js:81`, `:88`).
    fn principals(&self, seed_public: bool) -> Vec<ParseValue> {
        // `null` matches a row with no permission column, i.e. a public row.
        let mut out = vec![ParseValue::Null];
        if seed_public {
            out.push(ParseValue::String("*".to_string()));
        }
        out.extend(self.acl_group().into_iter().map(ParseValue::String));
        out
    }

    /// The constraint to add to a read.
    ///
    /// `None` for [`AclScope::Unrestricted`], which is the only case where no constraint is
    /// applied. Returning `Option` makes the master case explicit at every call site instead of
    /// being the absence of a step.
    ///
    /// UPSTREAM-QUIRK: the emitted list carries `'*'` twice, once seeded by `addReadACL`
    /// (`DatabaseController.js:88`) and once already present in the `aclGroup`
    /// (`RestQuery.js:427`). A duplicate in an `$in` changes nothing, and removing it would make
    /// the compiled query differ from upstream's for no gain.
    pub fn read_constraint(&self) -> Option<Constraint> {
        match self {
            AclScope::Unrestricted => None,
            _ => Some(Constraint {
                field: "_rperm".to_string(),
                comparison: Comparison::In(self.principals(true)),
            }),
        }
    }

    /// The constraint to add to a write.
    ///
    /// Note the asymmetry with reads: `addWriteACL` omits `'*'` from the injected list, because
    /// `getUserAndRoleACL` already seeds it for every non-master caller. Reproduced rather than
    /// unified, since the two functions are not symmetric upstream and a caller path that builds
    /// its own list would behave differently.
    pub fn write_constraint(&self) -> Option<Constraint> {
        match self {
            AclScope::Unrestricted => None,
            _ => Some(Constraint {
                field: "_wperm".to_string(),
                comparison: Comparison::In(self.principals(false)),
            }),
        }
    }
}

/// Resolve a class's declared default ACL into the value a create should carry.
///
/// `RestWrite.js:385-391`. The declared block is copied, and if it names `currentUser` then the
/// caller's objectId gains a copy of that entry and the `currentUser` key is removed.
///
/// Three details are load-bearing and each fails silently if it is got wrong.
///
/// **`currentUser` is resolved, never stored.** An ACL containing the literal string
/// `currentUser` as a principal matches nobody, so the row is unreadable by everyone including
/// the user it was meant for, and the configuration reads as though it worked.
///
/// **An anonymous caller loses the entry rather than keeping it.** Upstream's `delete` is outside
/// the `if (this.auth.user?.id)` guard, so with no caller there is no substitute id and the key
/// simply goes. A class whose only declared entry is `currentUser` therefore produces an ACL with
/// no entries at all for an anonymous create, which is a row only master can read. That is
/// upstream's behavior and it is the restrictive direction.
///
/// **Key order is preserved, and it is not upstream's order.** The `_rperm` and `_wperm` arrays
/// are built by walking the ACL, so their element order comes from here, and a mixed fleet compares
/// stored rows. Substituting the caller's id in place of `currentUser` rather than appending would
/// reorder them, so the substitution appends as upstream's assignment does.
///
/// That is where the resemblance stops. **Upstream enumerates a JavaScript object, so an
/// integer-like key sorts ahead of every string key regardless of insertion order**, and an
/// objectId of `1234567890` is integer-like. parse-rust preserves wire order throughout, so the
/// stored arrays differ for any ACL naming such a principal. Measured, and not fixed here: it has
/// no authorization consequence and the fix belongs in `lower_acl` with the array case.
pub fn default_acl_for_create(declared: &ParseValue, caller: Option<&str>) -> ParseValue {
    let ParseValue::Object(map) = declared else {
        // A truthy non-object is assigned verbatim upstream and lowered by the same rule any
        // client-supplied non-object ACL is: two empty columns, a master-only row.
        return declared.clone();
    };
    let mut acl = map.clone();
    let Some(current_user) = acl.get("currentUser").cloned() else {
        return ParseValue::Object(acl);
    };
    // `if (acl.currentUser)`: a falsy entry is left in place and not resolved, because upstream's
    // guard is truthiness rather than presence.
    if !parse_rust_core::is_js_truthy(&current_user) {
        return ParseValue::Object(acl);
    }
    if let Some(caller) = caller {
        acl.insert(caller.to_string(), current_user);
    }
    acl.shift_remove("currentUser");
    ParseValue::Object(acl)
}

/// Split an `ACL` field out of a row into the two storage columns.
///
/// Returns the row with `ACL` removed and the columns added. A row with no `ACL` gets no columns,
/// which is what makes it public.
///
/// **The test upstream applies is falsiness, not "is it an object"** (`DatabaseController.js:94-96`,
/// literally `if (!ACL) return result`). Everything truthy falls through to a `for...in` that reads
/// `.read` and `.write` off each entry, so a string, a number or an array yields no principals but
/// **still writes both columns as empty arrays**, which is a master-only row. Skipping the columns
/// instead writes a row with no `_rperm`/`_wperm` at all, and an absent column is public.
///
/// Getting this wrong is not a cosmetic divergence. Nothing type-checks `ACL` on either side, by
/// design (`SchemaController.js:1312-1315`), so `{"ACL":"x"}` reaches here from any client. The
/// consequential class is `_Role`: its required-column check tests presence and truthiness only, so
/// a non-object `ACL` would satisfy it and then produce a world-writable role that any caller can
/// add itself to.
///
/// The update path applies the same test, in `lower_acl_into_update`. It did not until a review:
/// it tested for `null` alone, so `false`, `0` and `""` fell through and cleared both columns on a
/// row that already had permissions. Both paths now branch on truthiness, and the tests on each
/// side loop over the falsy values rather than checking one, because checking one is what let the
/// other three through.
pub fn lower_acl(mut row: ParseMap) -> ParseMap {
    let Some(acl_value) = row.shift_remove("ACL") else {
        return row;
    };
    if !parse_rust_core::is_js_truthy(&acl_value) {
        return row;
    }
    // `None` here is a truthy non-object, which upstream's loop walks and takes nothing from.
    let acl = acl_from_value(&acl_value).unwrap_or_default();
    let (rperm, wperm) = acl.to_perms();
    row.insert(
        "_rperm".to_string(),
        ParseValue::Array(rperm.into_iter().map(ParseValue::String).collect()),
    );
    row.insert(
        "_wperm".to_string(),
        ParseValue::Array(wperm.into_iter().map(ParseValue::String).collect()),
    );
    row
}

/// Rebuild the `ACL` field from the two storage columns, then drop them.
///
/// Reproduces `untransformObjectACL` exactly, including that both columns absent produce **no
/// `ACL` key at all** rather than `null` or `{}`.
pub fn raise_acl(mut row: ParseMap) -> ParseMap {
    let rperm = take_string_array(&mut row, "_rperm");
    let wperm = take_string_array(&mut row, "_wperm");

    let Some(acl) = Acl::from_perms(rperm.as_deref(), wperm.as_deref()) else {
        return row;
    };

    let mut map = ParseMap::new();
    for (principal, perms) in acl.iter() {
        if perms.is_empty() {
            continue;
        }
        let mut entry = ParseMap::new();
        // Only true flags are emitted. UPSTREAM-QUIRK, see `parse_rust_core::acl`.
        if perms.read {
            entry.insert("read".to_string(), ParseValue::Bool(true));
        }
        if perms.write {
            entry.insert("write".to_string(), ParseValue::Bool(true));
        }
        map.insert(principal.as_key(), ParseValue::Object(entry));
    }
    row.insert("ACL".to_string(), ParseValue::Object(map));
    row
}

fn take_string_array(row: &mut ParseMap, key: &str) -> Option<Vec<String>> {
    match row.shift_remove(key) {
        Some(ParseValue::Array(items)) => Some(
            items
                .into_iter()
                .filter_map(|v| match v {
                    ParseValue::String(s) => Some(s),
                    _ => None,
                })
                .collect(),
        ),
        _ => None,
    }
}

/// Read a client-supplied `ACL` value.
fn acl_from_value(value: &ParseValue) -> Option<Acl> {
    let ParseValue::Object(map) = value else {
        return None;
    };
    let mut acl = Acl::new();
    for (key, entry) in map {
        let ParseValue::Object(flags) = entry else {
            continue;
        };
        let flag = |name: &str| matches!(flags.get(name), Some(ParseValue::Bool(true)));
        acl.set(
            Principal::parse(key),
            Permissions {
                read: flag("read"),
                write: flag("write"),
            },
        );
    }
    Some(acl)
}

#[cfg(test)]
mod tests {
    use super::*;

    fn row(pairs: Vec<(&str, ParseValue)>) -> ParseMap {
        let mut m = ParseMap::new();
        for (k, v) in pairs {
            m.insert(k.to_string(), v);
        }
        m
    }

    /// The single most important assertion in this module.
    #[test]
    fn the_read_constraint_includes_null_so_public_rows_stay_visible() {
        let c = AclScope::Anonymous
            .read_constraint()
            .expect("anonymous is constrained");
        assert_eq!(c.field, "_rperm");
        match c.comparison {
            Comparison::In(values) => {
                assert!(
                    values.iter().any(|v| matches!(v, ParseValue::Null)),
                    "null must be in the list, or every row saved without an ACL becomes invisible"
                );
                assert!(values
                    .iter()
                    .any(|v| matches!(v, ParseValue::String(s) if s == "*")));
            }
            other => panic!("expected In, got {other:?}"),
        }
    }

    /// **The array case that is a known divergence, pinned so the eventual fix is visible.**
    ///
    /// The test above uses `[]` and `["*"]`, neither of which carries a permission, so it passes
    /// whether or not arrays are enumerated. `[{"read":true}]` is the case that separates the two:
    /// upstream's `for...in` grants principal `"0"`, because an array index is a property name, and
    /// `lower_acl` reads principals from a map only, so it writes two empty columns and grants
    /// nobody. Measured at the pin, on signup, on a `_User` update and as a CLP-declared default.
    ///
    /// This asserts today's behavior rather than upstream's. When `lower_acl` learns to enumerate
    /// an array, this test fails and is the reminder to move the divergence row with it.
    #[test]
    fn a_permission_bearing_array_currently_grants_nobody() {
        let mut entry = ParseMap::new();
        entry.insert("read".into(), ParseValue::Bool(true));
        let lowered = lower_acl(row(vec![(
            "ACL",
            ParseValue::Array(vec![ParseValue::Object(entry)]),
        )]));
        for column in ["_rperm", "_wperm"] {
            assert!(
                matches!(lowered.get(column), Some(ParseValue::Array(a)) if a.is_empty()),
                "upstream grants principal \"0\" here; parse-rust grants nobody, and the row \
                 records it. Got {:?} for {column}",
                lowered.get(column)
            );
        }
    }

    /// The falsy-versus-not-an-object distinction, which is the whole of `lower_acl`'s contract.
    ///
    /// A truthy non-object must produce two empty arrays, which is a master-only row. Producing no
    /// columns instead is a public row, and on `_Role` that is a world-writable role any caller can
    /// add itself to, because the required-column check tests truthiness and stops there.
    #[test]
    fn a_truthy_non_object_acl_writes_empty_columns_rather_than_none() {
        for value in [
            ParseValue::String("x".into()),
            ParseValue::Number(1.0),
            ParseValue::Array(vec![]),
            ParseValue::Array(vec![ParseValue::String("*".into())]),
            ParseValue::Bool(true),
            ParseValue::Object(ParseMap::new()),
        ] {
            let lowered = lower_acl(row(vec![("ACL", value.clone())]));
            for column in ["_rperm", "_wperm"] {
                assert!(
                    matches!(lowered.get(column), Some(ParseValue::Array(a)) if a.is_empty()),
                    "a truthy ACL must write an empty {column}, got {:?} for {value:?}",
                    lowered.get(column)
                );
            }
            assert!(!lowered.contains_key("ACL"));
        }
    }

    /// The other half. Falsy means no columns, which is a public row, and that is upstream's
    /// `if (!ACL) return result`.
    #[test]
    fn a_falsy_or_absent_acl_writes_no_columns() {
        for value in [
            ParseValue::Null,
            ParseValue::Bool(false),
            ParseValue::Number(0.0),
            ParseValue::String(String::new()),
        ] {
            let lowered = lower_acl(row(vec![("ACL", value.clone())]));
            assert!(!lowered.contains_key("_rperm"), "falsy ACL: {value:?}");
            assert!(!lowered.contains_key("_wperm"), "falsy ACL: {value:?}");
        }
        let untouched = lower_acl(row(vec![("title", ParseValue::String("x".into()))]));
        assert!(!untouched.contains_key("_rperm"));
        assert!(!untouched.contains_key("_wperm"));
    }

    #[test]
    fn master_applies_no_constraint_at_all() {
        assert!(AclScope::Unrestricted.read_constraint().is_none());
        assert!(AclScope::Unrestricted.write_constraint().is_none());
    }

    #[test]
    fn a_user_scope_carries_its_object_id() {
        let c = AclScope::user("u1", vec![])
            .expect("plain id")
            .read_constraint()
            .expect("constrained");
        match c.comparison {
            Comparison::In(values) => assert!(values
                .iter()
                .any(|v| matches!(v, ParseValue::String(s) if s == "u1"))),
            other => panic!("expected In, got {other:?}"),
        }
    }

    /// 0.1.0 emitted no `role:` entry at all, so a `role:Admins` entry in `_rperm` matched
    /// nobody and every role-protected row was invisible to its own members.
    #[test]
    fn a_role_entry_now_matches() {
        let scope = AclScope::user("u1", vec!["Admins".into(), "Editors".into()]).expect("scope");
        let c = scope.read_constraint().expect("constrained");
        match c.comparison {
            Comparison::In(values) => {
                let strings: Vec<&str> = values
                    .iter()
                    .filter_map(|v| match v {
                        ParseValue::String(s) => Some(s.as_str()),
                        _ => None,
                    })
                    .collect();
                assert!(strings.contains(&"role:Admins"), "{strings:?}");
                assert!(strings.contains(&"role:Editors"), "{strings:?}");
                assert!(strings.contains(&"u1"));
            }
            other => panic!("expected In, got {other:?}"),
        }
    }

    /// Upstream order: `['*']`, then roles, then the user id. `addPointerPermissions` recovers
    /// the single user id by filtering the first two out, so the shape is load bearing.
    #[test]
    fn the_acl_group_is_star_then_roles_then_the_user() {
        let scope = AclScope::user("u1", vec!["A".into()]).expect("scope");
        assert_eq!(scope.acl_group(), vec!["*", "role:A", "u1"]);
        assert_eq!(AclScope::Anonymous.acl_group(), vec!["*"]);
        assert!(AclScope::Unrestricted.acl_group().is_empty());
    }

    #[test]
    fn accessors_answer_for_every_variant() {
        let scope = AclScope::user("u1", vec!["A".into()]).expect("scope");
        assert!(scope.has_role("A"));
        assert!(!scope.has_role("role:A"), "roles are stored bare");
        assert!(!scope.has_role("B"));
        assert_eq!(scope.user_id(), Some("u1"));
        assert!(!scope.is_master());
        assert!(AclScope::Unrestricted.is_master());
        assert_eq!(AclScope::Anonymous.user_id(), None);
        assert!(!AclScope::Anonymous.has_role("A"));
    }

    /// The `role:` objectId collision, guarded at the one place a scope can be built.
    #[test]
    fn a_role_prefixed_object_id_is_refused() {
        let e = AclScope::user("role:Admins", vec![]).unwrap_err();
        assert_eq!(e.code, parse_rust_core::ErrorCode::InternalServerError);
        assert_eq!(e.message, "Invalid object ID.");
    }

    #[test]
    fn acl_lowers_to_two_columns_and_raises_back() {
        let mut acl_map = ParseMap::new();
        let mut public = ParseMap::new();
        public.insert("read".into(), ParseValue::Bool(true));
        acl_map.insert("*".into(), ParseValue::Object(public));
        let mut owner = ParseMap::new();
        owner.insert("read".into(), ParseValue::Bool(true));
        owner.insert("write".into(), ParseValue::Bool(true));
        acl_map.insert("u1".into(), ParseValue::Object(owner));

        let lowered = lower_acl(row(vec![
            ("title", ParseValue::String("x".into())),
            ("ACL", ParseValue::Object(acl_map)),
        ]));
        assert!(
            lowered.get("ACL").is_none(),
            "ACL must not be stored as a field"
        );
        assert!(matches!(lowered.get("_rperm"), Some(ParseValue::Array(a)) if a.len() == 2));
        assert!(matches!(lowered.get("_wperm"), Some(ParseValue::Array(a)) if a.len() == 1));

        let raised = raise_acl(lowered);
        assert!(raised.get("_rperm").is_none() && raised.get("_wperm").is_none());
        let ParseValue::Object(acl) = raised.get("ACL").expect("ACL restored") else {
            panic!("ACL should be an object");
        };
        assert!(acl.contains_key("*") && acl.contains_key("u1"));
    }

    /// UPSTREAM-QUIRK, reproduced end to end.
    #[test]
    fn a_false_flag_disappears_on_the_round_trip() {
        let mut entry = ParseMap::new();
        entry.insert("read".into(), ParseValue::Bool(true));
        entry.insert("write".into(), ParseValue::Bool(false));
        let mut acl_map = ParseMap::new();
        acl_map.insert("*".into(), ParseValue::Object(entry));

        let raised = raise_acl(lower_acl(row(vec![("ACL", ParseValue::Object(acl_map))])));
        let ParseValue::Object(acl) = raised.get("ACL").expect("ACL") else {
            panic!()
        };
        let ParseValue::Object(star) = acl.get("*").expect("*") else {
            panic!()
        };
        assert!(star.contains_key("read"));
        assert!(
            !star.contains_key("write"),
            "the false key is dropped, matching untransformObjectACL"
        );
    }

    // -----------------------------------------------------------------------------------------
    // The CLP-declared default ACL
    // -----------------------------------------------------------------------------------------

    fn declared(json: &str) -> ParseValue {
        parse_rust_core::decode::classify(
            serde_json::from_str(json).expect("test literal must be valid JSON"),
        )
        .expect("classify")
    }

    /// What the resolved ACL looks like on the wire, after a round trip through the two columns.
    /// Asserted this way rather than on the intermediate map, because the columns are what the
    /// row actually carries and an entry that survives resolution but not lowering grants nothing.
    fn principals(acl: ParseValue) -> (Vec<String>, Vec<String>) {
        let mut carrier = ParseMap::new();
        carrier.insert("ACL".to_string(), acl);
        let lowered = lower_acl(carrier);
        let read = take_string_array(&mut lowered.clone(), "_rperm").unwrap_or_default();
        let write = take_string_array(&mut lowered.clone(), "_wperm").unwrap_or_default();
        (read, write)
    }

    /// The headline case, and the one the release is named for: a class declared private, an
    /// object created by user A, and the caller's own id in both columns.
    #[test]
    fn current_user_resolves_to_the_callers_object_id() {
        let acl = default_acl_for_create(
            &declared(r#"{"currentUser":{"read":true,"write":true}}"#),
            Some("userA"),
        );
        let ParseValue::Object(map) = &acl else {
            panic!("expected an object")
        };
        assert!(
            !map.contains_key("currentUser"),
            "the literal key matches nobody and must not be stored"
        );
        assert_eq!(
            principals(acl),
            (vec!["userA".to_string()], vec!["userA".to_string()])
        );
    }

    /// **Both columns, not just `_rperm`.** They are written separately, so an implementation that
    /// resolved the read entry and dropped the write one would hide the row from user B and pass a
    /// read-only test while leaving it writable by everybody.
    #[test]
    fn a_read_only_declaration_produces_a_read_only_row() {
        let acl = default_acl_for_create(&declared(r#"{"currentUser":{"read":true}}"#), Some("u1"));
        assert_eq!(principals(acl), (vec!["u1".to_string()], Vec::new()));
    }

    /// With no caller there is no substitute id, and upstream's `delete` is outside the guard, so
    /// the entry goes. A class whose only declared entry is `currentUser` therefore yields an ACL
    /// with no principals at all for an anonymous create: readable by master and nobody else.
    #[test]
    fn an_anonymous_create_loses_the_current_user_entry_rather_than_keeping_it() {
        let acl = default_acl_for_create(
            &declared(r#"{"currentUser":{"read":true,"write":true}}"#),
            None,
        );
        let ParseValue::Object(map) = &acl else {
            panic!("expected an object")
        };
        assert!(map.is_empty(), "the literal key must not survive: {map:?}");
        assert_eq!(principals(acl), (Vec::new(), Vec::new()));
    }

    /// Entries other than `currentUser` are carried through untouched, and the caller's own entry
    /// is appended after them, which is where a JS property assignment puts a new key. The column
    /// order is observable in a stored row.
    #[test]
    fn other_entries_survive_and_the_caller_is_appended() {
        let acl = default_acl_for_create(
            &declared(
                r#"{"role:Admins":{"read":true,"write":true},"currentUser":{"read":true},"*":{"read":true}}"#,
            ),
            Some("u1"),
        );
        let (read, write) = principals(acl);
        assert_eq!(read, vec!["role:Admins", "*", "u1"]);
        assert_eq!(write, vec!["role:Admins"]);
    }

    /// An id already present is updated in place rather than moved to the end, which is what a
    /// JavaScript assignment to an existing key does.
    #[test]
    fn a_caller_already_named_keeps_its_position() {
        let acl = default_acl_for_create(
            &declared(
                r#"{"u1":{"read":true},"*":{"read":true},"currentUser":{"read":true,"write":true}}"#,
            ),
            Some("u1"),
        );
        let (read, write) = principals(acl);
        assert_eq!(read, vec!["u1", "*"]);
        assert_eq!(write, vec!["u1"], "the currentUser entry replaced it");
    }

    /// `if (acl.currentUser)` is truthiness, so a falsy entry is neither resolved nor deleted.
    #[test]
    fn a_falsy_current_user_entry_is_left_alone() {
        let acl = default_acl_for_create(&declared(r#"{"currentUser":null}"#), Some("u1"));
        let ParseValue::Object(map) = &acl else {
            panic!("expected an object")
        };
        assert!(map.contains_key("currentUser"));
        assert!(!map.contains_key("u1"));
    }

    /// A truthy non-object is assigned verbatim and lowered like any other truthy non-object ACL:
    /// two empty columns, which is a master-only row rather than a public one.
    #[test]
    fn a_truthy_non_object_declaration_yields_a_master_only_row() {
        let acl = default_acl_for_create(&declared(r#""nonsense""#), Some("u1"));
        assert!(matches!(&acl, ParseValue::String(s) if s == "nonsense"));
        let mut carrier = ParseMap::new();
        carrier.insert("ACL".to_string(), acl);
        let lowered = lower_acl(carrier);
        for column in ["_rperm", "_wperm"] {
            assert!(matches!(lowered.get(column), Some(ParseValue::Array(a)) if a.is_empty()));
        }
    }

    #[test]
    fn a_row_with_no_acl_gets_no_columns_and_no_acl_key_back() {
        let lowered = lower_acl(row(vec![("title", ParseValue::String("x".into()))]));
        assert!(lowered.get("_rperm").is_none());
        let raised = raise_acl(lowered);
        assert!(
            raised.get("ACL").is_none(),
            "absent columns produce no ACL key at all, not null and not an empty object"
        );
    }
}