liminal_server/http/
admin_drafts.rs

1//! The existing operator HTTP listener dispatches this exact route.
2//!
3//! Signed passes carry expanded actions, not relation names. The schema's
4//! dedicated pull action must belong only to network administrator and operator;
5//! an unrelated call/deploy grant cannot be mistaken for either relation.
6
7use std::num::NonZeroUsize;
8use std::sync::Arc;
9
10use lys_pass::{Decision, Target};
11use serde_json::{Value, json};
12
13use crate::auth_lys::LysPassConsumer;
14use crate::auth_lys::live::drafts::{
15    AtomicDraftStore, DraftError, DraftExecutor, DraftLimits, PullReport, PullSignal,
16    valid_metadata, validate_bounds,
17};
18use crate::auth_lys::lys_pass_adapter::LysPassAdapter;
19
20/// One read-only administrative pull, after the person approves in Lys.
21pub const PULL_PATH: &str = "/admin/drafts/pull";
22/// A person's outright agent act, taken at once rather than held by draft.
23pub const CREATE_PATH: &str = "/admin/agents/create";
24/// This action is granted only by the network's administrator/operator relations.
25pub const PULL_ACTION: &str = "pull_drafts";
26
27/// Explicit HTTP input bounds, without a timer or an implicit default.
28#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Deserialize)]
29#[serde(deny_unknown_fields)]
30pub struct AdminLimits {
31    /// Bounds one bearer before signature verification and allocation.
32    pub credential_bytes: NonZeroUsize,
33    /// Bounds the empty-object request body, refusing extra authority fields.
34    pub body_bytes: NonZeroUsize,
35}
36
37/// Borrowed from the HTTP parser; diagnostics never include its bearer or body.
38pub struct AdminRequest<'a> {
39    /// The parsed request method.
40    pub method: &'a str,
41    /// The exact path, with no accepted query parameters.
42    pub path: &'a str,
43    /// The one Authorization header; the HTTP parser refuses duplicates.
44    pub authorization: Option<&'a str>,
45    /// Empty bytes or an empty JSON object.
46    pub body: &'a [u8],
47    /// The server's operation time, never a caller's body field.
48    pub at_ms: u64,
49}
50
51/// A response containing counts or a stable refusal, never prepared bytes.
52pub struct AdminResponse {
53    /// HTTP status.
54    pub status: u16,
55    /// JSON response body.
56    pub body: Value,
57}
58
59/// The route uses its server-configured network and borrowed machine connector.
60pub struct AdminDrafts<S> {
61    puller: DraftPuller<S>,
62    adapter: Arc<LysPassAdapter>,
63    network: String,
64    limits: AdminLimits,
65    ask: Option<String>,
66}
67
68impl<S: AtomicDraftStore> AdminDrafts<S> {
69    pub(crate) async fn signal(
70        &mut self,
71        connector: &str,
72        signal: PullSignal,
73        at_ms: u64,
74    ) -> Result<PullReport, DraftError> {
75        self.puller.pull(connector, signal, at_ms).await
76    }
77
78    /// Uses the same pass verifier and exact network as the wire admission.
79    ///
80    /// # Errors
81    /// The configured network is absent, controlled or oversized.
82    pub fn new(
83        puller: DraftPuller<S>,
84        adapter: Arc<LysPassAdapter>,
85        network: String,
86        limits: AdminLimits,
87    ) -> Result<Self, DraftError> {
88        if network.is_empty()
89            || network.chars().any(char::is_control)
90            || network.len() > limits.credential_bytes.get()
91        {
92            return Err(DraftError::Bound);
93        }
94        Ok(Self {
95            puller,
96            adapter,
97            network,
98            limits,
99            ask: None,
100        })
101    }
102
103    /// Asks Lys at the configured API address (`api_url`) rather than the
104    /// issuer origin a pass names, as every other Lys call this server makes.
105    #[must_use]
106    pub fn asking_at(mut self, api_url: Option<String>) -> Self {
107        self.ask = api_url;
108        self
109    }
110
111    /// Handles the route without accepting authority or credentials in JSON.
112    /// A held right is refused; this read cannot create another approval draft.
113    /// The person's pass authorizes the pull, while only the machine connector
114    /// reaches the product draft endpoints.
115    ///
116    /// # Errors
117    /// An admitted pull failed; the endpoint must return this named failure,
118    /// never a success response or a timer-based retry.
119    pub async fn handle(
120        &mut self,
121        request: AdminRequest<'_>,
122        connector_pass: &str,
123    ) -> Result<AdminResponse, DraftError> {
124        if request.path != PULL_PATH {
125            return Ok(refused(404, "draft_pull_route_unknown"));
126        }
127        if request.method != "POST" {
128            return Ok(refused(405, "draft_pull_method_refused"));
129        }
130        if request.body.len() > self.limits.body_bytes.get() {
131            return Ok(refused(413, "draft_pull_body_bound"));
132        }
133        if !request.body.is_empty() {
134            let body: Value = match serde_json::from_slice(request.body) {
135                Ok(body) => body,
136                Err(_) => return Ok(refused(400, "draft_pull_body_invalid")),
137            };
138            if !body.as_object().is_some_and(serde_json::Map::is_empty) {
139                return Ok(refused(400, "draft_pull_body_invalid"));
140            }
141        }
142        let Some(pass) = request
143            .authorization
144            .and_then(|header| header.strip_prefix("Bearer "))
145        else {
146            return Ok(refused(401, "draft_pull_pass_missing"));
147        };
148        if pass.is_empty()
149            || pass.len() > self.limits.credential_bytes.get()
150            || pass.chars().any(char::is_whitespace)
151            || pass.chars().any(char::is_control)
152        {
153            return Ok(refused(401, "draft_pull_pass_invalid"));
154        }
155        let now = request.at_ms / 1000;
156        let Ok(verified) = self.adapter.verify_offline(pass.as_bytes(), "liminal", now) else {
157            return Ok(refused(401, "draft_pull_pass_invalid"));
158        };
159        match verified
160            .verified()
161            .evaluate("liminal.network", &self.network, PULL_ACTION, now)
162        {
163            Ok(Decision::Allowed { .. }) => {}
164            Ok(Decision::Held { .. } | Decision::Refused) => {
165                return Ok(refused(403, "draft_pull_network_operator_required"));
166            }
167            Err(_) => return Ok(refused(401, "draft_pull_pass_invalid")),
168        }
169        // Check the current grant, so revocation cannot be bypassed by an
170        // otherwise unexpired offline pass on the administration plane.
171        let http = reqwest::Client::builder()
172            .no_proxy()
173            .redirect(reqwest::redirect::Policy::none())
174            .build()
175            .map_err(|_| DraftError::Lys("transport_failed".to_owned()))?;
176        let issuer = crate::auth_lys::live::lys_origin(
177            self.ask.as_deref().unwrap_or_else(|| self.adapter.issuer()),
178        )
179        .map_err(|_| DraftError::Contract)?;
180        let client = lys_pass::Client::new(http, issuer)
181            .map_err(|error| DraftError::Lys(error.name().to_owned()))?;
182        let target = Target::new("liminal.network", &self.network, PULL_ACTION)
183            .map_err(|_| DraftError::Contract)?;
184        let answer = client
185            // Lys's live check takes the app's own credential (lys-app), never
186            // the person's pass, whose holder is named as the subject instead.
187            .check_batch(connector_pass, verified.holder(), &[target], None)
188            .await
189            .map_err(|error| DraftError::Lys(error.name().to_owned()))?;
190        let Some(result) = answer.results.first() else {
191            return Err(DraftError::Contract);
192        };
193        if answer.degraded.is_some() || result.degraded.is_some() {
194            return Ok(refused(503, "draft_pull_permission_degraded"));
195        }
196        if !result.allowed || result.mode != Some(lys_pass::Mode::Outright) {
197            if let Some(name) = &result.refusal {
198                return Ok(AdminResponse {
199                    status: 403,
200                    body: json!({"refusal":name, "reason":result.reason}),
201                });
202            }
203            return Ok(refused(403, "draft_pull_network_operator_required"));
204        }
205        let report = self
206            .puller
207            .pull(connector_pass, PullSignal::Console, request.at_ms)
208            .await?;
209        Ok(AdminResponse {
210            status: 200,
211            body: serde_json::to_value(report).map_err(|_| DraftError::Store)?,
212        })
213    }
214}
215
216#[path = "admin_outright.rs"]
217mod outright;
218pub use outright::OutrightAsk;
219
220fn refused(status: u16, name: &str) -> AdminResponse {
221    AdminResponse {
222        status,
223        body: json!({"refusal":name}),
224    }
225}
226
227use lys_pass::drafts::{ApprovedDraft, Execution, Executor, execute_once};
228use std::collections::BTreeSet;
229/// One serialized owner. Borrow its machine credential for each told pull;
230/// nothing here holds a person's session or starts work in the background.
231pub struct DraftPuller<S> {
232    client: lys_pass::Client,
233    pub(crate) executor: DraftExecutor<S>,
234    startup_seen: bool,
235}
236
237impl<S> std::fmt::Debug for DraftPuller<S> {
238    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
239        formatter
240            .debug_struct("DraftPuller")
241            .finish_non_exhaustive()
242    }
243}
244
245impl<S: AtomicDraftStore> DraftPuller<S> {
246    /// Connects only to the configured issuer; redirects and environment
247    /// proxies cannot route its machine credential elsewhere.
248    ///
249    /// # Errors
250    /// The issuer or HTTP client could not be constructed.
251    pub fn new(issuer: &str, store: S, limits: DraftLimits) -> Result<Self, DraftError> {
252        let http = reqwest::Client::builder()
253            .no_proxy()
254            .redirect(reqwest::redirect::Policy::none())
255            .build()
256            .map_err(|_| DraftError::Lys("transport_failed".to_owned()))?;
257        let issuer = crate::auth_lys::live::lys_origin(issuer).map_err(|_| DraftError::Contract)?;
258        let client = lys_pass::Client::new(http, issuer)
259            .map_err(|error| DraftError::Lys(error.name().to_owned()))?;
260        Ok(Self {
261            client,
262            executor: DraftExecutor {
263                store,
264                limits,
265                at_ms: 0,
266            },
267            startup_seen: false,
268        })
269    }
270
271    /// Pulls and closes approved drafts only on an explicit signal. An error
272    /// ends the pull; a later caller may reuse the durable receipt. No automatic
273    /// retry is scheduled. The target store revalidates stale prepared acts.
274    ///
275    /// # Errors
276    /// Lys, immutable contract, configured bounds, target storage or delivery.
277    pub async fn pull(
278        &mut self,
279        connector_pass: &str,
280        signal: PullSignal,
281        at_ms: u64,
282    ) -> Result<PullReport, DraftError> {
283        let limits = self.executor.limits;
284        if !valid_metadata(connector_pass, limits.metadata_bytes) {
285            return Err(DraftError::Bound);
286        }
287        if signal == PullSignal::Startup {
288            if self.startup_seen {
289                return Err(DraftError::StartupRepeated);
290            }
291            self.startup_seen = true;
292        }
293        self.executor.at_ms = at_ms;
294        let mut report = PullReport::default();
295        let mut after: Option<String> = None;
296        let mut cursors = BTreeSet::new();
297        let mut ids = BTreeSet::new();
298        let pending = self
299            .acknowledge_pending(connector_pass, &mut report, &mut ids)
300            .await?;
301        loop {
302            if report.pages >= limits.pages.get() {
303                return Err(DraftError::Bound);
304            }
305            let page = self
306                .client
307                .approved_drafts(connector_pass, crate::auth_lys::LYS_APP, after.as_deref())
308                .await
309                .map_err(|error| DraftError::Lys(error.name().to_owned()))?;
310            report.pages += 1;
311            let mut page_ids = BTreeSet::new();
312            for draft in &page.drafts {
313                validate_bounds(draft, limits)?;
314                if !page_ids.insert(&draft.id) {
315                    return Err(DraftError::IdRepeated);
316                }
317            }
318            let additional = page
319                .drafts
320                .iter()
321                .filter(|draft| !ids.contains(&draft.id))
322                .count();
323            if ids.len().saturating_add(additional) > limits.drafts.get() {
324                return Err(DraftError::Bound);
325            }
326            if let Some(next) = &page.next {
327                if !valid_metadata(next, limits.metadata_bytes) {
328                    return Err(DraftError::Bound);
329                }
330                if !cursors.insert(next.clone()) {
331                    return Err(DraftError::CursorRepeated);
332                }
333            }
334            for draft in &page.drafts {
335                if pending.contains(&draft.id) {
336                    self.execution(draft)?;
337                    continue;
338                }
339                if !ids.insert(draft.id.clone()) {
340                    return Err(DraftError::IdRepeated);
341                }
342                let replay = self.executor.receipt(&draft.id)?.is_some();
343                let receipt = self.execution(draft)?;
344                self.executor.store.deliver_outcome(draft, &receipt)?;
345                if replay {
346                    report.replayed += 1;
347                } else {
348                    match &receipt {
349                        Execution::Executed { .. } => report.executed += 1,
350                        Execution::RefusedOnExecution { .. } => report.refused += 1,
351                    }
352                }
353                if !self.executor.store.acknowledged(&draft.id)? {
354                    self.client
355                        .record_execution(connector_pass, draft, &receipt)
356                        .await
357                        .map_err(|error| DraftError::Acknowledgment {
358                            draft: draft.id.clone(),
359                            name: error.name().to_owned(),
360                        })?;
361                    self.executor.store.acknowledge(&draft.id, &receipt)?;
362                    report.acknowledged += 1;
363                }
364            }
365            match page.next {
366                Some(next) => after = Some(next),
367                None => return Ok(report),
368            }
369        }
370    }
371    fn execution(&mut self, draft: &ApprovedDraft) -> Result<Execution, DraftError> {
372        execute_once(draft, &mut self.executor).map_err(|error| match error {
373            lys_pass::drafts::ExecuteError::DigestMismatch => DraftError::ReceiptMismatch,
374            lys_pass::drafts::ExecuteError::Product(error) => error,
375            lys_pass::drafts::ExecuteError::Contract(_) => DraftError::Contract,
376        })
377    }
378
379    async fn acknowledge_pending(
380        &mut self,
381        pass: &str,
382        report: &mut PullReport,
383        ids: &mut BTreeSet<String>,
384    ) -> Result<BTreeSet<String>, DraftError> {
385        let limits = self.executor.limits;
386        loop {
387            let pending = self.executor.store.pending_acknowledgments(limits.drafts)?;
388            if pending.is_empty() {
389                return Ok(ids.clone());
390            }
391            if ids.len().saturating_add(pending.len()) > limits.drafts.get() {
392                return Err(DraftError::Bound);
393            }
394            for draft in pending {
395                validate_bounds(&draft, limits)?;
396                if !ids.insert(draft.id.clone()) {
397                    return Err(DraftError::IdRepeated);
398                }
399                if self.executor.receipt(&draft.id)?.is_none() {
400                    return Err(DraftError::ReceiptMismatch);
401                }
402                let receipt = self.execution(&draft)?;
403                self.executor.store.deliver_outcome(&draft, &receipt)?;
404                self.client
405                    .record_execution(pass, &draft, &receipt)
406                    .await
407                    .map_err(|error| DraftError::Acknowledgment {
408                        draft: draft.id.clone(),
409                        name: error.name().to_owned(),
410                    })?;
411                self.executor.store.acknowledge(&draft.id, &receipt)?;
412                report.replayed += 1;
413                report.acknowledged += 1;
414            }
415        }
416    }
417}
418
419#[cfg(test)]
420#[path = "admin_drafts_fixture.rs"]
421mod fixture;
422#[cfg(test)]
423pub(crate) use fixture::fixture_site_drafts;
424
425#[cfg(test)]
426#[path = "admin_drafts_tests.rs"]
427pub(crate) mod tests;
428
429#[cfg(test)]
430mod signal_tests {
431    use super::*;
432    type TestResult = Result<(), Box<dyn std::error::Error>>;
433    use crate::auth_lys::live::drafts::tests::{Issuer, TargetStore, limits};
434    #[tokio::test]
435    async fn startup_is_one_signal_and_unknown_stream_rows_do_not_pull() -> TestResult {
436        let dir = tempfile::tempdir()?;
437        let issuer = Issuer::new(vec![(
438            200,
439            serde_json::json!({"drafts":[],"next":null,"total":0}),
440        )])?;
441        let mut puller = DraftPuller::new(
442            &issuer.api,
443            TargetStore::new(dir.path().join("site.json")),
444            limits()?,
445        )?;
446        assert_eq!(issuer.calls()?, 0);
447        puller
448            .pull("connector-fixture", PullSignal::Startup, 1)
449            .await?;
450        assert!(matches!(
451            puller
452                .pull("connector-fixture", PullSignal::Startup, 2)
453                .await,
454            Err(DraftError::StartupRepeated)
455        ));
456        assert_eq!(
457            crate::auth_lys::live::drafts::stream_signal("placement"),
458            None
459        );
460        assert_eq!(
461            crate::auth_lys::live::drafts::stream_signal("new_unknown_kind"),
462            None
463        );
464        assert_eq!(issuer.calls()?, 1);
465        issuer.finish()
466    }
467}
liminal_server/http/admin_drafts.rs

liminal_server/http/
admin_drafts.rs
