diff options
Diffstat (limited to 'crates/tor-dirserver')
| -rw-r--r-- | crates/tor-dirserver/Cargo.toml | 7 | ||||
| -rw-r--r-- | crates/tor-dirserver/src/http.rs | 93 | ||||
| -rw-r--r-- | crates/tor-dirserver/src/mirror.rs | 41 |
3 files changed, 138 insertions, 3 deletions
diff --git a/crates/tor-dirserver/Cargo.toml b/crates/tor-dirserver/Cargo.toml index b1080d4cd..0b744f74d 100644 --- a/crates/tor-dirserver/Cargo.toml +++ b/crates/tor-dirserver/Cargo.toml @@ -24,6 +24,9 @@ full = [ "tor-llcrypto/full", "tor-checkable/full", ] +experimental = ["dir-plugin-backend"] +dir-plugin-backend = ["__is_experimental"] +__is_experimental = [] [dependencies] bytes = "1.10.1" @@ -33,6 +36,7 @@ futures = "0.3.31" hex = "0.4.3" http = "1.3.1" http-body = "1.0.1" +http-body-util = "0.1.3" hyper = { version = "1.7.0", default-features = false, features = ["server", "http1"] } hyper-util = { version = "0.1.16", features = ["full"] } lzma-rs = "0.3.0" @@ -51,7 +55,7 @@ tokio-util = { version = "0.7.16", features = ["compat"] } tor-basic-utils = { version = "0.45.0", path = "../tor-basic-utils" } tor-checkable = { path = "../tor-checkable", version = "0.45.0" } tor-dirclient = { version = "0.45.0", path = "../tor-dirclient" } -tor-dircommon = { version = "0.45.0", path = "../tor-dircommon" } +tor-dircommon = { version = "0.45.0", path = "../tor-dircommon", features = ["experimental"] } tor-error = { version = "0.45.0", path = "../tor-error" } tor-llcrypto = { version = "0.45.0", path = "../tor-llcrypto" } tor-netdoc = { version = "0.45.0", path = "../tor-netdoc", features = ["full", "experimental"] } @@ -61,7 +65,6 @@ weak-table = "0.4.0" zstd = "0.13.3" [dev-dependencies] -http-body-util = "0.1.3" lazy_static = "1.5.0" tempfile = "3.23.0" tokio-stream = { version = "0.1.17", features = ["full"] } diff --git a/crates/tor-dirserver/src/http.rs b/crates/tor-dirserver/src/http.rs index ce53a4fe5..eded32780 100644 --- a/crates/tor-dirserver/src/http.rs +++ b/crates/tor-dirserver/src/http.rs @@ -6,6 +6,8 @@ use cache::StoreCache; use r2d2::Pool; use r2d2_sqlite::SqliteConnectionManager; +#[cfg(feature = "dir-plugin-backend")] +use tor_dircommon::dir_plugin_backend::DirBackendPlugin; use tor_error::internal; use std::{ @@ -36,6 +38,11 @@ use tokio::{ }; use tracing::warn; +#[cfg(feature = "dir-plugin-backend")] +use http_body_util::Full; +#[cfg(feature = "dir-plugin-backend")] +use std::io::Cursor; + use crate::database::{self, ContentEncoding, DocumentId, sql}; mod cache; @@ -140,6 +147,92 @@ impl HttpServer { Self { endpoints, pool } } + /// Bluntly launches an HTTP server only serving from the given backend. + /// + /// Absolutely not suited for anything in production as it comes with + /// various limitations. Primarily intended as an intermediate abstraction + /// for relay development. + #[cfg(feature = "dir-plugin-backend")] + pub(crate) async fn serve_backend<I, S, E, B>( + mut listener: I, + backend: B, + ) -> Result<(), tor_error::Bug> + where + I: Stream<Item = Result<S, E>> + Unpin, + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, + E: std::error::Error, + B: DirBackendPlugin, + { + // Creates a failing HTTP resposne while satisfying the hyper requirements. + let failure = |code| -> _ { + Response::builder() + .status(code) + .body(Default::default()) + .expect("response builder should not fail") + }; + + // We need to wrap the backend as an Arc, as the value would otherwise + // not live long enough. + let backend = Arc::new(backend); + let mut tasks: JoinSet<Result<(), hyper::Error>> = JoinSet::new(); + loop { + tokio::select! { + res = listener.next() => match res { + // Connection successfully accepted. + Some(Ok(s)) => { + let stream = TokioIo::new(s); + + // Two Arc clones required. First is to be able to run + // this in an endless loop and second one is required + // because hyper requires the function to be Fn, i.e. + // meaning it may not capture from it's surrounding + // state. + let backend = backend.clone(); + let service = service_fn(move |requ: Request<Incoming>| { + let backend = backend.clone(); + async move { + if requ.method() != Method::GET { + warn!("Unsupported method: {}", requ.method()); + // dir-spec does not allow StatusCode::METHOD_NOT_ALLOWED. + return Ok(failure(StatusCode::BAD_REQUEST)); + } + if !requ.body().is_end_stream() { + warn!("HTTP GET with non-empty body?"); + return Ok(failure(StatusCode::BAD_REQUEST)); + } + // Convert Request::<Incoming> to Request::<()>. + let requ = requ.map(|_| ()); + + // Convert the Box<[u8]> to something hyper accepts. + backend + .get(&requ) + .map(|resp| resp.map(|body| Full::new(Cursor::new(body)))) + } + }); + tasks.spawn(http1::Builder::new().serve_connection(stream, service)); + }, + + // There has been an error in accepting the connection. + Some(Err(e)) => { + warn!("listener accept failure: {e}"); + continue; + } + + // This should not happen due to ownership. + None => return Err(internal!("listener was closed externally?")), + }, + + // A hyper task we monitored in our tasks has exiteed. + Some(res) = tasks.join_next() => match res { + Ok(Ok(())) => {}, + Ok(Err(e)) => warn!("client task encountered an error: {e}"), + Err(e) => warn!("client task exited ungracefully: {e}"), + }, + + } + } + } + /// Runs the server endlessly in the current task. /// /// This function does not fail, because all errors that could potentially diff --git a/crates/tor-dirserver/src/mirror.rs b/crates/tor-dirserver/src/mirror.rs index 5c26ae4d2..f82024757 100644 --- a/crates/tor-dirserver/src/mirror.rs +++ b/crates/tor-dirserver/src/mirror.rs @@ -52,6 +52,9 @@ use tor_dircommon::{ config::{DirTolerance, DownloadScheduleConfig}, }; +#[cfg(feature = "dir-plugin-backend")] +use tor_dircommon::dir_plugin_backend::DirBackendPlugin; + mod operation; /// Core data type of a directory mirror. @@ -87,6 +90,43 @@ pub struct DirMirror { tolerance: DirTolerance, } +/// Insecure [`DirMirror`] abstraction providing a custom backend. +/// +/// Intended for relay development as a medium-term abstraction. +#[cfg(feature = "dir-plugin-backend")] +#[non_exhaustive] +pub struct DirMirrorWithBackend<B> { + /// The original [`DirMirror`]. + mirror: DirMirror, + /// The backend to use for handling requests instead. + backend: B, +} + +#[cfg(feature = "dir-plugin-backend")] +impl<B: DirBackendPlugin> DirMirrorWithBackend<B> { + /// Creates a new [`DirMirrorWithBackend`] from a given [`DirMirror`] and + /// a given [`DirBackendPlugin`]. + pub fn new(mirror: DirMirror, backend: B) -> Self { + Self { mirror, backend } + } + + /// Consumes this [`DirMirror`] by running endlessly in the current task. + /// + /// Be aware of the limitations and also see [`DirMirror::serve()`]. + pub async fn serve<S, T, E>(self, listener: S) -> Result<(), Infallible> + where + S: Stream<Item = Result<T, E>> + Unpin, + T: AsyncRead + AsyncWrite + Unpin + Send + 'static, + E: std::error::Error, + { + let res = crate::http::HttpServer::serve_backend(listener, self.backend).await; + if let Err(e) = res { + tracing::error!("HTTP backend failed unexpectedly: {e}"); + } + Ok(()) + } +} + impl DirMirror { /// Creates a new [`DirMirror`] with a given set of configuration options. /// @@ -152,7 +192,6 @@ impl DirMirror { // the stream over to DirMirror::serve(). // // See https://gitlab.torproject.org/tpo/core/arti/-/merge_requests/4222#note_3437135 - #[allow(clippy::unused_async)] // TODO pub async fn serve<S, T, E>(self, mut listener: S) -> Result<(), Infallible> where S: Stream<Item = Result<T, E>> + Unpin, |