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#[path = "drafts.rs"]
75pub mod drafts;
76
77/// Exact prepared operation; the target owner retains these immutable bytes.
78pub struct PreparedAct<'a> {
79    /// The holder proved by the admitted pass, never taken from a JSON body.
80    pub holder: &'a str,
81    /// The authoritative resource and deliberate action.
82    pub target: &'a Target,
83    /// Stable id of this prepared operation, reused only for the same bytes.
84    pub operation: &'a str,
85    /// Exact words presented to the approver, unchanged.
86    pub words: &'a str,
87    /// The server's preparation instant in epoch milliseconds.
88    pub at_ms: u64,
89}
90
91/// The target registry keeps preparation before the remote draft is created,
92///
93/// then binds its returned id and waiting state durably. A failed handoff is
94/// named; it never turns the held decision into permission to execute.
95pub trait PreparedDraftKeeper: Send + Sync {
96    /// Retains the immutable prepared act under its operation and digest.
97    ///
98    /// # Errors
99    /// A conflicting operation, configured bound or store failure.
100    fn prepare(&self, holder: &str, request: &DraftRequest, at_ms: u64) -> Result<(), Error>;
101    /// Binds the retained draft and writes waiting through the agent state writer.
102    ///
103    /// # Errors
104    /// Binding or durable waiting-state failure.
105    fn held(&self, holder: &str, id: &str, request: &DraftRequest, at_ms: u64)
106    -> Result<(), Error>;
107}
108
109/// Prepared answers carry held words as data and exclude them from diagnostics.
110pub struct PreparedAnswer {
111    /// The actual Lys decision, still held when approval is required.
112    pub decision: LiveAnswer,
113    /// Exact words of a held act, absent for outright or refused decisions.
114    pub words: Option<String>,
115}
116
117impl std::fmt::Debug for PreparedAnswer {
118    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
119        formatter
120            .debug_struct("PreparedAnswer")
121            .finish_non_exhaustive()
122    }
123}
124/// `lys-pass`'s client behind the [`LiveLysAsk`] seam.
125pub struct HttpLysAsk {
126    client: Client,
127    metrics: Arc<AskMetrics>,
128    handle: tokio::runtime::Handle,
129    /// This liminal's Lys app credential, which the batch check is asked
130    /// with when present; absent, the holder's pass is the bearer.
131    asker: Option<String>,
132    keeper: Option<Arc<dyn PreparedDraftKeeper>>,
133}
134
135/// The Lys origin the lys-pass client takes (from 0.3.3 it mounts `/api`
136/// itself): a configured API mount (`api_url`, `.../api/`) names its origin.
137///
138/// # Errors
139/// The value is not a URL.
140pub fn lys_origin(url: &str) -> Result<reqwest::Url, Error> {
141    let mut url: reqwest::Url = url
142        .parse()
143        .map_err(|_| Error::Invalid("issuer is not a URL"))?;
144    if matches!(url.path(), "/api" | "/api/") {
145        url.set_path("/");
146    }
147    Ok(url)
148}
149
150impl std::fmt::Debug for HttpLysAsk {
151    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
152        formatter.debug_struct("HttpLysAsk").finish_non_exhaustive()
153    }
154}
155
156impl HttpLysAsk {
157    /// Asks `issuer` through a fresh HTTP client driven by `handle`'s runtime.
158    ///
159    /// # Errors
160    /// The issuer is not an HTTP origin.
161    pub fn new(issuer: &str, handle: tokio::runtime::Handle) -> Result<Self, Error> {
162        let issuer = lys_origin(issuer)?;
163        let metrics = Arc::new(AskMetrics::default());
164        {
165            let mut clients = ASK_METRICS
166                .lock()
167                .map_err(|_| Error::CannotAsk("Lys ask metrics registry poisoned"))?;
168            clients.retain(|client| client.strong_count() > 0);
169            clients.push(Arc::downgrade(&metrics));
170        }
171        Ok(Self {
172            client: Client::new(
173                reqwest::Client::builder()
174                    .no_proxy()
175                    .redirect(reqwest::redirect::Policy::none())
176                    .tcp_keepalive(Duration::from_secs(30))
177                    .pool_idle_timeout(None)
178                    .build()
179                    .map_err(|error| Error::Transport(Box::new(error)))?,
180                issuer,
181            )?,
182            handle,
183            metrics,
184            asker: None,
185            keeper: None,
186        })
187    }
188
189    /// Asks the batch check as this liminal's app, `credential`, about the
190    /// holder, instead of with the holder's own pass.
191    #[must_use]
192    pub fn asking_as(mut self, credential: String) -> Self {
193        self.asker = Some(credential);
194        self
195    }
196
197    /// Installs the target registry preparation and waiting writer.
198    #[must_use]
199    pub fn with_drafts(mut self, keeper: Arc<dyn PreparedDraftKeeper>) -> Self {
200        self.keeper = Some(keeper);
201        self
202    }
203
204    #[cfg(test)]
205    fn statistics(&self) -> AskStatistics {
206        self.metrics.snapshot()
207    }
208
209    /// Runs one ask to completion on its own thread against the runtime.
210    fn block<T, F>(&self, route: &'static str, future: F) -> Result<T, Error>
211    where
212        T: Send,
213        F: std::future::Future<Output = Result<T, Error>> + Send,
214    {
215        let started = Instant::now();
216        let answer = drafts::block(&self.handle, future);
217        let elapsed = started.elapsed();
218        self.metrics.record(elapsed);
219        tracing::debug!(target: "liminal::lys_ask", route, ok = answer.is_ok(),
220            elapsed_ms = elapsed.as_secs_f64() * 1000.0, "Lys ask completed");
221        answer
222    }
223
224    fn draft(
225        &self,
226        pass: &str,
227        act: &PreparedAct<'_>,
228        grant: &str,
229        judgment: &CheckAnswer,
230    ) -> Result<String, Error> {
231        let draft = DraftRequest {
232            operation: act.operation.to_owned(),
233            grant: grant.to_owned(),
234            target: Target::new(&act.target.kind, &act.target.id, &act.target.action)?,
235            request_digest: request_digest(act.words),
236            words: act.words.to_owned(),
237        };
238        if let Some(keeper) = &self.keeper {
239            keeper.prepare(act.holder, &draft, act.at_ms)?;
240        }
241        let created = self.block(
242            "/api/product-drafts",
243            self.client.create_draft(pass, &draft, judgment),
244        )?;
245        if let Some(keeper) = &self.keeper {
246            keeper.held(act.holder, &created.draft, &draft, act.at_ms)?;
247        }
248        Ok(created.draft)
249    }
250    /// Asks about a complete prepared act, preserving the exact approver words.
251    ///
252    /// # Errors
253    /// Missing keeper, invalid or oversized input, or Lys's live refusal.
254    pub fn ask_prepared(
255        &self,
256        pass: &str,
257        act: &PreparedAct<'_>,
258        limits: drafts::DraftLimits,
259    ) -> Result<PreparedAnswer, Error> {
260        if self.keeper.is_none() {
261            return Err(Error::CannotAsk("draft preparation writer is missing"));
262        }
263        // The pass is a credential, bounded at admission (credential_bytes), not draft metadata.
264        if pass.is_empty()
265            || act.words.is_empty()
266            || act.words.len() > limits.words_bytes.get()
267            || [
268                act.holder,
269                act.operation,
270                &act.target.kind,
271                &act.target.id,
272                &act.target.action,
273            ]
274            .iter()
275            .any(|value| {
276                value.is_empty()
277                    || value.len() > limits.metadata_bytes.get()
278                    || value.chars().any(char::is_control)
279            })
280        {
281            return Err(Error::Invalid("prepared draft bound exceeded"));
282        }
283        let decision = self.prepared_decision(pass, act)?;
284        let words = matches!(decision, LiveAnswer::Held { .. }).then(|| act.words.to_owned());
285        Ok(PreparedAnswer { decision, words })
286    }
287
288    fn prepared_decision(&self, pass: &str, act: &PreparedAct<'_>) -> Result<LiveAnswer, Error> {
289        let answer = self.block(
290            "/api/grants/check/batch",
291            self.client.check_batch(
292                self.asker.as_deref().unwrap_or(pass),
293                act.holder,
294                std::slice::from_ref(act.target),
295                None,
296            ),
297        )?;
298        if answer.degraded.is_some() {
299            return Err(Error::CannotAsk("live permission projection degraded"));
300        }
301        let Some(result) = answer.results.into_iter().next() else {
302            return Err(Error::CannotAsk("the live answer carried no result"));
303        };
304        if result.degraded.is_some() {
305            return Err(Error::CannotAsk("live permission projection degraded"));
306        }
307        match (
308            result.allowed,
309            result.mode.as_ref(),
310            result.grant.as_deref(),
311            result.refusal.as_deref(),
312        ) {
313            (true, Some(Mode::Outright), Some(grant), None) => Ok(LiveAnswer::Outright {
314                grant: grant.to_owned(),
315            }),
316            (false, Some(Mode::ByDraft), Some(grant), None) => Ok(LiveAnswer::Held {
317                mode: GrantMode::ByDraft,
318                grant: grant.to_owned(),
319                draft: self.draft(pass, act, grant, &result)?,
320            }),
321            (false, Some(Mode::ByTwo), Some(grant), None) => Ok(LiveAnswer::Held {
322                mode: GrantMode::ByTwo,
323                grant: grant.to_owned(),
324                draft: self.draft(pass, act, grant, &result)?,
325            }),
326            (false, None, None, Some(refusal)) => Ok(LiveAnswer::Refused {
327                refusal: refusal.to_owned(),
328            }),
329            _ => Err(Error::CannotAsk("inconsistent live permission answer")),
330        }
331    }
332}
333
334impl LiveLysAsk for HttpLysAsk {
335    fn ask(
336        &self,
337        pass: &str,
338        holder: &str,
339        target: &Target,
340        now: u64,
341    ) -> Result<LiveAnswer, Error> {
342        let words = format!(
343            "{holder} asks {} on {}:{}",
344            target.action, target.kind, target.id
345        );
346        let operation = format!(
347            "liminal:{holder}:{}:{}:{}",
348            target.kind, target.id, target.action
349        );
350        let at_ms = now
351            .checked_mul(1000)
352            .ok_or(Error::Invalid("draft instant overflow"))?;
353        self.prepared_decision(
354            pass,
355            &PreparedAct {
356                holder,
357                target,
358                operation: &operation,
359                words: &words,
360                at_ms,
361            },
362        )
363    }
364}
365
366#[path = "live/drafts_join.rs"]
367mod drafts_join;
368pub use drafts_join::PreparedOwner;
369
370#[cfg(test)]
371#[path = "live/timing_tests.rs"]
372mod timing_tests;
liminal_server/auth_lys/live.rs

liminal_server/auth_lys/
live.rs
