liminal_server/auth_lys/
live.rs

1//! The live Lys question over HTTP (ACCESS-002 R1, ACCESS-003 R1).
2//!
3//! `lys-pass`'s batch route is asynchronous; the connection's decision path
4//! is not. Each ask runs its future to completion on a scoped thread against
5//! the server's tokio runtime, so a connection process never blocks the
6//! runtime's own workers and never nests a runtime. Deliberate asks are rare
7//! by definition (opening a link, deploying a service, an administrator's
8//! act), so a thread per ask is the honest cost; nothing is cached, and no
9//! failure falls back to the hot seam. A held answer is recorded as a draft
10//! with the holder's own pass before it is reported, so a held open is a
11//! draft Lys retains, never a promise liminal invented.
12
13use super::GrantMode;
14use super::lys_pass_adapter::{LiveAnswer, LiveLysAsk};
15use lys_pass::deliberate::CheckAnswer;
16use lys_pass::drafts::{DraftRequest, request_digest};
17use lys_pass::{Client, Error, Mode, Target};
18use std::sync::Arc;
19use std::sync::atomic::{AtomicU64, Ordering};
20use std::sync::{Mutex, Weak};
21use std::time::{Duration, Instant};
22
23static ASK_METRICS: Mutex<Vec<Weak<AskMetrics>>> = Mutex::new(Vec::new());
24
25#[derive(Debug, Default)]
26struct AskMetrics {
27    count: AtomicU64,
28    maximum_ns: AtomicU64,
29}
30
31/// Completed requests across the live Lys clients in this process.
32#[derive(Clone, Copy, Debug, Default, serde::Serialize)]
33pub(crate) struct AskStatistics {
34    pub count: u64,
35    pub max_elapsed_ms: f64,
36}
37
38impl AskMetrics {
39    fn record(&self, elapsed: Duration) {
40        let nanos = u64::try_from(elapsed.as_nanos()).unwrap_or(u64::MAX);
41        self.maximum_ns.fetch_max(nanos, Ordering::Relaxed);
42        self.count.fetch_add(1, Ordering::Release);
43    }
44
45    fn snapshot(&self) -> AskStatistics {
46        AskStatistics {
47            count: self.count.load(Ordering::Acquire),
48            max_elapsed_ms: Duration::from_nanos(self.maximum_ns.load(Ordering::Relaxed))
49                .as_secs_f64()
50                * 1000.0,
51        }
52    }
53}
54
55/// Reads the same per-client counters the request path updates, without a task or timer.
56pub(crate) fn ask_statistics() -> Result<AskStatistics, Error> {
57    let mut clients = ASK_METRICS
58        .lock()
59        .map_err(|_| Error::CannotAsk("Lys ask metrics registry poisoned"))?;
60    let mut total = AskStatistics::default();
61    clients.retain(|client| {
62        let Some(client) = client.upgrade() else {
63            return false;
64        };
65        let snapshot = client.snapshot();
66        total.count = total.count.saturating_add(snapshot.count);
67        total.max_elapsed_ms = total.max_elapsed_ms.max(snapshot.max_elapsed_ms);
68        true
69    });
70    drop(clients);
71    Ok(total)
72}
73
74/// `lys-pass`'s client behind the [`LiveLysAsk`] seam.
75pub struct HttpLysAsk {
76    client: Client,
77    metrics: Arc<AskMetrics>,
78    handle: tokio::runtime::Handle,
79    /// This liminal's Lys app credential, which the batch check is asked
80    /// with when present; absent, the holder's pass is the bearer.
81    asker: Option<String>,
82}
83
84/// The Lys origin the lys-pass client takes (from 0.3.3 it mounts `/api`
85/// itself): a configured API mount (`api_url`, `.../api/`) names its origin.
86///
87/// # Errors
88/// The value is not a URL.
89pub fn lys_origin(url: &str) -> Result<reqwest::Url, Error> {
90    let mut url: reqwest::Url = url
91        .parse()
92        .map_err(|_| Error::Invalid("issuer is not a URL"))?;
93    if matches!(url.path(), "/api" | "/api/") {
94        url.set_path("/");
95    }
96    Ok(url)
97}
98
99impl std::fmt::Debug for HttpLysAsk {
100    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
101        formatter.debug_struct("HttpLysAsk").finish_non_exhaustive()
102    }
103}
104
105impl HttpLysAsk {
106    /// Asks `issuer` through a fresh HTTP client driven by `handle`'s runtime.
107    ///
108    /// # Errors
109    /// The issuer is not an HTTP origin.
110    pub fn new(issuer: &str, handle: tokio::runtime::Handle) -> Result<Self, Error> {
111        let issuer = lys_origin(issuer)?;
112        let metrics = Arc::new(AskMetrics::default());
113        {
114            let mut clients = ASK_METRICS
115                .lock()
116                .map_err(|_| Error::CannotAsk("Lys ask metrics registry poisoned"))?;
117            clients.retain(|client| client.strong_count() > 0);
118            clients.push(Arc::downgrade(&metrics));
119        }
120        Ok(Self {
121            client: Client::new(
122                reqwest::Client::builder()
123                    .tcp_keepalive(Duration::from_secs(30))
124                    .pool_idle_timeout(None)
125                    .build()
126                    .map_err(|error| Error::Transport(Box::new(error)))?,
127                issuer,
128            )?,
129            handle,
130            metrics,
131            asker: None,
132        })
133    }
134
135    /// Asks the batch check as this liminal's app, `credential`, about the
136    /// holder, instead of with the holder's own pass.
137    #[must_use]
138    pub fn asking_as(mut self, credential: String) -> Self {
139        self.asker = Some(credential);
140        self
141    }
142
143    #[cfg(test)]
144    fn statistics(&self) -> AskStatistics {
145        self.metrics.snapshot()
146    }
147
148    /// Runs one ask to completion on its own thread against the runtime.
149    fn block<T, F>(&self, route: &'static str, future: F) -> Result<T, Error>
150    where
151        T: Send,
152        F: std::future::Future<Output = Result<T, Error>> + Send,
153    {
154        let started = Instant::now();
155        let handle = &self.handle;
156        let answer =
157            std::thread::scope(|scope| scope.spawn(move || handle.block_on(future)).join())
158                .unwrap_or_else(|_| Err(Error::CannotAsk("the live ask did not complete")));
159        let elapsed = started.elapsed();
160        self.metrics.record(elapsed);
161        tracing::debug!(target: "liminal::lys_ask", route, ok = answer.is_ok(),
162            elapsed_ms = elapsed.as_secs_f64() * 1000.0, "Lys ask completed");
163        answer
164    }
165
166    /// Records the held act as a draft with the holder's own pass, returning
167    /// the draft Lys retained.
168    fn draft(
169        &self,
170        pass: &str,
171        holder: &str,
172        target: &Target,
173        grant: &str,
174        judgment: &CheckAnswer,
175    ) -> Result<String, Error> {
176        let words = format!(
177            "{holder} asks {} on {}:{}",
178            target.action, target.kind, target.id
179        );
180        let draft = DraftRequest {
181            operation: format!(
182                "liminal:{holder}:{}:{}:{}",
183                target.kind, target.id, target.action
184            ),
185            grant: grant.to_owned(),
186            target: Target::new(&target.kind, &target.id, &target.action)?,
187            request_digest: request_digest(&words),
188            words,
189        };
190        self.block(
191            "/api/product-drafts",
192            self.client.create_draft(pass, &draft, judgment),
193        )
194        .map(|created| created.draft)
195    }
196}
197
198impl LiveLysAsk for HttpLysAsk {
199    fn ask(
200        &self,
201        pass: &str,
202        holder: &str,
203        target: &Target,
204        _now: u64,
205    ) -> Result<LiveAnswer, Error> {
206        let answer = self.block(
207            "/api/grants/check/batch",
208            self.client.check_batch(
209                self.asker.as_deref().unwrap_or(pass),
210                holder,
211                std::slice::from_ref(target),
212                None,
213            ),
214        )?;
215        let Some(result) = answer.results.into_iter().next() else {
216            return Err(Error::CannotAsk("the live answer carried no result"));
217        };
218        match (
219            result.allowed,
220            result.mode.as_ref(),
221            result.grant.as_deref(),
222            result.refusal.as_deref(),
223        ) {
224            (true, Some(Mode::Outright), Some(grant), None) => Ok(LiveAnswer::Outright {
225                grant: grant.to_owned(),
226            }),
227            (false, Some(Mode::ByDraft), Some(grant), None) => Ok(LiveAnswer::Held {
228                mode: GrantMode::ByDraft,
229                grant: grant.to_owned(),
230                draft: self.draft(pass, holder, target, grant, &result)?,
231            }),
232            (false, Some(Mode::ByTwo), Some(grant), None) => Ok(LiveAnswer::Held {
233                mode: GrantMode::ByTwo,
234                grant: grant.to_owned(),
235                draft: self.draft(pass, holder, target, grant, &result)?,
236            }),
237            (false, None, None, Some(refusal)) => Ok(LiveAnswer::Refused {
238                refusal: refusal.to_owned(),
239            }),
240            _ => Err(Error::CannotAsk("inconsistent live permission answer")),
241        }
242    }
243}
244
245#[cfg(test)]
246#[path = "live/timing_tests.rs"]
247mod timing_tests;
liminal_server/auth_lys/live.rs

liminal_server/auth_lys/
live.rs
