1use std::collections::BTreeMap;
17use std::sync::{Arc, Condvar, Mutex, OnceLock, PoisonError, Weak};
18use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
19
20use lys_pass::deliberate::CheckAnswer;
21use lys_pass::drafts::{DraftRequest, RunError, RunReport, request_digest};
22use lys_pass::{Decision, Error, KeySet, Mode, Refusal, Target, VerifiedPass};
23use serde_json::{Value, json};
24
25use crate::serve::authorize::subjects;
26use crate::serve::drafts::Drafts;
27use crate::serve::protocol::Operation;
28use crate::serve::route_matrix::{Class, Route};
29use crate::serve::service::Service;
30use crate::serve::storage::Failure;
31
32pub const AUDIENCE: &str = "haematite";
34
35pub const PASS_MISSING: &str = "pass_missing";
37
38pub const LYS_KEYS_PENDING: &str = "lys_keys_pending";
41
42const KEYS_RETRY_FIRST: Duration = Duration::from_millis(250);
45
46const KEYS_RETRY_MOST: Duration = Duration::from_secs(30);
48
49#[derive(Clone, Copy)]
52pub struct Trust<'a> {
53 pub keys: &'a KeySet,
55 pub issuer: &'a str,
57 pub audience: &'a str,
59}
60
61pub trait LiveCheck {
64 fn ask(&self, token: &str, holder: &str, target: &Target) -> Result<CheckAnswer, Error>;
71
72 fn draft(
80 &self,
81 token: &str,
82 request: &DraftRequest,
83 judgment: &CheckAnswer,
84 ) -> Result<String, Error>;
85
86 fn run_drafts(
94 &self,
95 _connector: &str,
96 _drafts: &mut Drafts<'_>,
97 _after: Option<&str>,
98 ) -> Result<RunReport, Box<RunError<Failure>>> {
99 Err(Box::new(RunError::Pull(Error::CannotAsk(
100 "this live check runs no drafts",
101 ))))
102 }
103}
104
105pub struct Unreachable;
108
109impl LiveCheck for Unreachable {
110 fn ask(&self, _token: &str, _holder: &str, _target: &Target) -> Result<CheckAnswer, Error> {
111 Err(Error::CannotAsk("no live check is configured"))
112 }
113
114 fn draft(
115 &self,
116 _token: &str,
117 _request: &DraftRequest,
118 _judgment: &CheckAnswer,
119 ) -> Result<String, Error> {
120 Err(Error::CannotAsk("no live check is configured"))
121 }
122}
123
124#[derive(Debug, Clone, PartialEq, Eq)]
126pub enum Verdict {
127 Allowed {
129 grant: String,
131 },
132 Held {
135 grant: String,
137 mode: Mode,
139 judgment: Box<CheckAnswer>,
141 },
142 Refused(Refusal),
144 Named(String),
146}
147
148impl Verdict {
149 fn named(error: &Error) -> Self {
150 Self::Named(error.name().to_owned())
151 }
152
153 fn refused(trust: Trust<'_>, target: &Target) -> Self {
154 match Refusal::new(
155 trust.audience,
156 &target.kind,
157 &target.id,
158 &target.action,
159 trust.issuer,
160 ) {
161 Ok(refusal) => Self::Refused(refusal),
162 Err(error) => Self::named(&error),
163 }
164 }
165}
166
167pub fn target(route: &Route, database: &str, name: Option<&str>) -> Result<Target, Error> {
178 let kind = format!("{AUDIENCE}.{}", route.kind);
179 match (route.kind, name) {
180 ("database", _) => Target::new(&kind, database, route.action),
181 (_, Some(name)) => Target::new(&kind, &format!("{database}.{name}"), route.action),
182 (_, None) => Err(Error::Invalid("a placed resource needs its name")),
183 }
184}
185
186#[must_use]
190pub fn decide(
191 trust: Trust<'_>,
192 token: Option<&str>,
193 target: &Target,
194 class: Class,
195 live: &dyn LiveCheck,
196 now: u64,
197) -> Verdict {
198 let Some(token) = token else {
199 return Verdict::Named(PASS_MISSING.to_owned());
200 };
201 let pass = match VerifiedPass::verify(token, trust.keys, trust.issuer, trust.audience, now) {
202 Ok(pass) => pass,
203 Err(error) => return Verdict::named(&error),
204 };
205 match class {
206 Class::Hot => match pass.evaluate(&target.kind, &target.id, &target.action, now) {
207 Ok(Decision::Allowed { grant }) => Verdict::Allowed {
208 grant: grant.to_owned(),
209 },
210 Ok(Decision::Held { .. } | Decision::Refused) => Verdict::refused(trust, target),
211 Err(error) => Verdict::named(&error),
212 },
213 Class::Deliberate => match live.ask(token, &pass.claims().holder.id, target) {
214 Ok(answer) => deliberate_verdict(trust, target, &answer),
215 Err(error) => Verdict::named(&error),
216 },
217 }
218}
219
220fn deliberate_verdict(trust: Trust<'_>, target: &Target, answer: &CheckAnswer) -> Verdict {
221 match (answer.allowed, answer.grant.as_deref(), answer.mode) {
222 (true, Some(grant), Some(Mode::Outright)) => Verdict::Allowed {
223 grant: grant.to_owned(),
224 },
225 (false, Some(grant), Some(mode @ (Mode::ByDraft | Mode::ByTwo))) => Verdict::Held {
226 grant: grant.to_owned(),
227 mode,
228 judgment: Box::new(answer.clone()),
229 },
230 (false, None, None) => Verdict::refused(trust, target),
231 _ => Verdict::Named(
232 Error::CannotAsk("inconsistent live permission answer")
233 .name()
234 .to_owned(),
235 ),
236 }
237}
238
239pub struct Gate {
242 keys: Arc<OnceLock<Mutex<Arc<KeySet>>>>,
244 refresh: Option<Mutex<KeyRefresh>>,
245 issuer: String,
246 database: String,
247 live: Arc<dyn LiveCheck + Send + Sync>,
248 made: Arc<Made>,
251}
252
253type FetchKeys = Arc<dyn Fn() -> Result<KeySet, Error> + Send + Sync>;
254
255struct KeyRefresh {
256 fetch: FetchKeys,
257 last: Option<Instant>,
258 interval: Duration,
259}
260
261const PLACED_KEPT_SECS: u64 = 24 * 60 * 60;
265
266const PLACEMENT_WAIT: Duration = Duration::from_secs(10);
270
271#[derive(Debug, Default)]
278pub struct Made {
279 state: Mutex<MadeState>,
280 landed: Condvar,
281}
282
283#[derive(Debug, Default)]
284struct MadeState {
285 made: BTreeMap<(String, String), (u64, bool)>,
287 waits: u64,
289}
290
291impl Made {
292 fn committed(&self, kind: &str, name: &str, now: u64, placing: bool) {
293 let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
294 state
295 .made
296 .retain(|_, (at, _)| now.saturating_sub(*at) < PLACED_KEPT_SECS);
297 state
298 .made
299 .insert((kind.to_owned(), name.to_owned()), (now, !placing));
300 }
301
302 pub fn landed(&self, kind: &str, name: &str) {
305 if let Some((_, landed)) = self
306 .state
307 .lock()
308 .unwrap_or_else(PoisonError::into_inner)
309 .made
310 .get_mut(&(kind.to_owned(), name.to_owned()))
311 {
312 *landed = true;
313 }
314 self.landed.notify_all();
315 }
316
317 fn made_at(&self, kind: &str, name: &str, within: Duration) -> Option<u64> {
320 let key = (kind.to_owned(), name.to_owned());
321 {
322 let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
323 state.waits = state.waits.saturating_add(1);
324 }
325 self.landed.notify_all();
326 let (ended, _) = self
328 .landed
329 .wait_timeout_while(
330 self.state.lock().unwrap_or_else(PoisonError::into_inner),
331 within,
332 |state| state.made.get(&key).is_some_and(|(_, landed)| !landed),
333 )
334 .unwrap_or_else(PoisonError::into_inner);
335 let made = ended.made.get(&key).map(|(at, _)| *at);
336 drop(ended);
337 made
338 }
339}
340
341#[cfg(test)]
342impl Made {
343 pub(super) fn await_waiting(&self) {
345 drop(
346 self.landed
347 .wait_while(
348 self.state.lock().unwrap_or_else(PoisonError::into_inner),
349 |state| state.waits == 0,
350 )
351 .unwrap_or_else(PoisonError::into_inner),
352 );
353 }
354}
355
356impl Gate {
357 pub fn new(
364 keys_json: &str,
365 issuer: &str,
366 database: &str,
367 live: Arc<dyn LiveCheck + Send + Sync>,
368 ) -> Result<Self, Error> {
369 Ok(Self::with_keys(
370 KeySet::from_json(keys_json)?,
371 issuer,
372 database,
373 live,
374 ))
375 }
376
377 #[must_use]
379 pub fn with_keys(
380 keys: KeySet,
381 issuer: &str,
382 database: &str,
383 live: Arc<dyn LiveCheck + Send + Sync>,
384 ) -> Self {
385 Self {
386 keys: Arc::new(OnceLock::from(Mutex::new(Arc::new(keys)))),
387 refresh: None,
388 issuer: issuer.to_owned(),
389 database: database.to_owned(),
390 live,
391 made: Arc::new(Made::default()),
392 }
393 }
394
395 pub fn fetching(
405 fetch: impl Fn() -> Result<KeySet, Error> + Send + Sync + 'static,
406 issuer: &str,
407 database: &str,
408 live: Arc<dyn LiveCheck + Send + Sync>,
409 ) -> std::io::Result<Self> {
410 let interval = crate::door::config::positive_env("HAEMATITE_LYS_KEYS_REFRESH_SECONDS", 30)
412 .map_err(std::io::Error::other)?;
413 let fetch: FetchKeys = Arc::new(fetch);
414 let retry = Arc::clone(&fetch);
415 let keys = Arc::new(OnceLock::new());
416 let landing = Arc::downgrade(&keys);
417 let from = issuer.to_owned();
418 std::thread::Builder::new()
419 .name("haematite-lys-keys".to_owned())
420 .spawn(move || read_keys(retry.as_ref(), &landing, &from))?;
421 Ok(Self {
422 keys,
423 refresh: Some(Mutex::new(KeyRefresh {
424 fetch,
425 last: None,
426 interval: Duration::from_secs(interval.get()),
427 })),
428 issuer: issuer.to_owned(),
429 database: database.to_owned(),
430 live,
431 made: Arc::new(Made::default()),
432 })
433 }
434
435 #[must_use]
437 pub fn made(&self) -> Arc<Made> {
438 Arc::clone(&self.made)
439 }
440
441 pub fn placed(&self, kind: &str, name: &str, now: u64, placing: bool) {
445 self.made.committed(kind, name, now, placing);
446 }
447
448 fn placed_after_pass(&self, token: Option<&str>, target: &Target, now: u64) -> Option<String> {
452 let token = token?;
453 let keys = self.keys_for(Some(token), now).ok()?;
454 let trust = self.trust(&keys);
455 let name = target.id.strip_prefix(&format!("{}.", self.database))?;
456 let made = self.made.made_at(&target.kind, name, PLACEMENT_WAIT)?;
457 let pass = VerifiedPass::verify(token, trust.keys, trust.issuer, AUDIENCE, now).ok()?;
458 if made < pass.claims().iat {
459 return None;
460 }
461 let answer = self
462 .live
463 .ask(token, &pass.claims().holder.id, target)
464 .ok()?;
465 match deliberate_verdict(trust, target, &answer) {
466 Verdict::Allowed { grant } => Some(grant),
467 _ => None,
468 }
469 }
470
471 fn database_right(&self, token: Option<&str>, action: &str, now: u64) -> Option<String> {
473 let target = Target::new(&format!("{AUDIENCE}.database"), &self.database, action).ok()?;
474 let keys = self.keys_for(token, now).ok()?;
475 match decide(
476 self.trust(&keys),
477 token,
478 &target,
479 Class::Hot,
480 self.live.as_ref(),
481 now,
482 ) {
483 Verdict::Allowed { grant } => Some(grant),
484 _ => None,
485 }
486 }
487
488 fn snapshot(&self) -> Result<Arc<KeySet>, Failure> {
491 let held = self
492 .keys
493 .get()
494 .ok_or_else(|| named_failure(LYS_KEYS_PENDING))?;
495 held.lock()
496 .map(|keys| Arc::clone(&keys))
497 .map_err(|_| Failure::new("lys_keys_unavailable", "Lys key state was poisoned"))
498 }
499
500 fn keys_for(&self, token: Option<&str>, now: u64) -> Result<Arc<KeySet>, Failure> {
501 let keys = self.snapshot()?;
502 let Some(token) = token else {
503 return Ok(keys);
504 };
505 let unknown = |keys: &KeySet| {
506 VerifiedPass::verify(token, keys, &self.issuer, AUDIENCE, now)
507 .is_err_and(|error| error.name() == "unpublished_key")
508 };
509 if !unknown(&keys) {
510 return Ok(keys);
511 }
512 let Some(refresh) = &self.refresh else {
513 return Ok(keys);
514 };
515 let mut refresh = refresh
516 .lock()
517 .map_err(|_| Failure::new("lys_keys_unavailable", "Lys key refresh was poisoned"))?;
518 let current = self.snapshot()?;
519 if !unknown(¤t)
520 || refresh
521 .last
522 .is_some_and(|last| last.elapsed() < refresh.interval)
523 {
524 return Ok(current);
525 }
526 refresh.last = Some(Instant::now());
527 let fresh = (refresh.fetch)().map_err(|error| {
528 eprintln!("haem/1 lys key refresh refused: {}", error.name());
529 named_failure(error.name())
530 })?;
531 let fresh = Arc::new(fresh);
532 let held = self
533 .keys
534 .get()
535 .ok_or_else(|| named_failure(LYS_KEYS_PENDING))?;
536 *held
537 .lock()
538 .map_err(|_| Failure::new("lys_keys_unavailable", "Lys key state was poisoned"))? =
539 Arc::clone(&fresh);
540 drop(refresh);
542 eprintln!("haem/1 lys keys refreshed");
543 Ok(fresh)
544 }
545
546 fn trust<'a>(&'a self, keys: &'a KeySet) -> Trust<'a> {
547 Trust {
548 keys,
549 issuer: &self.issuer,
550 audience: AUDIENCE,
551 }
552 }
553
554 pub fn present(&self, token: &str, now: u64) -> Result<Value, Failure> {
560 let keys = self.keys_for(Some(token), now)?;
561 let trust = self.trust(&keys);
562 let pass = VerifiedPass::verify(token, trust.keys, trust.issuer, AUDIENCE, now)
563 .map_err(|error| named_failure(error.name()))?;
564 let claims = pass.claims();
565 Ok(json!({"holder": claims.holder.id, "expires": claims.exp}))
566 }
567
568 pub fn authorize(
578 &self,
579 token: Option<&str>,
580 operation: &Operation,
581 words: Option<&str>,
582 now: u64,
583 ) -> Result<Vec<String>, Failure> {
584 self.authorize_with(token, operation, words, now, &|_, _| false)
585 }
586
587 pub fn authorize_with(
598 &self,
599 token: Option<&str>,
600 operation: &Operation,
601 words: Option<&str>,
602 now: u64,
603 absent: &dyn Fn(&str, &str) -> bool,
604 ) -> Result<Vec<String>, Failure> {
605 let subjects = subjects(operation)?;
606 if subjects.is_empty() {
607 return Ok(Vec::new());
608 }
609 let keys = self.keys_for(token, now)?;
610 let alone = subjects.len() == 1;
611 let mut grants = Vec::new();
612 for subject in subjects {
613 let target = target(subject.route, &self.database, subject.name.as_deref())
614 .map_err(|error| named_failure(error.name()))?;
615 match decide(
616 self.trust(&keys),
617 token,
618 &target,
619 subject.route.class,
620 self.live.as_ref(),
621 now,
622 ) {
623 Verdict::Allowed { grant } => grants.push(grant),
624 Verdict::Held { judgment, .. } if alone => {
625 return Err(self.hold(token, operation, words, &target, &judgment));
626 }
627 Verdict::Held { grant, .. } => {
628 return Err(Failure::new(
629 "held_for_approval",
630 format!(
631 "held under grant {grant}; an act on more than one resource is never drafted"
632 ),
633 ));
634 }
635 Verdict::Refused(_)
636 if subject.route.class == Class::Hot
637 && let Some(grant) = self.placed_after_pass(token, &target, now) =>
638 {
639 grants.push(grant);
640 }
641 Verdict::Refused(_)
642 if subject.route.class == Class::Hot
643 && subject
644 .name
645 .as_deref()
646 .is_some_and(|name| absent(subject.route.kind, name))
647 && let Some(grant) =
648 self.database_right(token, subject.route.action, now) =>
649 {
650 grants.push(grant);
651 }
652 Verdict::Refused(refusal) => {
653 let shape = serde_json::to_string(&refusal).map_err(Failure::storage)?;
654 return Err(Failure::new("permission_refused", shape));
655 }
656 Verdict::Named(name) => return Err(named_failure(&name)),
657 }
658 }
659 Ok(grants)
660 }
661
662 fn hold(
668 &self,
669 token: Option<&str>,
670 operation: &Operation,
671 words: Option<&str>,
672 target: &Target,
673 judgment: &CheckAnswer,
674 ) -> Failure {
675 let Some(token) = token else {
676 return named_failure(PASS_MISSING);
677 };
678 let (Some(grant), Some(mode)) = (judgment.grant.as_deref(), judgment.mode) else {
680 return named_failure(Error::CannotAsk("inconsistent live permission answer").name());
681 };
682 let (Operation::Mutate(mutation), Some(words)) = (operation, words) else {
683 return Failure::new(
684 "held_for_approval",
685 format!("held under grant {grant}; only a change is drafted for approval"),
686 );
687 };
688 let request = DraftRequest {
689 operation: format!(
690 "{AUDIENCE}:{}:{}:{}",
691 self.database, mutation.branch, mutation.request_id
692 ),
693 grant: grant.to_owned(),
694 target: target.clone(),
695 request_digest: request_digest(words),
696 words: words.to_owned(),
697 };
698 match self.live.draft(token, &request, judgment) {
699 Ok(draft) => Failure::new(
700 "held_for_approval",
701 json!({"draft": draft, "grant": grant, "mode": mode}).to_string(),
702 ),
703 Err(error) => named_failure(error.name()),
704 }
705 }
706
707 pub fn run_drafts(
717 &self,
718 service: &Service,
719 token: Option<&str>,
720 after: Option<&str>,
721 ) -> Result<Value, Failure> {
722 let Some(token) = token else {
723 return Err(named_failure(PASS_MISSING));
724 };
725 let mut drafts = Drafts::new(service, &self.database);
726 match self
727 .live
728 .run_drafts(token, &mut drafts, after)
729 .map_err(|error| *error)
730 {
731 Ok(report) => Ok(json!({"acknowledged": report.acknowledged, "next": report.next})),
732 Err(RunError::Pull(error)) => Err(named_failure(error.name())),
733 Err(RunError::Execution { draft, source }) => Err(Failure::new(
734 "draft_execution_failed",
735 format!("draft {draft} was not applied: {source}"),
736 )),
737 Err(RunError::Record { draft, source, .. }) => Err(Failure::new(
738 "draft_acknowledgment_failed",
739 format!(
740 "draft {draft} was applied but Lys did not take its outcome ({}); \
741 the next run answers it again from its receipt",
742 source.name()
743 ),
744 )),
745 }
746 }
747}
748
749#[must_use]
751pub fn now() -> u64 {
752 SystemTime::now()
753 .duration_since(UNIX_EPOCH)
754 .map_or(0, |elapsed| elapsed.as_secs())
755}
756
757fn read_keys(
760 fetch: &dyn Fn() -> Result<KeySet, Error>,
761 landing: &Weak<OnceLock<Mutex<Arc<KeySet>>>>,
762 issuer: &str,
763) {
764 let mut wait = KEYS_RETRY_FIRST;
765 loop {
766 if landing.strong_count() == 0 {
767 return;
768 }
769 match fetch() {
770 Ok(keys) => {
771 if landing
772 .upgrade()
773 .is_some_and(|held| held.set(Mutex::new(Arc::new(keys))).is_ok())
774 {
775 eprintln!("haem/1 lys keys read from {issuer}");
776 }
777 return;
778 }
779 Err(error) => {
780 eprintln!(
781 "haem/1 lys keys pending: {} from {issuer}; asking again in {} ms",
782 error.name(),
783 wait.as_millis()
784 );
785 std::thread::sleep(wait);
786 wait = wait.saturating_mul(2).min(KEYS_RETRY_MOST);
787 }
788 }
789 }
790}
791
792fn named_failure(name: &str) -> Failure {
793 let words = match name {
794 "lys_could_not_be_asked" => "Lys could not be asked",
795 PASS_MISSING => "present a Lys pass with session.pass first",
796 LYS_KEYS_PENDING => "Lys's published keys have not been read yet",
797 _ => "the Lys pass was refused",
798 };
799 Failure::new(name, words)
800}
801
802#[cfg(test)]
803#[path = "pass_tests.rs"]
804mod tests;
805
806#[cfg(test)]
807#[path = "pass_draft_tests.rs"]
808mod draft_tests;
809
810#[cfg(test)]
811#[path = "pass_collection_tests.rs"]
812mod collection_tests;
813
814#[cfg(test)]
815#[path = "pass_merge_tests.rs"]
816mod merge_tests;
817
818#[cfg(test)]
819#[path = "pass_keys_tests.rs"]
820mod keys_tests;
821
822#[cfg(test)]
823#[path = "pass_rotation_tests.rs"]
824mod rotation_tests;