lexe_connect/
http.rs

1//! HTTP delivery for the WALLET and a mailbox client for the REQUESTER.
2
3use std::time::{Duration, Instant};
4
5use anyhow::{Context, anyhow, ensure};
6use lexe_common::env::DeployEnv;
7use lexe_uri::Uri;
8use reqwest::StatusCode;
9
10use crate::{mailbox::Address, seal::Blob, wallet::HttpDelivery};
11
12/// A client for REQUESTER servers and mailboxes.
13///
14/// - Trusts Mozilla's webpki roots.
15/// - Follows no redirects, so a redirecting server can't re-route a plaintext
16///   response.
17/// - Requires `https://` in staging and prod.
18//
19// Spec: `post_url` delivery.
20#[derive(Clone)]
21pub struct LexeConnectClient(reqwest::Client);
22
23impl LexeConnectClient {
24    const HTTP_TIMEOUT: Duration = Duration::from_secs(15);
25
26    pub fn new(deploy_env: DeployEnv) -> anyhow::Result<Self> {
27        // Use the default ring CryptoProvider with webpki roots for broad
28        // compatibility with REQUESTER servers.
29        #[allow(clippy::disallowed_methods)]
30        let tls_config = rustls::ClientConfig::builder_with_protocol_versions(
31            lexe_tls_core::LEXE_TLS_PROTOCOL_VERSIONS,
32        )
33        .with_root_certificates(lexe_tls_core::WEBPKI_ROOT_CERTS.clone())
34        .with_no_client_auth();
35
36        let client = reqwest::Client::builder()
37            .https_only(deploy_env.is_staging_or_prod())
38            .redirect(reqwest::redirect::Policy::none())
39            .timeout(Self::HTTP_TIMEOUT)
40            .use_preconfigured_tls(tls_config)
41            .build()
42            .context("Failed to build LexeConnect reqwest client")?;
43
44        Ok(Self(client))
45    }
46
47    /// Performs an [`HttpDelivery`]. Any non-2xx status is an error,
48    /// including a mailbox's `409 Conflict` for an address holding a
49    /// different blob.
50    pub async fn deliver(&self, delivery: &HttpDelivery) -> anyhow::Result<()> {
51        let resp = self
52            .0
53            .post(&delivery.url)
54            .header(reqwest::header::CONTENT_TYPE, delivery.content_type)
55            .body(delivery.body.clone())
56            .send()
57            .await
58            .context("Failed to deliver response")?;
59
60        let status = resp.status();
61        ensure!(status.is_success(), "Delivery failed: HTTP {status}");
62
63        Ok(())
64    }
65
66    /// Fetches the blob at `address`, or `None` if it hasn't arrived.
67    pub async fn mailbox_get(
68        &self,
69        mailbox_url: &str,
70        address: &Address,
71    ) -> anyhow::Result<Option<Blob>> {
72        let mut url = Uri::parse(mailbox_url).context("Invalid mailbox url")?;
73        url.push_param("address", address.to_string());
74
75        let resp = self
76            .0
77            .get(url.to_string())
78            .send()
79            .await
80            .context("Failed to query mailbox")?;
81
82        match resp.status() {
83            StatusCode::NOT_FOUND => Ok(None),
84            status if status.is_success() => {
85                let bytes =
86                    resp.bytes().await.context("Failed to read blob")?;
87                Ok(Some(Blob(bytes.to_vec())))
88            }
89            status => Err(anyhow!("Mailbox query failed: HTTP {status}")),
90        }
91    }
92
93    /// Polls `address` every `interval` until a blob arrives or `timeout`
94    /// elapses.
95    pub async fn mailbox_poll(
96        &self,
97        mailbox_url: &str,
98        address: &Address,
99        interval: Duration,
100        timeout: Duration,
101    ) -> anyhow::Result<Blob> {
102        let deadline = Instant::now() + timeout;
103        loop {
104            if let Some(blob) = self.mailbox_get(mailbox_url, address).await? {
105                return Ok(blob);
106            }
107
108            ensure!(
109                Instant::now() + interval <= deadline,
110                "Timed out waiting for the mailbox"
111            );
112            tokio::time::sleep(interval).await;
113        }
114    }
115}
lexe_connect/http.rs

lexe_connect/
http.rs
