1use std::collections::{BTreeMap, BTreeSet};
51use std::sync::Arc;
52
53use cheers_core::{
54 Error, Lease, PrincipalId, ResourceRevocationKey, Revoked, SchemaRegistry, SetSnapshot, SnapshotMember,
55 StoreError, Subject,
56};
57
58use crate::codec::PasetoV4SecretMinter;
59use crate::ownership::{OwnershipRow, OwnershipStore, OwnershipTuples};
60use crate::revocation::RevocationWriter;
61
62pub async fn revoke_ownership<O, W>(store: &O, revocations: &W, id: &str, now: i64) -> Result<u64, StoreError>
87where
88 O: OwnershipStore + ?Sized,
89 W: RevocationWriter + ?Sized,
90{
91 let row = store.get(id).await?.ok_or(StoreError::NotFound)?;
92 let version = store.revoke_by_id(id, now).await?;
93 let Subject::Principal(principal) = &row.subject else {
94 return Ok(version);
95 };
96 if !principal.kind.is_member_kind() {
97 return Ok(version);
98 }
99 let still_direct = store
100 .list_for_resource(&row.resource_kind, &row.resource_id)
101 .await?
102 .iter()
103 .any(|r| !r.is_revoked() && r.subject.principal() == Some(principal));
104 if !still_direct {
105 let entry = Revoked::membership(&row.resource_kind, &row.resource_id, principal.clone(), version);
106 revocations.revoke(&entry).await?;
107 }
108 Ok(version)
109}
110
111pub async fn revoke_principal_ownership<O, W>(
135 store: &O,
136 revocations: &W,
137 principal: &PrincipalId,
138 now: i64,
139) -> Result<u64, StoreError>
140where
141 O: OwnershipStore + ?Sized,
142 W: RevocationWriter + ?Sized,
143{
144 let version = store.revoke_by_principal(principal, now).await?;
145 if !principal.kind.is_member_kind() {
146 return Ok(version);
147 }
148 let resource = |r: OwnershipRow| (r.resource_kind, r.resource_id);
149 let live: BTreeSet<_> = store.list_for_principal(principal).await?.into_iter().map(resource).collect();
150 let held: BTreeSet<_> = store.list_history_for_principal(principal).await?.into_iter().map(resource).collect();
151 for (kind, id) in held.difference(&live) {
152 revocations.revoke(&Revoked::membership(kind.as_str(), id.as_str(), principal.clone(), version)).await?;
153 }
154 Ok(version)
155}
156
157#[derive(Debug, Clone, PartialEq, Eq)]
159pub struct SignedSetSnapshot {
160 pub snapshot: SetSnapshot,
161 pub token: String,
164}
165
166pub struct SnapshotIssuer {
173 store: Arc<dyn OwnershipStore>,
174 schema: Arc<SchemaRegistry>,
175 minter: PasetoV4SecretMinter,
176 issuer: String,
177 kid: String,
178 refresh_after_seconds: i64,
179}
180
181impl std::fmt::Debug for SnapshotIssuer {
182 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
183 f.debug_struct("SnapshotIssuer")
184 .field("issuer", &self.issuer)
185 .field("kid", &self.kid)
186 .field("refresh_after_seconds", &self.refresh_after_seconds)
187 .finish_non_exhaustive()
188 }
189}
190
191impl SnapshotIssuer {
192 pub const DEFAULT_REFRESH_AFTER_SECONDS: i64 = 7 * 24 * 60 * 60;
196
197 pub const MINT_ATTEMPTS: usize = 5;
200
201 pub fn new(
202 store: impl OwnershipStore + 'static,
203 schema: Arc<SchemaRegistry>,
204 minter: PasetoV4SecretMinter,
205 issuer: impl Into<String>,
206 kid: impl Into<String>,
207 ) -> Self {
208 Self {
209 store: Arc::new(store),
210 schema,
211 minter,
212 issuer: issuer.into(),
213 kid: kid.into(),
214 refresh_after_seconds: Self::DEFAULT_REFRESH_AFTER_SECONDS,
215 }
216 }
217
218 pub fn with_refresh_after(mut self, seconds: i64) -> Self {
219 self.refresh_after_seconds = seconds;
220 self
221 }
222
223 pub fn issuer(&self) -> &str {
224 &self.issuer
225 }
226
227 async fn member_leases(&self, kind: &str, id: &str, now: i64) -> Result<BTreeMap<PrincipalId, Option<Lease>>, Error> {
231 let mut out: BTreeMap<PrincipalId, Option<Lease>> = BTreeMap::new();
232 for row in self.store.list_for_resource(kind, id).await? {
233 let Some(p) = row.subject.principal().cloned() else { continue };
234 if !row.is_live_at(now) {
235 continue;
236 }
237 let lease = row.lease.map(|l| l.lease);
238 out.entry(p)
239 .and_modify(|held| {
240 *held = match (*held, lease) {
241 (Some(a), Some(b)) => Some(if b.exp() > a.exp() { b } else { a }),
242 _ => None,
243 }
244 })
245 .or_insert(lease);
246 }
247 Ok(out)
248 }
249
250 pub async fn mint(&self, kind: &str, id: &str, now: i64) -> Result<SignedSetSnapshot, Error> {
253 for _ in 0..Self::MINT_ATTEMPTS {
254 let before = self.store.current_version().await?;
255 let members = self.schema.members(&OwnershipTuples::at(&*self.store, now), kind, id).await?;
256 let policy = self.store.admission_policy(kind, id).await?;
258 let policy = policy.filter(|p| self.schema.validate_admission_policy(kind, p).is_ok());
261 if self.store.current_version().await? != before {
262 continue;
263 }
264 let leases = self.member_leases(kind, id, now).await?;
265 let here = [(kind.to_owned(), id.to_owned())];
266 let members: Vec<SnapshotMember> = members
267 .into_iter()
268 .filter(|m| m.principal.kind.is_member_kind())
269 .filter(|m| self.schema.is_membership(kind, &m.relation))
274 .map(|m| {
275 let via: Vec<(String, String)> = m.via.into_iter().collect();
276 let lease = if via[..] == here[..] { leases.get(&m.principal).copied().flatten() } else { None };
280 SnapshotMember { principal: m.principal, relation: m.relation, via, lease }
281 })
282 .collect();
283 let mut resources: BTreeSet<(String, String)> = members.iter().flat_map(|m| m.via.iter().cloned()).collect();
286 resources.insert((kind.to_owned(), id.to_owned()));
287 let mut keys = Vec::with_capacity(resources.len());
288 for (k, i) in resources {
289 let key = self.store.revocation_key(&k, &i).await?;
290 keys.push(ResourceRevocationKey { kind: k, id: i, key });
291 }
292 let lease = Lease::new(now, now.saturating_add(self.refresh_after_seconds), None)?;
293 let snapshot = SetSnapshot::new(&self.issuer, kind, id, before, members, keys, now, lease).with_policy(policy);
294 let token = self.minter.mint_artifact(&snapshot, &self.kid)?;
295 return Ok(SignedSetSnapshot { snapshot, token });
296 }
297 Err(Error::SnapshotContended {
298 kind: kind.to_owned(),
299 id: id.to_owned(),
300 attempts: Self::MINT_ATTEMPTS,
301 })
302 }
303}
304
305#[cfg(test)]
306mod tests {
307 use super::*;
308 use cheers_core::LeaseState;
309 use crate::ownership::{Inserted, MemoryOwnershipStore, NewOwnership};
310 use crate::revocation::{MemoryRevocationStore, RevocationPublisher, RevocationSnapshot};
311 use async_trait::async_trait;
312 use cheers_core::{RelationDef, ResourceSchema, ScopeRegistry};
313 use cheers_verify::{IssuerTrust, ReplicatedRevocations, SnapshotVerifier, VerifiedSnapshot};
314 use std::sync::atomic::{AtomicUsize, Ordering};
315
316 const ISS: &str = "https://c.test";
317
318 const NAMESPACE: ResourceSchema = ResourceSchema {
320 kind: "namespace",
321 relations: &[
322 RelationDef {
323 name: "owner",
324 membership: true,
325 implies: &["admin"],
326 scopes: &[],
327 grants: &["owner", "admin", "publisher", "member", "guest"],
328 },
329 RelationDef {
330 name: "admin",
331 membership: true,
332 implies: &["publisher"],
333 scopes: &[],
334 grants: &["admin", "publisher", "member", "guest"],
335 },
336 RelationDef { name: "publisher", membership: true, implies: &["member"], scopes: &[], grants: &[] },
337 RelationDef { name: "member", membership: true, implies: &["guest"], scopes: &[], grants: &[] },
338 RelationDef { name: "guest", membership: true, implies: &[], scopes: &[], grants: &[] },
339 RelationDef { name: "allow", membership: false, implies: &[], scopes: &[], grants: &[] },
342 RelationDef { name: "steward", membership: false, implies: &["guest"], scopes: &[], grants: &[] },
343 ],
344 kind_relations: &[],
345 };
346
347 fn schema() -> Arc<SchemaRegistry> {
348 let scopes = ScopeRegistry::builder().build().unwrap();
349 Arc::new(SchemaRegistry::build(&[NAMESPACE], &scopes).unwrap())
350 }
351
352 fn ns(who: impl Into<Subject>, id: &str, rel: &str) -> NewOwnership {
353 NewOwnership::new(who, "namespace", id, rel, PrincipalId::service("noisetable-account"), None).unwrap()
354 }
355
356 fn user(id: &str) -> PrincipalId {
357 PrincipalId::user(id)
358 }
359
360 async fn institution(store: &MemoryOwnershipStore) {
363 store.insert(&ns(Subject::set("namespace", "museum", "admin"), "ed", "admin"), 10).await.unwrap();
364 store.insert(&ns(Subject::set("namespace", "museum", "member"), "ed", "guest"), 10).await.unwrap();
365 }
366
367 struct Rig {
368 store: MemoryOwnershipStore,
369 revocations: MemoryRevocationStore,
370 issuer: SnapshotIssuer,
371 publisher: RevocationPublisher<MemoryRevocationStore>,
372 replica: Arc<ReplicatedRevocations>,
373 edge: SnapshotVerifier<Arc<ReplicatedRevocations>>,
374 }
375
376 impl Rig {
377 async fn gossip(&self, now: i64) {
379 let set = self.publisher.current(now).await.unwrap();
380 self.replica.adopt(&set.token).await.unwrap();
381 }
382 }
383
384 fn rig() -> Rig {
385 let (minter, verifier) = PasetoV4SecretMinter::generate().unwrap();
386 let store = MemoryOwnershipStore::new();
387 let revocations = MemoryRevocationStore::new();
388 let key: &[u8; 64] = minter.secret_key_bytes().try_into().unwrap();
389 let same_key = PasetoV4SecretMinter::from_secret_key(key).unwrap();
390 let issuer = SnapshotIssuer::new(store.clone(), schema(), minter, ISS, "k1").with_refresh_after(60);
391 let publisher = RevocationPublisher::new(revocations.clone(), Arc::new(store.clone()), same_key, ISS, "k1");
392 let trust = IssuerTrust::pinned(ISS, verifier);
393 let replica = Arc::new(ReplicatedRevocations::new(trust.clone()));
394 Rig {
395 edge: SnapshotVerifier::new(trust, replica.clone()),
396 store,
397 revocations,
398 issuer,
399 publisher,
400 replica,
401 }
402 }
403
404 fn admitted(v: &VerifiedSnapshot) -> Vec<String> {
405 v.admitted.iter().map(|m| format!("{}#{}", m.principal.id, m.relation)).collect()
406 }
407
408 #[test]
409 fn mint_lists_users_closure_expanded_with_via_at_the_store_version() {
410 pollster::block_on(async {
411 let r = rig();
412 institution(&r.store).await;
413 r.store.insert(&ns(user("alice"), "museum", "member"), 100).await.unwrap();
414 r.store.insert(&ns(user("carol"), "ed", "admin"), 100).await.unwrap();
415 r.store.insert(&ns(PrincipalId::service("bot"), "ed", "member"), 100).await.unwrap();
417 let version = r.store.current_version().await.unwrap();
418
419 let s = r.issuer.mint("namespace", "ed", 1_000).await.unwrap();
420 assert_eq!(s.snapshot.epoch, version);
421 assert_eq!((s.snapshot.iat, s.snapshot.lease.refresh_after()), (1_000, 1_060));
422 let listed: Vec<_> = s
423 .snapshot
424 .members
425 .iter()
426 .map(|m| (m.principal.id.as_str(), m.relation.as_str(), m.via.clone()))
427 .collect();
428 let ed = || vec![("namespace".to_owned(), "ed".to_owned())];
429 let museum = || vec![("namespace".to_owned(), "museum".to_owned())];
430 assert_eq!(
431 listed,
432 vec![
433 ("alice", "guest", museum()),
434 ("carol", "admin", ed()),
435 ("carol", "guest", ed()),
436 ("carol", "member", ed()),
437 ("carol", "publisher", ed()),
438 ]
439 );
440 let carried: Vec<_> =
442 s.snapshot.revocation_keys.iter().map(|k| (k.kind.as_str(), k.id.as_str(), k.key.clone())).collect();
443 assert_eq!(
444 carried,
445 vec![
446 ("namespace", "ed", r.store.revocation_key("namespace", "ed").await.unwrap()),
447 ("namespace", "museum", r.store.revocation_key("namespace", "museum").await.unwrap()),
448 ]
449 );
450 let ok = r.edge.verify_snapshot_at(&s.token, 1_000).await.unwrap();
451 assert!(ok.holds(&PrincipalId::user("alice"), "guest"));
452 assert!(!ok.holds(&PrincipalId::user("alice"), "member"));
453 assert_eq!(ok.lease, LeaseState::Current);
454 });
455 }
456
457 fn listed(s: &SignedSetSnapshot) -> Vec<String> {
458 s.snapshot.members.iter().map(|m| format!("{}#{}", m.principal.id, m.relation)).collect()
459 }
460
461 #[test]
462 fn a_non_membership_row_is_never_minted() {
463 pollster::block_on(async {
464 let r = rig();
465 r.store.insert(&ns(user("alice"), "ed", "member"), 100).await.unwrap();
466 r.store.insert(&ns(user("bob"), "ed", "allow"), 100).await.unwrap();
467 let s = r.issuer.mint("namespace", "ed", 1_000).await.unwrap();
468 assert_eq!(listed(&s), vec!["alice#guest", "alice#member"]);
469 });
470 }
471
472 #[test]
473 fn a_non_membership_relation_still_mints_the_membership_one_it_implies() {
474 pollster::block_on(async {
475 let r = rig();
476 r.store.insert(&ns(user("sam"), "ed", "steward"), 100).await.unwrap();
477 let s = r.issuer.mint("namespace", "ed", 1_000).await.unwrap();
478 assert_eq!(listed(&s), vec!["sam#guest"]);
479 });
480 }
481
482 #[test]
483 fn a_subject_set_member_at_a_membership_relation_is_minted() {
484 pollster::block_on(async {
485 let r = rig();
486 institution(&r.store).await;
487 r.store.insert(&ns(user("alice"), "museum", "member"), 100).await.unwrap();
488 r.store.insert(&ns(user("bob"), "museum", "allow"), 100).await.unwrap();
490 let s = r.issuer.mint("namespace", "ed", 1_000).await.unwrap();
491 assert_eq!(listed(&s), vec!["alice#guest"]);
492 });
493 }
494
495 #[test]
496 fn a_policy_naming_a_non_membership_relation_is_minted_as_none() {
497 use cheers_core::{AdmissionMode, AdmissionPolicy, PolicyError};
498 let reg = schema();
499 let ok = AdmissionPolicy::new(AdmissionMode::Knock, "guest").with_admitters(["admin"]);
500 assert_eq!(reg.validate_admission_policy("namespace", &ok), Ok(()));
501 let bad_floor = AdmissionPolicy::new(AdmissionMode::Open, "allow");
502 let bad_admitter = AdmissionPolicy::new(AdmissionMode::Knock, "guest").with_admitters(["admin", "steward"]);
503 assert_eq!(
504 reg.validate_admission_policy("namespace", &bad_floor),
505 Err(PolicyError::NotMembership { kind: "namespace".into(), relation: "allow".into() })
506 );
507 assert_eq!(
508 reg.validate_admission_policy("namespace", &bad_admitter),
509 Err(PolicyError::NotMembership { kind: "namespace".into(), relation: "steward".into() })
510 );
511 pollster::block_on(async {
512 let r = rig();
513 for (p, want) in [(&bad_floor, None), (&bad_admitter, None), (&ok, Some(&ok))] {
514 r.store.set_admission_policy("namespace", "ed", Some(p), 1_000).await.unwrap();
515 let s = r.issuer.mint("namespace", "ed", 1_000).await.unwrap();
516 assert_eq!(s.snapshot.policy.as_ref(), want);
517 }
518 });
519 }
520
521 #[test]
522 fn the_policy_row_is_minted_and_its_change_advances_the_epoch() {
523 use cheers_core::{AdmissionMode, AdmissionPolicy};
524 pollster::block_on(async {
525 let r = rig();
526 institution(&r.store).await;
527 let bare = r.issuer.mint("namespace", "ed", 1_000).await.unwrap();
528 assert_eq!(bare.snapshot.policy, None);
529 let open = AdmissionPolicy::new(AdmissionMode::Open, "guest");
530 let v = r.store.set_admission_policy("namespace", "ed", Some(&open), 1_000).await.unwrap();
531 assert!(v > bare.snapshot.epoch);
532 let s = r.issuer.mint("namespace", "ed", 1_000).await.unwrap();
533 assert_eq!((s.snapshot.epoch, s.snapshot.policy.as_ref()), (v, Some(&open)));
534 let ok = r.edge.verify_snapshot_at(&s.token, 1_000).await.unwrap();
535 assert_eq!(ok.admission_policy(), Some(&open));
536 let v2 = r.store.set_admission_policy("namespace", "ed", None, 1_000).await.unwrap();
538 let s = r.issuer.mint("namespace", "ed", 1_000).await.unwrap();
539 assert_eq!((s.snapshot.epoch, s.snapshot.policy), (v2, None));
540 assert!(v2 > v);
541 });
542 }
543
544 #[test]
545 fn a_key_principal_is_minted_held_and_revoked_like_a_user() {
546 pollster::block_on(async {
547 let r = rig();
548 institution(&r.store).await;
549 let k = PrincipalId::from_public_key(&[0x4b; 32]);
550 let tuple = r.store.insert(&ns(k.clone(), "ed", "member"), 100).await.unwrap();
551 let held = r.issuer.mint("namespace", "ed", 1_000).await.unwrap();
552 assert!(held.snapshot.lists(&k, "member"));
553 let ok = r.edge.verify_snapshot_at(&held.token, 1_000).await.unwrap();
554 assert!(ok.holds(&k, "member") && ok.holds(&k, "guest"));
555 assert!(!ok.holds(&PrincipalId::user(k.id.clone()), "member"));
556
557 let v = revoke_ownership(&r.store, &r.revocations, &tuple.row.id, 2_000).await.unwrap();
560 assert!(r.revocations.snapshot().await.unwrap().revoked.contains(&Revoked::membership("namespace", "ed", k.clone(), v)));
561 r.gossip(2_000).await;
562 let ok = r.edge.verify_snapshot_at(&held.token, 2_000).await.unwrap();
563 assert!(!ok.holds(&k, "member"));
564 });
565 }
566
567 #[test]
568 fn removal_from_the_parent_drops_an_inherited_member_and_keeps_a_direct_one() {
569 pollster::block_on(async {
570 let r = rig();
571 institution(&r.store).await;
572 let alice = r.store.insert(&ns(user("alice"), "museum", "member"), 100).await.unwrap();
573 let bob = r.store.insert(&ns(user("bob"), "museum", "member"), 100).await.unwrap();
574 r.store.insert(&ns(user("bob"), "ed", "member"), 100).await.unwrap();
575 let held = r.issuer.mint("namespace", "ed", 1_000).await.unwrap();
576 let ok = r.edge.verify_snapshot_at(&held.token, 1_000).await.unwrap();
577 assert_eq!(admitted(&ok), ["alice#guest", "bob#guest", "bob#member"]);
578
579 let va = revoke_ownership(&r.store, &r.revocations, &alice.row.id, 2_000).await.unwrap();
581 let vb = revoke_ownership(&r.store, &r.revocations, &bob.row.id, 2_000).await.unwrap();
582 assert!(va > held.snapshot.epoch && vb > va);
583 r.gossip(2_000).await;
584
585 let ok = r.edge.verify_snapshot_at(&held.token, 2_000).await.unwrap();
588 assert_eq!(admitted(&ok), ["bob#guest", "bob#member"]);
589 assert_eq!(ok.snapshot.members.len(), 3, "the signed list itself is untouched");
590
591 let fresh = r.issuer.mint("namespace", "ed", 2_000).await.unwrap();
593 assert_eq!(fresh.snapshot.epoch, vb);
594 assert_eq!(admitted(&r.edge.verify_snapshot_at(&fresh.token, 2_000).await.unwrap()), [
595 "bob#guest",
596 "bob#member"
597 ]);
598 assert!(r.edge.verify_snapshot_at(&held.token, 2_000).await.is_err());
599
600 r.store.insert(&ns(user("alice"), "museum", "member"), 3_000).await.unwrap();
603 let readded = r.issuer.mint("namespace", "ed", 3_000).await.unwrap();
604 let ok = r.edge.verify_snapshot_at(&readded.token, 3_000).await.unwrap();
605 assert!(ok.holds(&PrincipalId::user("alice"), "guest"));
606 });
607 }
608
609 #[test]
610 fn recording_fires_only_when_the_last_direct_tuple_goes() {
611 pollster::block_on(async {
612 let r = rig();
613 institution(&r.store).await;
614 let owner = r.store.insert(&ns(user("carol"), "ed", "owner"), 100).await.unwrap();
615 let member = r.store.insert(&ns(user("carol"), "ed", "member"), 100).await.unwrap();
616 let bot = r.store.insert(&ns(PrincipalId::service("bot"), "ed", "member"), 100).await.unwrap();
617 let set_rows = r.store.list_for_subject_set("namespace", "museum").await.unwrap();
618 let epoch = || async { r.revocations.snapshot().await.unwrap() };
619
620 revoke_ownership(&r.store, &r.revocations, &owner.row.id, 200).await.unwrap();
622 assert!(epoch().await.revoked.is_empty());
623 revoke_ownership(&r.store, &r.revocations, &set_rows[0].id, 200).await.unwrap();
625 revoke_ownership(&r.store, &r.revocations, &bot.row.id, 200).await.unwrap();
626 assert!(epoch().await.revoked.is_empty());
627
628 let v = revoke_ownership(&r.store, &r.revocations, &member.row.id, 200).await.unwrap();
630 assert_eq!(epoch().await.revoked, vec![Revoked::membership("namespace", "ed", PrincipalId::user("carol"), v)]);
631
632 assert_eq!(revoke_ownership(&r.store, &r.revocations, &member.row.id, 300).await.unwrap(), v);
635 assert!(matches!(
636 revoke_ownership(&r.store, &r.revocations, "ghost", 300).await,
637 Err(StoreError::NotFound)
638 ));
639 });
640 }
641
642 #[test]
646 fn deleting_a_user_drops_them_from_every_held_snapshot() {
647 pollster::block_on(async {
648 let r = rig();
649 institution(&r.store).await;
650 r.store.insert(&ns(user("dana"), "band", "member"), 100).await.unwrap();
652 r.store.insert(&ns(user("dana"), "museum", "member"), 100).await.unwrap();
653 r.store.insert(&ns(user("erin"), "band", "member"), 100).await.unwrap();
654 r.store.insert(&ns(user("erin"), "museum", "member"), 100).await.unwrap();
655 let band = r.issuer.mint("namespace", "band", 1_000).await.unwrap();
656 let ed = r.issuer.mint("namespace", "ed", 1_000).await.unwrap();
657 for held in [&band, &ed] {
658 let ok = r.edge.verify_snapshot_at(&held.token, 1_000).await.unwrap();
659 assert!(admitted(&ok).iter().any(|m| m.starts_with("dana#")));
660 }
661
662 let v = revoke_principal_ownership(&r.store, &r.revocations, &user("dana"), 2_000).await.unwrap();
663 assert!(r.store.list_for_principal(&user("dana")).await.unwrap().is_empty());
664 let mut recorded = r.revocations.snapshot().await.unwrap().revoked;
665 recorded.sort_by(|a, b| a.identity().cmp(&b.identity()));
666 assert_eq!(
667 recorded,
668 vec![
669 Revoked::membership("namespace", "band", PrincipalId::user("dana"), v),
670 Revoked::membership("namespace", "museum", PrincipalId::user("dana"), v),
671 ]
672 );
673 r.gossip(2_000).await;
674
675 assert_eq!(admitted(&r.edge.verify_snapshot_at(&band.token, 2_000).await.unwrap()), ["erin#guest", "erin#member"]);
677 assert_eq!(admitted(&r.edge.verify_snapshot_at(&ed.token, 2_000).await.unwrap()), ["erin#guest"]);
678
679 r.store.insert(&ns(PrincipalId::service("bot"), "band", "admin"), 3_000).await.unwrap();
681 revoke_principal_ownership(&r.store, &r.revocations, &PrincipalId::service("bot"), 3_000).await.unwrap();
682 assert_eq!(r.revocations.snapshot().await.unwrap().revoked.len(), 2);
683 });
684 }
685
686 struct Flaky {
688 inner: MemoryRevocationStore,
689 failures: AtomicUsize,
690 }
691
692 #[async_trait]
693 impl RevocationWriter for Flaky {
694 async fn revoke(&self, entry: &Revoked) -> Result<(), StoreError> {
695 if self.failures.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| n.checked_sub(1)).is_ok() {
696 return Err(StoreError::Backend("revocation log unavailable".into()));
697 }
698 self.inner.revoke(entry).await
699 }
700 async fn snapshot(&self) -> Result<RevocationSnapshot, StoreError> {
701 self.inner.snapshot().await
702 }
703 }
704
705 #[test]
706 fn a_failed_cascade_recording_is_healed_by_re_running_it() {
707 pollster::block_on(async {
708 let store = MemoryOwnershipStore::new();
709 store.insert(&ns(user("dana"), "band", "member"), 100).await.unwrap();
710 store.insert(&ns(user("dana"), "ed", "admin"), 100).await.unwrap();
711 let log = Flaky { inner: MemoryRevocationStore::new(), failures: AtomicUsize::new(1) };
712
713 assert!(revoke_principal_ownership(&store, &log, &user("dana"), 2_000).await.is_err());
716 assert!(store.list_for_principal(&user("dana")).await.unwrap().is_empty());
717 assert!(log.snapshot().await.unwrap().revoked.is_empty());
718
719 let v = revoke_principal_ownership(&store, &log, &user("dana"), 3_000).await.unwrap();
721 assert_eq!(v, store.current_version().await.unwrap());
722 let mut recorded = log.snapshot().await.unwrap().revoked;
723 recorded.sort_by(|a, b| a.identity().cmp(&b.identity()));
724 assert_eq!(
725 recorded,
726 vec![
727 Revoked::membership("namespace", "band", PrincipalId::user("dana"), v),
728 Revoked::membership("namespace", "ed", PrincipalId::user("dana"), v),
729 ]
730 );
731 });
732 }
733
734 struct Racing {
737 inner: MemoryOwnershipStore,
738 writes: AtomicUsize,
739 }
740
741 #[async_trait]
742 impl OwnershipStore for Racing {
743 async fn insert(&self, o: &NewOwnership, now: i64) -> Result<Inserted, StoreError> {
744 self.inner.insert(o, now).await
745 }
746 async fn get(&self, id: &str) -> Result<Option<OwnershipRow>, StoreError> {
747 self.inner.get(id).await
748 }
749 async fn revoke_by_id(&self, id: &str, now: i64) -> Result<u64, StoreError> {
750 self.inner.revoke_by_id(id, now).await
751 }
752 async fn revoke_by_principal(&self, p: &PrincipalId, now: i64) -> Result<u64, StoreError> {
753 self.inner.revoke_by_principal(p, now).await
754 }
755 async fn list_history_for_principal(&self, p: &PrincipalId) -> Result<Vec<OwnershipRow>, StoreError> {
756 self.inner.list_history_for_principal(p).await
757 }
758 async fn current_version(&self) -> Result<u64, StoreError> {
759 self.inner.current_version().await
760 }
761 async fn list_for_principal(&self, p: &PrincipalId) -> Result<Vec<OwnershipRow>, StoreError> {
762 self.inner.list_for_principal(p).await
763 }
764 async fn list_for_resource(&self, kind: &str, id: &str) -> Result<Vec<OwnershipRow>, StoreError> {
765 let rows = self.inner.list_for_resource(kind, id).await;
766 if self.writes.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| n.checked_sub(1)).is_ok() {
767 let n = self.writes.load(Ordering::SeqCst);
768 self.inner.insert(&ns(user(&format!("late-{n}")), "ed", "member"), 500).await?;
769 }
770 rows
771 }
772 async fn list_for_subject_set(&self, kind: &str, id: &str) -> Result<Vec<OwnershipRow>, StoreError> {
773 self.inner.list_for_subject_set(kind, id).await
774 }
775 async fn list_for_kind(&self, kind: &str) -> Result<Vec<OwnershipRow>, StoreError> {
776 self.inner.list_for_kind(kind).await
777 }
778 async fn revocation_key(&self, kind: &str, id: &str) -> Result<cheers_core::RevocationKey, StoreError> {
779 self.inner.revocation_key(kind, id).await
780 }
781 async fn admission_policy(&self, kind: &str, id: &str) -> Result<Option<cheers_core::AdmissionPolicy>, StoreError> {
782 self.inner.admission_policy(kind, id).await
783 }
784 async fn set_admission_policy(
785 &self,
786 kind: &str,
787 id: &str,
788 policy: Option<&cheers_core::AdmissionPolicy>,
789 now: i64,
790 ) -> Result<u64, StoreError> {
791 self.inner.set_admission_policy(kind, id, policy, now).await
792 }
793 }
794
795 fn racing_issuer(writes: usize) -> (MemoryOwnershipStore, SnapshotIssuer) {
796 let inner = MemoryOwnershipStore::new();
797 let (minter, _) = PasetoV4SecretMinter::generate().unwrap();
798 let store = Racing { inner: inner.clone(), writes: AtomicUsize::new(writes) };
799 (inner, SnapshotIssuer::new(store, schema(), minter, ISS, "k1"))
800 }
801
802 #[test]
803 fn a_write_under_the_walk_retries_and_mints_the_later_state() {
804 pollster::block_on(async {
805 let (store, issuer) = racing_issuer(2);
806 store.insert(&ns(user("alice"), "ed", "member"), 100).await.unwrap();
807 let s = issuer.mint("namespace", "ed", 1_000).await.unwrap();
808 assert_eq!(s.snapshot.epoch, store.current_version().await.unwrap());
810 let users: Vec<_> = s.snapshot.members.iter().map(|m| m.principal.id.as_str()).collect();
811 assert!(users.contains(&"late-0") && users.contains(&"late-1"), "{users:?}");
812 });
813 }
814
815 #[test]
816 fn a_walk_that_never_settles_is_refused() {
817 pollster::block_on(async {
818 let (store, issuer) = racing_issuer(usize::MAX);
819 store.insert(&ns(user("alice"), "ed", "member"), 100).await.unwrap();
820 match issuer.mint("namespace", "ed", 1_000).await {
821 Err(Error::SnapshotContended { attempts, .. }) => assert_eq!(attempts, SnapshotIssuer::MINT_ATTEMPTS),
822 other => panic!("expected SnapshotContended, got {other:?}"),
823 }
824 });
825 }
826}