Add Async OHTTP client

Co-authored-by Alex Lewin <[email protected]>

[OHTTP](https://datatracker.ietf.org/doc/rfc9458/) lets a client
send encrypted requests through a relay so the server can’t see
who sent them and the relay can’t see what they contain.
The following commit adds optional configurations to enable
clients to proxy their requests through an OHTTP relay and gateway.

OHTTP functionality is feature flagged off behind `async-ohttp`. If a
client provided OHTTP config it will attempt to use the relay
instead of the target resource directly.

Author: arminsabouri

Cargo.toml

@@ -29,6 +29,11 @@ reqwest = { version = "0.12",  features = ["json"], default-features = false, op
 
 # default async runtime
 tokio = { version = "1", features = ["time"], optional = true }
+bitcoin-ohttp = { version = "0.6.0", optional = true}
+url = {version = "2.5.7", optional = true}
+bhttp = { version = "0.6.1", optional = true}
+http = { version = "1.3.1", optional = true}
+
 
 [dev-dependencies]
 serde_json = "1.0"
@@ -43,6 +48,7 @@ blocking-https = ["blocking", "minreq/https"]
 blocking-https-rustls = ["blocking", "minreq/https-rustls"]
 blocking-https-native = ["blocking", "minreq/https-native"]
 blocking-https-bundled = ["blocking", "minreq/https-bundled"]
+async-ohttp = ["async", "bitcoin-ohttp", "bhttp", "reqwest", "tokio", "url", "http"]
 
 tokio = ["dep:tokio"]
 async = ["reqwest", "reqwest/socks", "tokio?/time"]

src/async.rs

@@ -26,6 +26,8 @@ use log::{debug, error, info, trace};
 
 use reqwest::{header, Client, Response};
 
+#[cfg(feature = "async-ohttp")]
+use crate::ohttp::OhttpClient;
 use crate::{
     AddressStats, BlockInfo, BlockStatus, BlockSummary, Builder, Error, MempoolRecentTx,
     MempoolStats, MerkleProof, OutputStatus, ScriptHashStats, Tx, TxStatus, Utxo,
@@ -43,6 +45,9 @@ pub struct AsyncClient<S = DefaultSleeper> {
 
     /// Marker for the type of sleeper used
     marker: PhantomData<S>,
+    /// Ohttp config
+    #[cfg(feature = "async-ohttp")]
+    ohttp_client: Option<OhttpClient>,
 }
 
 impl<S: Sleeper> AsyncClient<S> {
@@ -77,6 +82,8 @@ impl<S: Sleeper> AsyncClient<S> {
             client: client_builder.build()?,
             max_retries: builder.max_retries,
             marker: PhantomData,
+            #[cfg(feature = "async-ohttp")]
+            ohttp_client: None,
         })
     }
 
@@ -86,9 +93,17 @@ impl<S: Sleeper> AsyncClient<S> {
             client,
             max_retries: crate::DEFAULT_MAX_RETRIES,
             marker: PhantomData,
+            #[cfg(feature = "async-ohttp")]
+            ohttp_client: None,
         }
     }
 
+    #[cfg(feature = "async-ohttp")]
+    pub(crate) fn set_ohttp_client(mut self, ohttp_client: OhttpClient) -> Self {
+        self.ohttp_client = Some(ohttp_client);
+        self
+    }
+
     /// Make an HTTP GET request to given URL, deserializing to any `T` that
     /// implement [`bitcoin::consensus::Decodable`].
     ///
@@ -557,12 +572,32 @@ impl<S: Sleeper> AsyncClient<S> {
         let mut attempts = 0;
 
         loop {
-            match self.client.get(url).send().await? {
+            let res = {
+                #[cfg(feature = "async-ohttp")]
+                if let Some(ohttp_client) = &self.ohttp_client {
+                    let (body, ctx) = ohttp_client.ohttp_encapsulate("get", url, None)?;
+                    let res = self
+                        .client
+                        .post(ohttp_client.relay_url().to_string())
+                        .header("Content-Type", "message/ohttp-req")
+                        .body(body)
+                        .send()
+                        .await?;
+                    let body = res.bytes().await?.to_vec();
+                    ohttp_client.ohttp_decapsulate(ctx, body)?.into()
+                } else {
+                    self.client.get(url).send().await?
+                }
+                #[cfg(not(feature = "async-ohttp"))]
+                self.client.get(url).send().await?
+            };
+            match res {
                 resp if attempts < self.max_retries && is_status_retryable(resp.status()) => {
                     S::sleep(delay).await;
                     attempts += 1;
                     delay *= 2;
                 }
+
                 resp => return Ok(resp),
             }
         }

src/lib.rs

@@ -83,6 +83,9 @@ pub mod r#async;
 #[cfg(feature = "blocking")]
 pub mod blocking;
 
+#[cfg(feature = "async-ohttp")]
+pub(crate) mod ohttp;
+
 pub use api::*;
 #[cfg(feature = "blocking")]
 pub use blocking::BlockingClient;
@@ -195,6 +198,20 @@ impl Builder {
     pub fn build_async_with_sleeper<S: Sleeper>(self) -> Result<AsyncClient<S>, Error> {
         AsyncClient::from_builder(self)
     }
+
+    #[cfg(feature = "async-ohttp")]
+    pub async fn build_async_with_ohttp(
+        self,
+        ohttp_relay_url: &str,
+        ohttp_gateway_url: &str,
+    ) -> Result<AsyncClient, Error> {
+        use crate::ohttp::OhttpClient;
+
+        let ohttp_client = OhttpClient::new(ohttp_relay_url, ohttp_gateway_url).await?;
+        Ok(self
+            .build_async_with_sleeper()?
+            .set_ohttp_client(ohttp_client))
+    }
 }
 
 /// Errors that can happen during a request to `Esplora` servers.
@@ -230,6 +247,18 @@ pub enum Error {
     InvalidHttpHeaderValue(String),
     /// The server sent an invalid response
     InvalidResponse,
+    /// Error from Ohttp library
+    #[cfg(feature = "async-ohttp")]
+    Ohttp(bitcoin_ohttp::Error),
+    /// Error when reading and writing to bhttp payloads
+    #[cfg(feature = "async-ohttp")]
+    Bhttp(bhttp::Error),
+    /// Error when converting the http response to and from bhttp response
+    #[cfg(feature = "async-ohttp")]
+    Http(http::Error),
+    /// Error when parsing the URL
+    #[cfg(feature = "async-ohttp")]
+    UrlParsing(url::ParseError),
 }
 
 impl fmt::Display for Error {

src/ohttp.rs

@@ -0,0 +1,102 @@
+use crate::Error;
+use bitcoin_ohttp as ohttp;
+use reqwest::Client;
+use url::Url;
+
+#[derive(Debug, Clone)]
+pub struct OhttpClient {
+    key_config: ohttp::KeyConfig,
+    relay_url: Url,
+}
+
+impl OhttpClient {
+    /// Will attempt to fetch the key config from the gateway and then create a new client.
+    /// Keyconfig is fetched directly from the gateway thus revealing our network metadata.
+    /// TODO: use the relay HTTP connect proxy to fetch to.
+    pub(crate) async fn new(relay_url: &str, ohttp_gateway_url: &str) -> Result<Self, Error> {
+        let gateway_url = Url::parse(ohttp_gateway_url).map_err(Error::UrlParsing)?;
+        let res = Client::new()
+            .get(gateway_url)
+            .send()
+            .await
+            .map_err(Error::Reqwest)?;
+        let body = res.bytes().await.map_err(Error::Reqwest)?;
+        let key_config = ohttp::KeyConfig::decode(&body).map_err(Error::Ohttp)?;
+        Ok(Self {
+            key_config,
+            relay_url: Url::parse(relay_url).map_err(Error::UrlParsing)?,
+        })
+    }
+
+    pub(crate) fn relay_url(&self) -> &Url {
+        &self.relay_url
+    }
+
+    pub(crate) fn ohttp_encapsulate(
+        &self,
+        method: &str,
+        target_resource: &str,
+        body: Option<&[u8]>,
+    ) -> Result<(Vec<u8>, ohttp::ClientResponse), Error> {
+        use std::fmt::Write;
+
+        // Bitcoin-hpke takes keyconfig as mutable ref but it doesnt mutate it should fix it
+        // upstream but for now we can clone it to avoid changing self to mutable self
+        let mut key_config = self.key_config.clone();
+
+        let ctx = ohttp::ClientRequest::from_config(&mut key_config).map_err(Error::Ohttp)?;
+        let url = url::Url::parse(target_resource).map_err(Error::UrlParsing)?;
+        let authority_bytes = url.host().map_or_else(Vec::new, |host| {
+            let mut authority = host.to_string();
+            if let Some(port) = url.port() {
+                write!(authority, ":{port}").unwrap();
+            }
+            authority.into_bytes()
+        });
+        let mut bhttp_message = bhttp::Message::request(
+            method.as_bytes().to_vec(),
+            url.scheme().as_bytes().to_vec(),
+            authority_bytes,
+            url.path().as_bytes().to_vec(),
+        );
+        // TODO: do we need to add headers?
+        if let Some(body) = body {
+            bhttp_message.write_content(body);
+        }
+
+        let mut bhttp_req = Vec::new();
+        bhttp_message
+            .write_bhttp(bhttp::Mode::IndeterminateLength, &mut bhttp_req)
+            .map_err(Error::Bhttp)?;
+        let (encapsulated, ohttp_ctx) = ctx.encapsulate(&bhttp_req).map_err(Error::Ohttp)?;
+
+        Ok((encapsulated, ohttp_ctx))
+    }
+
+    pub(crate) fn ohttp_decapsulate(
+        &self,
+        res_ctx: ohttp::ClientResponse,
+        ohttp_body: Vec<u8>,
+    ) -> Result<http::Response<Vec<u8>>, Error> {
+        let bhttp_body = res_ctx.decapsulate(&ohttp_body).map_err(Error::Ohttp)?;
+        let mut r = std::io::Cursor::new(bhttp_body);
+        let m: bhttp::Message = bhttp::Message::read_bhttp(&mut r).map_err(Error::Bhttp)?;
+        let mut builder = http::Response::builder();
+        for field in m.header().iter() {
+            builder = builder.header(field.name(), field.value());
+        }
+        builder
+            .status({
+                let code = m
+                    .control()
+                    .status()
+                    .ok_or(bhttp::Error::InvalidStatus)
+                    .map_err(Error::Bhttp)?;
+                http::StatusCode::from_u16(code.code())
+                    .map_err(|_| bhttp::Error::InvalidStatus)
+                    .map_err(Error::Bhttp)?
+            })
+            .body(m.content().to_vec())
+            .map_err(Error::Http)
+    }
+}