liminal-server 0.25.4

//! The existing operator HTTP listener dispatches this exact route.
//!
//! Signed passes carry expanded actions, not relation names. The schema's
//! dedicated pull action must belong only to network administrator and operator;
//! an unrelated call/deploy grant cannot be mistaken for either relation.

use std::num::NonZeroUsize;
use std::sync::Arc;

use lys_pass::{Decision, Target};
use serde_json::{Value, json};

use crate::auth_lys::LysPassConsumer;
use crate::auth_lys::live::drafts::{
    AtomicDraftStore, DraftError, DraftExecutor, DraftLimits, PullReport, PullSignal,
    valid_metadata, validate_bounds,
};
use crate::auth_lys::lys_pass_adapter::LysPassAdapter;

/// One read-only administrative pull, after the person approves in Lys.
pub const PULL_PATH: &str = "/admin/drafts/pull";
/// A person's outright agent act, taken at once rather than held by draft.
pub const CREATE_PATH: &str = "/admin/agents/create";
/// This action is granted only by the network's administrator/operator relations.
pub const PULL_ACTION: &str = "pull_drafts";

/// Explicit HTTP input bounds, without a timer or an implicit default.
#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AdminLimits {
    /// Bounds one bearer before signature verification and allocation.
    pub credential_bytes: NonZeroUsize,
    /// Bounds the empty-object request body, refusing extra authority fields.
    pub body_bytes: NonZeroUsize,
}

/// Borrowed from the HTTP parser; diagnostics never include its bearer or body.
pub struct AdminRequest<'a> {
    /// The parsed request method.
    pub method: &'a str,
    /// The exact path, with no accepted query parameters.
    pub path: &'a str,
    /// The one Authorization header; the HTTP parser refuses duplicates.
    pub authorization: Option<&'a str>,
    /// Empty bytes or an empty JSON object.
    pub body: &'a [u8],
    /// The server's operation time, never a caller's body field.
    pub at_ms: u64,
}

/// A response containing counts or a stable refusal, never prepared bytes.
pub struct AdminResponse {
    /// HTTP status.
    pub status: u16,
    /// JSON response body.
    pub body: Value,
}

/// The route uses its server-configured network and borrowed machine connector.
pub struct AdminDrafts<S> {
    puller: DraftPuller<S>,
    adapter: Arc<LysPassAdapter>,
    network: String,
    limits: AdminLimits,
    ask: Option<String>,
}

impl<S: AtomicDraftStore> AdminDrafts<S> {
    pub(crate) async fn signal(
        &mut self,
        connector: &str,
        signal: PullSignal,
        at_ms: u64,
    ) -> Result<PullReport, DraftError> {
        self.puller.pull(connector, signal, at_ms).await
    }

    /// Uses the same pass verifier and exact network as the wire admission.
    ///
    /// # Errors
    /// The configured network is absent, controlled or oversized.
    pub fn new(
        puller: DraftPuller<S>,
        adapter: Arc<LysPassAdapter>,
        network: String,
        limits: AdminLimits,
    ) -> Result<Self, DraftError> {
        if network.is_empty()
            || network.chars().any(char::is_control)
            || network.len() > limits.credential_bytes.get()
        {
            return Err(DraftError::Bound);
        }
        Ok(Self {
            puller,
            adapter,
            network,
            limits,
            ask: None,
        })
    }

    /// Asks Lys at the configured API address (`api_url`) rather than the
    /// issuer origin a pass names, as every other Lys call this server makes.
    #[must_use]
    pub fn asking_at(mut self, api_url: Option<String>) -> Self {
        self.ask = api_url;
        self
    }

    /// Handles the route without accepting authority or credentials in JSON.
    /// A held right is refused; this read cannot create another approval draft.
    /// The person's pass authorizes the pull, while only the machine connector
    /// reaches the product draft endpoints.
    ///
    /// # Errors
    /// An admitted pull failed; the endpoint must return this named failure,
    /// never a success response or a timer-based retry.
    pub async fn handle(
        &mut self,
        request: AdminRequest<'_>,
        connector_pass: &str,
    ) -> Result<AdminResponse, DraftError> {
        if request.path != PULL_PATH {
            return Ok(refused(404, "draft_pull_route_unknown"));
        }
        if request.method != "POST" {
            return Ok(refused(405, "draft_pull_method_refused"));
        }
        if request.body.len() > self.limits.body_bytes.get() {
            return Ok(refused(413, "draft_pull_body_bound"));
        }
        if !request.body.is_empty() {
            let body: Value = match serde_json::from_slice(request.body) {
                Ok(body) => body,
                Err(_) => return Ok(refused(400, "draft_pull_body_invalid")),
            };
            if !body.as_object().is_some_and(serde_json::Map::is_empty) {
                return Ok(refused(400, "draft_pull_body_invalid"));
            }
        }
        let Some(pass) = request
            .authorization
            .and_then(|header| header.strip_prefix("Bearer "))
        else {
            return Ok(refused(401, "draft_pull_pass_missing"));
        };
        if pass.is_empty()
            || pass.len() > self.limits.credential_bytes.get()
            || pass.chars().any(char::is_whitespace)
            || pass.chars().any(char::is_control)
        {
            return Ok(refused(401, "draft_pull_pass_invalid"));
        }
        let now = request.at_ms / 1000;
        let Ok(verified) = self.adapter.verify_offline(pass.as_bytes(), "liminal", now) else {
            return Ok(refused(401, "draft_pull_pass_invalid"));
        };
        match verified
            .verified()
            .evaluate("liminal.network", &self.network, PULL_ACTION, now)
        {
            Ok(Decision::Allowed { .. }) => {}
            Ok(Decision::Held { .. } | Decision::Refused) => {
                return Ok(refused(403, "draft_pull_network_operator_required"));
            }
            Err(_) => return Ok(refused(401, "draft_pull_pass_invalid")),
        }
        // Check the current grant, so revocation cannot be bypassed by an
        // otherwise unexpired offline pass on the administration plane.
        let http = reqwest::Client::builder()
            .no_proxy()
            .redirect(reqwest::redirect::Policy::none())
            .build()
            .map_err(|_| DraftError::Lys("transport_failed".to_owned()))?;
        let issuer = crate::auth_lys::live::lys_origin(
            self.ask.as_deref().unwrap_or_else(|| self.adapter.issuer()),
        )
        .map_err(|_| DraftError::Contract)?;
        let client = lys_pass::Client::new(http, issuer)
            .map_err(|error| DraftError::Lys(error.name().to_owned()))?;
        let target = Target::new("liminal.network", &self.network, PULL_ACTION)
            .map_err(|_| DraftError::Contract)?;
        let answer = client
            // Lys's live check takes the app's own credential (lys-app), never
            // the person's pass, whose holder is named as the subject instead.
            .check_batch(connector_pass, verified.holder(), &[target], None)
            .await
            .map_err(|error| DraftError::Lys(error.name().to_owned()))?;
        let Some(result) = answer.results.first() else {
            return Err(DraftError::Contract);
        };
        if answer.degraded.is_some() || result.degraded.is_some() {
            return Ok(refused(503, "draft_pull_permission_degraded"));
        }
        if !result.allowed || result.mode != Some(lys_pass::Mode::Outright) {
            if let Some(name) = &result.refusal {
                return Ok(AdminResponse {
                    status: 403,
                    body: json!({"refusal":name, "reason":result.reason}),
                });
            }
            return Ok(refused(403, "draft_pull_network_operator_required"));
        }
        let report = self
            .puller
            .pull(connector_pass, PullSignal::Console, request.at_ms)
            .await?;
        Ok(AdminResponse {
            status: 200,
            body: serde_json::to_value(report).map_err(|_| DraftError::Store)?,
        })
    }
}

#[path = "admin_outright.rs"]
mod outright;
pub use outright::OutrightAsk;

fn refused(status: u16, name: &str) -> AdminResponse {
    AdminResponse {
        status,
        body: json!({"refusal":name}),
    }
}

use lys_pass::drafts::{ApprovedDraft, Execution, Executor, execute_once};
use std::collections::BTreeSet;
/// One serialized owner. Borrow its machine credential for each told pull;
/// nothing here holds a person's session or starts work in the background.
pub struct DraftPuller<S> {
    client: lys_pass::Client,
    pub(crate) executor: DraftExecutor<S>,
    startup_seen: bool,
}

impl<S> std::fmt::Debug for DraftPuller<S> {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("DraftPuller")
            .finish_non_exhaustive()
    }
}

impl<S: AtomicDraftStore> DraftPuller<S> {
    /// Connects only to the configured issuer; redirects and environment
    /// proxies cannot route its machine credential elsewhere.
    ///
    /// # Errors
    /// The issuer or HTTP client could not be constructed.
    pub fn new(issuer: &str, store: S, limits: DraftLimits) -> Result<Self, DraftError> {
        let http = reqwest::Client::builder()
            .no_proxy()
            .redirect(reqwest::redirect::Policy::none())
            .build()
            .map_err(|_| DraftError::Lys("transport_failed".to_owned()))?;
        let issuer = crate::auth_lys::live::lys_origin(issuer).map_err(|_| DraftError::Contract)?;
        let client = lys_pass::Client::new(http, issuer)
            .map_err(|error| DraftError::Lys(error.name().to_owned()))?;
        Ok(Self {
            client,
            executor: DraftExecutor {
                store,
                limits,
                at_ms: 0,
            },
            startup_seen: false,
        })
    }

    /// Pulls and closes approved drafts only on an explicit signal. An error
    /// ends the pull; a later caller may reuse the durable receipt. No automatic
    /// retry is scheduled. The target store revalidates stale prepared acts.
    ///
    /// # Errors
    /// Lys, immutable contract, configured bounds, target storage or delivery.
    pub async fn pull(
        &mut self,
        connector_pass: &str,
        signal: PullSignal,
        at_ms: u64,
    ) -> Result<PullReport, DraftError> {
        let limits = self.executor.limits;
        if !valid_metadata(connector_pass, limits.metadata_bytes) {
            return Err(DraftError::Bound);
        }
        if signal == PullSignal::Startup {
            if self.startup_seen {
                return Err(DraftError::StartupRepeated);
            }
            self.startup_seen = true;
        }
        self.executor.at_ms = at_ms;
        let mut report = PullReport::default();
        let mut after: Option<String> = None;
        let mut cursors = BTreeSet::new();
        let mut ids = BTreeSet::new();
        let pending = self
            .acknowledge_pending(connector_pass, &mut report, &mut ids)
            .await?;
        loop {
            if report.pages >= limits.pages.get() {
                return Err(DraftError::Bound);
            }
            let page = self
                .client
                .approved_drafts(connector_pass, crate::auth_lys::LYS_APP, after.as_deref())
                .await
                .map_err(|error| DraftError::Lys(error.name().to_owned()))?;
            report.pages += 1;
            let mut page_ids = BTreeSet::new();
            for draft in &page.drafts {
                validate_bounds(draft, limits)?;
                if !page_ids.insert(&draft.id) {
                    return Err(DraftError::IdRepeated);
                }
            }
            let additional = page
                .drafts
                .iter()
                .filter(|draft| !ids.contains(&draft.id))
                .count();
            if ids.len().saturating_add(additional) > limits.drafts.get() {
                return Err(DraftError::Bound);
            }
            if let Some(next) = &page.next {
                if !valid_metadata(next, limits.metadata_bytes) {
                    return Err(DraftError::Bound);
                }
                if !cursors.insert(next.clone()) {
                    return Err(DraftError::CursorRepeated);
                }
            }
            for draft in &page.drafts {
                if pending.contains(&draft.id) {
                    self.execution(draft)?;
                    continue;
                }
                if !ids.insert(draft.id.clone()) {
                    return Err(DraftError::IdRepeated);
                }
                let replay = self.executor.receipt(&draft.id)?.is_some();
                let receipt = self.execution(draft)?;
                self.executor.store.deliver_outcome(draft, &receipt)?;
                if replay {
                    report.replayed += 1;
                } else {
                    match &receipt {
                        Execution::Executed { .. } => report.executed += 1,
                        Execution::RefusedOnExecution { .. } => report.refused += 1,
                    }
                }
                if !self.executor.store.acknowledged(&draft.id)? {
                    self.client
                        .record_execution(connector_pass, draft, &receipt)
                        .await
                        .map_err(|error| DraftError::Acknowledgment {
                            draft: draft.id.clone(),
                            name: error.name().to_owned(),
                        })?;
                    self.executor.store.acknowledge(&draft.id, &receipt)?;
                    report.acknowledged += 1;
                }
            }
            match page.next {
                Some(next) => after = Some(next),
                None => return Ok(report),
            }
        }
    }
    fn execution(&mut self, draft: &ApprovedDraft) -> Result<Execution, DraftError> {
        execute_once(draft, &mut self.executor).map_err(|error| match error {
            lys_pass::drafts::ExecuteError::DigestMismatch => DraftError::ReceiptMismatch,
            lys_pass::drafts::ExecuteError::Product(error) => error,
            lys_pass::drafts::ExecuteError::Contract(_) => DraftError::Contract,
        })
    }

    async fn acknowledge_pending(
        &mut self,
        pass: &str,
        report: &mut PullReport,
        ids: &mut BTreeSet<String>,
    ) -> Result<BTreeSet<String>, DraftError> {
        let limits = self.executor.limits;
        loop {
            let pending = self.executor.store.pending_acknowledgments(limits.drafts)?;
            if pending.is_empty() {
                return Ok(ids.clone());
            }
            if ids.len().saturating_add(pending.len()) > limits.drafts.get() {
                return Err(DraftError::Bound);
            }
            for draft in pending {
                validate_bounds(&draft, limits)?;
                if !ids.insert(draft.id.clone()) {
                    return Err(DraftError::IdRepeated);
                }
                if self.executor.receipt(&draft.id)?.is_none() {
                    return Err(DraftError::ReceiptMismatch);
                }
                let receipt = self.execution(&draft)?;
                self.executor.store.deliver_outcome(&draft, &receipt)?;
                self.client
                    .record_execution(pass, &draft, &receipt)
                    .await
                    .map_err(|error| DraftError::Acknowledgment {
                        draft: draft.id.clone(),
                        name: error.name().to_owned(),
                    })?;
                self.executor.store.acknowledge(&draft.id, &receipt)?;
                report.replayed += 1;
                report.acknowledged += 1;
            }
        }
    }
}

#[cfg(test)]
#[path = "admin_drafts_fixture.rs"]
mod fixture;
#[cfg(test)]
pub(crate) use fixture::fixture_site_drafts;

#[cfg(test)]
#[path = "admin_drafts_tests.rs"]
pub(crate) mod tests;

#[cfg(test)]
mod signal_tests {
    use super::*;
    type TestResult = Result<(), Box<dyn std::error::Error>>;
    use crate::auth_lys::live::drafts::tests::{Issuer, TargetStore, limits};
    #[tokio::test]
    async fn startup_is_one_signal_and_unknown_stream_rows_do_not_pull() -> TestResult {
        let dir = tempfile::tempdir()?;
        let issuer = Issuer::new(vec![(
            200,
            serde_json::json!({"drafts":[],"next":null,"total":0}),
        )])?;
        let mut puller = DraftPuller::new(
            &issuer.api,
            TargetStore::new(dir.path().join("site.json")),
            limits()?,
        )?;
        assert_eq!(issuer.calls()?, 0);
        puller
            .pull("connector-fixture", PullSignal::Startup, 1)
            .await?;
        assert!(matches!(
            puller
                .pull("connector-fixture", PullSignal::Startup, 2)
                .await,
            Err(DraftError::StartupRepeated)
        ));
        assert_eq!(
            crate::auth_lys::live::drafts::stream_signal("placement"),
            None
        );
        assert_eq!(
            crate::auth_lys::live::drafts::stream_signal("new_unknown_kind"),
            None
        );
        assert_eq!(issuer.calls()?, 1);
        issuer.finish()
    }
}