1use std::collections::{BTreeMap, BTreeSet};
43use std::sync::atomic::{AtomicU64, Ordering};
44use std::sync::{Arc, Mutex, MutexGuard, OnceLock};
45
46use lys_pass::membership::{
47 CONTRACT_UNSUPPORTED, CONTRACT_VERSION, GrantLog, LOG_MISMATCH, MembershipDecision,
48 MembershipRequest, Subject, Verdict,
49};
50use lys_pass::membership_admission::{AdmissionDecision, AdmissionRequest};
51use lys_pass::membership_pages::{
52 CURSOR_FOREIGN, CURSOR_RESET, MembershipPage, PageBounds, PageOutcome, PageRow,
53 RecipientPageRequest, ResourcePageRequest,
54};
55use lys_pass::rights::{Mode, Resource};
56
57use crate::ServerError;
58use crate::auth_lys::HotAction;
59use crate::auth_lys::session::LysSession;
60use crate::config::types::MembershipConfig;
61
62pub const DECISION_ROUTE: &str = "grants/membership";
64pub const ADMISSION_ROUTE: &str = "grants/membership/admission";
66pub const RESOURCES_ROUTE: &str = "grants/membership/resources";
68pub const RECIPIENTS_ROUTE: &str = "grants/membership/recipients";
70
71#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
74pub enum MembershipAction {
75 Read,
78 Post,
80}
81
82impl MembershipAction {
83 #[must_use]
85 pub const fn as_str(self) -> &'static str {
86 match self {
87 Self::Read => "read",
88 Self::Post => "post",
89 }
90 }
91
92 #[must_use]
97 pub const fn for_hot(action: HotAction) -> Self {
98 match action {
99 HotAction::Publish | HotAction::Call => Self::Post,
100 HotAction::Subscribe | HotAction::Serve => Self::Read,
101 }
102 }
103}
104
105#[derive(Clone, Debug, PartialEq, Eq)]
111pub enum MembershipRefusal {
112 HolderUnverified,
114 HolderKindUnknown,
116 HolderMalformed,
119 AuthorityMissing,
122 NotReady,
124 Unavailable {
126 cause: String,
128 },
129 Gap {
131 after: u64,
133 missing: u64,
135 },
136 Reset,
138 ResumePending,
140 ChangeApplying,
142 LogMismatch,
144 ContractUnsupported {
146 contract: u32,
148 },
149 AnswerSubstituted,
152 RevisionStale {
154 required: u64,
156 decided: Option<u64>,
158 },
159 ScopeOutsideMapping,
161 Refused {
163 refusal: String,
165 },
166 Held {
168 mode: Mode,
170 },
171 CapacityExceeded {
173 limit: usize,
175 },
176 IndexPoisoned,
178 WorkspaceNotMapped,
180 ChannelNotMapped,
182 CursorForeign,
184 CursorStale,
186 PageOverBound {
188 rows: u64,
190 bytes: u64,
192 },
193 PageForeignRow,
195 PageDuplicateRow,
197 PageInconsistent,
199}
200
201impl MembershipRefusal {
202 #[must_use]
207 pub const fn holds(&self) -> bool {
208 matches!(
209 self,
210 Self::AuthorityMissing
211 | Self::NotReady
212 | Self::Unavailable { .. }
213 | Self::Gap { .. }
214 | Self::Reset
215 | Self::ResumePending
216 | Self::ChangeApplying
217 | Self::IndexPoisoned
218 )
219 }
220
221 #[must_use]
223 pub fn name(&self) -> &str {
224 match self {
225 Self::HolderUnverified => "membership_holder_unverified",
226 Self::HolderKindUnknown => "membership_holder_kind_unknown",
227 Self::HolderMalformed => "membership_holder_malformed",
228 Self::AuthorityMissing => "membership_authority_missing",
229 Self::NotReady => "membership_not_ready",
230 Self::Unavailable { .. } => "membership_unavailable",
231 Self::Gap { .. } => "membership_gap",
232 Self::Reset => "membership_log_reset",
233 Self::ResumePending => "membership_resume_pending",
234 Self::ChangeApplying => "membership_change_applying",
235 Self::LogMismatch => LOG_MISMATCH,
236 Self::ContractUnsupported { .. } => CONTRACT_UNSUPPORTED,
237 Self::AnswerSubstituted => "membership_answer_substituted",
238 Self::RevisionStale { .. } => "membership_revision_stale",
239 Self::ScopeOutsideMapping => "membership_scope_outside_mapping",
240 Self::Refused { refusal } => refusal,
241 Self::Held { .. } => "membership_held",
242 Self::CapacityExceeded { .. } => "membership_capacity_exceeded",
243 Self::IndexPoisoned => "membership_index_poisoned",
244 Self::WorkspaceNotMapped => "membership_workspace_not_mapped",
245 Self::ChannelNotMapped => "membership_channel_not_mapped",
246 Self::CursorForeign => CURSOR_FOREIGN,
247 Self::CursorStale => "membership_cursor_stale",
248 Self::PageOverBound { .. } => "membership_page_over_bound",
249 Self::PageForeignRow => "membership_page_foreign_row",
250 Self::PageDuplicateRow => "membership_page_duplicate_row",
251 Self::PageInconsistent => "membership_page_inconsistent",
252 }
253 }
254}
255
256impl std::fmt::Display for MembershipRefusal {
257 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
258 match self {
259 Self::Unavailable { cause } => write!(formatter, "{}: {cause}", self.name()),
260 Self::Gap { after, missing } => write!(
261 formatter,
262 "{}: {missing} revisions missing after {after}",
263 self.name()
264 ),
265 Self::ContractUnsupported { contract } => write!(
266 formatter,
267 "{}: contract {contract}; this liminal speaks {CONTRACT_VERSION}",
268 self.name()
269 ),
270 Self::RevisionStale { required, decided } => write!(
271 formatter,
272 "{}: required {required}, decided {decided:?}",
273 self.name()
274 ),
275 Self::Held { mode } => write!(formatter, "{}: {mode:?}", self.name()),
276 Self::CapacityExceeded { limit } => {
277 write!(formatter, "{}: limit {limit}", self.name())
278 }
279 Self::PageOverBound { rows, bytes } => {
280 write!(formatter, "{}: {rows} rows, {bytes} bytes", self.name())
281 }
282 _ => formatter.write_str(self.name()),
283 }
284 }
285}
286
287impl std::error::Error for MembershipRefusal {}
288
289fn malformed(value: &str) -> bool {
290 value.is_empty() || value.chars().any(char::is_control)
291}
292
293fn invalid(message: impl Into<String>) -> ServerError {
294 ServerError::ConfigValidation {
295 message: message.into(),
296 }
297}
298
299fn bump(counter: &AtomicU64) {
300 counter.fetch_add(1, Ordering::Relaxed);
301}
302
303fn bump_by(counter: &AtomicU64, count: u64) {
304 counter.fetch_add(count, Ordering::Relaxed);
305}
306
307#[derive(Clone, PartialEq, Eq, PartialOrd, Ord)]
311pub struct MembershipSubject {
312 id: String,
313 kind: String,
314}
315
316impl MembershipSubject {
317 pub fn from_session(session: &dyn LysSession) -> Result<Self, MembershipRefusal> {
324 let kind = session
325 .holder_kind()
326 .ok_or(MembershipRefusal::HolderKindUnknown)?;
327 let id = session.holder();
328 if malformed(id) || malformed(kind) {
329 return Err(MembershipRefusal::HolderMalformed);
330 }
331 Ok(Self {
332 id: id.to_owned(),
333 kind: kind.to_owned(),
334 })
335 }
336
337 #[must_use]
339 pub fn kind(&self) -> &str {
340 &self.kind
341 }
342
343 #[must_use]
345 pub fn to_contract(&self) -> Subject {
346 Subject {
347 id: self.id.clone(),
348 kind: self.kind.clone(),
349 }
350 }
351
352 fn is(&self, subject: &Subject) -> bool {
353 self.id == subject.id && self.kind == subject.kind
354 }
355}
356
357impl std::fmt::Debug for MembershipSubject {
358 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
359 formatter
360 .debug_struct("MembershipSubject")
361 .field("id", &crate::auth_pass::ByteLength(self.id.len()))
362 .field("kind", &self.kind)
363 .finish()
364 }
365}
366
367#[derive(Clone, Debug, PartialEq, Eq)]
369pub struct MappedChannel {
370 workspace: Resource,
371 resource: Resource,
372}
373
374impl MappedChannel {
375 #[must_use]
377 pub const fn workspace(&self) -> &Resource {
378 &self.workspace
379 }
380
381 #[must_use]
383 pub const fn resource(&self) -> &Resource {
384 &self.resource
385 }
386}
387
388#[derive(Debug)]
390struct ChannelMap {
391 channels: BTreeMap<String, MappedChannel>,
392 conversations: BTreeMap<u64, String>,
393 by_resource: BTreeMap<String, String>,
394 workspaces: BTreeSet<String>,
395}
396
397fn channel_map(config: &MembershipConfig) -> Result<ChannelMap, ServerError> {
398 if config.channels.is_empty() {
399 return Err(invalid(
400 "auth.lys.membership.channels: at least one protected channel is required",
401 ));
402 }
403 let mut channels = BTreeMap::new();
404 let mut by_resource = BTreeMap::new();
405 let mut workspaces = BTreeSet::new();
406 for entry in &config.channels {
407 for (member, value) in [
408 ("channel", entry.channel.as_str()),
409 ("workspace", entry.workspace.as_str()),
410 ("resource", entry.resource.as_str()),
411 ] {
412 if malformed(value) {
413 return Err(invalid(format!(
414 "auth.lys.membership.channels: a {member} is empty or contains control characters"
415 )));
416 }
417 }
418 let mapped = MappedChannel {
419 workspace: Resource {
420 kind: config.workspace_kind.clone(),
421 id: entry.workspace.clone(),
422 },
423 resource: Resource {
424 kind: config.channel_kind.clone(),
425 id: entry.resource.clone(),
426 },
427 };
428 if channels.insert(entry.channel.clone(), mapped).is_some() {
429 return Err(invalid(
430 "auth.lys.membership.channels: a channel is listed twice",
431 ));
432 }
433 if by_resource
434 .insert(entry.resource.clone(), entry.channel.clone())
435 .is_some()
436 {
437 return Err(invalid(
438 "auth.lys.membership.channels: a placed resource is mapped to two channels",
439 ));
440 }
441 workspaces.insert(entry.workspace.clone());
442 }
443 let mut conversations = BTreeMap::new();
444 for entry in &config.conversations {
445 if !channels.contains_key(&entry.channel) {
446 return Err(invalid(
447 "auth.lys.membership.conversations: a conversation names an unlisted channel",
448 ));
449 }
450 if conversations
451 .insert(entry.conversation, entry.channel.clone())
452 .is_some()
453 {
454 return Err(invalid(
455 "auth.lys.membership.conversations: a conversation is listed twice",
456 ));
457 }
458 }
459 Ok(ChannelMap {
460 channels,
461 conversations,
462 by_resource,
463 workspaces,
464 })
465}
466
467fn validated_bounds(config: &MembershipConfig) -> Result<PageBounds, ServerError> {
468 if config.contract != CONTRACT_VERSION {
469 return Err(invalid(format!(
470 "auth.lys.membership.contract {} is not served; this liminal speaks {CONTRACT_VERSION}",
471 config.contract
472 )));
473 }
474 for (member, value) in [
475 ("log_identity", config.log_identity.as_str()),
476 ("workspace_kind", config.workspace_kind.as_str()),
477 ("channel_kind", config.channel_kind.as_str()),
478 ] {
479 if malformed(value) {
480 return Err(invalid(format!(
481 "auth.lys.membership.{member} is empty or contains control characters"
482 )));
483 }
484 }
485 if config.workspace_kind == config.channel_kind {
486 return Err(invalid(
487 "auth.lys.membership: workspace_kind and channel_kind must differ",
488 ));
489 }
490 if config.page_rows == 0 || config.page_bytes == 0 || config.max_decisions == 0 {
491 return Err(invalid(
492 "auth.lys.membership: page_rows, page_bytes and max_decisions must each be positive",
493 ));
494 }
495 Ok(PageBounds {
496 rows: config.page_rows,
497 bytes: config.page_bytes,
498 })
499}
500
501#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
503#[error("membership provider could not be asked: {cause}")]
504pub struct ProviderError {
505 pub cause: String,
507}
508
509fn provider_cause(cause: &str) -> ProviderError {
510 ProviderError {
511 cause: cause.to_owned(),
512 }
513}
514
515pub trait MembershipProvider: Send + Sync + std::fmt::Debug {
522 fn decide(&self, request: &MembershipRequest) -> Result<MembershipDecision, ProviderError>;
528
529 fn admission(&self, request: &AdmissionRequest) -> Result<AdmissionDecision, ProviderError>;
534
535 fn resources(&self, request: &ResourcePageRequest) -> Result<MembershipPage, ProviderError>;
540
541 fn recipients(&self, request: &RecipientPageRequest) -> Result<MembershipPage, ProviderError>;
546}
547
548pub struct HttpMembershipProvider {
555 http: reqwest::Client,
556 base: reqwest::Url,
557 credential: zeroize::Zeroizing<String>,
558 handle: tokio::runtime::Handle,
559}
560
561impl std::fmt::Debug for HttpMembershipProvider {
562 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
563 formatter
564 .debug_struct("HttpMembershipProvider")
565 .field("base", &self.base.as_str())
566 .field(
567 "credential",
568 &crate::auth_pass::ByteLength(self.credential.len()),
569 )
570 .finish_non_exhaustive()
571 }
572}
573
574impl HttpMembershipProvider {
575 pub fn new(
582 issuer: &str,
583 credential: zeroize::Zeroizing<String>,
584 handle: tokio::runtime::Handle,
585 ) -> Result<Self, ProviderError> {
586 let mut base =
587 reqwest::Url::parse(issuer).map_err(|_| provider_cause("issuer_not_a_url"))?;
588 if !matches!(base.scheme(), "http" | "https")
589 || base.host_str().is_none()
590 || !base.username().is_empty()
591 || base.password().is_some()
592 || base.query().is_some()
593 || base.fragment().is_some()
594 {
595 return Err(provider_cause("issuer_not_an_http_origin"));
596 }
597 if !base.path().ends_with('/') {
598 let path = format!("{}/", base.path());
599 base.set_path(&path);
600 }
601 if credential.trim().is_empty() {
602 return Err(provider_cause("membership_credential_empty"));
603 }
604 let http = reqwest::Client::builder()
605 .build()
606 .map_err(|_| provider_cause("http_client_unavailable"))?;
607 Ok(Self {
608 http,
609 base,
610 credential,
611 handle,
612 })
613 }
614
615 fn post<Q, A>(&self, route: &str, question: &Q) -> Result<A, ProviderError>
616 where
617 Q: serde::Serialize + ?Sized,
618 A: serde::de::DeserializeOwned + Send,
619 {
620 let url = self
621 .base
622 .join(route)
623 .map_err(|_| provider_cause("membership_route_unjoinable"))?;
624 let request = self
625 .http
626 .post(url)
627 .bearer_auth(self.credential.as_str())
628 .json(question);
629 let handle = &self.handle;
630 std::thread::scope(|scope| {
631 scope
632 .spawn(move || {
633 handle.block_on(async move {
634 let response = request
635 .send()
636 .await
637 .map_err(|_| provider_cause("lys_could_not_be_asked"))?;
638 let status = response.status();
639 if !status.is_success() {
640 return Err(ProviderError {
641 cause: format!("lys_http_status_{}", status.as_u16()),
642 });
643 }
644 response
645 .json::<A>()
646 .await
647 .map_err(|_| provider_cause("membership_answer_unreadable"))
648 })
649 })
650 .join()
651 })
652 .unwrap_or_else(|_| Err(provider_cause("membership_ask_did_not_complete")))
653 }
654}
655
656impl MembershipProvider for HttpMembershipProvider {
657 fn decide(&self, request: &MembershipRequest) -> Result<MembershipDecision, ProviderError> {
658 self.post(DECISION_ROUTE, request)
659 }
660
661 fn admission(&self, request: &AdmissionRequest) -> Result<AdmissionDecision, ProviderError> {
662 self.post(ADMISSION_ROUTE, request)
663 }
664
665 fn resources(&self, request: &ResourcePageRequest) -> Result<MembershipPage, ProviderError> {
666 self.post(RESOURCES_ROUTE, request)
667 }
668
669 fn recipients(&self, request: &RecipientPageRequest) -> Result<MembershipPage, ProviderError> {
670 self.post(RECIPIENTS_ROUTE, request)
671 }
672}
673
674#[derive(Clone, Debug, PartialEq, Eq)]
677pub enum AuthorityState {
678 NotReady,
680 Ready {
683 log: GrantLog,
685 revision: u64,
687 generation: u64,
689 },
690 Unavailable {
692 cause: String,
694 },
695 Gap {
697 after: u64,
699 missing: u64,
701 },
702 Reset,
704}
705
706pub trait MembershipAuthority: Send + Sync + std::fmt::Debug {
713 fn state(&self) -> AuthorityState;
715}
716
717#[derive(Clone, Debug, PartialEq, Eq)]
719pub struct MembershipChange {
720 pub log: GrantLog,
722 pub revision: u64,
724 pub subject: Option<Subject>,
727 pub resource: Resource,
729}
730
731#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
733pub struct AppliedChange {
734 pub duplicate: bool,
736 pub affected: u64,
738 pub redecided: u64,
740 pub failed: u64,
743}
744
745#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
747pub struct ResumeReport {
748 pub rechecked: u64,
750 pub failed: u64,
752 pub settled: bool,
755}
756
757pub trait MembershipListener: Send + Sync {
764 fn changed(&self, change: &MembershipChange) -> Result<AppliedChange, MembershipRefusal>;
770
771 fn resumed(&self, generation: u64, revision: u64) -> Result<ResumeReport, MembershipRefusal>;
777}
778
779#[derive(Clone, Debug, PartialEq, Eq)]
781pub enum Decided {
782 NotProtected,
784 Allowed {
786 grant: String,
788 revision: u64,
790 },
791}
792
793#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
797pub struct MembershipCounts {
798 pub decisions_asked: u64,
800 pub admissions_asked: u64,
802 pub pages_asked: u64,
804 pub page_rows: u64,
806 pub local_lookups: u64,
808 pub release_checks: u64,
810 pub refused: u64,
812 pub affected_visited: u64,
814 pub resume_rechecks: u64,
816 pub released: u64,
818 pub locks: u64,
820}
821
822#[derive(Debug, Default)]
823struct Counters {
824 decisions_asked: AtomicU64,
825 admissions_asked: AtomicU64,
826 pages_asked: AtomicU64,
827 page_rows: AtomicU64,
828 local_lookups: AtomicU64,
829 release_checks: AtomicU64,
830 refused: AtomicU64,
831 affected_visited: AtomicU64,
832 resume_rechecks: AtomicU64,
833 released: AtomicU64,
834 locks: AtomicU64,
835}
836
837impl Counters {
838 fn snapshot(&self) -> MembershipCounts {
839 let read = |counter: &AtomicU64| counter.load(Ordering::Relaxed);
840 MembershipCounts {
841 decisions_asked: read(&self.decisions_asked),
842 admissions_asked: read(&self.admissions_asked),
843 pages_asked: read(&self.pages_asked),
844 page_rows: read(&self.page_rows),
845 local_lookups: read(&self.local_lookups),
846 release_checks: read(&self.release_checks),
847 refused: read(&self.refused),
848 affected_visited: read(&self.affected_visited),
849 resume_rechecks: read(&self.resume_rechecks),
850 released: read(&self.released),
851 locks: read(&self.locks),
852 }
853 }
854}
855
856#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
857struct DecisionKey {
858 subject: MembershipSubject,
859 channel: String,
860 action: MembershipAction,
861}
862
863#[derive(Clone, Debug, PartialEq, Eq)]
865enum Indexed {
866 Applying { ticket: u64 },
867 Allowed { grant: String, revision: u64 },
868 Refused { refusal: String },
869 Held { mode: Mode },
870}
871
872impl Indexed {
873 fn result(&self) -> Result<Decided, MembershipRefusal> {
874 match self {
875 Self::Applying { .. } => Err(MembershipRefusal::ChangeApplying),
876 Self::Allowed { grant, revision } => Ok(Decided::Allowed {
877 grant: grant.clone(),
878 revision: *revision,
879 }),
880 Self::Refused { refusal } => Err(MembershipRefusal::Refused {
881 refusal: refusal.clone(),
882 }),
883 Self::Held { mode } => Err(MembershipRefusal::Held { mode: *mode }),
884 }
885 }
886}
887
888type ResourceKey = (String, String);
889
890fn resource_key(resource: &Resource) -> ResourceKey {
891 (resource.kind.clone(), resource.id.clone())
892}
893
894#[derive(Debug)]
895struct Entry {
896 verdict: Indexed,
897 connections: BTreeSet<u64>,
898 reached_by: [ResourceKey; 2],
899}
900
901#[derive(Debug, Default)]
902struct DecisionIndex {
903 generation: Option<u64>,
904 applied: u64,
905 next_ticket: u64,
906 entries: BTreeMap<DecisionKey, Entry>,
907 by_resource: BTreeMap<ResourceKey, BTreeSet<DecisionKey>>,
908 by_connection: BTreeMap<u64, BTreeSet<DecisionKey>>,
909}
910
911impl DecisionIndex {
912 const fn adopt(&mut self, generation: u64) -> Result<(), MembershipRefusal> {
913 match self.generation {
914 None => {
915 self.generation = Some(generation);
916 Ok(())
917 }
918 Some(current) if current == generation => Ok(()),
919 Some(_) => Err(MembershipRefusal::ResumePending),
920 }
921 }
922
923 fn ticket(&mut self) -> Result<u64, MembershipRefusal> {
924 let ticket = self.next_ticket;
925 self.next_ticket = ticket
926 .checked_add(1)
927 .ok_or(MembershipRefusal::CapacityExceeded { limit: usize::MAX })?;
928 Ok(ticket)
929 }
930
931 fn known(&mut self, key: &DecisionKey, pid: u64) -> Option<Indexed> {
933 let verdict = self.entries.get_mut(key).map(|entry| {
934 entry.connections.insert(pid);
935 entry.verdict.clone()
936 })?;
937 self.by_connection
938 .entry(pid)
939 .or_default()
940 .insert(key.clone());
941 Some(verdict)
942 }
943
944 fn reserve(
946 &mut self,
947 key: &DecisionKey,
948 mapped: &MappedChannel,
949 pid: u64,
950 limit: usize,
951 ) -> Result<u64, MembershipRefusal> {
952 if self.entries.len() >= limit {
953 return Err(MembershipRefusal::CapacityExceeded { limit });
954 }
955 let ticket = self.ticket()?;
956 let reached_by = [
957 resource_key(&mapped.resource),
958 resource_key(&mapped.workspace),
959 ];
960 for reach in &reached_by {
961 self.by_resource
962 .entry(reach.clone())
963 .or_default()
964 .insert(key.clone());
965 }
966 self.by_connection
967 .entry(pid)
968 .or_default()
969 .insert(key.clone());
970 self.entries.insert(
971 key.clone(),
972 Entry {
973 verdict: Indexed::Applying { ticket },
974 connections: BTreeSet::from([pid]),
975 reached_by,
976 },
977 );
978 Ok(ticket)
979 }
980
981 fn reapply(&mut self, key: &DecisionKey) -> Result<Option<u64>, MembershipRefusal> {
983 if !self.entries.contains_key(key) {
984 return Ok(None);
985 }
986 let ticket = self.ticket()?;
987 if let Some(entry) = self.entries.get_mut(key) {
988 entry.verdict = Indexed::Applying { ticket };
989 }
990 Ok(Some(ticket))
991 }
992
993 fn settle(&mut self, key: &DecisionKey, ticket: u64, verdict: Indexed) -> bool {
996 match self.entries.get_mut(key) {
997 Some(entry) if entry.verdict == (Indexed::Applying { ticket }) => {
998 entry.verdict = verdict;
999 true
1000 }
1001 Some(_) | None => false,
1002 }
1003 }
1004
1005 fn forget(&mut self, key: &DecisionKey, ticket: u64) {
1007 let held = self
1008 .entries
1009 .get(key)
1010 .is_some_and(|entry| entry.verdict == (Indexed::Applying { ticket }));
1011 if held {
1012 self.remove(key);
1013 }
1014 }
1015
1016 fn remove(&mut self, key: &DecisionKey) {
1017 let Some(entry) = self.entries.remove(key) else {
1018 return;
1019 };
1020 for reach in &entry.reached_by {
1021 let emptied = self.by_resource.get_mut(reach).is_some_and(|keys| {
1022 keys.remove(key);
1023 keys.is_empty()
1024 });
1025 if emptied {
1026 self.by_resource.remove(reach);
1027 }
1028 }
1029 for pid in &entry.connections {
1030 let emptied = self.by_connection.get_mut(pid).is_some_and(|keys| {
1031 keys.remove(key);
1032 keys.is_empty()
1033 });
1034 if emptied {
1035 self.by_connection.remove(pid);
1036 }
1037 }
1038 }
1039
1040 fn release(&mut self, pid: u64) -> u64 {
1043 let Some(keys) = self.by_connection.remove(&pid) else {
1044 return 0;
1045 };
1046 let mut released = 0_u64;
1047 for key in keys {
1048 let unheld = self.entries.get_mut(&key).is_some_and(|entry| {
1049 entry.connections.remove(&pid);
1050 entry.connections.is_empty()
1051 });
1052 if unheld {
1053 self.remove(&key);
1054 released = released.saturating_add(1);
1055 }
1056 }
1057 released
1058 }
1059}
1060
1061#[derive(Clone, Debug, PartialEq, Eq)]
1063enum PageQuestion {
1064 Resources {
1065 workspace: String,
1066 subject: MembershipSubject,
1067 action: MembershipAction,
1068 },
1069 Recipients {
1070 channel: String,
1071 action: MembershipAction,
1072 },
1073}
1074
1075#[derive(Clone, PartialEq, Eq)]
1078pub struct PageCursor {
1079 token: String,
1080 question: PageQuestion,
1081 generation: u64,
1082}
1083
1084impl std::fmt::Debug for PageCursor {
1085 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1086 formatter
1087 .debug_struct("PageCursor")
1088 .field("token", &crate::auth_pass::ByteLength(self.token.len()))
1089 .field("generation", &self.generation)
1090 .finish_non_exhaustive()
1091 }
1092}
1093
1094#[derive(Clone, Debug, PartialEq, Eq)]
1096pub struct ChannelPage {
1097 pub channels: Vec<String>,
1099 pub returned: u64,
1101 pub skipped: u64,
1103 pub held: u64,
1105 pub unmapped: u64,
1107 pub bytes: u64,
1109 pub revision: u64,
1111 pub complete: bool,
1113 pub next: Option<PageCursor>,
1115}
1116
1117#[derive(Clone, Debug, PartialEq, Eq)]
1119pub struct RecipientPage {
1120 pub recipients: Vec<Subject>,
1122 pub returned: u64,
1124 pub skipped: u64,
1126 pub held: u64,
1128 pub bytes: u64,
1130 pub revision: u64,
1132 pub complete: bool,
1134 pub next: Option<PageCursor>,
1136}
1137
1138struct PageParts<'p> {
1140 rows: &'p [PageRow],
1141 skipped: u64,
1142 complete: bool,
1143 next: Option<String>,
1144 revision: u64,
1145 bytes: u64,
1146}
1147
1148pub struct MembershipGate {
1151 served: GrantLog,
1152 workspace_kind: String,
1153 channel_kind: String,
1154 map: ChannelMap,
1155 bounds: PageBounds,
1156 max_decisions: usize,
1157 connect_workspace: Option<String>,
1158 provider: Arc<dyn MembershipProvider>,
1159 authority: OnceLock<Arc<dyn MembershipAuthority>>,
1160 index: Mutex<DecisionIndex>,
1161 counters: Counters,
1162}
1163
1164impl std::fmt::Debug for MembershipGate {
1165 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1166 formatter
1167 .debug_struct("MembershipGate")
1168 .field("served", &self.served)
1169 .field("channels", &self.map.channels.len())
1170 .field("bounds", &self.bounds)
1171 .field("max_decisions", &self.max_decisions)
1172 .field("authority", &self.authority.get().is_some())
1173 .finish_non_exhaustive()
1174 }
1175}
1176
1177impl MembershipGate {
1178 pub fn new(
1190 config: &MembershipConfig,
1191 provider: Arc<dyn MembershipProvider>,
1192 ) -> Result<Self, ServerError> {
1193 let bounds = validated_bounds(config)?;
1194 let map = channel_map(config)?;
1195 if let Some(workspace) = &config.connect_workspace
1196 && !map.workspaces.contains(workspace)
1197 {
1198 return Err(invalid(
1199 "auth.lys.membership.connect_workspace is not a mapped workspace",
1200 ));
1201 }
1202 Ok(Self {
1203 served: GrantLog {
1204 identity: config.log_identity.clone(),
1205 epoch: config.log_epoch,
1206 },
1207 workspace_kind: config.workspace_kind.clone(),
1208 channel_kind: config.channel_kind.clone(),
1209 map,
1210 bounds,
1211 max_decisions: config.max_decisions,
1212 connect_workspace: config.connect_workspace.clone(),
1213 provider,
1214 authority: OnceLock::new(),
1215 index: Mutex::new(DecisionIndex::default()),
1216 counters: Counters::default(),
1217 })
1218 }
1219
1220 pub fn install_authority(
1226 &self,
1227 authority: Arc<dyn MembershipAuthority>,
1228 ) -> Result<(), Arc<dyn MembershipAuthority>> {
1229 self.authority.set(authority)
1230 }
1231
1232 #[must_use]
1234 pub const fn served_log(&self) -> &GrantLog {
1235 &self.served
1236 }
1237
1238 #[must_use]
1240 pub fn protects(&self, channel: &str) -> bool {
1241 self.map.channels.contains_key(channel)
1242 }
1243
1244 #[must_use]
1246 pub fn mapped(&self, channel: &str) -> Option<&MappedChannel> {
1247 self.map.channels.get(channel)
1248 }
1249
1250 #[must_use]
1253 pub fn channel_for_conversation(&self, conversation: u64) -> Option<&str> {
1254 self.map
1255 .conversations
1256 .get(&conversation)
1257 .map(String::as_str)
1258 }
1259
1260 #[must_use]
1262 pub fn counts(&self) -> MembershipCounts {
1263 self.counters.snapshot()
1264 }
1265
1266 pub fn retained(&self) -> Result<usize, MembershipRefusal> {
1271 Ok(self.lock()?.entries.len())
1272 }
1273
1274 fn lock(&self) -> Result<MutexGuard<'_, DecisionIndex>, MembershipRefusal> {
1275 bump(&self.counters.locks);
1276 self.index
1277 .lock()
1278 .map_err(|_| MembershipRefusal::IndexPoisoned)
1279 }
1280
1281 fn ready(&self) -> Result<(u64, u64), MembershipRefusal> {
1284 let authority = self
1285 .authority
1286 .get()
1287 .ok_or(MembershipRefusal::AuthorityMissing)?;
1288 match authority.state() {
1289 AuthorityState::Ready {
1290 log,
1291 revision,
1292 generation,
1293 } => {
1294 if log == self.served {
1295 Ok((revision, generation))
1296 } else {
1297 Err(MembershipRefusal::LogMismatch)
1298 }
1299 }
1300 AuthorityState::NotReady => Err(MembershipRefusal::NotReady),
1301 AuthorityState::Unavailable { cause } => Err(MembershipRefusal::Unavailable { cause }),
1302 AuthorityState::Gap { after, missing } => {
1303 Err(MembershipRefusal::Gap { after, missing })
1304 }
1305 AuthorityState::Reset => Err(MembershipRefusal::Reset),
1306 }
1307 }
1308
1309 pub fn check(
1320 &self,
1321 pid: u64,
1322 holder: Option<&MembershipSubject>,
1323 channel: &str,
1324 action: MembershipAction,
1325 ) -> Result<Decided, MembershipRefusal> {
1326 let Some(mapped) = self.map.channels.get(channel) else {
1327 return Ok(Decided::NotProtected);
1328 };
1329 let decided = self.check_mapped(pid, holder, channel, mapped, action);
1330 if decided.is_err() {
1331 bump(&self.counters.refused);
1332 }
1333 decided
1334 }
1335
1336 pub fn check_release(
1342 &self,
1343 pid: u64,
1344 holder: Option<&MembershipSubject>,
1345 channel: &str,
1346 action: MembershipAction,
1347 ) -> Result<Decided, MembershipRefusal> {
1348 bump(&self.counters.release_checks);
1349 self.check(pid, holder, channel, action)
1350 }
1351
1352 fn check_mapped(
1353 &self,
1354 pid: u64,
1355 holder: Option<&MembershipSubject>,
1356 channel: &str,
1357 mapped: &MappedChannel,
1358 action: MembershipAction,
1359 ) -> Result<Decided, MembershipRefusal> {
1360 let holder = holder.ok_or(MembershipRefusal::HolderUnverified)?;
1361 let (revision, generation) = self.ready()?;
1362 let key = DecisionKey {
1363 subject: holder.clone(),
1364 channel: channel.to_owned(),
1365 action,
1366 };
1367 let (ticket, at_least) = {
1368 let mut index = self.lock()?;
1369 index.adopt(generation)?;
1370 if let Some(verdict) = index.known(&key, pid) {
1371 drop(index);
1372 bump(&self.counters.local_lookups);
1373 return verdict.result();
1374 }
1375 let at_least = revision.max(index.applied);
1376 let ticket = index.reserve(&key, mapped, pid, self.max_decisions)?;
1377 drop(index);
1378 (ticket, at_least)
1379 };
1380 let asked = self.ask(holder, mapped, action, at_least);
1381 let mut index = self.lock()?;
1382 match asked {
1383 Ok(verdict) => {
1384 let current = index.generation == Some(generation)
1385 && index.settle(&key, ticket, verdict.clone());
1386 if !current {
1387 index.forget(&key, ticket);
1390 return Err(MembershipRefusal::ChangeApplying);
1391 }
1392 drop(index);
1393 verdict.result()
1394 }
1395 Err(refusal) => {
1396 index.forget(&key, ticket);
1397 drop(index);
1398 Err(refusal)
1399 }
1400 }
1401 }
1402
1403 fn ask(
1404 &self,
1405 holder: &MembershipSubject,
1406 mapped: &MappedChannel,
1407 action: MembershipAction,
1408 at_least: u64,
1409 ) -> Result<Indexed, MembershipRefusal> {
1410 let request = MembershipRequest {
1411 contract: CONTRACT_VERSION,
1412 log: self.served.clone(),
1413 workspace: mapped.workspace.clone(),
1414 subject: holder.to_contract(),
1415 resource: mapped.resource.clone(),
1416 action: action.as_str().to_owned(),
1417 at_least,
1418 };
1419 bump(&self.counters.decisions_asked);
1420 let decision = self
1421 .provider
1422 .decide(&request)
1423 .map_err(|error| MembershipRefusal::Unavailable { cause: error.cause })?;
1424 judged(&request, &self.served, mapped, &decision)
1425 }
1426
1427 pub fn admit_connect(
1436 &self,
1437 holder: &MembershipSubject,
1438 ) -> Result<Option<u64>, MembershipRefusal> {
1439 let Some(workspace) = &self.connect_workspace else {
1440 return Ok(None);
1441 };
1442 let (revision, _) = self.ready()?;
1443 let request = AdmissionRequest {
1444 contract: CONTRACT_VERSION,
1445 log: self.served.clone(),
1446 workspace: Resource {
1447 kind: self.workspace_kind.clone(),
1448 id: workspace.clone(),
1449 },
1450 subject: holder.to_contract(),
1451 at_least: revision,
1452 };
1453 bump(&self.counters.admissions_asked);
1454 let decision = self
1455 .provider
1456 .admission(&request)
1457 .map_err(|error| MembershipRefusal::Unavailable { cause: error.cause })?;
1458 admitted(&request, &self.served, &decision).map(Some)
1459 }
1460
1461 pub fn release_connection(&self, pid: u64) {
1465 match self.lock() {
1466 Ok(mut index) => {
1467 let released = index.release(pid);
1468 drop(index);
1469 bump_by(&self.counters.released, released);
1470 }
1471 Err(refusal) => {
1472 tracing::warn!(
1473 pid,
1474 refusal = refusal.name(),
1475 "membership bindings could not be released"
1476 );
1477 }
1478 }
1479 }
1480
1481 fn redecide(&self, tickets: Vec<(DecisionKey, u64)>, at_least: u64) -> (u64, u64) {
1484 let mut settled = 0_u64;
1485 let mut failed = 0_u64;
1486 for (key, ticket) in tickets {
1487 let asked = self
1488 .map
1489 .channels
1490 .get(&key.channel)
1491 .ok_or(MembershipRefusal::ChannelNotMapped)
1492 .and_then(|mapped| self.ask(&key.subject, mapped, key.action, at_least));
1493 let Ok(mut index) = self.lock() else {
1494 failed = failed.saturating_add(1);
1495 continue;
1496 };
1497 let kept = asked.is_ok_and(|verdict| index.settle(&key, ticket, verdict));
1498 if kept {
1499 settled = settled.saturating_add(1);
1500 } else {
1501 index.forget(&key, ticket);
1502 failed = failed.saturating_add(1);
1503 }
1504 }
1505 (settled, failed)
1506 }
1507
1508 pub fn resource_page(
1518 &self,
1519 holder: Option<&MembershipSubject>,
1520 workspace: &str,
1521 action: MembershipAction,
1522 after: Option<&PageCursor>,
1523 ) -> Result<ChannelPage, MembershipRefusal> {
1524 let holder = holder.ok_or(MembershipRefusal::HolderUnverified)?;
1525 if !self.map.workspaces.contains(workspace) {
1526 return Err(MembershipRefusal::WorkspaceNotMapped);
1527 }
1528 let (revision, generation) = self.ready()?;
1529 let question = PageQuestion::Resources {
1530 workspace: workspace.to_owned(),
1531 subject: holder.clone(),
1532 action,
1533 };
1534 let request = ResourcePageRequest {
1535 contract: CONTRACT_VERSION,
1536 log: self.served.clone(),
1537 workspace: Resource {
1538 kind: self.workspace_kind.clone(),
1539 id: workspace.to_owned(),
1540 },
1541 subject: holder.to_contract(),
1542 kind: self.channel_kind.clone(),
1543 action: action.as_str().to_owned(),
1544 at_least: revision,
1545 bounds: self.bounds,
1546 after: continuation(after, &question, generation)?,
1547 };
1548 bump(&self.counters.pages_asked);
1549 let page = self
1550 .provider
1551 .resources(&request)
1552 .map_err(|error| MembershipRefusal::Unavailable { cause: error.cause })?;
1553 let parts = self.page_parts(&page, revision)?;
1554 let mut channels = Vec::with_capacity(parts.rows.len());
1555 let mut seen = BTreeSet::new();
1556 let mut held = 0_u64;
1557 let mut unmapped = 0_u64;
1558 for row in parts.rows {
1559 if !holder.is(&row.subject) || row.resource.kind != self.channel_kind {
1560 return Err(MembershipRefusal::PageForeignRow);
1561 }
1562 if !seen.insert(row.resource.id.as_str()) {
1563 return Err(MembershipRefusal::PageDuplicateRow);
1564 }
1565 let Some(channel) = self.map.by_resource.get(&row.resource.id) else {
1566 unmapped = unmapped.saturating_add(1);
1567 continue;
1568 };
1569 let placed_here = self
1570 .map
1571 .channels
1572 .get(channel)
1573 .is_some_and(|mapped| mapped.workspace.id == workspace);
1574 if !placed_here {
1575 return Err(MembershipRefusal::PageForeignRow);
1576 }
1577 if row.mode == Mode::Outright {
1578 channels.push(channel.clone());
1579 } else {
1580 held = held.saturating_add(1);
1581 }
1582 }
1583 Ok(ChannelPage {
1584 returned: u64::try_from(parts.rows.len())
1585 .map_err(|_| MembershipRefusal::PageInconsistent)?,
1586 channels,
1587 skipped: parts.skipped,
1588 held,
1589 unmapped,
1590 bytes: parts.bytes,
1591 revision: parts.revision,
1592 complete: parts.complete,
1593 next: parts.next.map(|token| PageCursor {
1594 token,
1595 question,
1596 generation,
1597 }),
1598 })
1599 }
1600
1601 pub fn recipient_page(
1608 &self,
1609 channel: &str,
1610 action: MembershipAction,
1611 after: Option<&PageCursor>,
1612 ) -> Result<RecipientPage, MembershipRefusal> {
1613 let mapped = self
1614 .map
1615 .channels
1616 .get(channel)
1617 .ok_or(MembershipRefusal::ChannelNotMapped)?;
1618 let (revision, generation) = self.ready()?;
1619 let question = PageQuestion::Recipients {
1620 channel: channel.to_owned(),
1621 action,
1622 };
1623 let request = RecipientPageRequest {
1624 contract: CONTRACT_VERSION,
1625 log: self.served.clone(),
1626 workspace: mapped.workspace.clone(),
1627 resource: mapped.resource.clone(),
1628 action: action.as_str().to_owned(),
1629 at_least: revision,
1630 bounds: self.bounds,
1631 after: continuation(after, &question, generation)?,
1632 };
1633 bump(&self.counters.pages_asked);
1634 let page = self
1635 .provider
1636 .recipients(&request)
1637 .map_err(|error| MembershipRefusal::Unavailable { cause: error.cause })?;
1638 let parts = self.page_parts(&page, revision)?;
1639 let mut recipients = Vec::with_capacity(parts.rows.len());
1640 let mut seen = BTreeSet::new();
1641 let mut held = 0_u64;
1642 for row in parts.rows {
1643 if row.resource != mapped.resource {
1644 return Err(MembershipRefusal::PageForeignRow);
1645 }
1646 if !seen.insert((row.subject.id.as_str(), row.subject.kind.as_str())) {
1647 return Err(MembershipRefusal::PageDuplicateRow);
1648 }
1649 if row.mode == Mode::Outright {
1650 recipients.push(row.subject.clone());
1651 } else {
1652 held = held.saturating_add(1);
1653 }
1654 }
1655 Ok(RecipientPage {
1656 returned: u64::try_from(parts.rows.len())
1657 .map_err(|_| MembershipRefusal::PageInconsistent)?,
1658 recipients,
1659 skipped: parts.skipped,
1660 held,
1661 bytes: parts.bytes,
1662 revision: parts.revision,
1663 complete: parts.complete,
1664 next: parts.next.map(|token| PageCursor {
1665 token,
1666 question,
1667 generation,
1668 }),
1669 })
1670 }
1671
1672 fn page_parts<'p>(
1673 &self,
1674 page: &'p MembershipPage,
1675 at_least: u64,
1676 ) -> Result<PageParts<'p>, MembershipRefusal> {
1677 if page.contract != CONTRACT_VERSION {
1678 return Err(MembershipRefusal::ContractUnsupported {
1679 contract: page.contract,
1680 });
1681 }
1682 if page.log != self.served {
1683 return Err(MembershipRefusal::LogMismatch);
1684 }
1685 let (rows, returned, skipped, complete, next) = match &page.outcome {
1686 PageOutcome::Refused { refusal, .. } => return Err(page_refusal(refusal)),
1687 PageOutcome::Page {
1688 rows,
1689 returned,
1690 skipped,
1691 complete,
1692 next,
1693 } => (rows, *returned, *skipped, *complete, next),
1694 };
1695 let revision = page.revision.ok_or(MembershipRefusal::RevisionStale {
1696 required: at_least,
1697 decided: None,
1698 })?;
1699 if revision < at_least {
1700 return Err(MembershipRefusal::RevisionStale {
1701 required: at_least,
1702 decided: Some(revision),
1703 });
1704 }
1705 let count = u64::try_from(rows.len()).map_err(|_| MembershipRefusal::PageInconsistent)?;
1706 bump_by(&self.counters.page_rows, count);
1707 if count != returned || complete == next.is_some() {
1708 return Err(MembershipRefusal::PageInconsistent);
1709 }
1710 let bytes = encoded_bytes(rows)?;
1711 if count > u64::from(self.bounds.rows) || bytes > u64::from(self.bounds.bytes) {
1712 return Err(MembershipRefusal::PageOverBound { rows: count, bytes });
1713 }
1714 Ok(PageParts {
1715 rows,
1716 skipped,
1717 complete,
1718 next: next.clone(),
1719 revision,
1720 bytes,
1721 })
1722 }
1723}
1724
1725impl MembershipListener for MembershipGate {
1726 fn changed(&self, change: &MembershipChange) -> Result<AppliedChange, MembershipRefusal> {
1727 if change.log != self.served {
1728 return Err(MembershipRefusal::LogMismatch);
1729 }
1730 let reached = resource_key(&change.resource);
1731 let tickets = {
1732 let mut index = self.lock()?;
1733 if change.revision <= index.applied {
1734 return Ok(AppliedChange {
1737 duplicate: true,
1738 ..AppliedChange::default()
1739 });
1740 }
1741 index.applied = change.revision;
1742 let keys: Vec<DecisionKey> = index
1743 .by_resource
1744 .get(&reached)
1745 .map(|keys| {
1746 keys.iter()
1747 .filter(|key| {
1748 change
1749 .subject
1750 .as_ref()
1751 .is_none_or(|subject| key.subject.is(subject))
1752 })
1753 .cloned()
1754 .collect()
1755 })
1756 .unwrap_or_default();
1757 let mut tickets = Vec::with_capacity(keys.len());
1758 for key in keys {
1759 if let Some(ticket) = index.reapply(&key)? {
1760 tickets.push((key, ticket));
1761 }
1762 }
1763 tickets
1764 };
1765 let affected = u64::try_from(tickets.len()).unwrap_or(u64::MAX);
1766 bump_by(&self.counters.affected_visited, affected);
1767 let (redecided, failed) = self.redecide(tickets, change.revision);
1768 Ok(AppliedChange {
1769 duplicate: false,
1770 affected,
1771 redecided,
1772 failed,
1773 })
1774 }
1775
1776 fn resumed(&self, generation: u64, revision: u64) -> Result<ResumeReport, MembershipRefusal> {
1777 let tickets = {
1778 let mut index = self.lock()?;
1779 let keys: Vec<DecisionKey> = index.entries.keys().cloned().collect();
1780 let mut tickets = Vec::with_capacity(keys.len());
1781 for key in keys {
1782 if let Some(ticket) = index.reapply(&key)? {
1783 tickets.push((key, ticket));
1784 }
1785 }
1786 tickets
1787 };
1788 let rechecked = u64::try_from(tickets.len()).unwrap_or(u64::MAX);
1789 bump_by(&self.counters.resume_rechecks, rechecked);
1790 let (_, failed) = self.redecide(tickets, revision);
1791 let settled = failed == 0;
1792 if settled {
1793 let mut index = self.lock()?;
1794 index.generation = Some(generation);
1795 index.applied = index.applied.max(revision);
1796 }
1797 Ok(ResumeReport {
1798 rechecked,
1799 failed,
1800 settled,
1801 })
1802 }
1803}
1804
1805fn continuation(
1808 after: Option<&PageCursor>,
1809 question: &PageQuestion,
1810 generation: u64,
1811) -> Result<Option<String>, MembershipRefusal> {
1812 let Some(cursor) = after else {
1813 return Ok(None);
1814 };
1815 if &cursor.question != question {
1816 return Err(MembershipRefusal::CursorForeign);
1817 }
1818 if cursor.generation != generation {
1819 return Err(MembershipRefusal::CursorStale);
1820 }
1821 Ok(Some(cursor.token.clone()))
1822}
1823
1824fn page_refusal(refusal: &str) -> MembershipRefusal {
1825 match refusal {
1826 CURSOR_FOREIGN => MembershipRefusal::CursorForeign,
1827 CURSOR_RESET => MembershipRefusal::CursorStale,
1828 LOG_MISMATCH => MembershipRefusal::LogMismatch,
1829 other => MembershipRefusal::Refused {
1830 refusal: other.to_owned(),
1831 },
1832 }
1833}
1834
1835fn encoded_bytes(rows: &[PageRow]) -> Result<u64, MembershipRefusal> {
1836 rows.iter().try_fold(0_u64, |total, row| {
1837 let length = serde_json::to_vec(row)
1838 .map(|encoded| encoded.len())
1839 .map_err(|_| MembershipRefusal::PageInconsistent)?;
1840 u64::try_from(length)
1841 .ok()
1842 .and_then(|length| total.checked_add(length))
1843 .ok_or(MembershipRefusal::PageInconsistent)
1844 })
1845}
1846
1847fn judged(
1853 request: &MembershipRequest,
1854 served: &GrantLog,
1855 mapped: &MappedChannel,
1856 decision: &MembershipDecision,
1857) -> Result<Indexed, MembershipRefusal> {
1858 if decision.contract != CONTRACT_VERSION {
1859 return Err(MembershipRefusal::ContractUnsupported {
1860 contract: decision.contract,
1861 });
1862 }
1863 if &decision.log != served {
1864 return Err(MembershipRefusal::LogMismatch);
1865 }
1866 if &decision.request != request {
1867 return Err(MembershipRefusal::AnswerSubstituted);
1868 }
1869 let Some(revision) = decision.revision else {
1870 return Err(match &decision.verdict {
1871 Verdict::Refused { refusal, .. } => page_refusal(refusal),
1872 Verdict::Allowed { .. } | Verdict::Held { .. } => MembershipRefusal::RevisionStale {
1873 required: request.at_least,
1874 decided: None,
1875 },
1876 });
1877 };
1878 if revision < request.at_least {
1879 return Err(MembershipRefusal::RevisionStale {
1880 required: request.at_least,
1881 decided: Some(revision),
1882 });
1883 }
1884 match &decision.verdict {
1885 Verdict::Allowed { grant, scope, .. } => {
1886 if scope == &mapped.resource || scope == &mapped.workspace {
1887 Ok(Indexed::Allowed {
1888 grant: grant.clone(),
1889 revision,
1890 })
1891 } else {
1892 Err(MembershipRefusal::ScopeOutsideMapping)
1893 }
1894 }
1895 Verdict::Held { mode, .. } => Ok(Indexed::Held { mode: *mode }),
1896 Verdict::Refused { refusal, .. } => Ok(Indexed::Refused {
1897 refusal: refusal.clone(),
1898 }),
1899 }
1900}
1901
1902fn admitted(
1904 request: &AdmissionRequest,
1905 served: &GrantLog,
1906 decision: &AdmissionDecision,
1907) -> Result<u64, MembershipRefusal> {
1908 if decision.contract != CONTRACT_VERSION {
1909 return Err(MembershipRefusal::ContractUnsupported {
1910 contract: decision.contract,
1911 });
1912 }
1913 if &decision.log != served {
1914 return Err(MembershipRefusal::LogMismatch);
1915 }
1916 if &decision.request != request {
1917 return Err(MembershipRefusal::AnswerSubstituted);
1918 }
1919 match (&decision.verdict, decision.revision) {
1920 (Verdict::Refused { refusal, .. }, _) => Err(page_refusal(refusal)),
1921 (Verdict::Held { mode, .. }, _) => Err(MembershipRefusal::Held { mode: *mode }),
1922 (Verdict::Allowed { .. }, Some(revision)) if revision >= request.at_least => Ok(revision),
1923 (Verdict::Allowed { .. }, decided) => Err(MembershipRefusal::RevisionStale {
1924 required: request.at_least,
1925 decided,
1926 }),
1927 }
1928}
1929
1930#[derive(Debug, Default)]
1937pub struct ConnectionMembership {
1938 bound: Option<BoundMembership>,
1939}
1940
1941#[derive(Debug)]
1942struct BoundMembership {
1943 gate: Arc<MembershipGate>,
1944 pid: u64,
1945 holder: Result<MembershipSubject, MembershipRefusal>,
1946}
1947
1948impl ConnectionMembership {
1949 #[must_use]
1952 pub fn bind(
1953 gate: Option<Arc<MembershipGate>>,
1954 pid: u64,
1955 session: Option<&dyn LysSession>,
1956 ) -> Self {
1957 Self {
1958 bound: gate.map(|gate| BoundMembership {
1959 gate,
1960 pid,
1961 holder: session.map_or(
1962 Err(MembershipRefusal::HolderUnverified),
1963 MembershipSubject::from_session,
1964 ),
1965 }),
1966 }
1967 }
1968
1969 #[must_use]
1971 pub fn gate(&self) -> Option<&Arc<MembershipGate>> {
1972 self.bound.as_ref().map(|bound| &bound.gate)
1973 }
1974
1975 pub fn holder(&self) -> Result<&MembershipSubject, MembershipRefusal> {
1981 self.bound
1982 .as_ref()
1983 .ok_or(MembershipRefusal::HolderUnverified)
1984 .and_then(|bound| bound.holder.as_ref().map_err(Clone::clone))
1985 }
1986
1987 pub fn admit_connect(&self) -> Result<Option<u64>, MembershipRefusal> {
1992 let Some(bound) = &self.bound else {
1993 return Ok(None);
1994 };
1995 if bound.gate.connect_workspace.is_none() {
1996 return Ok(None);
1997 }
1998 let holder = bound.holder.as_ref().map_err(Clone::clone)?;
1999 bound.gate.admit_connect(holder)
2000 }
2001
2002 pub fn check(
2008 &self,
2009 channel: &str,
2010 action: MembershipAction,
2011 ) -> Result<Decided, MembershipRefusal> {
2012 self.decide(channel, action, false)
2013 }
2014
2015 pub fn release(
2020 &self,
2021 channel: &str,
2022 action: MembershipAction,
2023 ) -> Result<Decided, MembershipRefusal> {
2024 self.decide(channel, action, true)
2025 }
2026
2027 pub fn release_conversation(&self, conversation: u64) -> Result<Decided, MembershipRefusal> {
2033 let Some(bound) = &self.bound else {
2034 return Ok(Decided::NotProtected);
2035 };
2036 let Some(channel) = bound.gate.channel_for_conversation(conversation) else {
2037 return Ok(Decided::NotProtected);
2038 };
2039 self.decide(channel, MembershipAction::Read, true)
2040 }
2041
2042 fn decide(
2043 &self,
2044 channel: &str,
2045 action: MembershipAction,
2046 release: bool,
2047 ) -> Result<Decided, MembershipRefusal> {
2048 let Some(bound) = &self.bound else {
2049 return Ok(Decided::NotProtected);
2050 };
2051 if !bound.gate.protects(channel) {
2052 return Ok(Decided::NotProtected);
2053 }
2054 let holder = match &bound.holder {
2055 Ok(holder) => holder,
2056 Err(refusal) => {
2057 bump(&bound.gate.counters.refused);
2058 return Err(refusal.clone());
2059 }
2060 };
2061 if release {
2062 bound
2063 .gate
2064 .check_release(bound.pid, Some(holder), channel, action)
2065 } else {
2066 bound.gate.check(bound.pid, Some(holder), channel, action)
2067 }
2068 }
2069}
2070
2071impl ConnectionMembership {
2072 #[must_use]
2077 pub fn release_check(&self, channel: &str, action: MembershipAction) -> Option<ReleaseCheck> {
2078 let bound = self.bound.as_ref()?;
2079 if !bound.gate.protects(channel) {
2080 return None;
2081 }
2082 let holder = bound.holder.as_ref().ok()?;
2083 Some(ReleaseCheck {
2084 gate: Arc::clone(&bound.gate),
2085 pid: bound.pid,
2086 holder: holder.clone(),
2087 channel: channel.to_owned(),
2088 action,
2089 })
2090 }
2091}
2092
2093#[derive(Clone, Debug)]
2100pub struct ReleaseCheck {
2101 gate: Arc<MembershipGate>,
2102 pid: u64,
2103 holder: MembershipSubject,
2104 channel: String,
2105 action: MembershipAction,
2106}
2107
2108impl ReleaseCheck {
2109 pub fn check(&self) -> Result<Decided, MembershipRefusal> {
2114 self.gate
2115 .check_release(self.pid, Some(&self.holder), &self.channel, self.action)
2116 }
2117}
2118
2119impl Drop for ConnectionMembership {
2120 fn drop(&mut self) {
2121 if let Some(bound) = &self.bound {
2122 bound.gate.release_connection(bound.pid);
2123 }
2124 }
2125}
2126
2127pub fn configured_gate(
2135 config: &MembershipConfig,
2136 issuer: &str,
2137) -> Result<Arc<MembershipGate>, ServerError> {
2138 let raw = zeroize::Zeroizing::new(std::fs::read_to_string(&config.credential_path).map_err(
2139 |error| {
2140 invalid(format!(
2141 "auth.lys.membership.credential_path {}: {}",
2142 config.credential_path.display(),
2143 error.kind()
2144 ))
2145 },
2146 )?);
2147 let credential = zeroize::Zeroizing::new(raw.trim().to_owned());
2148 let handle = tokio::runtime::Handle::try_current()
2149 .map_err(|_| invalid("auth.lys.membership needs the server's tokio runtime at start"))?;
2150 let provider = HttpMembershipProvider::new(issuer, credential, handle).map_err(|error| {
2151 invalid(format!(
2152 "auth.lys.membership provider refused: {}",
2153 error.cause
2154 ))
2155 })?;
2156 MembershipGate::new(config, Arc::new(provider)).map(Arc::new)
2157}
2158
2159#[cfg(any(test, feature = "test-support"))]
2162pub mod fixture;
2163
2164#[cfg(test)]
2165#[path = "auth_membership_tests.rs"]
2166mod tests;