From 25d2a9351fbaea54dd4681ac4b1ef042662efd8d Mon Sep 17 00:00:00 2001 From: Erick Tryzelaar Date: Tue, 14 Jul 2026 02:43:39 +0000 Subject: [PATCH] Extract HttpRepository into tuf-hyper, port to hyper 1.x This extracts HttpRepository into its own crate, so its easier for users to not pull in the http stack unless it is necessary. It also updates hyper to 1.x, and adds some examples on how to create a client using tuf-hyper. --- Cargo.toml | 1 + interop-tests/Cargo.toml | 2 +- tuf-hyper/Cargo.toml | 39 + tuf-hyper/examples/client-smol.rs | 281 +++++ tuf-hyper/examples/client-tokio.rs | 206 ++++ tuf-hyper/examples/server-tokio.rs | 253 +++++ tuf-hyper/src/enforce_minimum_bitrate.rs | 103 ++ tuf-hyper/src/lib.rs | 27 + tuf-hyper/src/repository.rs | 958 ++++++++++++++++++ tuf/Cargo.toml | 10 +- tuf/src/client.rs | 37 +- tuf/src/{util.rs => enforce_size_and_hash.rs} | 144 +-- tuf/src/error.rs | 53 +- tuf/src/lib.rs | 2 +- tuf/src/repository.rs | 20 +- tuf/src/repository/http.rs | 460 --------- 16 files changed, 1963 insertions(+), 633 deletions(-) create mode 100644 tuf-hyper/Cargo.toml create mode 100644 tuf-hyper/examples/client-smol.rs create mode 100644 tuf-hyper/examples/client-tokio.rs create mode 100644 tuf-hyper/examples/server-tokio.rs create mode 100644 tuf-hyper/src/enforce_minimum_bitrate.rs create mode 100644 tuf-hyper/src/lib.rs create mode 100644 tuf-hyper/src/repository.rs rename tuf/src/{util.rs => enforce_size_and_hash.rs} (60%) delete mode 100644 tuf/src/repository/http.rs diff --git a/Cargo.toml b/Cargo.toml index 5274d89..df00dc0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,5 +2,6 @@ resolver = "2" members = [ "tuf", + "tuf-hyper", "interop-tests", ] diff --git a/interop-tests/Cargo.toml b/interop-tests/Cargo.toml index 024b405..ff82254 100644 --- a/interop-tests/Cargo.toml +++ b/interop-tests/Cargo.toml @@ -12,7 +12,7 @@ publish = false [dependencies] chrono = { version = "0.4.42", features = [ "serde" ] } -data-encoding = "2.9.0" +data-encoding = "2.10" futures-executor = "0.3.31" serde = "1" serde_json = "1" diff --git a/tuf-hyper/Cargo.toml b/tuf-hyper/Cargo.toml new file mode 100644 index 0000000..278c6cf --- /dev/null +++ b/tuf-hyper/Cargo.toml @@ -0,0 +1,39 @@ +[package] +name = "tuf-hyper" +version = "0.3.0-beta14" +edition = "2024" +rust-version = "1.88.0" +authors = [ "Erick Tryzelaar " ] +description = "HTTP repository support for TUF using Hyper" +homepage = "https://github.com/theupdateframework/rust-tuf" +repository = "https://github.com/theupdateframework/rust-tuf" +license = "MIT/Apache-2.0" + +[dependencies] +futures-io = "0.3.31" +futures-util = { version = "0.3.31", features = ["io"] } +http = "1" +http-body-util = "0.1" +hyper = { version = "1", default-features = false, features = ["client", "http1"] } +percent-encoding = "2.1" +tower = { version = "0.4", default-features = false, features = ["util"] } +tuf = { path = "../tuf", default-features = false } + +[dev-dependencies] +assert_matches = "1.5.0" +clap = { version = "4", features = ["derive", "env"] } +futures-executor = "0.3.31" +hyper = { version = "1", features = ["client", "server", "http1"] } +hyper-util = { version = "0.1", features = ["client-legacy", "client-pool", "http1", "tokio", "server", "server-auto", "service"] } +pretty_assertions = "1" +smol = "2" +tokio = { version = "1", features = ["rt", "rt-multi-thread", "macros", "net"] } + +[[example]] +name = "server-tokio" + +[[example]] +name = "client-tokio" + +[[example]] +name = "client-smol" diff --git a/tuf-hyper/examples/client-smol.rs b/tuf-hyper/examples/client-smol.rs new file mode 100644 index 0000000..f50ab30 --- /dev/null +++ b/tuf-hyper/examples/client-smol.rs @@ -0,0 +1,281 @@ +//! A TUF client using the Smol runtime (`smol::block_on`) and hyper legacy client. +//! +//! Demonstrates implementing a custom `SmolExecutor` (`hyper::rt::Executor`), +//! `SmolStream` (`hyper::rt::Read` + `hyper::rt::Write` + `Connection`), and +//! `SmolConnector` (`Service`), constructing an `HttpRepositoryBuilder` on Smol, +//! setting up a local `EphemeralRepository` or `FileSystemRepository`, initializing +//! the client with trusted root keys, running `.update().await`, and fetching target files. +//! +//! # Usage +//! ```bash +//! # Make sure `server-tokio` is running first: +//! # cargo run --example server-tokio +//! +//! cargo run --example client-smol -- [options] +//! +//! Options: +//! --url, -u TUF server URL (default: http://127.0.0.1:8080) +//! --dir, -d Local metadata/target storage directory (FileSystemRepository) +//! --help, -h Print usage and exit +//! ``` + +use clap::Parser; +use futures_io::{AsyncRead, AsyncWrite}; +use futures_util::io::AsyncReadExt as _; +use http::{Request, Response, Uri}; +use hyper::body::{Bytes, Incoming}; +use hyper::rt::{Read, ReadBufCursor, Write}; +use hyper_util::client::pool::cache; +use std::future::Future; +use std::io; +use std::mem::MaybeUninit; +use std::path::PathBuf; +use std::pin::Pin; +use std::task::{Context, Poll}; +use tower::{Service, ServiceBuilder, ServiceExt as _}; +use tuf::client::{Client, Config}; +use tuf::crypto::{Ed25519PrivateKey, PrivateKey}; +use tuf::metadata::{MetadataVersion, TargetPath}; +use tuf::pouf::Pouf1; +use tuf::repository::{ + EphemeralRepository, FileSystemRepository, RepositoryProvider, RepositoryStorage, +}; +use tuf_hyper::HttpRepositoryBuilder; + +type BoxError = Box; + +#[derive(Parser, Debug)] +#[command(about = "A TUF client using the Smol runtime and hyper pooling client.")] +struct Cli { + /// TUF server URL + #[arg( + short = 'u', + long, + env = "TUF_SERVER_URL", + default_value = "http://127.0.0.1:8080" + )] + url: Uri, + + /// Local metadata/target storage directory (FileSystemRepository) + #[arg(short = 'd', long, env = "TUF_LOCAL_DIR")] + dir: Option, +} + +const ED25519_1_PK8: &[u8] = include_bytes!("../../tuf/tests/ed25519/ed25519-1.pk8.der"); + +/// A custom hyper runtime executor powered by `smol::spawn`. +#[derive(Clone, Copy, Debug)] +pub struct SmolExecutor; + +impl hyper::rt::Executor for SmolExecutor +where + F: std::future::Future + Send + 'static, + F::Output: Send + 'static, +{ + fn execute(&self, fut: F) { + smol::spawn(async move { + fut.await; + }) + .detach(); + } +} + +/// A wrapper around `smol::net::TcpStream` that implements hyper's `Read` and `Write`. +pub struct SmolStream { + inner: smol::net::TcpStream, +} + +impl Read for SmolStream { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + mut cursor: ReadBufCursor<'_>, + ) -> Poll> { + // SAFETY: Ensure that all bytes are initialized. + // + // FIXME(https://github.com/hyperium/hyper/pull/4115): We can replace the unsafe block with + // this once it is in a release. + // let slice = cursor.initialize_unfilled(); + let slice = unsafe { + let uninit = cursor.as_mut(); + uninit.fill(MaybeUninit::new(0)); + + std::slice::from_raw_parts_mut(uninit.as_mut_ptr() as *mut u8, uninit.len()) + }; + match Pin::new(&mut self.inner).poll_read(cx, slice) { + Poll::Ready(Ok(n)) => { + // SAFETY: We initialized all the bytes earlier in the function. + unsafe { cursor.advance(n) }; + + Poll::Ready(Ok(())) + } + Poll::Ready(Err(e)) => Poll::Ready(Err(e)), + Poll::Pending => Poll::Pending, + } + } +} + +impl Write for SmolStream { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.inner).poll_write(cx, buf) + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_flush(cx) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_close(cx) + } +} + +/// Wrapper around `hyper::client::conn::http1::SendRequest` implementing `Service>`. +#[derive(Debug)] +pub struct ConnectionService(pub hyper::client::conn::http1::SendRequest); + +impl Service> for ConnectionService +where + B: hyper::body::Body + Send + 'static + Unpin, + B::Data: Send, + B::Error: Into, +{ + type Response = Response; + type Error = BoxError; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_ready(cx).map_err(Into::into) + } + + fn call(&mut self, req: Request) -> Self::Future { + let fut = self.0.send_request(req); + Box::pin(async move { fut.await.map_err(Into::into) }) + } +} + +/// A custom hyper `Service` connector that establishes TCP connections via `smol::net::TcpStream` +/// and performs HTTP/1 handshakes. +#[derive(Clone, Copy, Debug)] +pub struct SmolConnector; + +impl Service for SmolConnector { + type Response = ConnectionService>; + type Error = BoxError; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, uri: Uri) -> Self::Future { + Box::pin(async move { + let host = uri.host().ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidInput, "Missing host in URI") + })?; + let port = uri.port_u16().unwrap_or(80); + let addr_str = format!("{}:{}", host, port); + let stream = smol::net::TcpStream::connect(addr_str).await?; + let (sender, conn) = + hyper::client::conn::http1::handshake(SmolStream { inner: stream }).await?; + smol::spawn(async move { + if let Err(err) = conn.await { + eprintln!("Connection failed: {:?}", err); + } + }) + .detach(); + Ok(ConnectionService(sender)) + }) + } +} + +async fn async_main() -> std::result::Result<(), Box> { + let cli = Cli::parse(); + let uri = cli.url; + let local_dir = cli.dir; + println!("Connecting to TUF HTTP repository at {} via Smol", uri); + + // Build the pooling client using hyper_util::client::pool::cache and SmolExecutor + let pool = cache::builder().executor(SmolExecutor).build(SmolConnector); + let hyper_client = + ServiceBuilder::new().service_fn(move |req: Request>| { + let mut pool = pool.clone(); + async move { + let mut conn = pool.call(req.uri().clone()).await?; + conn.ready().await?.call(req).await + } + }); + + let remote = HttpRepositoryBuilder::<_, Pouf1>::new(uri, hyper_client) + .user_agent("tuf-client-smol-example/0.1") + .build(); + + // Load the trusted root public key used by our example server. + let root_key = Ed25519PrivateKey::from_pkcs8(ED25519_1_PK8)?; + let root_public_key = root_key.public(); + + let config = Config::default(); + + match local_dir { + Some(dir) => { + println!("Using local FileSystemRepository at {:?}", dir); + let local = FileSystemRepository::::new(dir); + run_client(config, root_public_key, local, remote).await?; + } + None => { + println!("Using local EphemeralRepository (in-memory)"); + let local = EphemeralRepository::::new(); + run_client(config, root_public_key, local, remote).await?; + } + } + + Ok(()) +} + +async fn run_client( + config: Config, + root_public_key: &tuf::crypto::PublicKey, + local: L, + remote: R, +) -> std::result::Result<(), Box> +where + L: RepositoryProvider + RepositoryStorage + Send + Sync + 'static, + R: RepositoryProvider + Send + Sync + 'static, +{ + let mut client = Client::with_trusted_root_keys( + config, + MetadataVersion::Number(1), + 1, + [root_public_key], + local, + remote, + ) + .await?; + + println!("Updating client metadata..."); + let _ = client.update().await?; + println!("Client metadata updated successfully."); + + // Fetch the target file into the local repository and verify its content. + let target_path = TargetPath::new("foo-bar")?; + println!("Fetching target '{}' to local storage...", target_path); + client.fetch_target_to_local(&target_path).await?; + + let mut target_reader = client.fetch_target(&target_path).await?; + let mut target_data = Vec::new(); + target_reader.read_to_end(&mut target_data).await?; + println!( + "Successfully fetched target '{}': {:?}", + target_path, + String::from_utf8_lossy(&target_data) + ); + + Ok(()) +} + +fn main() -> std::result::Result<(), Box> { + smol::block_on(async_main()) +} diff --git a/tuf-hyper/examples/client-tokio.rs b/tuf-hyper/examples/client-tokio.rs new file mode 100644 index 0000000..48a5088 --- /dev/null +++ b/tuf-hyper/examples/client-tokio.rs @@ -0,0 +1,206 @@ +//! A TUF client using the Tokio runtime (`#[tokio::main]`) and hyper legacy client. +//! +//! Demonstrates constructing an `HttpRepositoryBuilder` with the pooling client +//! using `TokioExecutor`, setting up a local `EphemeralRepository` or `FileSystemRepository`, +//! initializing `Client::with_trusted_root_keys(...)`, running `.update().await`, +//! and downloading target files. +//! +//! # Usage +//! ```bash +//! # Make sure `server-tokio` is running first: +//! # cargo run --example server-tokio +//! +//! cargo run --example client-tokio -- [options] +//! +//! Options: +//! --url, -u TUF server URL (default: http://127.0.0.1:8080) +//! --dir, -d Local metadata/target storage directory (FileSystemRepository) +//! --help, -h Print usage and exit +//! ``` + +use clap::Parser; +use futures_util::io::AsyncReadExt as _; +use http::{Request, Response, Uri}; +use hyper::body::{Bytes, Incoming}; +use hyper_util::client::pool::cache; +use hyper_util::rt::{TokioExecutor, TokioIo}; +use std::future::Future; +use std::io; +use std::path::PathBuf; +use std::pin::Pin; +use std::task::{Context, Poll}; +use tower::{Service, ServiceBuilder, ServiceExt as _}; +use tuf::client::{Client, Config}; +use tuf::crypto::{Ed25519PrivateKey, PrivateKey}; +use tuf::metadata::{MetadataVersion, TargetPath}; +use tuf::pouf::Pouf1; +use tuf::repository::{ + EphemeralRepository, FileSystemRepository, RepositoryProvider, RepositoryStorage, +}; +use tuf_hyper::HttpRepositoryBuilder; + +type BoxError = Box; + +#[derive(Parser, Debug)] +#[command(about = "A TUF client using the Tokio runtime and hyper pooling client.")] +struct Cli { + /// TUF server URL + #[arg( + short = 'u', + long, + env = "TUF_SERVER_URL", + default_value = "http://127.0.0.1:8080" + )] + url: Uri, + + /// Local metadata/target storage directory (FileSystemRepository) + #[arg(short = 'd', long, env = "TUF_LOCAL_DIR")] + dir: Option, +} + +const ED25519_1_PK8: &[u8] = include_bytes!("../../tuf/tests/ed25519/ed25519-1.pk8.der"); + +/// Wrapper around `hyper::client::conn::http1::SendRequest` implementing `Service>`. +#[derive(Debug)] +pub struct ConnectionService(pub hyper::client::conn::http1::SendRequest); + +impl Service> for ConnectionService +where + B: hyper::body::Body + Send + 'static + Unpin, + B::Data: Send, + B::Error: Into, +{ + type Response = Response; + type Error = BoxError; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_ready(cx).map_err(Into::into) + } + + fn call(&mut self, req: Request) -> Self::Future { + let fut = self.0.send_request(req); + Box::pin(async move { fut.await.map_err(Into::into) }) + } +} + +/// A custom hyper `Service` connector for Tokio that establishes TCP connections +/// and performs HTTP/1 handshakes. +#[derive(Clone, Copy, Debug)] +pub struct TokioConnector; + +impl Service for TokioConnector { + type Response = ConnectionService>; + type Error = BoxError; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, uri: Uri) -> Self::Future { + Box::pin(async move { + let host = uri.host().ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidInput, "Missing host in URI") + })?; + let port = uri.port_u16().unwrap_or(80); + let stream = tokio::net::TcpStream::connect((host, port)).await?; + let io = TokioIo::new(stream); + let (sender, conn) = hyper::client::conn::http1::handshake(io).await?; + tokio::spawn(async move { + if let Err(err) = conn.await { + eprintln!("Connection failed: {:?}", err); + } + }); + Ok(ConnectionService(sender)) + }) + } +} + +#[tokio::main] +async fn main() -> std::result::Result<(), Box> { + let cli = Cli::parse(); + let uri = cli.url; + let local_dir = cli.dir; + println!("Connecting to TUF HTTP repository at {} via Tokio", uri); + + // Build the pooling client using hyper_util::client::pool::cache and TokioExecutor + let pool = cache::builder() + .executor(TokioExecutor::new()) + .build(TokioConnector); + let hyper_client = + ServiceBuilder::new().service_fn(move |req: Request>| { + let mut pool = pool.clone(); + async move { + let mut conn = pool.call(req.uri().clone()).await?; + conn.ready().await?.call(req).await + } + }); + + // Construct the remote HTTP repository. + let remote = HttpRepositoryBuilder::<_, Pouf1>::new(uri, hyper_client) + .user_agent("tuf-client-tokio-example/0.1") + .build(); + + // Load the trusted root public key used by our example server. + let root_key = Ed25519PrivateKey::from_pkcs8(ED25519_1_PK8)?; + let root_public_key = root_key.public(); + + let config = Config::default(); + + match local_dir { + Some(dir) => { + println!("Using local FileSystemRepository at {:?}", dir); + let local = FileSystemRepository::::new(dir); + run_client(config, root_public_key, local, remote).await?; + } + None => { + println!("Using local EphemeralRepository (in-memory)"); + let local = EphemeralRepository::::new(); + run_client(config, root_public_key, local, remote).await?; + } + } + + Ok(()) +} + +async fn run_client( + config: Config, + root_public_key: &tuf::crypto::PublicKey, + local: L, + remote: R, +) -> std::result::Result<(), Box> +where + L: RepositoryProvider + RepositoryStorage + Send + Sync + 'static, + R: RepositoryProvider + Send + Sync + 'static, +{ + let mut client = Client::with_trusted_root_keys( + config, + MetadataVersion::Number(1), + 1, + [root_public_key], + local, + remote, + ) + .await?; + + println!("Updating client metadata..."); + let _ = client.update().await?; + println!("Client metadata updated successfully."); + + // Fetch the target file into the local repository and verify its content. + let target_path = TargetPath::new("foo-bar")?; + println!("Fetching target '{}' to local storage...", target_path); + client.fetch_target_to_local(&target_path).await?; + + let mut target_reader = client.fetch_target(&target_path).await?; + let mut target_data = Vec::new(); + target_reader.read_to_end(&mut target_data).await?; + println!( + "Successfully fetched target '{}': {:?}", + target_path, + String::from_utf8_lossy(&target_data) + ); + + Ok(()) +} diff --git a/tuf-hyper/examples/server-tokio.rs b/tuf-hyper/examples/server-tokio.rs new file mode 100644 index 0000000..6c160bb --- /dev/null +++ b/tuf-hyper/examples/server-tokio.rs @@ -0,0 +1,253 @@ +//! An HTTP TUF repository server using the Tokio runtime (`#[tokio::main]`) and +//! `hyper` / `http-body-util` 1.x. +//! +//! It serves an ephemeral repository with TUF metadata and targets over HTTP, +//! drawing inspiration from `tuf/tests/simple_example.rs`, or optionally serves +//! from a local filesystem directory (`FileSystemRepository`). +//! +//! # Usage +//! ```bash +//! cargo run --example server-tokio -- [options] +//! +//! Options: +//! --addr, -a Listen address and port (default: 127.0.0.1:8080) +//! --dir, -d Serve from a local directory (FileSystemRepository) +//! --help, -h Print usage and exit +//! ``` + +use clap::Parser; +use futures_util::io::{AsyncReadExt as _, Cursor}; +use http::{Method, Request, Response, StatusCode}; +use http_body_util::Full; +use hyper::body::{Bytes, Incoming}; +use hyper::server::conn::http1; +use hyper::service::service_fn; +use hyper_util::rt::TokioIo; +use std::convert::Infallible; +use std::net::SocketAddr; +use std::path::PathBuf; +use std::sync::Arc; +use tokio::net::TcpListener; +use tuf::Result; +use tuf::crypto::Ed25519PrivateKey; +use tuf::metadata::{MetadataPath, MetadataVersion, TargetPath}; +use tuf::pouf::{Pouf as _, Pouf1}; +use tuf::repo_builder::RepoBuilder; +use tuf::repository::{EphemeralRepository, FileSystemRepository, RepositoryProvider}; + +#[derive(Parser, Debug)] +#[command(about = "An HTTP TUF repository server using the Tokio runtime and hyper 1.x")] +struct Cli { + /// Listen address and port + #[arg( + short = 'a', + long, + env = "TUF_SERVER_ADDR", + default_value = "127.0.0.1:8080" + )] + addr: SocketAddr, + + /// Serve from a local directory (FileSystemRepository) + #[arg(short = 'd', long, env = "TUF_SERVER_DIR")] + dir: Option, +} + +const ED25519_1_PK8: &[u8] = include_bytes!("../../tuf/tests/ed25519/ed25519-1.pk8.der"); +const ED25519_2_PK8: &[u8] = include_bytes!("../../tuf/tests/ed25519/ed25519-2.pk8.der"); +const ED25519_3_PK8: &[u8] = include_bytes!("../../tuf/tests/ed25519/ed25519-3.pk8.der"); +const ED25519_4_PK8: &[u8] = include_bytes!("../../tuf/tests/ed25519/ed25519-4.pk8.der"); + +pub enum ServerRepo { + Ephemeral(EphemeralRepository), + FileSystem(FileSystemRepository), +} + +impl ServerRepo { + pub async fn fetch(&self, path: &str) -> Option> { + match self { + ServerRepo::Ephemeral(repo) => fetch_from_repo_provider(repo, path).await, + ServerRepo::FileSystem(repo) => fetch_from_repo_provider(repo, path).await, + } + } +} + +async fn run() -> std::result::Result<(), Box> { + let cli = Cli::parse(); + let addr = cli.addr; + let dir_path = cli.dir; + + let repo = match dir_path { + Some(dir) => { + println!("Serving TUF repository from local directory: {:?}", dir); + Arc::new(ServerRepo::FileSystem(FileSystemRepository::::new( + dir, + ))) + } + None => { + println!("Serving sample in-memory TUF repository (EphemeralRepository)"); + let mut remote = EphemeralRepository::::new(); + init_server(&mut remote, true).await?; + Arc::new(ServerRepo::Ephemeral(remote)) + } + }; + + let listener = TcpListener::bind(addr).await?; + println!("TUF HTTP server listening on http://{}", addr); + + loop { + let (tcp_stream, remote_addr) = match listener.accept().await { + Ok(conn) => conn, + Err(err) => { + eprintln!("Failed to accept connection: {}", err); + continue; + } + }; + + let io = TokioIo::new(tcp_stream); + let repo = Arc::clone(&repo); + + tokio::spawn(async move { + let service = service_fn(move |req| { + let repo = Arc::clone(&repo); + async move { handle_request(req, repo).await } + }); + + if let Err(err) = http1::Builder::new().serve_connection(io, service).await { + eprintln!("Error serving connection from {}: {}", remote_addr, err); + } + }); + } +} + +/// Initialize the server's repository with signed metadata and a sample target file. +async fn init_server( + remote: &mut EphemeralRepository, + consistent_snapshot: bool, +) -> Result<()> { + let root_key = Ed25519PrivateKey::from_pkcs8(ED25519_1_PK8)?; + let snapshot_key = Ed25519PrivateKey::from_pkcs8(ED25519_2_PK8)?; + let targets_key = Ed25519PrivateKey::from_pkcs8(ED25519_3_PK8)?; + let timestamp_key = Ed25519PrivateKey::from_pkcs8(ED25519_4_PK8)?; + + let target_path = TargetPath::new("foo-bar")?; + let target_file: &[u8] = b"things fade, alternatives exclude"; + + RepoBuilder::create(&mut *remote) + .trusted_root_keys(&[&root_key]) + .trusted_snapshot_keys(&[&snapshot_key]) + .trusted_targets_keys(&[&targets_key]) + .trusted_timestamp_keys(&[×tamp_key]) + .stage_root_with_builder(|builder| builder.consistent_snapshot(consistent_snapshot)) + .unwrap() + .add_target(target_path, Cursor::new(target_file)) + .await + .unwrap() + .commit() + .await + .unwrap(); + + Ok(()) +} + +/// Handle incoming HTTP requests and serve metadata or target files from the repository. +async fn handle_request( + req: Request, + repo: Arc, +) -> std::result::Result>, Infallible> { + if req.method() != Method::GET { + let mut resp = Response::new(Full::new(Bytes::from("Method Not Allowed"))); + *resp.status_mut() = StatusCode::METHOD_NOT_ALLOWED; + return Ok(resp); + } + + let path = req.uri().path().trim_start_matches('/'); + if let Some(buf) = repo.fetch(path).await { + let resp = Response::new(Full::new(Bytes::from(buf))); + Ok(resp) + } else { + let mut resp = Response::new(Full::new(Bytes::from("Not Found"))); + *resp.status_mut() = StatusCode::NOT_FOUND; + Ok(resp) + } +} + +/// Attempt to fetch either a metadata file or a target file from any `RepositoryProvider`. +/// Includes prefix-stripping fallback for paths requested with custom metadata/targets prefixes. +async fn fetch_from_repo_provider>( + repo: &R, + path: &str, +) -> Option> { + let ext = format!(".{}", Pouf1::extension()); + if path.ends_with(&ext) { + let (dir_prefix, filename) = match path.rsplit_once('/') { + Some((dir, file)) => (dir, file), + None => ("", path), + }; + + let stem = &filename[..filename.len() - ext.len()]; + let (version, role_str) = match stem.split_once('.') { + Some((ver_str, r_str)) => match ver_str.parse::() { + Ok(num) => (MetadataVersion::Number(num), r_str), + Err(_) => (MetadataVersion::None, stem), + }, + None => (MetadataVersion::None, stem), + }; + + let role_path_str = if dir_prefix.is_empty() { + role_str.to_string() + } else { + format!("{}/{}", dir_prefix, role_str) + }; + + // Try exact role path first. + if let Ok(meta_path) = MetadataPath::new(role_path_str) + && let Ok(mut reader) = repo.fetch_metadata(&meta_path, version).await + { + let mut buf = Vec::new(); + if reader.read_to_end(&mut buf).await.is_ok() { + return Some(buf); + } + } + + // Fallback: if dir_prefix was non-empty (e.g. client requested `metadata/1.root.json`), + // try querying without the leading directory prefix. + if !dir_prefix.is_empty() + && let Ok(meta_path) = MetadataPath::new(role_str.to_string()) + && let Ok(mut reader) = repo.fetch_metadata(&meta_path, version).await + { + let mut buf = Vec::new(); + if reader.read_to_end(&mut buf).await.is_ok() { + return Some(buf); + } + } + } + + // Try exact target path. + if let Ok(target_path) = TargetPath::new(path) + && let Ok(mut reader) = repo.fetch_target(&target_path).await + { + let mut buf = Vec::new(); + if reader.read_to_end(&mut buf).await.is_ok() { + return Some(buf); + } + } + + // Fallback: if target path had a leading directory prefix (e.g. `targets/foo-bar`), + // try querying with just the filename. + if let Some((_, filename)) = path.rsplit_once('/') + && let Ok(target_path) = TargetPath::new(filename) + && let Ok(mut reader) = repo.fetch_target(&target_path).await + { + let mut buf = Vec::new(); + if reader.read_to_end(&mut buf).await.is_ok() { + return Some(buf); + } + } + + None +} + +#[tokio::main] +async fn main() -> std::result::Result<(), Box> { + run().await +} diff --git a/tuf-hyper/src/enforce_minimum_bitrate.rs b/tuf-hyper/src/enforce_minimum_bitrate.rs new file mode 100644 index 0000000..6feb6e3 --- /dev/null +++ b/tuf-hyper/src/enforce_minimum_bitrate.rs @@ -0,0 +1,103 @@ +//! Utility functions and adapters for async readers and bitrate enforcement. + +use futures_io::AsyncRead; +use std::io::{self, ErrorKind}; +use std::pin::Pin; +use std::task::{Context, Poll}; +use std::time::{Duration, Instant}; + +/// Wraps an `AsyncRead` to detect and fail transfers slower than a minimum bitrate. +pub(crate) struct EnforceMinimumBitrate { + inner: R, + min_bytes_per_second: u32, + start_time: Option, + bytes_read: u64, +} + +impl EnforceMinimumBitrate { + /// Create a new `EnforceMinimumBitrate`. + pub(crate) fn new(read: R, min_bytes_per_second: u32) -> Self { + Self { + inner: read, + min_bytes_per_second, + start_time: None, + bytes_read: 0, + } + } +} + +#[cfg(not(test))] +const BITRATE_GRACE_PERIOD: Duration = Duration::from_secs(30); +#[cfg(test)] +const BITRATE_GRACE_PERIOD: Duration = Duration::from_secs(1); + +impl AsyncRead for EnforceMinimumBitrate { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context, + buf: &mut [u8], + ) -> Poll> { + // FIXME(#272) transfers that stall out completely won't enforce the minimum bit rate. + let read_bytes = futures_util::ready!(Pin::new(&mut self.inner).poll_read(cx, buf))?; + + let start_time = *self.start_time.get_or_insert_with(Instant::now); + + if read_bytes == 0 { + return Poll::Ready(Ok(0)); + } + + self.bytes_read += read_bytes as u64; + + // allow a grace period before we start checking the bitrate + let duration = start_time.elapsed(); + if duration >= BITRATE_GRACE_PERIOD { + if (self.bytes_read as f32) / duration.as_secs_f32() < self.min_bytes_per_second as f32 + { + return Poll::Ready(Err(io::Error::new( + ErrorKind::TimedOut, + "Read aborted. Bitrate too low.", + ))); + } + } + + Poll::Ready(Ok(read_bytes)) + } +} + +#[cfg(test)] +mod test { + use super::*; + use futures_executor::block_on; + use futures_util::AsyncReadExt; + + #[test] + fn enforce_minimum_bitrate_is_identity_for_fast_transfers() { + block_on(async { + let bytes: &[u8] = &[0x42; 64 * 1024]; + + let mut reader = EnforceMinimumBitrate::new(bytes, 100); + + let mut buf = Vec::new(); + assert!(reader.read_to_end(&mut buf).await.is_ok()); + assert_eq!(bytes, &buf[..]); + }) + } + + #[test] + fn enforce_minimum_bitrate_is_fails_when_reader_is_too_slow() { + block_on(async { + let bytes: &[u8] = &[0x42; 64 * 1024]; + + let mut reader = EnforceMinimumBitrate::new(bytes, 100); + + let mut buf = vec![0; 50]; + + assert!(reader.read_exact(&mut buf).await.is_ok()); + assert_eq!(buf, &[0x42; 50][..]); + + std::thread::sleep(BITRATE_GRACE_PERIOD); + + assert!(reader.read_to_end(&mut buf).await.is_err()); + }) + } +} diff --git a/tuf-hyper/src/lib.rs b/tuf-hyper/src/lib.rs new file mode 100644 index 0000000..f92ae17 --- /dev/null +++ b/tuf-hyper/src/lib.rs @@ -0,0 +1,27 @@ +//! HTTP repository support for TUF using Hyper. +//! +//! # Example +//! +//! ```no_run +//! use http::Uri; +//! use hyper_util::client::legacy::Client; +//! use hyper_util::rt::TokioExecutor; +//! use tuf::pouf::Pouf1; +//! use tuf_hyper::HttpRepositoryBuilder; +//! +//! let client = Client::builder(TokioExecutor::new()).build_http(); +//! let repository = HttpRepositoryBuilder::<_, Pouf1>::new( +//! Uri::from_static("https://example.com/tuf"), +//! client, +//! ) +//! .user_agent("tuf-client/1.0") +//! .build(); +//! ``` + +#![deny(missing_docs)] +#![allow(clippy::collapsible_if, clippy::type_complexity)] + +mod enforce_minimum_bitrate; +mod repository; + +pub use crate::repository::{HttpRepository, HttpRepositoryBuilder}; diff --git a/tuf-hyper/src/repository.rs b/tuf-hyper/src/repository.rs new file mode 100644 index 0000000..d38bc0d --- /dev/null +++ b/tuf-hyper/src/repository.rs @@ -0,0 +1,958 @@ +//! Read-only Repository implementation backed by a web server. + +use futures_io::AsyncRead; +use futures_util::future::{BoxFuture, FutureExt as _}; +use futures_util::stream::TryStreamExt; +use http::{Request, Response, StatusCode, Uri}; +use http_body_util::BodyExt; +use hyper::body::{Body, Bytes}; +use percent_encoding::utf8_percent_encode; +use std::borrow::Cow; +use std::io; +use std::marker::PhantomData; +use tower::{Service, ServiceExt as _}; + +use crate::enforce_minimum_bitrate::EnforceMinimumBitrate; +use tuf::Result; +use tuf::error::Error; +use tuf::metadata::{MetadataPath, MetadataVersion, TargetPath}; +use tuf::pouf::Pouf; +use tuf::repository::RepositoryProvider; + +/// A builder to create a repository accessible over HTTP. +/// +/// # Example +/// +/// ```no_run +/// use http::Uri; +/// use hyper_util::client::legacy::Client; +/// use hyper_util::rt::TokioExecutor; +/// use tuf::pouf::Pouf1; +/// use tuf_hyper::HttpRepositoryBuilder; +/// +/// let client = Client::builder(TokioExecutor::new()).build_http(); +/// let repository = HttpRepositoryBuilder::<_, Pouf1>::new( +/// Uri::from_static("https://example.com/tuf"), +/// client, +/// ) +/// .user_agent("tuf-client/1.0") +/// .build(); +/// ``` +pub struct HttpRepositoryBuilder< + S, + D, + B = http_body_util::Empty, + ResBody = hyper::body::Incoming, +> where + S: Service, Response = Response>, + D: Pouf, +{ + uri: Uri, + client: S, + user_agent: Option, + metadata_prefix: Option>, + targets_prefix: Option>, + min_bytes_per_second: u32, + _pouf: PhantomData<(D, fn() -> (B, ResBody))>, +} + +impl HttpRepositoryBuilder, ResBody> +where + S: Service>, Response = Response>, + D: Pouf, +{ + /// Create a new repository with the given `Uri` and `Service` client. + pub fn new(uri: Uri, client: S) -> Self { + HttpRepositoryBuilder { + uri, + client, + user_agent: None, + metadata_prefix: None, + targets_prefix: None, + min_bytes_per_second: 4096, + _pouf: PhantomData, + } + } + + /// Create a new repository with the given `Uri` and `Service` client. + pub fn new_with_uri(uri: Uri, client: S) -> Self { + Self::new(uri, client) + } +} + +impl HttpRepositoryBuilder +where + S: Service, Response = Response>, + D: Pouf, +{ + /// Create a new repository with the given `Uri` and a `Service` client with a custom request body type `B`. + pub fn new_with_body(uri: Uri, client: S) -> Self { + HttpRepositoryBuilder { + uri, + client, + user_agent: None, + metadata_prefix: None, + targets_prefix: None, + min_bytes_per_second: 4096, + _pouf: PhantomData, + } + } + + /// Create a new repository with the given `Uri` and a `Service` client with a custom request body type `B`. + pub fn new_with_uri_and_body(uri: Uri, client: S) -> Self { + Self::new_with_body(uri, client) + } + + /// Set the User-Agent prefix. + /// + /// Callers *should* include a custom User-Agent prefix to help maintainers of TUF repositories + /// keep track of which client versions exist in the field. + /// + pub fn user_agent>(mut self, user_agent: T) -> Self { + self.user_agent = Some(user_agent.into()); + self + } + + /// The argument `metadata_prefix` is used to provide an alternate path where metadata is + /// stored on the repository. If `None`, this defaults to `/`. For example, if there is a TUF + /// repository at `https://tuf.example.com/`, but all metadata is stored at `/meta/`, then + /// passing the arg `Some("meta".into())` would cause `root.json` to be fetched from + /// `https://tuf.example.com/meta/root.json`. + pub fn metadata_prefix(mut self, metadata_prefix: Vec) -> Self { + self.metadata_prefix = Some(metadata_prefix); + self + } + + /// The argument `targets_prefix` is used to provide an alternate path where targets is + /// stored on the repository. If `None`, this defaults to `/`. For example, if there is a TUF + /// repository at `https://tuf.example.com/`, but all targets are stored at `/targets/`, then + /// passing the arg `Some("targets".into())` would cause `hello-world` to be fetched from + /// `https://tuf.example.com/targets/hello-world`. + pub fn targets_prefix(mut self, targets_prefix: Vec) -> Self { + self.targets_prefix = Some(targets_prefix); + self + } + + /// Set the minimum bytes per second for a read to be considered good. + pub fn min_bytes_per_second(mut self, min: u32) -> Self { + self.min_bytes_per_second = min; + self + } + + /// Build a `HttpRepository`. + pub fn build(self) -> HttpRepository { + let user_agent = match self.user_agent { + Some(user_agent) => user_agent, + None => "rust-tuf".into(), + }; + + HttpRepository { + uri: self.uri, + client: self.client, + user_agent, + metadata_prefix: self.metadata_prefix, + targets_prefix: self.targets_prefix, + min_bytes_per_second: self.min_bytes_per_second, + _pouf: PhantomData, + } + } +} + +/// A repository accessible over HTTP. +#[derive(Debug)] +pub struct HttpRepository, ResBody = hyper::body::Incoming> +where + S: Service, Response = Response>, + D: Pouf, +{ + uri: Uri, + client: S, + user_agent: String, + metadata_prefix: Option>, + targets_prefix: Option>, + min_bytes_per_second: u32, + _pouf: PhantomData<(D, fn() -> (B, ResBody))>, +} + +// Configuration for urlencoding URI path elements. +// From https://url.spec.whatwg.org/#path-percent-encode-set +const URLENCODE_FRAGMENT: &percent_encoding::AsciiSet = &percent_encoding::CONTROLS + .add(b' ') + .add(b'"') + .add(b'<') + .add(b'>') + .add(b'`'); +const URLENCODE_PATH: &percent_encoding::AsciiSet = &URLENCODE_FRAGMENT + .add(b'#') + .add(b'?') + .add(b'{') + .add(b'}') + .add(b'%'); + +fn extend_uri(uri: &Uri, prefix: &Option>, components: &[String]) -> Result { + let uri = uri.clone(); + let mut uri_parts = uri.into_parts(); + + let (path, query) = match &uri_parts.path_and_query { + Some(path_and_query) => (path_and_query.path(), path_and_query.query()), + None => ("", None), + }; + + let modified_path = path.strip_suffix('/').unwrap_or(path); + let mut path_split: Vec> = modified_path.split('/').map(Cow::Borrowed).collect(); + + if let Some(prefix) = prefix { + path_split.extend(prefix.iter().map(|s| Cow::Borrowed(s.as_str()))); + } + path_split.extend( + components + .iter() + .map(|s| utf8_percent_encode(s, URLENCODE_PATH).into()), + ); + let constructed_path = path_split.join("/"); + + uri_parts.path_and_query = + match query { + Some(query) => Some(format!("{}?{}", constructed_path, query).parse().map_err( + |_| { + Error::IllegalArgument(format!( + "Invalid path and query: {:?}, {:?}", + constructed_path, query + )) + }, + )?), + None => Some(constructed_path.parse().map_err(|_| { + Error::IllegalArgument(format!("Invalid URI path: {:?}", constructed_path)) + })?), + }; + + Uri::from_parts(uri_parts).map_err(|_| { + Error::IllegalArgument(format!( + "Invalid URI parts: {:?}, {:?}, {:?}", + constructed_path, prefix, components + )) + }) +} + +impl HttpRepository +where + S: Service, Response = Response> + Clone + Send + Sync + 'static, + S::Error: Into> + Send + Sync + 'static, + S::Future: Send + 'static, + D: Pouf, + B: Body + Default + Send + 'static, + B::Data: Send, + B::Error: Into>, + ResBody: Body + Send + Unpin + 'static, + ResBody::Data: AsRef<[u8]> + Send, + ResBody::Error: Into>, +{ + async fn get(&self, uri: &Uri) -> Result> { + let req = Request::builder() + .uri(uri) + .header("User-Agent", &*self.user_agent) + .body(B::default()) + .map_err(|err| Error::Http { + uri: uri.to_string(), + err, + })?; + + self.client + .clone() + .oneshot(req) + .await + .map_err(|err| Error::Repository { + uri: uri.to_string(), + err: err.into(), + }) + } +} + +impl RepositoryProvider for HttpRepository +where + S: Service, Response = Response> + Clone + Send + Sync + 'static, + S::Error: Into> + Send + Sync + 'static, + S::Future: Send + 'static, + D: Pouf, + B: Body + Default + Send + 'static, + B::Data: Send, + B::Error: Into>, + ResBody: Body + Send + Unpin + 'static, + ResBody::Data: AsRef<[u8]> + Send, + ResBody::Error: Into>, +{ + fn fetch_metadata<'a>( + &'a self, + meta_path: &MetadataPath, + version: MetadataVersion, + ) -> BoxFuture<'a, Result>> { + let meta_path = meta_path.clone(); + let components = meta_path.components::(version); + let uri = extend_uri(&self.uri, &self.metadata_prefix, &components); + + async move { + // TODO(#278) check content length if known and fail early if the payload is too large. + + let uri = uri?; + let resp = self.get(&uri).await?; + + let status = resp.status(); + if status == StatusCode::OK { + let reader = EnforceMinimumBitrate::new( + resp.into_body() + .into_data_stream() + .map_err(io::Error::other) + .into_async_read(), + self.min_bytes_per_second, + ); + + let reader: Box = Box::new(reader); + Ok(reader) + } else if status == StatusCode::NOT_FOUND { + Err(Error::MetadataNotFound { + path: meta_path, + version, + }) + } else { + Err(Error::BadHttpStatus { + uri: uri.to_string(), + code: status, + }) + } + } + .boxed() + } + + fn fetch_target<'a>( + &'a self, + target_path: &TargetPath, + ) -> BoxFuture<'a, Result>> { + let target_path = target_path.clone(); + let components = target_path.components(); + let uri = extend_uri(&self.uri, &self.targets_prefix, &components); + + async move { + // TODO(#278) check content length if known and fail early if the payload is too large. + + let uri = uri?; + let resp = self.get(&uri).await?; + + let status = resp.status(); + if status == StatusCode::OK { + let reader = EnforceMinimumBitrate::new( + resp.into_body() + .into_data_stream() + .map_err(io::Error::other) + .into_async_read(), + self.min_bytes_per_second, + ); + + let reader: Box = Box::new(reader); + Ok(reader) + } else if status == StatusCode::NOT_FOUND { + Err(Error::TargetNotFound(target_path)) + } else { + Err(Error::BadHttpStatus { + uri: uri.to_string(), + code: status, + }) + } + } + .boxed() + } +} + +#[cfg(test)] +mod test { + use super::*; + + // Old behavior of the `HttpRepository::get` extension + // functionality + #[test] + fn http_repository_uri_construction() { + let base_uri = "http://example.com/one"; + + let prefix = Some(vec![String::from("prefix")]); + let components = [ + String::from("components_one"), + String::from("components_two"), + ]; + + let uri = base_uri.parse::().unwrap(); + let extended_uri = extend_uri(&uri, &prefix, &components).unwrap(); + + assert_eq!( + extended_uri.to_string(), + "http://example.com/one/prefix/components_one/components_two" + ); + } + + #[test] + fn http_repository_uri_construction_encoded() { + let base_uri = "http://example.com/one"; + + let prefix = Some(vec![String::from("prefix")]); + let components = [String::from("chars to encode#?")]; + let uri = base_uri.parse::().unwrap(); + let extended_uri = extend_uri(&uri, &prefix, &components) + .expect("correctly generated a URI with a zone id"); + + assert_eq!( + extended_uri.to_string(), + "http://example.com/one/prefix/chars%20to%20encode%23%3F" + ); + } + + #[test] + fn http_repository_uri_construction_no_components() { + let base_uri = "http://example.com/one"; + + let prefix = Some(vec![String::from("prefix")]); + let components = []; + + let uri = base_uri.parse::().unwrap(); + let extended_uri = extend_uri(&uri, &prefix, &components).unwrap(); + + assert_eq!(extended_uri.to_string(), "http://example.com/one/prefix"); + } + + #[test] + fn http_repository_uri_construction_no_prefix() { + let base_uri = "http://example.com/one"; + + let prefix = None; + let components = [ + String::from("components_one"), + String::from("components_two"), + ]; + + let uri = base_uri.parse::().unwrap(); + let extended_uri = extend_uri(&uri, &prefix, &components).unwrap(); + + assert_eq!( + extended_uri.to_string(), + "http://example.com/one/components_one/components_two" + ); + } + + #[test] + fn http_repository_uri_construction_with_query() { + let base_uri = "http://example.com/one?test=1"; + + let prefix = None; + let components = [ + String::from("components_one"), + String::from("components_two"), + ]; + + let uri = base_uri.parse::().unwrap(); + let extended_uri = extend_uri(&uri, &prefix, &components).unwrap(); + + assert_eq!( + extended_uri.to_string(), + "http://example.com/one/components_one/components_two?test=1" + ); + } + + #[test] + fn http_repository_uri_construction_ipv6_zoneid() { + let base_uri = "http://[aaaa::aaaa:aaaa:aaaa:1234%252]:80"; + + let prefix = Some(vec![String::from("prefix")]); + let components = [ + String::from("componenents_one"), + String::from("components_two"), + ]; + let uri = base_uri.parse::().unwrap(); + let extended_uri = extend_uri(&uri, &prefix, &components) + .expect("correctly generated a URI with a zone id"); + assert_eq!( + extended_uri.to_string(), + "http://[aaaa::aaaa:aaaa:aaaa:1234%252]:80/prefix/componenents_one/components_two" + ); + } + + #[derive(Clone, Debug, PartialEq, Eq)] + struct MockServiceError(&'static str); + + impl std::fmt::Display for MockServiceError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.0) + } + } + + impl std::error::Error for MockServiceError {} + + #[derive(Clone)] + struct ReadinessFailingService; + + impl Service for ReadinessFailingService { + type Response = Response; + type Error = MockServiceError; + type Future = std::future::Ready>; + + fn poll_ready( + &mut self, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Err(MockServiceError("simulated poll_ready error"))) + } + + fn call(&mut self, _req: Req) -> Self::Future { + std::future::ready(Err(MockServiceError("should not be called"))) + } + } + + #[derive(Clone)] + struct CallFailingService; + + impl Service for CallFailingService { + type Response = Response; + type Error = MockServiceError; + type Future = std::future::Ready>; + + fn poll_ready( + &mut self, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Ok(())) + } + + fn call(&mut self, _req: Req) -> Self::Future { + std::future::ready(Err(MockServiceError("simulated call error"))) + } + } + + #[derive(Clone)] + struct CustomHeaderMiddleware { + inner: S, + } + + impl Service> for CustomHeaderMiddleware + where + S: Service>, + { + type Response = S::Response; + type Error = S::Error; + type Future = S::Future; + + fn poll_ready( + &mut self, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + self.inner.poll_ready(cx) + } + + fn call(&mut self, mut req: Request) -> Self::Future { + req.headers_mut() + .insert("X-Custom-Tuf-Header", "test-middleware".parse().unwrap()); + self.inner.call(req) + } + } + + #[tokio::test] + async fn http_repository_maps_service_poll_ready_error() { + let repo = HttpRepositoryBuilder::<_, tuf::pouf::Pouf1>::new( + Uri::from_static("http://example.com/repo"), + ReadinessFailingService, + ) + .build(); + + let err = repo + .fetch_metadata(&MetadataPath::root(), MetadataVersion::Number(1)) + .await + .err() + .expect("expected fetch_metadata to fail"); + + match err { + Error::Repository { uri, err } => { + assert_eq!(uri, "http://example.com/repo/1.root.json"); + let mock_err = err + .downcast_ref::() + .expect("should be downcastable to MockServiceError"); + assert_eq!(mock_err, &MockServiceError("simulated poll_ready error")); + } + other => panic!("expected Error::Backend, got {:?}", other), + } + } + + #[tokio::test] + async fn http_repository_maps_service_call_error() { + let repo = HttpRepositoryBuilder::<_, tuf::pouf::Pouf1>::new( + Uri::from_static("http://example.com/repo"), + CallFailingService, + ) + .build(); + + let err = repo + .fetch_metadata(&MetadataPath::root(), MetadataVersion::Number(1)) + .await + .err() + .expect("expected fetch_metadata to fail"); + + match err { + Error::Repository { uri, err } => { + assert_eq!(uri, "http://example.com/repo/1.root.json"); + let mock_err = err + .downcast_ref::() + .expect("should be downcastable to MockServiceError"); + assert_eq!(mock_err, &MockServiceError("simulated call error")); + } + other => panic!("expected Error::Backend, got {:?}", other), + } + } + + #[derive(Clone)] + struct PendingReadinessService { + polls: std::sync::Arc, + } + + impl Service for PendingReadinessService { + type Response = Response; + type Error = MockServiceError; + type Future = std::future::Ready>; + + fn poll_ready( + &mut self, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + let count = self.polls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if count == 0 { + cx.waker().wake_by_ref(); + std::task::Poll::Pending + } else { + std::task::Poll::Ready(Err(MockServiceError("ready after pending"))) + } + } + + fn call(&mut self, _req: Req) -> Self::Future { + std::future::ready(Err(MockServiceError("should not be called"))) + } + } + + #[tokio::test] + async fn http_repository_waits_for_poll_ready() { + let polls = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let repo = HttpRepositoryBuilder::<_, tuf::pouf::Pouf1>::new( + Uri::from_static("http://example.com/repo"), + PendingReadinessService { + polls: polls.clone(), + }, + ) + .build(); + + let err = repo + .fetch_metadata(&MetadataPath::root(), MetadataVersion::Number(1)) + .await + .err() + .expect("expected fetch_metadata to fail"); + + match err { + Error::Repository { uri, err } => { + assert_eq!(uri, "http://example.com/repo/1.root.json"); + let mock_err = err + .downcast_ref::() + .expect("should be downcastable to MockServiceError"); + assert_eq!(mock_err, &MockServiceError("ready after pending")); + assert_eq!(polls.load(std::sync::atomic::Ordering::SeqCst), 2); + } + other => panic!("expected Error::Backend, got {:?}", other), + } + } + + #[test] + fn http_repository_accepts_custom_middleware_wrapper() { + // Verify that wrapping an HTTP client with custom Tower middleware compiles and builds cleanly. + let client = + hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new()) + .build_http(); + let middleware = CustomHeaderMiddleware { inner: client }; + + let _repo = HttpRepositoryBuilder::<_, tuf::pouf::Pouf1>::new( + Uri::from_static("http://example.com/repo"), + middleware, + ) + .build(); + } + + #[derive(Clone)] + struct MockResponseService { + status: StatusCode, + body: &'static [u8], + } + + impl Service>> for MockResponseService { + type Response = Response>; + type Error = MockServiceError; + type Future = std::future::Ready>; + + fn poll_ready( + &mut self, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Ok(())) + } + + fn call(&mut self, _req: Request>) -> Self::Future { + let resp = Response::builder() + .status(self.status) + .body(http_body_util::Full::new(Bytes::from_static(self.body))) + .unwrap(); + std::future::ready(Ok(resp)) + } + } + + #[tokio::test] + async fn http_repository_fetch_metadata_success() { + use futures_util::io::AsyncReadExt; + let repo = HttpRepositoryBuilder::< + _, + tuf::pouf::Pouf1, + http_body_util::Empty, + http_body_util::Full, + >::new_with_body( + Uri::from_static("http://example.com/repo"), + MockResponseService { + status: StatusCode::OK, + body: b"root metadata payload", + }, + ) + .build(); + + let mut reader = repo + .fetch_metadata(&MetadataPath::root(), MetadataVersion::Number(1)) + .await + .expect("expected fetch_metadata to succeed"); + let mut buf = Vec::new(); + reader.read_to_end(&mut buf).await.unwrap(); + assert_eq!(buf, b"root metadata payload"); + } + + #[tokio::test] + async fn http_repository_fetch_target_success() { + use futures_util::io::AsyncReadExt; + let repo = HttpRepositoryBuilder::< + _, + tuf::pouf::Pouf1, + http_body_util::Empty, + http_body_util::Full, + >::new_with_body( + Uri::from_static("http://example.com/repo"), + MockResponseService { + status: StatusCode::OK, + body: b"target binary data", + }, + ) + .build(); + + let mut reader = repo + .fetch_target(&TargetPath::new("foo/bar.bin").unwrap()) + .await + .expect("expected fetch_target to succeed"); + let mut buf = Vec::new(); + reader.read_to_end(&mut buf).await.unwrap(); + assert_eq!(buf, b"target binary data"); + } + + #[tokio::test] + async fn http_repository_fetch_metadata_not_found() { + let repo = HttpRepositoryBuilder::< + _, + tuf::pouf::Pouf1, + http_body_util::Empty, + http_body_util::Full, + >::new_with_body( + Uri::from_static("http://example.com/repo"), + MockResponseService { + status: StatusCode::NOT_FOUND, + body: b"", + }, + ) + .build(); + + let err = repo + .fetch_metadata(&MetadataPath::root(), MetadataVersion::Number(1)) + .await + .err() + .expect("expected fetch_metadata to fail"); + + match err { + Error::MetadataNotFound { path, version } => { + assert_eq!(path, MetadataPath::root()); + assert_eq!(version, MetadataVersion::Number(1)); + } + other => panic!("expected Error::MetadataNotFound, got {:?}", other), + } + } + + #[tokio::test] + async fn http_repository_fetch_target_not_found() { + let target_path = TargetPath::new("missing.tar.gz").unwrap(); + let repo = HttpRepositoryBuilder::< + _, + tuf::pouf::Pouf1, + http_body_util::Empty, + http_body_util::Full, + >::new_with_body( + Uri::from_static("http://example.com/repo"), + MockResponseService { + status: StatusCode::NOT_FOUND, + body: b"", + }, + ) + .build(); + + let err = repo + .fetch_target(&target_path) + .await + .err() + .expect("expected fetch_target to fail"); + + match err { + Error::TargetNotFound(path) => { + assert_eq!(path, target_path); + } + other => panic!("expected Error::TargetNotFound, got {:?}", other), + } + } + + #[tokio::test] + async fn http_repository_fetch_bad_http_status() { + let repo = HttpRepositoryBuilder::< + _, + tuf::pouf::Pouf1, + http_body_util::Empty, + http_body_util::Full, + >::new_with_body( + Uri::from_static("http://example.com/repo"), + MockResponseService { + status: StatusCode::INTERNAL_SERVER_ERROR, + body: b"", + }, + ) + .build(); + + let err_meta = repo + .fetch_metadata(&MetadataPath::root(), MetadataVersion::Number(1)) + .await + .err() + .expect("expected fetch_metadata to fail on 500"); + match err_meta { + Error::BadHttpStatus { uri, code } => { + assert_eq!(uri, "http://example.com/repo/1.root.json"); + assert_eq!(code, StatusCode::INTERNAL_SERVER_ERROR); + } + other => panic!("expected Error::BadHttpStatus, got {:?}", other), + } + + let err_target = repo + .fetch_target(&TargetPath::new("file.dat").unwrap()) + .await + .err() + .expect("expected fetch_target to fail on 500"); + match err_target { + Error::BadHttpStatus { uri, code } => { + assert_eq!(uri, "http://example.com/repo/file.dat"); + assert_eq!(code, StatusCode::INTERNAL_SERVER_ERROR); + } + other => panic!("expected Error::BadHttpStatus, got {:?}", other), + } + } + + #[derive(Clone)] + struct HeaderCheckingService { + expected_header: &'static str, + expected_value: &'static str, + } + + impl Service>> for HeaderCheckingService { + type Response = Response>; + type Error = MockServiceError; + type Future = std::future::Ready>; + + fn poll_ready( + &mut self, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Ok(())) + } + + fn call(&mut self, req: Request>) -> Self::Future { + if let Some(val) = req.headers().get(self.expected_header) + && val == self.expected_value + { + let resp = Response::builder() + .status(StatusCode::OK) + .body(http_body_util::Full::new(Bytes::from_static(b"ok"))) + .unwrap(); + return std::future::ready(Ok(resp)); + } + std::future::ready(Err(MockServiceError("header check failed"))) + } + } + + #[tokio::test] + async fn http_repository_custom_middleware_wrapper_inserts_header() { + use futures_util::io::AsyncReadExt; + let inner = HeaderCheckingService { + expected_header: "X-Custom-Tuf-Header", + expected_value: "test-middleware", + }; + let middleware = CustomHeaderMiddleware { inner }; + + let repo = HttpRepositoryBuilder::< + _, + tuf::pouf::Pouf1, + http_body_util::Empty, + http_body_util::Full, + >::new_with_body(Uri::from_static("http://example.com/repo"), middleware) + .build(); + + let mut reader = repo + .fetch_metadata(&MetadataPath::root(), MetadataVersion::Number(1)) + .await + .expect("should succeed with injected header"); + let mut buf = Vec::new(); + reader.read_to_end(&mut buf).await.unwrap(); + assert_eq!(buf, b"ok"); + } + + #[test] + fn http_repository_builder_constructors() { + let client = MockResponseService { + status: StatusCode::OK, + body: b"", + }; + let uri = Uri::from_static("http://example.com/tuf"); + let builder1 = HttpRepositoryBuilder::< + _, + tuf::pouf::Pouf1, + http_body_util::Empty, + http_body_util::Full, + >::new(uri.clone(), client.clone()) + .user_agent("test-client/1.0") + .metadata_prefix(vec!["meta".into()]) + .targets_prefix(vec!["targets".into()]) + .min_bytes_per_second(1024); + let repo1 = builder1.build(); + assert_eq!(repo1.user_agent, "test-client/1.0"); + assert_eq!(repo1.min_bytes_per_second, 1024); + + let builder2 = HttpRepositoryBuilder::< + _, + tuf::pouf::Pouf1, + http_body_util::Empty, + http_body_util::Full, + >::new_with_body(uri, client.clone()); + let repo2 = builder2.build(); + assert_eq!(repo2.user_agent, "rust-tuf"); + + let builder3 = HttpRepositoryBuilder::< + _, + tuf::pouf::Pouf1, + http_body_util::Empty, + http_body_util::Full, + >::new_with_body(Uri::from_static("http://example.com/tuf"), client); + let _repo3 = builder3.build(); + } +} diff --git a/tuf/Cargo.toml b/tuf/Cargo.toml index e274d60..62ee089 100644 --- a/tuf/Cargo.toml +++ b/tuf/Cargo.toml @@ -19,15 +19,14 @@ path = "./src/lib.rs" [dependencies] chrono = { version = "0.4.42", features = [ "serde" ] } -data-encoding = "2.9.0" +data-encoding = "2.10" futures-io = "0.3.31" futures-util = { version = "0.3.31", features = [ "io" ] } -http = "0.2.4" -hyper = { version = "0.14.19", default-features = false, features = [ "stream", "client", "http1" ], optional = true } +http = "1" itoa = "1.0" log = "0.4" percent-encoding = "2.1" -ring = { version = "0.17" } +ring = "0.17" serde = { version = "1.0.204", features = ["derive"] } serde_json = "1" tempfile = "3" @@ -38,8 +37,7 @@ url = "2" [dev-dependencies] assert_matches = "1.5.0" futures-executor = "0.3.31" +http = "1" maplit = "1" pretty_assertions = "1" -[features] -default = ["hyper", "hyper/tcp"] diff --git a/tuf/src/client.rs b/tuf/src/client.rs index 9942207..60943d7 100644 --- a/tuf/src/client.rs +++ b/tuf/src/client.rs @@ -4,33 +4,24 @@ //! //! ```no_run //! # use futures_executor::block_on; -//! # use hyper::client::Client as HttpClient; //! # use std::path::PathBuf; -//! # use std::str::FromStr; -//! # use tuf::{Result, Database}; +//! # use tuf::Result; //! # use tuf::crypto::PublicKey; //! # use tuf::client::{Client, Config}; -//! # use tuf::metadata::{RootMetadata, Role, MetadataPath, MetadataVersion}; +//! # use tuf::metadata::MetadataVersion; //! # use tuf::pouf::Pouf1; -//! # use tuf::repository::{FileSystemRepository, HttpRepositoryBuilder}; +//! # use tuf::repository::{EphemeralRepository, FileSystemRepository}; //! # //! # const PUBLIC_KEY: &'static [u8] = include_bytes!("../tests/ed25519/ed25519-1.pub"); //! # //! # fn load_root_public_keys() -> Vec { //! # vec![PublicKey::from_ed25519(PUBLIC_KEY).unwrap()] //! # } -//! # //! # fn main() -> Result<()> { //! # block_on(async { //! let root_public_keys = load_root_public_keys(); //! let local = FileSystemRepository::::new(PathBuf::from("~/.rustup")); -//! -//! let remote = HttpRepositoryBuilder::new_with_uri( -//! "https://static.rust-lang.org/".parse::().unwrap(), -//! HttpClient::new(), -//! ) -//! .user_agent("rustup/1.4.0") -//! .build(); +//! let remote = EphemeralRepository::::new(); //! //! let mut client = Client::with_trusted_root_keys( //! Config::default(), @@ -1088,12 +1079,20 @@ where } } - let meta = self - .tuf - .trusted_delegations() - .get(delegation.name()) - .unwrap() - .clone(); + let meta = match self.tuf.trusted_delegations().get(delegation.name()) { + Some(m) => m.clone(), + None => { + let err = Error::MetadataNotFound { + path: delegation.name().clone(), + version: MetadataVersion::None, + }; + if delegation.terminating() { + return (true, Err(err)); + } else { + continue; + } + } + }; let f: Pin>> = Box::pin(self.lookup_target_description( start_time, diff --git a/tuf/src/util.rs b/tuf/src/enforce_size_and_hash.rs similarity index 60% rename from tuf/src/util.rs rename to tuf/src/enforce_size_and_hash.rs index ebc7ec7..0a334b8 100644 --- a/tuf/src/util.rs +++ b/tuf/src/enforce_size_and_hash.rs @@ -5,94 +5,10 @@ use std::io::{self, ErrorKind}; use std::marker::Unpin; use std::pin::Pin; use std::task::{Context, Poll}; -use std::time::{Duration, Instant}; use crate::Result; use crate::crypto::{HashAlgorithm, HashValue}; -pub(crate) trait SafeAsyncRead: AsyncRead + Sized + Unpin { - /// Creates an `AsyncRead` adapter which will fail transfers slower than - /// `min_bytes_per_second`. - fn enforce_minimum_bitrate(self, min_bytes_per_second: u32) -> EnforceMinimumBitrate { - EnforceMinimumBitrate::new(self, min_bytes_per_second) - } - - /// Creates an `AsyncRead` adapter that ensures the consumer can't read more than `max_length` - /// bytes. Also, when the underlying `AsyncRead` is fully consumed, the hash of the data is - /// optionally calculated and checked against `hash_data`. Consumers should purge and untrust - /// all read bytes if the returned `AsyncRead` ever returns an `Err`. - /// - /// It is **critical** that none of the bytes from this struct are used until it has been fully - /// consumed as the data is untrusted. - fn check_length_and_hash( - self, - max_length: u64, - hash_data: Vec<(&'static HashAlgorithm, HashValue)>, - ) -> Result> { - SafeReader::new(self, max_length, hash_data) - } -} - -impl SafeAsyncRead for R {} - -/// Wraps an `AsyncRead` to detect and fail transfers slower than a minimum bitrate. -pub(crate) struct EnforceMinimumBitrate { - inner: R, - min_bytes_per_second: u32, - start_time: Option, - bytes_read: u64, -} - -impl EnforceMinimumBitrate { - /// Create a new `EnforceMinimumBitrate`. - pub(crate) fn new(read: R, min_bytes_per_second: u32) -> Self { - Self { - inner: read, - min_bytes_per_second, - start_time: None, - bytes_read: 0, - } - } -} - -#[cfg(not(test))] -const BITRATE_GRACE_PERIOD: Duration = Duration::from_secs(30); -#[cfg(test)] -const BITRATE_GRACE_PERIOD: Duration = Duration::from_secs(1); - -impl AsyncRead for EnforceMinimumBitrate { - fn poll_read( - mut self: Pin<&mut Self>, - cx: &mut Context, - buf: &mut [u8], - ) -> Poll> { - // FIXME(#272) transfers that stall out completely won't enforce the minimum bit rate. - let read_bytes = ready!(Pin::new(&mut self.inner).poll_read(cx, buf))?; - - let start_time = *self.start_time.get_or_insert_with(Instant::now); - - if read_bytes == 0 { - return Poll::Ready(Ok(0)); - } - - self.bytes_read += read_bytes as u64; - - // allow a grace period before we start checking the bitrate - let duration = start_time.elapsed(); - if duration >= BITRATE_GRACE_PERIOD { - if (self.bytes_read as f32) / duration.as_secs_f32() < self.min_bytes_per_second as f32 - { - return Poll::Ready(Err(io::Error::new( - ErrorKind::TimedOut, - "Read aborted. Bitrate too low.", - ))); - } - } - - Poll::Ready(Ok(read_bytes)) - } -} - /// Wrapper to verify a byte stream as it is read. /// /// Wraps an `AsyncRead` to ensure that the consumer can't read more than a capped maximum number of @@ -103,14 +19,14 @@ impl AsyncRead for EnforceMinimumBitrate { /// /// It is **critical** that none of the bytes from this struct are used until it has been fully /// consumed as the data is untrusted. -pub(crate) struct SafeReader { +pub(crate) struct EnforceSizeAndHash { inner: R, max_size: u64, hashers: Vec<(digest::Context, HashValue)>, bytes_read: u64, } -impl SafeReader { +impl EnforceSizeAndHash { /// Create a new `SafeReader`. /// /// The argument `hash_data` takes a `HashAlgorithm` and expected `HashValue`. The given @@ -127,7 +43,7 @@ impl SafeReader { hashers.push((alg.digest_context()?, value)); } - Ok(SafeReader { + Ok(EnforceSizeAndHash { inner: read, max_size, hashers, @@ -136,7 +52,7 @@ impl SafeReader { } } -impl AsyncRead for SafeReader { +impl AsyncRead for EnforceSizeAndHash { fn poll_read( mut self: Pin<&mut Self>, cx: &mut Context, @@ -187,7 +103,7 @@ mod test { fn valid_read() { block_on(async { let bytes: &[u8] = &[0x00, 0x01, 0x02, 0x03]; - let mut reader = SafeReader::new(bytes, bytes.len() as u64, vec![]).unwrap(); + let mut reader = EnforceSizeAndHash::new(bytes, bytes.len() as u64, vec![]).unwrap(); let mut buf = Vec::new(); assert!(reader.read_to_end(&mut buf).await.is_ok()); assert_eq!(buf, bytes); @@ -198,7 +114,7 @@ mod test { fn valid_read_large_data() { block_on(async { let bytes: &[u8] = &[0x00; 64 * 1024]; - let mut reader = SafeReader::new(bytes, bytes.len() as u64, vec![]).unwrap(); + let mut reader = EnforceSizeAndHash::new(bytes, bytes.len() as u64, vec![]).unwrap(); let mut buf = Vec::new(); assert!(reader.read_to_end(&mut buf).await.is_ok()); assert_eq!(buf, bytes); @@ -209,7 +125,8 @@ mod test { fn valid_read_below_max_size() { block_on(async { let bytes: &[u8] = &[0x00, 0x01, 0x02, 0x03]; - let mut reader = SafeReader::new(bytes, (bytes.len() as u64) + 1, vec![]).unwrap(); + let mut reader = + EnforceSizeAndHash::new(bytes, (bytes.len() as u64) + 1, vec![]).unwrap(); let mut buf = Vec::new(); assert!(reader.read_to_end(&mut buf).await.is_ok()); assert_eq!(buf, bytes); @@ -220,7 +137,8 @@ mod test { fn invalid_read_above_max_size() { block_on(async { let bytes: &[u8] = &[0x00, 0x01, 0x02, 0x03]; - let mut reader = SafeReader::new(bytes, (bytes.len() as u64) - 1, vec![]).unwrap(); + let mut reader = + EnforceSizeAndHash::new(bytes, (bytes.len() as u64) - 1, vec![]).unwrap(); let mut buf = Vec::new(); assert!(reader.read_to_end(&mut buf).await.is_err()); }) @@ -230,7 +148,8 @@ mod test { fn invalid_read_above_max_size_large_data() { block_on(async { let bytes: &[u8] = &[0x00; 64 * 1024]; - let mut reader = SafeReader::new(bytes, (bytes.len() as u64) - 1, vec![]).unwrap(); + let mut reader = + EnforceSizeAndHash::new(bytes, (bytes.len() as u64) - 1, vec![]).unwrap(); let mut buf = Vec::new(); assert!(reader.read_to_end(&mut buf).await.is_err()); }) @@ -243,7 +162,7 @@ mod test { let mut context = digest::Context::new(&SHA256); context.update(bytes); let hash_value = HashValue::new(context.finish().as_ref().to_vec()); - let mut reader = SafeReader::new( + let mut reader = EnforceSizeAndHash::new( bytes, bytes.len() as u64, vec![(&HashAlgorithm::Sha256, hash_value)], @@ -263,7 +182,7 @@ mod test { context.update(bytes); context.update(&[0xFF]); // evil bytes let hash_value = HashValue::new(context.finish().as_ref().to_vec()); - let mut reader = SafeReader::new( + let mut reader = EnforceSizeAndHash::new( bytes, bytes.len() as u64, vec![(&HashAlgorithm::Sha256, hash_value)], @@ -281,7 +200,7 @@ mod test { let mut context = digest::Context::new(&SHA256); context.update(bytes); let hash_value = HashValue::new(context.finish().as_ref().to_vec()); - let mut reader = SafeReader::new( + let mut reader = EnforceSizeAndHash::new( bytes, bytes.len() as u64, vec![(&HashAlgorithm::Sha256, hash_value)], @@ -301,7 +220,7 @@ mod test { context.update(bytes); context.update(&[0xFF]); // evil bytes let hash_value = HashValue::new(context.finish().as_ref().to_vec()); - let mut reader = SafeReader::new( + let mut reader = EnforceSizeAndHash::new( bytes, bytes.len() as u64, vec![(&HashAlgorithm::Sha256, hash_value)], @@ -311,35 +230,4 @@ mod test { assert!(reader.read_to_end(&mut buf).await.is_err()); }) } - - #[test] - fn enforce_minimum_bitrate_is_identity_for_fast_transfers() { - block_on(async { - let bytes: &[u8] = &[0x42; 64 * 1024]; - - let mut reader = EnforceMinimumBitrate::new(bytes, 100); - - let mut buf = Vec::new(); - assert!(reader.read_to_end(&mut buf).await.is_ok()); - assert_eq!(bytes, &buf[..]); - }) - } - - #[test] - fn enforce_minimum_bitrate_is_fails_when_reader_is_too_slow() { - block_on(async { - let bytes: &[u8] = &[0x42; 64 * 1024]; - - let mut reader = EnforceMinimumBitrate::new(bytes, 100); - - let mut buf = vec![0; 50]; - - assert!(reader.read_exact(&mut buf).await.is_ok()); - assert_eq!(buf, &[0x42; 50][..]); - - std::thread::sleep(BITRATE_GRACE_PERIOD); - - assert!(reader.read_to_end(&mut buf).await.is_err()); - }) - } } diff --git a/tuf/src/error.rs b/tuf/src/error.rs index 4e6180d..b67e9fd 100644 --- a/tuf/src/error.rs +++ b/tuf/src/error.rs @@ -51,16 +51,15 @@ pub enum Error { err: http::Error, }, - /// Errors that can occur parsing HTTP streams. - #[cfg(feature = "hyper")] - #[error("hyper error for {uri}")] - Hyper { + /// Error occurring in a repository backend (e.g., HTTP, custom transport). + #[error("repository backend error for {uri}")] + Repository { /// URI Resource that resulted in the error. uri: String, /// The error. #[source] - err: hyper::Error, + err: Box, }, /// Unexpected HTTP response status. @@ -222,3 +221,47 @@ pub enum Error { role: MetadataPath, }, } + +#[cfg(test)] +mod test { + use super::*; + use std::error::Error as StdError; + use std::fmt; + + #[derive(Debug, PartialEq, Eq)] + struct CustomServiceError { + code: u32, + } + + impl fmt::Display for CustomServiceError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "custom service error code {}", self.code) + } + } + + impl StdError for CustomServiceError {} + + #[test] + fn repository_error_source_and_downcast() { + let inner = CustomServiceError { code: 404 }; + let err = Error::Repository { + uri: "https://example.com/repo".to_string(), + err: Box::new(inner), + }; + + assert_eq!( + err.to_string(), + "repository backend error for https://example.com/repo" + ); + + let source = err + .source() + .expect("Error::Backend should provide a source"); + assert_eq!(source.to_string(), "custom service error code 404"); + + let downcasted = source + .downcast_ref::() + .expect("should downcast cleanly to CustomServiceError"); + assert_eq!(downcasted, &CustomServiceError { code: 404 }); + } +} diff --git a/tuf/src/lib.rs b/tuf/src/lib.rs index b5246b0..60dad58 100644 --- a/tuf/src/lib.rs +++ b/tuf/src/lib.rs @@ -121,8 +121,8 @@ pub mod repo_builder; pub mod repository; pub mod verify; +mod enforce_size_and_hash; mod format_hex; -mod util; pub use crate::database::*; pub use crate::error::*; diff --git a/tuf/src/repository.rs b/tuf/src/repository.rs index 379bbd2..012fb7f 100644 --- a/tuf/src/repository.rs +++ b/tuf/src/repository.rs @@ -1,11 +1,11 @@ //! Interfaces for interacting with different types of TUF repositories. use crate::crypto::{self, HashAlgorithm, HashValue}; +use crate::enforce_size_and_hash::EnforceSizeAndHash; use crate::metadata::{ Metadata, MetadataPath, MetadataVersion, RawSignedMetadata, TargetDescription, TargetPath, }; use crate::pouf::Pouf; -use crate::util::SafeAsyncRead; use crate::{Error, Result}; use futures_io::AsyncRead; @@ -19,12 +19,6 @@ pub use self::file_system::{ FileSystemBatchUpdate, FileSystemRepository, FileSystemRepositoryBuilder, }; -#[cfg(feature = "hyper")] -mod http; - -#[cfg(feature = "hyper")] -pub use self::http::{HttpRepository, HttpRepositoryBuilder}; - mod ephemeral; pub use self::ephemeral::{EphemeralBatchUpdate, EphemeralRepository}; @@ -278,11 +272,11 @@ where // Fetch the metadata, verifying max_length and hashes (if provided), as // the repository implementation should only be trusted to use those as // hints to fail early. - let mut reader = self - .repository - .fetch_metadata(meta_path, version) - .await? - .check_length_and_hash(max_length.unwrap_or(usize::MAX) as u64, hashes)?; + let mut reader = EnforceSizeAndHash::new( + self.repository.fetch_metadata(meta_path, version).await?, + max_length.unwrap_or(usize::MAX) as u64, + hashes, + )?; let mut buf = Vec::new(); reader.read_to_end(&mut buf).await?; @@ -336,7 +330,7 @@ where self.repository.fetch_target(target_path).await? }; - target.check_length_and_hash(length, hashes) + EnforceSizeAndHash::new(target, length, hashes) } } diff --git a/tuf/src/repository/http.rs b/tuf/src/repository/http.rs deleted file mode 100644 index 645bc86..0000000 --- a/tuf/src/repository/http.rs +++ /dev/null @@ -1,460 +0,0 @@ -//! Read-only Repository implementation backed by a web server. - -use futures_io::AsyncRead; -use futures_util::future::{BoxFuture, FutureExt as _, TryFutureExt as _}; -use futures_util::stream::TryStreamExt; -use http::{Response, StatusCode, Uri}; -use hyper::Client; -use hyper::Request; -use hyper::body::Body; -use hyper::client::connect::Connect; -use percent_encoding::utf8_percent_encode; -use std::future::Future; -use std::io; -use std::marker::PhantomData; -use url::Url; - -use crate::Result; -use crate::error::Error; -use crate::metadata::{MetadataPath, MetadataVersion, TargetPath}; -use crate::pouf::Pouf; -use crate::repository::RepositoryProvider; -use crate::util::SafeAsyncRead; - -/// A builder to create a repository accessible over HTTP. -pub struct HttpRepositoryBuilder -where - C: Connect + Sync + 'static, - D: Pouf, -{ - uri: Uri, - client: Client, - user_agent: Option, - metadata_prefix: Option>, - targets_prefix: Option>, - min_bytes_per_second: u32, - _pouf: PhantomData, -} - -impl HttpRepositoryBuilder -where - C: Connect + Sync + 'static, - D: Pouf, -{ - /// Create a new repository with the given `Url` and `Client`. - pub fn new(url: Url, client: Client) -> Self { - HttpRepositoryBuilder { - uri: url.to_string().parse::().unwrap(), // This is dangerous, but will only exist for a short time as we migrate APIs. - client, - user_agent: None, - metadata_prefix: None, - targets_prefix: None, - min_bytes_per_second: 4096, - _pouf: PhantomData, - } - } - - /// Create a new repository with the given `Url` and `Client`. - pub fn new_with_uri(uri: Uri, client: Client) -> Self { - HttpRepositoryBuilder { - uri, - client, - user_agent: None, - metadata_prefix: None, - targets_prefix: None, - min_bytes_per_second: 4096, - _pouf: PhantomData, - } - } - - /// Set the User-Agent prefix. - /// - /// Callers *should* include a custom User-Agent prefix to help maintainers of TUF repositories - /// keep track of which client versions exist in the field. - /// - pub fn user_agent>(mut self, user_agent: T) -> Self { - self.user_agent = Some(user_agent.into()); - self - } - - /// The argument `metadata_prefix` is used to provide an alternate path where metadata is - /// stored on the repository. If `None`, this defaults to `/`. For example, if there is a TUF - /// repository at `https://tuf.example.com/`, but all metadata is stored at `/meta/`, then - /// passing the arg `Some("meta".into())` would cause `root.json` to be fetched from - /// `https://tuf.example.com/meta/root.json`. - pub fn metadata_prefix(mut self, metadata_prefix: Vec) -> Self { - self.metadata_prefix = Some(metadata_prefix); - self - } - - /// The argument `targets_prefix` is used to provide an alternate path where targets is - /// stored on the repository. If `None`, this defaults to `/`. For example, if there is a TUF - /// repository at `https://tuf.example.com/`, but all targets are stored at `/targets/`, then - /// passing the arg `Some("targets".into())` would cause `hello-world` to be fetched from - /// `https://tuf.example.com/targets/hello-world`. - pub fn targets_prefix(mut self, targets_prefix: Vec) -> Self { - self.targets_prefix = Some(targets_prefix); - self - } - - /// Set the minimum bytes per second for a read to be considered good. - pub fn min_bytes_per_second(mut self, min: u32) -> Self { - self.min_bytes_per_second = min; - self - } - - /// Build a `HttpRepository`. - pub fn build(self) -> HttpRepository { - let user_agent = match self.user_agent { - Some(user_agent) => user_agent, - None => "rust-tuf".into(), - }; - - HttpRepository { - uri: self.uri, - client: self.client, - user_agent, - metadata_prefix: self.metadata_prefix, - targets_prefix: self.targets_prefix, - min_bytes_per_second: self.min_bytes_per_second, - _pouf: PhantomData, - } - } -} - -/// A repository accessible over HTTP. -#[derive(Debug)] -pub struct HttpRepository -where - C: Connect + Sync + 'static, - D: Pouf, -{ - uri: Uri, - client: Client, - user_agent: String, - metadata_prefix: Option>, - targets_prefix: Option>, - min_bytes_per_second: u32, - _pouf: PhantomData, -} - -// Configuration for urlencoding URI path elements. -// From https://url.spec.whatwg.org/#path-percent-encode-set -const URLENCODE_FRAGMENT: &percent_encoding::AsciiSet = &percent_encoding::CONTROLS - .add(b' ') - .add(b'"') - .add(b'<') - .add(b'>') - .add(b'`'); -const URLENCODE_PATH: &percent_encoding::AsciiSet = - &URLENCODE_FRAGMENT.add(b'#').add(b'?').add(b'{').add(b'}'); - -fn extend_uri(uri: &Uri, prefix: &Option>, components: &[String]) -> Result { - let uri = uri.clone(); - let mut uri_parts = uri.into_parts(); - - let (path, query) = match &uri_parts.path_and_query { - Some(path_and_query) => (path_and_query.path(), path_and_query.query()), - None => ("", None), - }; - - let mut modified_path = path.to_owned(); - if modified_path.ends_with('/') { - modified_path.pop(); - } - - let mut path_split = modified_path - .split('/') - .map(String::from) - .collect::>(); - let mut new_path_elements: Vec<&str> = vec![]; - - if let Some(prefix) = prefix { - new_path_elements.extend(prefix.iter().map(String::as_str)); - } - new_path_elements.extend(components.iter().map(String::as_str)); - - // Urlencode new items to match behavior of PathSegmentsMut.extend from - // https://docs.rs/url/2.1.0/url/struct.PathSegmentsMut.html - let encoded_new_path_elements = new_path_elements - .into_iter() - .map(|path_segment| utf8_percent_encode(path_segment, URLENCODE_PATH).collect()); - path_split.extend(encoded_new_path_elements); - let constructed_path = path_split.join("/"); - - uri_parts.path_and_query = - match query { - Some(query) => Some(format!("{}?{}", constructed_path, query).parse().map_err( - |_| { - Error::IllegalArgument(format!( - "Invalid path and query: {:?}, {:?}", - constructed_path, query - )) - }, - )?), - None => Some(constructed_path.parse().map_err(|_| { - Error::IllegalArgument(format!("Invalid URI path: {:?}", constructed_path)) - })?), - }; - - Uri::from_parts(uri_parts).map_err(|_| { - Error::IllegalArgument(format!( - "Invalid URI parts: {:?}, {:?}, {:?}", - constructed_path, prefix, components - )) - }) -} - -impl HttpRepository -where - C: Connect + Clone + Send + Sync + 'static, - D: Pouf, -{ - fn get<'a>( - &self, - uri: &'a Uri, - ) -> Result>> + 'a + use<'a, C, D>> { - let req = Request::builder() - .uri(uri) - .header("User-Agent", &*self.user_agent) - .body(Body::default()) - .map_err(|err| Error::Http { - uri: uri.to_string(), - err, - })?; - - Ok(self.client.request(req).map_err(|err| Error::Hyper { - uri: uri.to_string(), - err, - })) - } -} - -impl RepositoryProvider for HttpRepository -where - C: Connect + Clone + Send + Sync + 'static, - D: Pouf, -{ - fn fetch_metadata<'a>( - &'a self, - meta_path: &MetadataPath, - version: MetadataVersion, - ) -> BoxFuture<'a, Result>> { - let meta_path = meta_path.clone(); - let components = meta_path.components::(version); - let uri = extend_uri(&self.uri, &self.metadata_prefix, &components); - - async move { - // TODO(#278) check content length if known and fail early if the payload is too large. - - let uri = uri?; - let resp = self.get(&uri)?.await?; - - let status = resp.status(); - if status == StatusCode::OK { - let reader = resp - .into_body() - .map_err(io::Error::other) - .into_async_read() - .enforce_minimum_bitrate(self.min_bytes_per_second); - - let reader: Box = Box::new(reader); - Ok(reader) - } else if status == StatusCode::NOT_FOUND { - Err(Error::MetadataNotFound { - path: meta_path, - version, - }) - } else { - Err(Error::BadHttpStatus { - uri: uri.to_string(), - code: status, - }) - } - } - .boxed() - } - - fn fetch_target<'a>( - &'a self, - target_path: &TargetPath, - ) -> BoxFuture<'a, Result>> { - let target_path = target_path.clone(); - let components = target_path.components(); - let uri = extend_uri(&self.uri, &self.targets_prefix, &components); - - async move { - // TODO(#278) check content length if known and fail early if the payload is too large. - - let uri = uri?; - let resp = self.get(&uri)?.await?; - - let status = resp.status(); - if status == StatusCode::OK { - let reader = resp - .into_body() - .map_err(io::Error::other) - .into_async_read() - .enforce_minimum_bitrate(self.min_bytes_per_second); - - let reader: Box = Box::new(reader); - Ok(reader) - } else if status == StatusCode::NOT_FOUND { - Err(Error::TargetNotFound(target_path)) - } else { - Err(Error::BadHttpStatus { - uri: uri.to_string(), - code: status, - }) - } - } - .boxed() - } -} - -#[cfg(test)] -mod test { - use super::*; - - // Old behavior of the `HttpRepository::get` extension - // functionality - fn http_repository_extend_using_url( - base_url: Url, - prefix: &Option>, - components: &[String], - ) -> url::Url { - let mut url = base_url; - { - let mut segments = url.path_segments_mut().unwrap(); - if let Some(prefix) = prefix { - segments.extend(prefix); - } - segments.extend(components); - } - url - } - - #[test] - fn http_repository_uri_construction() { - let base_uri = "http://example.com/one"; - - let prefix = Some(vec![String::from("prefix")]); - let components = [ - String::from("components_one"), - String::from("components_two"), - ]; - - let uri = base_uri.parse::().unwrap(); - let extended_uri = extend_uri(&uri, &prefix, &components).unwrap(); - - let url = - http_repository_extend_using_url(Url::parse(base_uri).unwrap(), &prefix, &components); - - assert_eq!(url.to_string(), extended_uri.to_string()); - assert_eq!( - extended_uri.to_string(), - "http://example.com/one/prefix/components_one/components_two" - ); - } - - #[test] - fn http_repository_uri_construction_encoded() { - let base_uri = "http://example.com/one"; - - let prefix = Some(vec![String::from("prefix")]); - let components = [String::from("chars to encode#?")]; - let uri = base_uri.parse::().unwrap(); - let extended_uri = extend_uri(&uri, &prefix, &components) - .expect("correctly generated a URI with a zone id"); - - let url = - http_repository_extend_using_url(Url::parse(base_uri).unwrap(), &prefix, &components); - - assert_eq!(url.to_string(), extended_uri.to_string()); - assert_eq!( - extended_uri.to_string(), - "http://example.com/one/prefix/chars%20to%20encode%23%3F" - ); - } - - #[test] - fn http_repository_uri_construction_no_components() { - let base_uri = "http://example.com/one"; - - let prefix = Some(vec![String::from("prefix")]); - let components = []; - - let uri = base_uri.parse::().unwrap(); - let extended_uri = extend_uri(&uri, &prefix, &components).unwrap(); - - let url = - http_repository_extend_using_url(Url::parse(base_uri).unwrap(), &prefix, &components); - - assert_eq!(url.to_string(), extended_uri.to_string()); - assert_eq!(extended_uri.to_string(), "http://example.com/one/prefix"); - } - - #[test] - fn http_repository_uri_construction_no_prefix() { - let base_uri = "http://example.com/one"; - - let prefix = None; - let components = [ - String::from("components_one"), - String::from("components_two"), - ]; - - let uri = base_uri.parse::().unwrap(); - let extended_uri = extend_uri(&uri, &prefix, &components).unwrap(); - - let url = - http_repository_extend_using_url(Url::parse(base_uri).unwrap(), &prefix, &components); - - assert_eq!(url.to_string(), extended_uri.to_string()); - assert_eq!( - extended_uri.to_string(), - "http://example.com/one/components_one/components_two" - ); - } - - #[test] - fn http_repository_uri_construction_with_query() { - let base_uri = "http://example.com/one?test=1"; - - let prefix = None; - let components = [ - String::from("components_one"), - String::from("components_two"), - ]; - - let uri = base_uri.parse::().unwrap(); - let extended_uri = extend_uri(&uri, &prefix, &components).unwrap(); - - let url = - http_repository_extend_using_url(Url::parse(base_uri).unwrap(), &prefix, &components); - - assert_eq!(url.to_string(), extended_uri.to_string()); - assert_eq!( - extended_uri.to_string(), - "http://example.com/one/components_one/components_two?test=1" - ); - } - - #[test] - fn http_repository_uri_construction_ipv6_zoneid() { - let base_uri = "http://[aaaa::aaaa:aaaa:aaaa:1234%252]:80"; - - let prefix = Some(vec![String::from("prefix")]); - let components = [ - String::from("componenents_one"), - String::from("components_two"), - ]; - let uri = base_uri.parse::().unwrap(); - let extended_uri = extend_uri(&uri, &prefix, &components) - .expect("correctly generated a URI with a zone id"); - assert_eq!( - extended_uri.to_string(), - "http://[aaaa::aaaa:aaaa:aaaa:1234%252]:80/prefix/componenents_one/components_two" - ); - } -}