liminal_server/auth_membership/

provider.rs

use super::{
    ADMISSION_ROUTE, AdmissionDecision, AdmissionRequest, DECISION_ROUTE, MembershipDecision,
    MembershipPage, MembershipRequest, RECIPIENTS_ROUTE, RESOURCES_ROUTE, RecipientPageRequest,
    ResourcePageRequest,
};

/// A provider failure before a contract answer was received.
#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
#[error("membership provider could not be asked: {cause}")]
pub struct ProviderError {
    /// The stable cause, naming no response body and no credential.
    pub cause: String,
}

pub(super) fn provider_cause(cause: &str) -> ProviderError {
    ProviderError {
        cause: cause.to_owned(),
    }
}

/// Lys's shared membership provider behind a synchronous seam
/// (ACCESS-006 R1, R4).
///
/// The live implementation is
/// [`HttpMembershipProvider`]; a fixture answers from a seeded world. No
/// implementation may answer from a liminal-side membership list.
pub trait MembershipProvider: Send + Sync + std::fmt::Debug {
    /// Asks one membership decision.
    ///
    /// # Errors
    /// [`ProviderError`] when Lys could not be asked or its answer could not
    /// be read.
    fn decide(&self, request: &MembershipRequest) -> Result<MembershipDecision, ProviderError>;

    /// Asks scoped guest admission to one workspace.
    ///
    /// # Errors
    /// [`ProviderError`] as for [`Self::decide`].
    fn admission(&self, request: &AdmissionRequest) -> Result<AdmissionDecision, ProviderError>;

    /// Asks one page of the resources a subject may act on.
    ///
    /// # Errors
    /// [`ProviderError`] as for [`Self::decide`].
    fn resources(&self, request: &ResourcePageRequest) -> Result<MembershipPage, ProviderError>;

    /// Asks one page of the subjects who may act on a resource.
    ///
    /// # Errors
    /// [`ProviderError`] as for [`Self::decide`].
    fn recipients(&self, request: &RecipientPageRequest) -> Result<MembershipPage, ProviderError>;
}

/// Lys's membership routes over HTTP, asked as this liminal's app with its
/// credential.
///
/// Each ask runs to completion on a scoped thread against the
/// server's runtime, as `HttpLysAsk` does; nothing is cached here and no
/// failure becomes an answer.
pub struct HttpMembershipProvider {
    pub(super) http: reqwest::Client,
    pub(super) base: reqwest::Url,
    pub(super) credential: zeroize::Zeroizing<String>,
    pub(super) handle: tokio::runtime::Handle,
    pub(super) max_response_bytes: usize,
    pub(super) stop: tokio::sync::watch::Receiver<bool>,
}

impl std::fmt::Debug for HttpMembershipProvider {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("HttpMembershipProvider")
            .field("base", &self.base.as_str())
            .field(
                "credential",
                &crate::auth_pass::ByteLength(self.credential.len()),
            )
            .finish_non_exhaustive()
    }
}

impl HttpMembershipProvider {
    /// Asks the Lys `issuer` as the app holding `credential`, on `handle`'s
    /// runtime.
    ///
    /// # Errors
    /// [`ProviderError`] when the issuer is not an HTTP origin, the credential
    /// is empty, or the HTTP client cannot be built.
    pub fn new(
        issuer: &str,
        credential: zeroize::Zeroizing<String>,
        handle: tokio::runtime::Handle,
        max_response_bytes: usize,
        stop: tokio::sync::watch::Receiver<bool>,
    ) -> Result<Self, ProviderError> {
        let mut base =
            reqwest::Url::parse(issuer).map_err(|_| provider_cause("issuer_not_a_url"))?;
        if !matches!(base.scheme(), "http" | "https")
            || base.host_str().is_none()
            || !base.username().is_empty()
            || base.password().is_some()
            || base.query().is_some()
            || base.fragment().is_some()
        {
            return Err(provider_cause("issuer_not_an_http_origin"));
        }
        if !base.path().ends_with('/') {
            let path = format!("{}/", base.path());
            base.set_path(&path);
        }
        if credential.trim().is_empty()
            || credential.chars().any(char::is_control)
            || max_response_bytes == 0
        {
            return Err(provider_cause("membership_credential_empty"));
        }
        // Only an active ask owns a socket; no idle expiration wakes this provider.
        let http = reqwest::Client::builder()
            .redirect(reqwest::redirect::Policy::none())
            .no_proxy()
            .pool_idle_timeout(None)
            .pool_max_idle_per_host(0)
            .build()
            .map_err(|_| provider_cause("http_client_unavailable"))?;
        Ok(Self {
            http,
            base,
            credential,
            handle,
            max_response_bytes,
            stop,
        })
    }

    pub(super) fn post<Q, A>(&self, route: &str, question: &Q) -> Result<A, ProviderError>
    where
        Q: serde::Serialize + ?Sized,
        A: serde::de::DeserializeOwned + Send,
    {
        let url = self
            .base
            .join(route)
            .map_err(|_| provider_cause("membership_route_unjoinable"))?;
        let request = self
            .http
            .post(url)
            .bearer_auth(self.credential.as_str())
            .json(question);
        let handle = &self.handle;
        std::thread::scope(|scope| {
            scope
                .spawn(move || {
                    handle.block_on(async move {
                        let mut stop = self.stop.clone();
                        if *stop.borrow() { return Err(provider_cause("membership_stopped")); }
                        tokio::select! {
                            biased;
                            _ = stop.changed() => Err(provider_cause("membership_stopped")),
                            answer = async {
                        let mut response = request
                            .send()
                            .await
                            .map_err(|_| provider_cause("lys_could_not_be_asked"))?;
                        let status = response.status();
                        if !status.is_success() {
                            return Err(ProviderError {
                                cause: format!("lys_http_status_{}", status.as_u16()),
                            });
                        }
                        let mut bytes = zeroize::Zeroizing::new(Vec::new());
                        while let Some(chunk) = response.chunk().await.map_err(|_| provider_cause("lys_could_not_be_asked"))? {
                            if chunk.len() > self.max_response_bytes.saturating_sub(bytes.len()) {
                                return Err(provider_cause("membership_answer_over_bound"));
                            }
                            bytes.extend_from_slice(&chunk);
                        }
                        serde_json::from_slice(&bytes).map_err(|_| provider_cause("membership_answer_unreadable"))
                            } => answer,
                        }
                    })
                })
                .join()
        })
        .unwrap_or_else(|_| Err(provider_cause("membership_ask_did_not_complete")))
    }
}

impl MembershipProvider for HttpMembershipProvider {
    fn decide(&self, request: &MembershipRequest) -> Result<MembershipDecision, ProviderError> {
        self.post(DECISION_ROUTE, request)
    }

    fn admission(&self, request: &AdmissionRequest) -> Result<AdmissionDecision, ProviderError> {
        self.post(ADMISSION_ROUTE, request)
    }

    fn resources(&self, request: &ResourcePageRequest) -> Result<MembershipPage, ProviderError> {
        self.post(RESOURCES_ROUTE, request)
    }

    fn recipients(&self, request: &RecipientPageRequest) -> Result<MembershipPage, ProviderError> {
        self.post(RECIPIENTS_ROUTE, request)
    }
}