liminal-server 0.24.5

//! One bounded HTTP exchange, cancellable through the reader's owned signal.

use super::StreamError;
use serde::{Serialize, de::DeserializeOwned};
use std::io::Read;
use std::path::Path;
use tokio::sync::watch;
use zeroize::Zeroizing;

pub(super) struct Http {
    client: reqwest::Client,
    base: reqwest::Url,
    credential: Zeroizing<String>,
    pub limit: usize,
    pub stop: watch::Receiver<bool>,
}

impl Http {
    pub fn new(
        base: &str,
        credential: Zeroizing<String>,
        limit: usize,
        stop: watch::Receiver<bool>,
    ) -> Result<Self, StreamError> {
        let mut base = reqwest::Url::parse(base)
            .map_err(|_| StreamError::named("grant_stream_api_invalid"))?;
        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(StreamError::named("grant_stream_api_invalid"));
        }
        if !base.path().ends_with('/') {
            base.set_path(&format!("{}/", base.path()));
        }
        if limit == 0 || credential.trim().is_empty() || credential.chars().any(char::is_control) {
            return Err(StreamError::named("grant_stream_credential_invalid"));
        }
        // A completed request must not retain an idle socket with an expiry timer.
        let client = reqwest::Client::builder()
            .redirect(reqwest::redirect::Policy::none())
            .no_proxy()
            .pool_idle_timeout(None)
            .pool_max_idle_per_host(0)
            .build()
            .map_err(|_| StreamError::named("grant_stream_http_unavailable"))?;
        Ok(Self {
            client,
            base,
            credential,
            limit,
            stop,
        })
    }

    pub async fn post<Q: Serialize + Sync + ?Sized, A: DeserializeOwned>(
        &self,
        route: &str,
        question: &Q,
    ) -> Result<A, StreamError> {
        let mut stop = self.stop.clone();
        if *stop.borrow() {
            return Err(StreamError::named("grant_stream_stopped"));
        }
        tokio::select! {
            biased;
            _ = stop.changed() => Err(StreamError::named("grant_stream_stopped")),
            answer = self.exchange(route, question) => answer,
        }
    }

    async fn exchange<Q: Serialize + Sync + ?Sized, A: DeserializeOwned>(
        &self,
        route: &str,
        question: &Q,
    ) -> Result<A, StreamError> {
        let url = self
            .base
            .join(route)
            .map_err(|_| StreamError::named("grant_stream_route_invalid"))?;
        let mut response = self
            .client
            .post(url)
            .bearer_auth(self.credential.as_str())
            .json(question)
            .send()
            .await
            .map_err(|error| {
                if error.is_builder() {
                    StreamError::named("grant_stream_request_invalid")
                } else if error.is_connect() {
                    StreamError::named("revocation_stream_unavailable")
                } else {
                    StreamError::transport()
                }
            })?;
        if !response.status().is_success() {
            let name = match response.status().as_u16() {
                401 => "grant_stream_credential_refused",
                403 => "grant_stream_not_admitted",
                404 => "grant_stream_route_missing",
                429 => "grant_stream_capacity_refused",
                500..=599 => "revocation_stream_unavailable",
                _ => "grant_stream_http_refused",
            };
            return Err(StreamError::named(name));
        }
        if response
            .content_length()
            .is_some_and(|length| length > u64::try_from(self.limit).unwrap_or(u64::MAX))
        {
            return Err(StreamError::named("grant_stream_response_over_bound"));
        }
        let mut bytes = Zeroizing::new(Vec::new());
        while let Some(chunk) = response
            .chunk()
            .await
            .map_err(|_| StreamError::transport())?
        {
            if chunk.len() > self.limit.saturating_sub(bytes.len()) {
                return Err(StreamError::named("grant_stream_response_over_bound"));
            }
            bytes.extend_from_slice(&chunk);
        }
        serde_json::from_slice(&bytes).map_err(|_| StreamError::named("grant_stream_frame_invalid"))
    }
}

/// Reads a bounded file whole, never silently truncates a credential or key cache.
pub(super) fn read_file(path: &Path, limit: usize) -> Result<Zeroizing<String>, StreamError> {
    let cap = u64::try_from(limit)
        .ok()
        .and_then(|limit| limit.checked_add(1))
        .ok_or(StreamError::named("grant_stream_file_over_bound"))?;
    let file = std::fs::File::open(path)
        .map_err(|_| StreamError::named("grant_stream_file_unreadable"))?;
    let mut bytes = Zeroizing::new(Vec::new());
    file.take(cap)
        .read_to_end(&mut bytes)
        .map_err(|_| StreamError::named("grant_stream_file_unreadable"))?;
    if bytes.len() > limit {
        return Err(StreamError::named("grant_stream_file_over_bound"));
    }
    let text =
        std::str::from_utf8(&bytes).map_err(|_| StreamError::named("grant_stream_file_invalid"))?;
    Ok(Zeroizing::new(text.trim().to_owned()))
}