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" - ); - } -}