From fb1b9dbce744240d8d386a87373ebbb8d1c830bc Mon Sep 17 00:00:00 2001 From: Kyle Brooks Date: Mon, 14 Sep 2026 07:54:04 -0400 Subject: [PATCH 1/6] refactor: extract the storage backends into a module archive.rs carried the Storage enum and both backends alongside the tar and index parsing. This moves them to src/storage.rs, leaving archive.rs responsible for the archive format only. The open_* functions return an ArchiveSource struct rather than a 4-tuple so that adding a backend does not churn every call site. Also drops the local backend's incidental dependency on aws_sdk_s3, which it used only to format an HTTP-date; httpdate is a zero-dependency IMF-fixdate formatter and lets the backends be built independently. No behaviour change. --- Cargo.lock | 1 + Cargo.toml | 1 + src/archive.rs | 204 +++----------------------------------------- src/main.rs | 1 + src/storage.rs | 225 +++++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 238 insertions(+), 194 deletions(-) create mode 100644 src/storage.rs diff --git a/Cargo.lock b/Cargo.lock index 58ac32a..1504fe5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1578,6 +1578,7 @@ dependencies = [ "flate2", "futures-util", "http-body 1.1.0", + "httpdate", "rustc-hash", "serde", "thiserror", diff --git a/Cargo.toml b/Cargo.toml index 3f7c687..73948a2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,6 +24,7 @@ clap = { version = "4.6", features = ["derive"] } flate2 = "1" futures-util = { version = "0.3", default-features = false, features = ["alloc"] } http-body = "1" +httpdate = "1.0.3" rustc-hash = "2" serde = { version = "1", features = ["derive"] } thiserror = "2" diff --git a/src/archive.rs b/src/archive.rs index b5eff83..a850bb8 100644 --- a/src/archive.rs +++ b/src/archive.rs @@ -1,5 +1,5 @@ -//! Tar archive backed by S3 or the local filesystem: index parsing, tile lookups, -//! and on-demand byte-range reads. +//! Tar archive index parsing and tile lookups, reading bytes through a [`Storage`] backend. +//! //! //! Loads the tar index via a handful of small range reads — no full download: //! @@ -14,11 +14,11 @@ //! 5. A 272-byte read (or the full first tile when it's compressed) to extract the //! dataset id from the `GraphTileHeader` ([`detect_dataset_id`]). -use std::sync::Arc; - use bytes::Bytes; use rustc_hash::FxHashMap; +use crate::storage::{ArchiveSource, Storage}; + /// Size of a POSIX tar header block. const TAR_BLOCK_SIZE: usize = 512; /// Size of a single index entry in bytes: u64 + u32 + u32. @@ -289,12 +289,12 @@ impl Archive { scan_index: bool, dataset_id_override: Option<&str>, ) -> Result<(Self, ArchiveMeta), Error> { - let (storage, etag, last_modified, archive_size) = - if let Some((bucket, key)) = parse_s3_url(source) { - open_s3(bucket, key).await? - } else { - open_local(source)? - }; + let (storage, src) = Storage::open(source).await?; + let ArchiveSource { + etag, + last_modified, + size: archive_size, + } = src; // Step 1: Read the first 512-byte tar header let header_bytes = storage.read_range(0, TAR_BLOCK_SIZE as u64).await?; @@ -395,167 +395,6 @@ impl Archive { } } -/// Storage backend for a tar archive. Hides whether bytes come from S3 or a local file — -/// every read goes through [`Storage::read_range`]; everything above this layer is unaware -/// of the source. -enum Storage { - S3 { - client: aws_sdk_s3::Client, - bucket: Box, - key: Box, - }, - Local { - // `Arc` lets us call `read_at` (which takes `&self`) concurrently - // from multiple `spawn_blocking` tasks without `try_clone()` syscalls per read. - file: Arc, - }, -} - -impl Storage { - /// Read `length` bytes at `offset`. The only place where the two backends diverge. - async fn read_range(&self, offset: u64, length: u64) -> Result { - if length == 0 { - return Ok(Bytes::new()); - } - - match self { - Self::S3 { - client, - bucket, - key, - } => read_s3_range(client, bucket, key, offset, length).await, - Self::Local { file } => read_local_range(file.clone(), offset, length).await, - } - } -} - -/// Open the archive from S3: HeadObject for ETag/Last-Modified/size, then hand back the -/// pieces `Archive::open` needs to read the rest. -async fn open_s3(bucket: &str, key: &str) -> Result<(Storage, Box, Box, u64), Error> { - let client = aws_sdk_s3::Client::new( - &aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await, - ); - - let head = client - .head_object() - .bucket(bucket) - .key(key) - .send() - .await - .map_err(|e| Error::Io(format!("HeadObject failed: {e}")))?; - - let etag: Box = head - .e_tag() - .ok_or_else(|| Error::Protocol("S3 HeadObject returned no ETag".into()))? - .into(); - - let last_modified: Box = head - .last_modified() - .and_then(|dt| { - dt.fmt(aws_sdk_s3::primitives::DateTimeFormat::HttpDate) - .ok() - }) - .ok_or_else(|| Error::Protocol("S3 HeadObject returned no Last-Modified".into()))? - .into(); - - let archive_size = head - .content_length() - .ok_or_else(|| Error::Protocol("S3 HeadObject returned no Content-Length".into()))? - as u64; - - Ok(( - Storage::S3 { - client, - bucket: bucket.into(), - key: key.into(), - }, - etag, - last_modified, - archive_size, - )) -} - -async fn read_s3_range( - client: &aws_sdk_s3::Client, - bucket: &str, - key: &str, - offset: u64, - length: u64, -) -> Result { - let range = format!("bytes={}-{}", offset, offset + length - 1); - let resp = client - .get_object() - .bucket(bucket) - .key(key) - .range(&range) - .send() - .await - .map_err(|e| Error::Io(format!("S3 GetObject failed: {e}")))?; - let data = resp - .body - .collect() - .await - .map_err(|e| Error::Io(format!("reading S3 response body: {e}")))? - .into_bytes(); - Ok(data) -} - -/// Open the archive from the local filesystem. ETag is synthesized from `mtime+size` -/// (matches a fresh value whenever the archive changes); Last-Modified is the file's -/// mtime formatted as an HTTP-date. -fn open_local(path: &str) -> Result<(Storage, Box, Box, u64), Error> { - let metadata = - std::fs::metadata(path).map_err(|e| Error::Io(format!("stat({path}) failed: {e}")))?; - if !metadata.is_file() { - return Err(Error::Protocol(format!("{path} is not a regular file"))); - } - let archive_size = metadata.len(); - let mtime = metadata - .modified() - .map_err(|e| Error::Io(format!("{path} has no mtime: {e}")))?; - let mtime_unix = mtime - .duration_since(std::time::UNIX_EPOCH) - .map_err(|_| Error::Protocol("file mtime is before UNIX epoch".into()))? - .as_secs(); - let etag: Box = format!("\"{mtime_unix}-{archive_size}\"").into(); - let last_modified: Box = aws_sdk_s3::primitives::DateTime::from(mtime) - .fmt(aws_sdk_s3::primitives::DateTimeFormat::HttpDate) - .map_err(|e| Error::Protocol(format!("format mtime as HttpDate: {e}")))? - .into(); - let file = - std::fs::File::open(path).map_err(|e| Error::Io(format!("open({path}) failed: {e}")))?; - - Ok(( - Storage::Local { - file: Arc::new(file), - }, - etag, - last_modified, - archive_size, - )) -} - -async fn read_local_range( - file: Arc, - offset: u64, - length: u64, -) -> Result { - let len = length as usize; - tokio::task::spawn_blocking(move || { - use std::os::unix::fs::FileExt; - let mut buf = vec![0u8; len]; - file.read_exact_at(&mut buf, offset) - .map(|()| Bytes::from(buf)) - }) - .await - .map_err(|e| Error::Io(format!("local read task panicked: {e}")))? - .map_err(|e| { - Error::Io(format!( - "local read_at(offset={offset}, len={length}) failed: {e}" - )) - }) -} - /// Decompress `data` according to `compression`. Returns the input unchanged for `TileCompression::None`. pub fn decompress(data: Bytes, compression: TileCompression) -> std::io::Result { use std::io::Read; @@ -781,11 +620,6 @@ fn parse_dataset_id(header: &[u8]) -> Result { Ok(u64::from_le_bytes(bytes)) } -fn parse_s3_url(url: &str) -> Option<(&str, &str)> { - let path = url.strip_prefix("s3://")?; - path.split_once('/') -} - #[derive(Debug, thiserror::Error)] pub enum TarError { #[error("index.bin size {size} is not a multiple of entry size {entry_size}")] @@ -1042,22 +876,4 @@ mod tests { let result = parse_dataset_id(&header).unwrap(); assert_eq!(result, 42); } - - #[test] - fn parse_s3_url_test() { - assert_eq!( - parse_s3_url("s3://my-bucket/path/to/file.tar"), - Some(("my-bucket", "path/to/file.tar")) - ); - assert_eq!( - parse_s3_url("s3://bucket/file.tar"), - Some(("bucket", "file.tar")) - ); - - assert_eq!(parse_s3_url("bucket/key"), None); - assert_eq!(parse_s3_url("https://wrong/scheme"), None); - assert_eq!(parse_s3_url("s3:/bad-url/format"), None); - assert_eq!(parse_s3_url("s3://bucket-only"), None); - assert_eq!(parse_s3_url("s3://file-only.tar"), None); - } } diff --git a/src/main.rs b/src/main.rs index da7ee6a..2b4bc00 100644 --- a/src/main.rs +++ b/src/main.rs @@ -18,6 +18,7 @@ use tokio::signal; use tracing::info; mod archive; +mod storage; use archive::TileCompression; diff --git a/src/storage.rs b/src/storage.rs new file mode 100644 index 0000000..7b5ac94 --- /dev/null +++ b/src/storage.rs @@ -0,0 +1,225 @@ +//! Storage backends for a tar archive. Hides whether bytes come from S3 or a local +//! file — every read goes through [`Storage::read_range`]; everything above this layer +//! is unaware of the source. + +use std::sync::Arc; + +use bytes::Bytes; + +use crate::archive::Error; + +/// Source metadata read once when the archive is opened. +pub struct ArchiveSource { + /// Source ETag — S3 object ETag, or synthesized `"-"` for local archives. + pub etag: Box, + /// Source Last-Modified as an HTTP-date string. + pub last_modified: Box, + /// Total size of the archive in bytes. + pub size: u64, +} + +pub enum Storage { + S3 { + client: aws_sdk_s3::Client, + bucket: Box, + key: Box, + }, + Local { + // `Arc` lets us call `read_at` (which takes `&self`) concurrently + // from multiple `spawn_blocking` tasks without `try_clone()` syscalls per read. + file: Arc, + }, +} + +impl Storage { + /// Open `source`: an S3 URL (`s3://bucket/key`) or a local filesystem path. + pub async fn open(source: &str) -> Result<(Self, ArchiveSource), Error> { + if let Some((bucket, key)) = parse_s3_url(source) { + return open_s3(bucket, key).await; + } + + open_local(source) + } + + /// Read `length` bytes at `offset`. The only place where the backends diverge. + pub async fn read_range(&self, offset: u64, length: u64) -> Result { + if length == 0 { + return Ok(Bytes::new()); + } + + match self { + Self::S3 { + client, + bucket, + key, + } => read_s3_range(client, bucket, key, offset, length).await, + Self::Local { file } => read_local_range(file.clone(), offset, length).await, + } + } +} + +/// Open the archive from the local filesystem. ETag is synthesized from `mtime+size` +/// (matches a fresh value whenever the archive changes); Last-Modified is the file's +/// mtime formatted as an HTTP-date. +fn open_local(path: &str) -> Result<(Storage, ArchiveSource), Error> { + let metadata = + std::fs::metadata(path).map_err(|e| Error::Io(format!("stat({path}) failed: {e}")))?; + if !metadata.is_file() { + return Err(Error::Protocol(format!("{path} is not a regular file"))); + } + let size = metadata.len(); + let mtime = metadata + .modified() + .map_err(|e| Error::Io(format!("{path} has no mtime: {e}")))?; + let mtime_unix = mtime + .duration_since(std::time::UNIX_EPOCH) + .map_err(|_| Error::Protocol("file mtime is before UNIX epoch".into()))? + .as_secs(); + let etag: Box = format!("\"{mtime_unix}-{size}\"").into(); + let last_modified: Box = httpdate::fmt_http_date(mtime).into(); + let file = + std::fs::File::open(path).map_err(|e| Error::Io(format!("open({path}) failed: {e}")))?; + + Ok(( + Storage::Local { + file: Arc::new(file), + }, + ArchiveSource { + etag, + last_modified, + size, + }, + )) +} + +async fn read_local_range( + file: Arc, + offset: u64, + length: u64, +) -> Result { + let len = length as usize; + tokio::task::spawn_blocking(move || { + use std::os::unix::fs::FileExt; + let mut buf = vec![0u8; len]; + file.read_exact_at(&mut buf, offset) + .map(|()| Bytes::from(buf)) + }) + .await + .map_err(|e| Error::Io(format!("local read task panicked: {e}")))? + .map_err(|e| { + Error::Io(format!( + "local read_at(offset={offset}, len={length}) failed: {e}" + )) + }) +} + +/// S3-backed tar archive: `HeadObject` for metadata, ranged `GetObject` for reads. +/// Split an `s3://bucket/key` URL into its bucket and key. +fn parse_s3_url(url: &str) -> Option<(&str, &str)> { + let path = url.strip_prefix("s3://")?; + path.split_once('/') +} + +/// Open the archive from S3: HeadObject for ETag/Last-Modified/size, then hand back the +/// pieces `Archive::open` needs to read the rest. +async fn open_s3(bucket: &str, key: &str) -> Result<(Storage, ArchiveSource), Error> { + let client = aws_sdk_s3::Client::new( + &aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await, + ); + + let head = client + .head_object() + .bucket(bucket) + .key(key) + .send() + .await + .map_err(|e| Error::Io(format!("HeadObject failed: {e}")))?; + + let etag: Box = head + .e_tag() + .ok_or_else(|| Error::Protocol("S3 HeadObject returned no ETag".into()))? + .into(); + + let last_modified: Box = head + .last_modified() + .and_then(|dt| { + dt.fmt(aws_sdk_s3::primitives::DateTimeFormat::HttpDate) + .ok() + }) + .ok_or_else(|| Error::Protocol("S3 HeadObject returned no Last-Modified".into()))? + .into(); + + let size = head + .content_length() + .ok_or_else(|| Error::Protocol("S3 HeadObject returned no Content-Length".into()))? + as u64; + + Ok(( + Storage::S3 { + client, + bucket: bucket.into(), + key: key.into(), + }, + ArchiveSource { + etag, + last_modified, + size, + }, + )) +} + +/// Read `length` bytes at `offset` from the object. +/// +/// Every read is conditional on the ETag captured when the archive was opened. The tile +/// index maps ids to byte offsets in one specific version of the object, so if the archive +/// is replaced while rati is running those offsets now point into different data. Without +/// the guard S3 happily returns whatever occupies that range, and rati serves it under the +/// old ETag and `Cache-Control: immutable` — silently poisoning client and CDN caches. +/// With it, S3 answers 412 and the read fails visibly. +async fn read_s3_range( + client: &aws_sdk_s3::Client, + bucket: &str, + key: &str, + offset: u64, + length: u64, +) -> Result { + let range = format!("bytes={}-{}", offset, offset + length - 1); + let resp = client + .get_object() + .bucket(bucket) + .key(key) + .range(&range) + .send() + .await + .map_err(|e| Error::Io(format!("S3 GetObject failed: {e}")))?; + let data = resp + .body + .collect() + .await + .map_err(|e| Error::Io(format!("reading S3 response body: {e}")))? + .into_bytes(); + Ok(data) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_s3_url_test() { + assert_eq!( + parse_s3_url("s3://my-bucket/path/to/file.tar"), + Some(("my-bucket", "path/to/file.tar")) + ); + assert_eq!( + parse_s3_url("s3://bucket/file.tar"), + Some(("bucket", "file.tar")) + ); + + assert_eq!(parse_s3_url("bucket/key"), None); + assert_eq!(parse_s3_url("https://wrong/scheme"), None); + assert_eq!(parse_s3_url("s3:/bad-url/format"), None); + assert_eq!(parse_s3_url("s3://bucket-only"), None); + assert_eq!(parse_s3_url("s3://file-only.tar"), None); + } +} From 2a281079e08b64153121c46937a2a8a6dd70e292 Mon Sep 17 00:00:00 2001 From: Kyle Brooks Date: Mon, 14 Sep 2026 07:54:04 -0400 Subject: [PATCH 2/6] fix: reject a short read from a ranged GET read_range hands its result to callers that index into it assuming it is exactly `length` bytes. scan_tar_headers computes `chunk.len() - local`, which underflows if the backend returns fewer bytes than the range asked for, and the index path reads fixed-size records straight out of the buffer. A ranged GET is not obliged to return the whole range, so treat a short read as an error naming the offset and both lengths. This changes the S3 backend from tolerating a short GetObject to failing hard, which is the point: a truncated buffer is worse than a visible error. Enforced once in Storage::read_range rather than per backend, so the invariant holds for any backend added later; the local backend already gets it from read_exact_at. --- src/storage.rs | 37 ++++++++++++++++++++++++++++++++++++- 1 file changed, 36 insertions(+), 1 deletion(-) diff --git a/src/storage.rs b/src/storage.rs index 7b5ac94..dfde718 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -47,14 +47,25 @@ impl Storage { return Ok(Bytes::new()); } - match self { + let data = match self { Self::S3 { client, bucket, key, } => read_s3_range(client, bucket, key, offset, length).await, Self::Local { file } => read_local_range(file.clone(), offset, length).await, + }?; + + // Callers index into the result assuming it is exactly `length` bytes; + // `scan_tar_headers` subtracts from `chunk.len()` and would underflow on a short + // read. A ranged GET may return less, so enforce it here once for every backend. + if data.len() as u64 != length { + return Err(Error::Io(format!( + "short read at offset={offset}: asked for {length} bytes, got {}", + data.len() + ))); } + Ok(data) } } @@ -205,6 +216,30 @@ async fn read_s3_range( mod tests { use super::*; + /// The exact-length contract every caller relies on: `scan_tar_headers` computes + /// `chunk.len() - local`, which underflows on a short read. + #[tokio::test] + async fn a_short_read_is_rejected() { + let dir = std::env::temp_dir().join(format!("rati-short-read-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("tiny.tar"); + std::fs::write(&path, b"0123456789").unwrap(); + + let (storage, _) = Storage::open(path.to_str().unwrap()).await.unwrap(); + + // Fully satisfiable reads are returned verbatim. + assert_eq!(storage.read_range(0, 4).await.unwrap().as_ref(), b"0123"); + // A zero-length read never reaches a backend. + assert!(storage.read_range(4, 0).await.unwrap().is_empty()); + + // Asking past the end must fail loudly rather than hand back a short buffer. + let err = storage.read_range(6, 8).await.unwrap_err(); + let msg = err.to_string(); + assert!(msg.contains("offset=6"), "unexpected error: {msg}"); + + std::fs::remove_dir_all(&dir).ok(); + } + #[test] fn parse_s3_url_test() { assert_eq!( From d49221ae2411d489ee79af80486335aeac0d51c5 Mon Sep 17 00:00:00 2001 From: Kyle Brooks Date: Mon, 14 Sep 2026 08:10:56 -0400 Subject: [PATCH 3/6] fix: pin S3 ranged reads to the object ETag The tile index maps ids to byte offsets inside one specific version of the object. Replace the archive under a running server and those offsets point into different data: GetObject returns whatever now occupies that range, and rati serves it under the *old* ETag with Cache-Control: immutable, so every client and CDN in front caches wrong bytes as if they were permanent. Nothing fails, which is what makes it dangerous. Reproduced against MinIO before the fix: replacing the object under a running server kept the tile endpoint at 200 while its sha256 changed, still labelled with the previous ETag and marked immutable. Carrying the ETag captured at open in the Storage::S3 variant and passing it as if_match turns that into a visible 500. --- src/storage.rs | 69 ++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 61 insertions(+), 8 deletions(-) diff --git a/src/storage.rs b/src/storage.rs index dfde718..d19bcbe 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -23,6 +23,8 @@ pub enum Storage { client: aws_sdk_s3::Client, bucket: Box, key: Box, + /// ETag the index was built against; every read is pinned to it. + etag: Box, }, Local { // `Arc` lets us call `read_at` (which takes `&self`) concurrently @@ -52,7 +54,8 @@ impl Storage { client, bucket, key, - } => read_s3_range(client, bucket, key, offset, length).await, + etag, + } => read_s3_range(client, bucket, key, etag, offset, length).await, Self::Local { file } => read_local_range(file.clone(), offset, length).await, }?; @@ -170,6 +173,7 @@ async fn open_s3(bucket: &str, key: &str) -> Result<(Storage, ArchiveSource), Er client, bucket: bucket.into(), key: key.into(), + etag: etag.clone(), }, ArchiveSource { etag, @@ -179,6 +183,23 @@ async fn open_s3(bucket: &str, key: &str) -> Result<(Storage, ArchiveSource), Er )) } +/// The `GetObject` request for one ranged read, pinned to `etag`. +fn ranged_read( + client: &aws_sdk_s3::Client, + bucket: &str, + key: &str, + etag: &str, + offset: u64, + length: u64, +) -> aws_sdk_s3::operation::get_object::builders::GetObjectFluentBuilder { + client + .get_object() + .bucket(bucket) + .key(key) + .range(format!("bytes={}-{}", offset, offset + length - 1)) + .if_match(etag) +} + /// Read `length` bytes at `offset` from the object. /// /// Every read is conditional on the ETag captured when the archive was opened. The tile @@ -191,18 +212,19 @@ async fn read_s3_range( client: &aws_sdk_s3::Client, bucket: &str, key: &str, + etag: &str, offset: u64, length: u64, ) -> Result { - let range = format!("bytes={}-{}", offset, offset + length - 1); - let resp = client - .get_object() - .bucket(bucket) - .key(key) - .range(&range) + let resp = ranged_read(client, bucket, key, etag, offset, length) .send() .await - .map_err(|e| Error::Io(format!("S3 GetObject failed: {e}")))?; + .map_err(|e| { + Error::Io(format!( + "S3 GetObject failed: {e} (a 412 here means the archive was replaced \ + while rati was running; restart it)" + )) + })?; let data = resp .body .collect() @@ -240,6 +262,37 @@ mod tests { std::fs::remove_dir_all(&dir).ok(); } + /// Every ranged read must be pinned to the ETag captured at open. Without it an + /// object replaced mid-flight is served at stale offsets under the old ETag and + /// `Cache-Control: immutable`, poisoning client and CDN caches. + #[test] + fn every_ranged_read_is_pinned_to_the_etag() { + let client = aws_sdk_s3::Client::from_conf( + aws_sdk_s3::Config::builder() + .behavior_version_latest() + .region(aws_sdk_s3::config::Region::new("us-east-1")) + .credentials_provider(aws_sdk_s3::config::Credentials::new( + "id", "secret", None, None, "test", + )) + .build(), + ); + + let req = ranged_read( + &client, + "bucket", + "planet.tar", + "\"abc123\"", + 1536, + 44_504_000, + ); + assert_eq!( + req.get_if_match().as_deref(), + Some("\"abc123\""), + "ranged reads must carry If-Match" + ); + assert_eq!(req.get_range().as_deref(), Some("bytes=1536-44505535")); + } + #[test] fn parse_s3_url_test() { assert_eq!( From 6197533f7d15939ea8ed91ce1c7af75047531b97 Mon Sep 17 00:00:00 2001 From: Kyle Brooks Date: Mon, 14 Sep 2026 08:11:07 -0400 Subject: [PATCH 4/6] feat: make the s3 backend an optional cargo feature A deployment that only reads local archives no longer pulls in the AWS SDK and its TLS stack. CI becomes a matrix over the feature combinations so both paths stay honest, and lints test code too via --all-targets. --- .github/workflows/sanity_checks.yml | 14 ++++++++++++-- Cargo.toml | 8 ++++++-- src/archive.rs | 4 +--- src/storage.rs | 16 ++++++++++++++++ 4 files changed, 35 insertions(+), 7 deletions(-) diff --git a/.github/workflows/sanity_checks.yml b/.github/workflows/sanity_checks.yml index 4965e7d..f63c200 100644 --- a/.github/workflows/sanity_checks.yml +++ b/.github/workflows/sanity_checks.yml @@ -6,6 +6,16 @@ jobs: build: runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + # Every combination that ships: the backend on, and off. + features: + - "" + - "--no-default-features" + + name: ${{ matrix.features || 'default features' }} + steps: - name: Checkout code uses: actions/checkout@v4 @@ -20,7 +30,7 @@ jobs: run: cargo fmt --all -- --check - name: Run Clippy - run: cargo clippy -- -Dwarnings + run: cargo clippy --all-targets ${{ matrix.features }} -- -Dwarnings - name: Run tests - run: cargo test + run: cargo test ${{ matrix.features }} diff --git a/Cargo.toml b/Cargo.toml index 73948a2..a18e290 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,6 +9,10 @@ repository = "https://github.com/valhalla/rati" license = "MIT OR Apache-2.0" include = ["**/*.rs"] +[features] +default = ["s3"] +s3 = ["dep:aws-config", "dep:aws-sdk-s3"] + # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [profile.release] @@ -16,8 +20,8 @@ strip = true lto = "thin" [dependencies] -aws-config = { version = "1", default-features = false, features = ["rt-tokio", "default-https-client", "sso"] } -aws-sdk-s3 = { version = "1", default-features = false, features = ["rt-tokio", "default-https-client"] } +aws-config = { version = "1", default-features = false, features = ["rt-tokio", "default-https-client", "sso"], optional = true } +aws-sdk-s3 = { version = "1", default-features = false, features = ["rt-tokio", "default-https-client"], optional = true } axum = "0.8" bytes = "1" clap = { version = "4.6", features = ["derive"] } diff --git a/src/archive.rs b/src/archive.rs index a850bb8..ac7f276 100644 --- a/src/archive.rs +++ b/src/archive.rs @@ -1,10 +1,8 @@ //! Tar archive index parsing and tile lookups, reading bytes through a [`Storage`] backend. //! -//! //! Loads the tar index via a handful of small range reads — no full download: //! -//! 1. Source-specific metadata: S3 `HeadObject` (ETag, Last-Modified, size) or -//! filesystem `metadata()` (mtime, size; ETag synthesized from mtime+size). +//! 1. Source metadata from the backend — see [`crate::storage`]. //! 2. First 512 bytes — the leading tar header, expected to name `index.bin`. If it //! doesn't and `--scan-index` is set, fall back to [`scan_tar_headers`] which walks //! the whole archive in 8 MB chunks with prefetch. diff --git a/src/storage.rs b/src/storage.rs index d19bcbe..bbcb08a 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -19,6 +19,7 @@ pub struct ArchiveSource { } pub enum Storage { + #[cfg(feature = "s3")] S3 { client: aws_sdk_s3::Client, bucket: Box, @@ -36,10 +37,18 @@ pub enum Storage { impl Storage { /// Open `source`: an S3 URL (`s3://bucket/key`) or a local filesystem path. pub async fn open(source: &str) -> Result<(Self, ArchiveSource), Error> { + #[cfg(feature = "s3")] if let Some((bucket, key)) = parse_s3_url(source) { return open_s3(bucket, key).await; } + #[cfg(not(feature = "s3"))] + if source.starts_with("s3://") { + return Err(Error::Protocol( + "S3 archives require the 's3' cargo feature".into(), + )); + } + open_local(source) } @@ -50,6 +59,7 @@ impl Storage { } let data = match self { + #[cfg(feature = "s3")] Self::S3 { client, bucket, @@ -128,12 +138,14 @@ async fn read_local_range( } /// S3-backed tar archive: `HeadObject` for metadata, ranged `GetObject` for reads. +#[cfg(feature = "s3")] /// Split an `s3://bucket/key` URL into its bucket and key. fn parse_s3_url(url: &str) -> Option<(&str, &str)> { let path = url.strip_prefix("s3://")?; path.split_once('/') } +#[cfg(feature = "s3")] /// Open the archive from S3: HeadObject for ETag/Last-Modified/size, then hand back the /// pieces `Archive::open` needs to read the rest. async fn open_s3(bucket: &str, key: &str) -> Result<(Storage, ArchiveSource), Error> { @@ -183,6 +195,7 @@ async fn open_s3(bucket: &str, key: &str) -> Result<(Storage, ArchiveSource), Er )) } +#[cfg(feature = "s3")] /// The `GetObject` request for one ranged read, pinned to `etag`. fn ranged_read( client: &aws_sdk_s3::Client, @@ -200,6 +213,7 @@ fn ranged_read( .if_match(etag) } +#[cfg(feature = "s3")] /// Read `length` bytes at `offset` from the object. /// /// Every read is conditional on the ETag captured when the archive was opened. The tile @@ -265,6 +279,7 @@ mod tests { /// Every ranged read must be pinned to the ETag captured at open. Without it an /// object replaced mid-flight is served at stale offsets under the old ETag and /// `Cache-Control: immutable`, poisoning client and CDN caches. + #[cfg(feature = "s3")] #[test] fn every_ranged_read_is_pinned_to_the_etag() { let client = aws_sdk_s3::Client::from_conf( @@ -293,6 +308,7 @@ mod tests { assert_eq!(req.get_range().as_deref(), Some("bytes=1536-44505535")); } + #[cfg(feature = "s3")] #[test] fn parse_s3_url_test() { assert_eq!( From f27569adc7e4d047d5dac054623b7abced0fa5f7 Mon Sep 17 00:00:00 2001 From: Kyle Brooks Date: Mon, 14 Sep 2026 08:11:50 -0400 Subject: [PATCH 5/6] feat: add an Azure Blob Storage backend Serves tiles from a tar in Azure Blob via ranged reads. The SDK's HttpRange::new(offset, length) maps 1:1 onto Storage::read_range, so nothing above the storage layer changes. rati has no generic HTTP backend and should not grow one, so routing needs no host inspection: any http(s) archive is Azure by elimination. Whether a host is trusted enough to be sent a credential is a separate question, and the only one that is load-bearing -- a bearer token goes out only to an https host carrying a known blob suffix, so an archive behind a CDN or a custom domain is read anonymously or with a SAS. That also means Azurite needs no special case at all. There are no Azure command-line flags. azure_identity 1.0 ships no DefaultAzureCredential, but that is a reason to pick sensible defaults rather than to put Azure's identity model in a tile server's --help. The credential comes from the same environment variables every other Azure SDK reads, in the order DefaultAzureCredential uses elsewhere: a projected token file means workload identity, a tenant/client/secret triple means a service principal, and otherwise managed identity -- user-assigned when AZURE_CLIENT_ID names one, system-assigned when it does not. That ordering is also what disambiguates AZURE_CLIENT_ID, which all three consult. An empty value is not a value: an empty id reaches IMDS as `client-id=&`, which can silently resolve to the system-assigned identity. A service principal is kept deliberately. A managed identity only exists on Azure compute, and workload identity federation covers a specific set of issuers, so without it a private archive would be readable only from Azure -- and it is also the local-development path against real blob storage. Metadata comes from a 512-byte ranged download rather than get_properties: BlobClientGetPropertiesResult is a unit struct that surfaces its values as raw headers, whereas download returns a typed BlobDownloadProperties. That probe is the first tar block, so it is handed back in ArchiveSource::prefetch and Archive::open reads it instead of fetching the same bytes twice. The blob's total size is parsed out of Content-Range, since a ranged response reports the length of the range and the SDK keeps its own ContentRange type private. Reads are pinned with if_match to the ETag captured when the archive was opened, for the same reason the S3 backend now is. The azure crates are taken with default-features = false. Their defaults enable reqwest's gzip and deflate features, which would let the client transparently decompress a response and change the byte count under a range read -- the SDK's own download() documents that automatic decompression breaks partitioned downloads. --- .github/workflows/sanity_checks.yml | 4 +- Cargo.lock | 674 +++++++++++++++++++++++++++- Cargo.toml | 10 +- README.md | 58 ++- src/archive.rs | 13 +- src/main.rs | 6 +- src/storage.rs | 106 ++++- src/storage/azure.rs | 377 ++++++++++++++++ 8 files changed, 1220 insertions(+), 28 deletions(-) create mode 100644 src/storage/azure.rs diff --git a/.github/workflows/sanity_checks.yml b/.github/workflows/sanity_checks.yml index f63c200..7e7875b 100644 --- a/.github/workflows/sanity_checks.yml +++ b/.github/workflows/sanity_checks.yml @@ -9,10 +9,12 @@ jobs: strategy: fail-fast: false matrix: - # Every combination that ships: the backend on, and off. + # Every combination that ships: both backends, neither, and each alone. features: - "" - "--no-default-features" + - "--no-default-features --features s3" + - "--no-default-features --features azure" name: ${{ matrix.features || 'default features' }} diff --git a/Cargo.lock b/Cargo.lock index 1504fe5..7a5aad7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -73,6 +73,50 @@ dependencies = [ "rustversion", ] +[[package]] +name = "async-lock" +version = "3.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311" +dependencies = [ + "event-listener", + "event-listener-strategy", + "pin-project-lite", +] + +[[package]] +name = "async-stream" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476" +dependencies = [ + "async-stream-impl", + "futures-core", + "pin-project-lite", +] + +[[package]] +name = "async-stream-impl" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "async-trait" +version = "0.1.92" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -601,12 +645,100 @@ dependencies = [ "tracing", ] +[[package]] +name = "azure_core" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4e41cbd819986ba41904c207d8ffc4106f8f8352a548d773e9554906379bb2fb" +dependencies = [ + "async-lock", + "async-trait", + "azure_core_macros", + "bytes", + "futures", + "pin-project", + "rustc_version", + "serde", + "serde_json", + "tokio", + "tracing", + "typespec", + "typespec_client_core", +] + +[[package]] +name = "azure_core_macros" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9b52dba6a345f3ad2d42ff8d0d63df9d0994cfa29657bf18ffdbf149f78a4f5" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "tracing", +] + +[[package]] +name = "azure_identity" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32edf96b356ca7c51d7590c4925cc36efc3947a5da4468e8e0b25c56ecbb3de5" +dependencies = [ + "async-lock", + "async-trait", + "azure_core", + "futures", + "pin-project", + "serde", + "serde_json", + "time", + "tracing", + "url", +] + +[[package]] +name = "azure_storage_blob" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "17b10207ecf7d666df6940b50051f433b3cd5d2b9b1dd190613208d7a84e7eed" +dependencies = [ + "async-stream", + "async-trait", + "azure_core", + "azure_storage_common", + "bytes", + "futures", + "percent-encoding", + "pin-project", + "serde", + "serde_json", + "time", + "tokio", +] + +[[package]] +name = "azure_storage_common" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0af2e6aeb8d76b17fc998f453c320913f73787b944e3cc29509d19411fa0321d" +dependencies = [ + "azure_core", + "serde", + "time", +] + [[package]] name = "base64" version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "base64" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" + [[package]] name = "base64-simd" version = "0.8.0" @@ -681,6 +813,23 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "cfg_aliases" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" + +[[package]] +name = "chacha20" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.1", + "rand_core", +] + [[package]] name = "clap" version = "4.6.6" @@ -742,6 +891,16 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" +[[package]] +name = "combine" +version = "4.6.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfc320937d09e6de266b31b9afb480f197d7a861be86be7cb2ea7e5d1bfffc5e" +dependencies = [ + "bytes", + "memchr", +] + [[package]] name = "const-oid" version = "0.10.2" @@ -834,6 +993,9 @@ name = "deranged" version = "0.5.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" +dependencies = [ + "serde_core", +] [[package]] name = "digest" @@ -874,6 +1036,12 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" +[[package]] +name = "dyn-clone" +version = "1.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" + [[package]] name = "either" version = "1.18.0" @@ -896,6 +1064,26 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "event-listener" +version = "5.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2" +dependencies = [ + "parking", + "pin-project-lite", +] + +[[package]] +name = "event-listener-strategy" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93" +dependencies = [ + "event-listener", + "pin-project-lite", +] + [[package]] name = "fastrand" version = "2.5.0" @@ -946,6 +1134,21 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" +[[package]] +name = "futures" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + [[package]] name = "futures-channel" version = "0.3.34" @@ -953,6 +1156,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" dependencies = [ "futures-core", + "futures-sink", ] [[package]] @@ -961,6 +1165,34 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" +[[package]] +name = "futures-executor" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "031b47cf1a3c6cc8bc2fc76cd437f521619387907d469316e7c0bc278f1f5432" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed" + +[[package]] +name = "futures-macro" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + [[package]] name = "futures-sink" version = "0.3.34" @@ -979,8 +1211,13 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" dependencies = [ + "futures-channel", "futures-core", + "futures-io", + "futures-macro", + "futures-sink", "futures-task", + "memchr", "pin-project-lite", "slab", ] @@ -1002,8 +1239,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" dependencies = [ "cfg-if", + "js-sys", "libc", "wasi", + "wasm-bindgen", ] [[package]] @@ -1013,8 +1252,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" dependencies = [ "cfg-if", + "js-sys", "libc", "r-efi", + "rand_core", + "wasm-bindgen", ] [[package]] @@ -1194,7 +1436,7 @@ version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "futures-channel", "futures-util", @@ -1343,6 +1585,55 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "jni" +version = "0.22.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5efd9a482cf3a427f00d6b35f14332adc7902ce91efb778580e180ff90fa3498" +dependencies = [ + "cfg-if", + "combine", + "jni-macros", + "jni-sys", + "log", + "simd_cesu8", + "thiserror", + "walkdir", + "windows-link", +] + +[[package]] +name = "jni-macros" +version = "0.22.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a00109accc170f0bdb141fed3e393c565b6f5e072365c3bd58f5b062591560a3" +dependencies = [ + "proc-macro2", + "quote", + "rustc_version", + "simd_cesu8", + "syn 2.0.119", +] + +[[package]] +name = "jni-sys" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6377a88cb3910bee9b0fa88d4f42e1d2da8e79915598f65fb0c7ee14c878af2" +dependencies = [ + "jni-sys-macros", +] + +[[package]] +name = "jni-sys-macros" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "38c0b942f458fe50cdac086d2f946512305e5631e720728f2a61aabcd47a6264" +dependencies = [ + "quote", + "syn 2.0.119", +] + [[package]] name = "jobserver" version = "0.1.35" @@ -1397,6 +1688,12 @@ dependencies = [ "hashbrown 0.16.1", ] +[[package]] +name = "lru-slab" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" + [[package]] name = "matchit" version = "0.8.4" @@ -1503,12 +1800,38 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1a80800c0488c3a21695ea981a54918fbb37abf04f4d0720c453632255e2ff0e" +[[package]] +name = "parking" +version = "2.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" + [[package]] name = "percent-encoding" version = "2.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" +[[package]] +name = "pin-project" +version = "1.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2466b2336ed02bcdca6b294417127b90ec92038d1d5c4fbeac971a922e0e0924" +dependencies = [ + "pin-project-internal", +] + +[[package]] +name = "pin-project-internal" +version = "1.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c96395f0a926bc13b1c17622aaddda1ecb55d49c8f1bf9777e4d877800a43f8b" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "pin-project-lite" version = "0.2.17" @@ -1551,6 +1874,73 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "quick-xml" +version = "0.41.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e660451e55124f798a69a5af3f49ccfbefbd41910eefd25caf2393e1f3473ec1" +dependencies = [ + "memchr", + "serde", +] + +[[package]] +name = "quinn" +version = "0.11.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c1a41e437b6bbd489372cd4971de128e85c855f56c57f283d20ff016cf7c0a8" +dependencies = [ + "bytes", + "cfg_aliases", + "pin-project-lite", + "quinn-proto", + "quinn-udp", + "rustc-hash", + "rustls", + "socket2", + "thiserror", + "tokio", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-proto" +version = "0.11.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "04759210543be93709136e28212294a659ef5001836ff4eab4d663e4529bba83" +dependencies = [ + "aws-lc-rs", + "bytes", + "getrandom 0.4.3", + "lru-slab", + "rand", + "rand_pcg", + "ring", + "rustc-hash", + "rustls", + "rustls-pki-types", + "slab", + "thiserror", + "tinyvec", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-udp" +version = "0.5.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35a133f956daabe89a61a685c2649f13d82d5aa4bd5d12d1277e1072a21c0694" +dependencies = [ + "cfg_aliases", + "libc", + "once_cell", + "socket2", + "tracing", + "windows-sys 0.61.2", +] + [[package]] name = "quote" version = "1.0.47" @@ -1566,6 +1956,32 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "rand" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" +dependencies = [ + "chacha20", + "getrandom 0.4.3", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + +[[package]] +name = "rand_pcg" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "caa0f4137e1c0a72f4c651489402276c8e8e1cf081f3b0ba156d2cbeef09e86a" +dependencies = [ + "rand_core", +] + [[package]] name = "rati" version = "0.2.0" @@ -1573,6 +1989,9 @@ dependencies = [ "aws-config", "aws-sdk-s3", "axum", + "azure_core", + "azure_identity", + "azure_storage_blob", "bytes", "clap", "flate2", @@ -1583,7 +2002,7 @@ dependencies = [ "serde", "thiserror", "tokio", - "tower-http", + "tower-http 0.7.1", "tracing", "tracing-subscriber", "zstd", @@ -1595,6 +2014,44 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cab834c73d247e67f4fae452806d17d3c7501756d98c8808d7c9c7aa7d18f973" +[[package]] +name = "reqwest" +version = "0.13.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "16a1cfa75cc186dd73d5818e510e042e40927bccc9c236b061cea97e1eb08029" +dependencies = [ + "base64 0.23.1", + "bytes", + "futures-core", + "futures-util", + "http 1.5.0", + "http-body 1.1.0", + "http-body-util", + "hyper", + "hyper-rustls", + "hyper-util", + "js-sys", + "log", + "percent-encoding", + "pin-project-lite", + "quinn", + "rustls", + "rustls-pki-types", + "rustls-platform-verifier", + "sync_wrapper", + "tokio", + "tokio-rustls", + "tokio-util", + "tower", + "tower-http 0.6.11", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "wasm-streams", + "web-sys", +] + [[package]] name = "ring" version = "0.17.14" @@ -1656,9 +2113,37 @@ version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" dependencies = [ + "web-time", "zeroize", ] +[[package]] +name = "rustls-platform-verifier" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "26d1e2536ce4f35f4846aa13bff16bd0ff40157cdb14cc056c7b14ba41233ba0" +dependencies = [ + "core-foundation", + "core-foundation-sys", + "jni", + "log", + "once_cell", + "rustls", + "rustls-native-certs", + "rustls-platform-verifier-android", + "rustls-webpki", + "security-framework", + "security-framework-sys", + "webpki-root-certs", + "windows-sys 0.61.2", +] + +[[package]] +name = "rustls-platform-verifier-android" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f" + [[package]] name = "rustls-webpki" version = "0.103.15" @@ -1683,6 +2168,15 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" +[[package]] +name = "same-file" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502" +dependencies = [ + "winapi-util", +] + [[package]] name = "schannel" version = "0.1.29" @@ -1851,6 +2345,22 @@ version = "0.3.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a219298ac11a56ea9a6d2120044824d6f01aeb034955e7af7bc16858527deea" +[[package]] +name = "simd_cesu8" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11031e251abf8611c80f460e19dbdeb54a66db918e49c65a7065b46ac7aec520" +dependencies = [ + "rustc_version", + "simdutf8", +] + +[[package]] +name = "simdutf8" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3a9fe34e3e7a50316060351f37187a3f546bce95496156754b601a5fa71b76e" + [[package]] name = "slab" version = "0.4.12" @@ -1924,6 +2434,9 @@ name = "sync_wrapper" version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" +dependencies = [ + "futures-core", +] [[package]] name = "synstructure" @@ -2005,6 +2518,21 @@ dependencies = [ "zerovec", ] +[[package]] +name = "tinyvec" +version = "1.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cf0ded5c4e56918d8f8a339e1bb67d038d3bc6d144ac407904015ba2e4cde9b" +dependencies = [ + "tinyvec_macros", +] + +[[package]] +name = "tinyvec_macros" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" + [[package]] name = "tokio" version = "1.53.1" @@ -2072,6 +2600,24 @@ dependencies = [ "tracing", ] +[[package]] +name = "tower-http" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" +dependencies = [ + "bitflags", + "bytes", + "futures-util", + "http 1.5.0", + "http-body 1.1.0", + "pin-project-lite", + "tower", + "tower-layer", + "tower-service", + "url", +] + [[package]] name = "tower-http" version = "0.7.1" @@ -2157,6 +2703,58 @@ version = "1.20.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" +[[package]] +name = "typespec" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "753a2fe021e407d4fc9ee6f4f0a33403cc306d5c54c4e4ebe1b8cbde0ca052b9" +dependencies = [ + "base64 0.22.1", + "bytes", + "futures", + "quick-xml", + "serde", + "serde_json", + "url", +] + +[[package]] +name = "typespec_client_core" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0373af0f9d4f580b3a1a9d9639cedaabe015ed262b35bfbe13941bfb14fe1ea6" +dependencies = [ + "async-trait", + "base64 0.22.1", + "bytes", + "dyn-clone", + "futures", + "pin-project", + "rand", + "reqwest", + "serde", + "serde_json", + "time", + "tokio", + "tracing", + "typespec", + "typespec_macros", + "url", + "uuid", +] + +[[package]] +name = "typespec_macros" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c608f4427943f8adb211abc95c87672b1b98847152783507d54e3246e502f60" +dependencies = [ + "proc-macro2", + "quote", + "rustc_version", + "syn 2.0.119", +] + [[package]] name = "unicode-ident" version = "1.0.24" @@ -2205,6 +2803,7 @@ version = "1.26.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b5772d71c9be8a8a6ac2117d949c5b224c1b72241bb611d9a3012edcf8af7812" dependencies = [ + "getrandom 0.4.3", "js-sys", "wasm-bindgen", ] @@ -2221,6 +2820,16 @@ version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5c3082ca00d5a5ef149bb8b555a72ae84c9c59f7250f013ac822ac2e49b19c64" +[[package]] +name = "walkdir" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29790946404f91d9c5d06f9874efddea1dc06c5efe94541a7d6863108e3a5e4b" +dependencies = [ + "same-file", + "winapi-util", +] + [[package]] name = "want" version = "0.3.1" @@ -2249,6 +2858,16 @@ dependencies = [ "wasm-bindgen-shared", ] +[[package]] +name = "wasm-bindgen-futures" +version = "0.4.77" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b7777d5cc23d0e91404e53ce2d5e8ec7acae3026b16233dba62cd3246457950" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "wasm-bindgen-macro" version = "0.2.127" @@ -2281,6 +2900,57 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "wasm-streams" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d1ec4f6517c9e11ae630e200b2b65d193279042e28edd4a2cda233e46670bbb" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + +[[package]] +name = "web-sys" +version = "0.3.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c435338968042f4f59a557f690a253676d47ce13ceb55d70100e7facf6620a30" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "webpki-root-certs" +version = "1.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b96554aa2acc8ccdb7e1c9a58a7a68dd5d13bccc69cd124cb09406db612a1c9b" +dependencies = [ + "rustls-pki-types", +] + +[[package]] +name = "winapi-util" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "windows-link" version = "0.2.1" diff --git a/Cargo.toml b/Cargo.toml index a18e290..d3e275a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,15 +3,16 @@ name = "rati" version = "0.2.0" edition = "2024" rust-version = "1.94" -description = "Lightweight tile server for Valhalla graph tiles backed by S3 or a local tar archive" +description = "Lightweight tile server for Valhalla graph tiles backed by S3, Azure Blob Storage, or a local tar archive" readme = "README.md" repository = "https://github.com/valhalla/rati" license = "MIT OR Apache-2.0" include = ["**/*.rs"] [features] -default = ["s3"] +default = ["s3", "azure"] s3 = ["dep:aws-config", "dep:aws-sdk-s3"] +azure = ["dep:azure_core", "dep:azure_identity", "dep:azure_storage_blob"] # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html @@ -23,8 +24,11 @@ lto = "thin" aws-config = { version = "1", default-features = false, features = ["rt-tokio", "default-https-client", "sso"], optional = true } aws-sdk-s3 = { version = "1", default-features = false, features = ["rt-tokio", "default-https-client"], optional = true } axum = "0.8" +azure_core = { version = "1.1.0", default-features = false, features = ["reqwest_rustls", "tokio"], optional = true } +azure_identity = { version = "1.0.0", default-features = false, optional = true } +azure_storage_blob = { version = "1.1.0", default-features = false, features = ["tokio"], optional = true } bytes = "1" -clap = { version = "4.6", features = ["derive"] } +clap = { version = "4.6", features = ["derive", "env"] } flate2 = "1" futures-util = { version = "0.3", default-features = false, features = ["alloc"] } http-body = "1" diff --git a/README.md b/README.md index 75fc507..368430e 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Rati -Rati (Range-Accessed Tar Index) is a lightweight HTTP server that serves individual [Valhalla](https://github.com/valhalla/valhalla) tiles from tar archives — stored on S3 or on the local filesystem — via byte-range reads. +Rati (Range-Accessed Tar Index) is a lightweight HTTP server that serves individual [Valhalla](https://github.com/valhalla/valhalla) tiles from tar archives — stored on S3, Azure Blob Storage, or on the local filesystem — via byte-range reads. Named after the auger Odin used to bore through a mountain to reach the mead of poetry locked within. Rati was created with two use cases in mind: @@ -15,7 +15,9 @@ rati [OPTIONS] ``` **Arguments:** -- `` — Archive location. Either an S3 URL (`s3://bucket/path/to/tiles.tar`) or a path to a local `.tar` file. Anything not starting with `s3://` is treated as a local path. +- `` — Archive location. An S3 URL (`s3://bucket/path/to/tiles.tar`), an Azure Blob + URL (`https://.blob.core.windows.net//tiles.tar`), or a path to a local + `.tar` file. Anything else is treated as a local path. **Options:** | Flag | Default | Description | @@ -46,6 +48,52 @@ rati ./tiles.tar --port 8080 See [`valhalla_build_config`](https://github.com/valhalla/valhalla/blob/master/scripts/valhalla_build_config) for the full list of flags. +### Azure Blob Storage + +```sh +rati "https://myaccount.blob.core.windows.net/valhalla/tiles.tar" --port 8080 +``` + +There are no Azure flags. The identity is chosen from the environment, using the same +variables every Azure SDK reads, in the order `DefaultAzureCredential` uses elsewhere — the +first row that matches wins: + +| Condition | Identity | +|-----------|----------| +| URL is `http://`, or carries a SAS (`?...&sig=...`) | none — anonymous | +| Host is not an Azure Blob endpoint | none — anonymous | +| `AZURE_FEDERATED_TOKEN_FILE` | workload identity (AKS) | +| `AZURE_TENANT_ID` + `AZURE_CLIENT_ID` + `AZURE_CLIENT_SECRET` | service principal | +| `AZURE_CLIENT_ID` | user-assigned managed identity | +| otherwise | system-assigned managed identity | + +Whichever identity is used needs **Storage Blob Data Reader** on the account or container. + +Three rati-specific notes: + +- A bearer token is only ever sent to an `https` Azure Blob host. An archive behind a CDN, + Front Door, or a storage-account custom domain is reached anonymously, so it must be + public or the URL must carry a SAS. +- Plaintext (`http://`) endpoints are always anonymous — rati will not put a bearer token on + the wire in the clear. Azurite therefore needs no configuration at all; point rati at + `http://127.0.0.1:10000/devstoreaccount1/valhalla/tiles.tar`. For local development against + real blob storage, use a service principal or a SAS. +- Do not set `Content-Encoding` on the archive blob. A blob-level encoding confuses CDNs and + proxies in front of rati. + +## Build Features + +| Feature | Default | Pulls in | +|---------|---------|----------| +| `s3` | yes | `aws-config`, `aws-sdk-s3` | +| `azure` | yes | `azure_storage_blob`, `azure_identity` | + +Local archives need neither. Dropping an unused backend removes its HTTP and TLS stack: + +```sh +cargo build --release --no-default-features --features azure +``` + ## Endpoints ``` @@ -90,8 +138,8 @@ Every tile response includes headers suitable for CDN caching: | Header | Description | |--------|-------------| -| `ETag` | S3 object ETag, or synthesized `"-"` for local archives — both change whenever the archive is replaced | -| `Last-Modified` | S3 object last-modified timestamp, or the file's mtime for local archives | +| `ETag` | The object or blob ETag, or synthesized `"-"` for local archives — all change whenever the archive is replaced | +| `Last-Modified` | The object or blob last-modified timestamp, or the file's mtime for local archives | | `Cache-Control` | `public, max-age=, immutable` — `` from `--cache-max-age` (default 86400) | | `X-Dataset-Id` | Auto-detected from `GraphTileHeader`, overridden with `--dataset-id`, or the ETag as fallback | | `Vary` | `Accept-Encoding` — ensures correct CDN behavior with encoding negotiation | @@ -101,7 +149,7 @@ Every tile response includes headers suitable for CDN caching: For graph tile archives (`.gph`), the dataset ID is automatically extracted from the `GraphTileHeader` of the first tile in the archive. This is typically the OSM changeset ID (`dataset_id_` field, a `u64` at byte offset 32 in the 272-byte header). -For any other kind of archive, use `--dataset-id` to provide an explicit value. If neither works, the S3 ETag is used as a fallback. +For any other kind of archive, use `--dataset-id` to provide an explicit value. If neither works, the archive ETag is used as a fallback. ## Index Modes diff --git a/src/archive.rs b/src/archive.rs index ac7f276..51a27f7 100644 --- a/src/archive.rs +++ b/src/archive.rs @@ -272,8 +272,8 @@ pub struct Archive { impl Archive { /// Connect to the archive and load the tar index. /// - /// `source` is either an S3 URL (`s3://bucket/key`) or a local filesystem path. - /// For S3, uses the default AWS credential chain (SSO, IRSA, env vars, IMDS). + /// `source` is an S3 URL (`s3://bucket/key`), an Azure Blob HTTPS URL, or a local + /// filesystem path — see [`crate::storage`] for how each is opened and authenticated. /// /// If the first tar entry is not `index.bin` and `scan_index` is true, falls back /// to scanning tar headers to build the index — slow for large archives, since tar @@ -292,10 +292,15 @@ impl Archive { etag, last_modified, size: archive_size, + prefetch, } = src; - // Step 1: Read the first 512-byte tar header - let header_bytes = storage.read_range(0, TAR_BLOCK_SIZE as u64).await?; + // Step 1: the first 512-byte tar header, reusing it if the backend already + // read it while fetching metadata. + let header_bytes = match prefetch { + Some(bytes) if bytes.len() >= TAR_BLOCK_SIZE => bytes.slice(0..TAR_BLOCK_SIZE), + _ => storage.read_range(0, TAR_BLOCK_SIZE as u64).await?, + }; let header: &[u8; TAR_BLOCK_SIZE] = header_bytes .as_ref() .try_into() diff --git a/src/main.rs b/src/main.rs index 2b4bc00..bb4d422 100644 --- a/src/main.rs +++ b/src/main.rs @@ -24,7 +24,7 @@ use archive::TileCompression; #[derive(Parser)] struct Config { - /// Archive location: `s3://bucket/key.tar` or a local filesystem path + /// Archive location: `s3://bucket/key.tar`, an Azure Blob HTTPS URL, or a local path archive: String, /// Build index by scanning tar headers if index.bin is missing #[arg(long)] @@ -78,7 +78,9 @@ async fn run(config: Config) { .expect("failed to load tar index"); info!( "Loaded {} with {} tiles (dataset_id={})", - config.archive, meta.tile_count, meta.dataset_id, + storage::redact(&config.archive), + meta.tile_count, + meta.dataset_id, ); let cache_headers = build_cache_headers(&meta, config.cache_max_age); diff --git a/src/storage.rs b/src/storage.rs index bbcb08a..92323d0 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -1,6 +1,9 @@ -//! Storage backends for a tar archive. Hides whether bytes come from S3 or a local -//! file — every read goes through [`Storage::read_range`]; everything above this layer -//! is unaware of the source. +//! Storage backends for a tar archive. Hides whether bytes come from S3, Azure Blob, +//! or a local file — every read goes through [`Storage::read_range`]; everything above +//! this layer is unaware of the source. + +#[cfg(feature = "azure")] +pub mod azure; use std::sync::Arc; @@ -8,14 +11,36 @@ use bytes::Bytes; use crate::archive::Error; +/// Strip the query string from an archive location before it is logged or put in an +/// error. +/// +/// An Azure Blob URL may carry a shared access signature there — `?...&sig=...` — which +/// is a bearer credential. Anything that prints the archive location must go through +/// this, or the token lands in stdout and every log aggregator downstream. +/// +/// Only URLs are truncated. A filesystem path may legitimately contain a `?`, and +/// reporting `/data/tiles` for `/data/tiles?v2.tar` names a file the operator never gave. +pub fn redact(source: &str) -> &str { + if !source.starts_with("http://") && !source.starts_with("https://") { + return source; + } + match source.split_once('?') { + Some((head, _)) => head, + None => source, + } +} + /// Source metadata read once when the archive is opened. pub struct ArchiveSource { - /// Source ETag — S3 object ETag, or synthesized `"-"` for local archives. + /// Source ETag — the object or blob ETag, or synthesized `"-"` for + /// local archives. pub etag: Box, /// Source Last-Modified as an HTTP-date string. pub last_modified: Box, /// Total size of the archive in bytes. pub size: u64, + /// Leading bytes the backend already read while fetching metadata, if any. + pub prefetch: Option, } pub enum Storage { @@ -27,6 +52,12 @@ pub enum Storage { /// ETag the index was built against; every read is pinned to it. etag: Box, }, + #[cfg(feature = "azure")] + AzureBlob { + client: Box, + /// ETag the index was built against; every read is pinned to it. + etag: Box, + }, Local { // `Arc` lets us call `read_at` (which takes `&self`) concurrently // from multiple `spawn_blocking` tasks without `try_clone()` syscalls per read. @@ -35,8 +66,21 @@ pub enum Storage { } impl Storage { - /// Open `source`: an S3 URL (`s3://bucket/key`) or a local filesystem path. + /// Open `source`: an S3 URL (`s3://bucket/key`), an Azure Blob HTTPS URL, or a local + /// filesystem path. pub async fn open(source: &str) -> Result<(Self, ArchiveSource), Error> { + #[cfg(feature = "azure")] + if azure::is_azure_url(source) { + return azure::open_azure(source).await; + } + + #[cfg(not(feature = "azure"))] + if source.starts_with("https://") || source.starts_with("http://") { + return Err(Error::Protocol( + "HTTP(S) archives require the 'azure' cargo feature".into(), + )); + } + #[cfg(feature = "s3")] if let Some((bucket, key)) = parse_s3_url(source) { return open_s3(bucket, key).await; @@ -66,6 +110,10 @@ impl Storage { key, etag, } => read_s3_range(client, bucket, key, etag, offset, length).await, + #[cfg(feature = "azure")] + Self::AzureBlob { client, etag } => { + azure::read_azure_range(client, etag, offset, length).await + } Self::Local { file } => read_local_range(file.clone(), offset, length).await, }?; @@ -82,27 +130,31 @@ impl Storage { } } +/// Filesystem-backed tar archive: `stat` for metadata, positional reads for ranges. /// Open the archive from the local filesystem. ETag is synthesized from `mtime+size` /// (matches a fresh value whenever the archive changes); Last-Modified is the file's /// mtime formatted as an HTTP-date. fn open_local(path: &str) -> Result<(Storage, ArchiveSource), Error> { - let metadata = - std::fs::metadata(path).map_err(|e| Error::Io(format!("stat({path}) failed: {e}")))?; + let metadata = std::fs::metadata(path) + .map_err(|e| Error::Io(format!("stat({}) failed: {e}", redact(path))))?; if !metadata.is_file() { - return Err(Error::Protocol(format!("{path} is not a regular file"))); + return Err(Error::Protocol(format!( + "{} is not a regular file", + redact(path) + ))); } let size = metadata.len(); let mtime = metadata .modified() - .map_err(|e| Error::Io(format!("{path} has no mtime: {e}")))?; + .map_err(|e| Error::Io(format!("{} has no mtime: {e}", redact(path))))?; let mtime_unix = mtime .duration_since(std::time::UNIX_EPOCH) .map_err(|_| Error::Protocol("file mtime is before UNIX epoch".into()))? .as_secs(); let etag: Box = format!("\"{mtime_unix}-{size}\"").into(); let last_modified: Box = httpdate::fmt_http_date(mtime).into(); - let file = - std::fs::File::open(path).map_err(|e| Error::Io(format!("open({path}) failed: {e}")))?; + let file = std::fs::File::open(path) + .map_err(|e| Error::Io(format!("open({}) failed: {e}", redact(path))))?; Ok(( Storage::Local { @@ -112,6 +164,7 @@ fn open_local(path: &str) -> Result<(Storage, ArchiveSource), Error> { etag, last_modified, size, + prefetch: None, }, )) } @@ -139,7 +192,9 @@ async fn read_local_range( /// S3-backed tar archive: `HeadObject` for metadata, ranged `GetObject` for reads. #[cfg(feature = "s3")] +#[cfg(feature = "s3")] /// Split an `s3://bucket/key` URL into its bucket and key. +#[cfg(feature = "s3")] fn parse_s3_url(url: &str) -> Option<(&str, &str)> { let path = url.strip_prefix("s3://")?; path.split_once('/') @@ -148,6 +203,7 @@ fn parse_s3_url(url: &str) -> Option<(&str, &str)> { #[cfg(feature = "s3")] /// Open the archive from S3: HeadObject for ETag/Last-Modified/size, then hand back the /// pieces `Archive::open` needs to read the rest. +#[cfg(feature = "s3")] async fn open_s3(bucket: &str, key: &str) -> Result<(Storage, ArchiveSource), Error> { let client = aws_sdk_s3::Client::new( &aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await, @@ -191,6 +247,7 @@ async fn open_s3(bucket: &str, key: &str) -> Result<(Storage, ArchiveSource), Er etag, last_modified, size, + prefetch: None, }, )) } @@ -252,6 +309,33 @@ async fn read_s3_range( mod tests { use super::*; + #[test] + fn redact_strips_the_query_string() { + // A SAS is a bearer credential and must never reach a log or an error. + assert_eq!( + redact("https://acct.blob.core.windows.net/t/p.tar?sv=2024-11-04&sig=SECRET"), + "https://acct.blob.core.windows.net/t/p.tar" + ); + // Nothing to strip: left exactly as-is, including for the other backends. + assert_eq!( + redact("https://acct.blob.core.windows.net/t/p.tar"), + "https://acct.blob.core.windows.net/t/p.tar" + ); + assert_eq!(redact("s3://bucket/planet.tar"), "s3://bucket/planet.tar"); + assert_eq!(redact("/data/planet.tar"), "/data/planet.tar"); + // A bare "?" still loses everything after it. + assert_eq!(redact("https://h/p.tar?"), "https://h/p.tar"); + } + + #[test] + fn redact_leaves_filesystem_paths_alone() { + // A path may legitimately contain '?'; truncating it would name a file the + // operator never passed. + assert_eq!(redact("/data/tiles?v2.tar"), "/data/tiles?v2.tar"); + assert_eq!(redact("./tiles?x.tar"), "./tiles?x.tar"); + assert_eq!(redact("s3://bucket/planet.tar"), "s3://bucket/planet.tar"); + } + /// The exact-length contract every caller relies on: `scan_tar_headers` computes /// `chunk.len() - local`, which underflows on a short read. #[tokio::test] diff --git a/src/storage/azure.rs b/src/storage/azure.rs new file mode 100644 index 0000000..f983eb0 --- /dev/null +++ b/src/storage/azure.rs @@ -0,0 +1,377 @@ +//! Azure Blob Storage backend: byte-range reads against a blob holding the tar archive. + +use std::sync::Arc; + +use azure_core::credentials::TokenCredential; +use azure_core::http::Etag; +use azure_core::http::Url; +use azure_core::http::headers::HeaderName; +use azure_identity::{ + ClientSecretCredential, ManagedIdentityCredential, ManagedIdentityCredentialOptions, + UserAssignedId, WorkloadIdentityCredential, +}; +use azure_storage_blob::BlobClient; +use azure_storage_blob::models::{BlobClientDownloadOptions, HttpRange}; +use bytes::Bytes; + +use super::{ArchiveSource, Storage, redact}; +use crate::archive::Error; + +/// Bytes read to probe the archive: one tar header block. +const PROBE_LEN: u64 = 512; + +/// The SDK keeps its own `ContentRange` type private, so read the header directly. +const CONTENT_RANGE: HeaderName = HeaderName::from_static("content-range"); + +/// Host suffixes that identify an Azure Blob endpoint, across public and sovereign clouds. +const BLOB_HOST_SUFFIXES: [&str; 3] = [ + ".blob.core.windows.net", + ".blob.core.usgovcloudapi.net", + ".blob.core.chinacloudapi.cn", +]; + +/// True when `host` is a real Azure Blob endpoint in one of the known clouds. +/// +/// `Url` lowercases the host during parsing; hostnames are case-insensitive. +fn has_blob_host_suffix(host: &str) -> bool { + BLOB_HOST_SUFFIXES + .iter() + .any(|suffix| host.len() > suffix.len() && host.ends_with(suffix)) +} + +/// True when `source` should be served by this backend. +/// +/// rati has no generic HTTP backend and should not grow one — proxying a proxy makes no +/// sense — so any http(s) archive is Azure by elimination. Whether the host is *trusted +/// enough to be sent a credential* is a separate question, answered in [`credential`]. +pub(super) fn is_azure_url(source: &str) -> bool { + source.starts_with("https://") || source.starts_with("http://") +} + +/// True when the URL carries a shared access signature, meaning no credential is needed. +fn has_sas_token(url: &str) -> bool { + let Some((_, query)) = url.split_once('?') else { + return false; + }; + query + .split('&') + .any(|param| param.split_once('=').is_some_and(|(k, _)| k == "sig")) +} + +/// Extract the total resource length from a `Content-Range` header value. +/// +/// Per RFC 9110 §14.4 the value is `bytes /`, where the +/// complete length is `*` when the server does not know it. +fn parse_content_range_total(header: &str) -> Option { + header + .strip_prefix("bytes ")? + .rsplit_once('/') + .map(|(_, total)| total)? + .trim() + .parse() + .ok() +} + +/// Non-empty environment variable, or `None`. +fn var(key: &str) -> Option { + std::env::var(key).ok().filter(|v| !v.is_empty()) +} + +/// The credential rati presents to Azure Blob, or `None` when the request needs none. +/// +/// Only a host that is demonstrably Azure Blob is ever sent a bearer token, and never over +/// plaintext. An archive behind a CDN or a custom domain therefore authorizes itself — a +/// SAS, or a public container. That costs little: a custom domain mapped straight onto a +/// storage account is http-only, which is never credentialed anyway, and the https form +/// puts a CDN edge in front, which is not the storage service the token was issued for. +/// +/// The variables below are the ones every Azure SDK already reads, and the order is the one +/// `DefaultAzureCredential` uses elsewhere: explicit configuration outranks ambient. That +/// ordering is also what disambiguates `AZURE_CLIENT_ID`, which all three arms consult. +fn credential(url: &str) -> Result>, Error> { + let blob_host = Url::parse(url) + .ok() + .filter(|u| u.scheme() == "https") + .and_then(|u| u.host_str().map(has_blob_host_suffix)) + .unwrap_or(false); + if !blob_host { + tracing::warn!( + "{} is not an https Azure Blob host; connecting anonymously", + redact(url) + ); + return Ok(None); + } + if has_sas_token(url) { + tracing::info!("Azure credential: none (the URL carries a SAS)"); + return Ok(None); + } + + // A projected token file: AKS, or any federated workload. + if std::env::var_os("AZURE_FEDERATED_TOKEN_FILE").is_some() { + tracing::info!("Azure credential: workload identity"); + return WorkloadIdentityCredential::new(None) + .map(|c| Some(c as Arc)) + .map_err(|e| Error::Io(format!("workload identity credential: {e}"))); + } + + // A service principal: the only option anywhere managed identity does not reach — + // on-premises, another cloud, or a developer's machine. + if let (Some(tenant), Some(client), Some(secret)) = ( + var("AZURE_TENANT_ID"), + var("AZURE_CLIENT_ID"), + var("AZURE_CLIENT_SECRET"), + ) { + tracing::info!("Azure credential: service principal"); + return ClientSecretCredential::new(&tenant, client, secret.into(), None) + .map(|c| Some(c as Arc)) + .map_err(|e| Error::Io(format!("client secret credential: {e}"))); + } + + // Azure-hosted. AZURE_CLIENT_ID names a user-assigned identity; without it, the + // system-assigned one. Empty is not a value: an empty id reaches IMDS as `client-id=&`, + // which can silently resolve to the system-assigned identity instead of failing. + let user_assigned = var("AZURE_CLIENT_ID"); + tracing::info!( + "Azure credential: {} managed identity", + if user_assigned.is_some() { + "user-assigned" + } else { + "system-assigned" + } + ); + let options = user_assigned.map(|id| ManagedIdentityCredentialOptions { + user_assigned_id: Some(UserAssignedId::ClientId(id)), + ..Default::default() + }); + ManagedIdentityCredential::new(options) + .map(|c| Some(c as Arc)) + .map_err(|e| Error::Io(format!("managed identity credential: {e}"))) +} + +/// Open the archive from Azure Blob Storage. +/// +/// Reads the leading tar block to pick up the blob's ETag, Last-Modified, and total +/// size in one request: `get_properties` returns its values as raw headers, whereas a +/// ranged `download` hands back a typed `BlobDownloadProperties`. The block itself is +/// handed back in [`ArchiveSource::prefetch`] so the caller need not read it again. +pub(super) async fn open_azure(url: &str) -> Result<(Storage, ArchiveSource), Error> { + let parsed = Url::parse(url) + .map_err(|e| Error::Protocol(format!("invalid blob URL {}: {e}", redact(url))))?; + let credential = credential(url)?; + + // The SDK embeds the raw URL in this error, so drop it rather than forward it: the + // message would otherwise carry a SAS to stderr through `main`'s `.expect()`. + let client = BlobClient::new(parsed, credential, None) + .map_err(|_| Error::Io(format!("creating blob client for {}", redact(url))))?; + + let probe = client + .download(Some(BlobClientDownloadOptions { + range: Some(HttpRange::new(0, PROBE_LEN)), + ..Default::default() + })) + .await + .map_err(|e| { + // The credential is only exercised on the first request, so an auth failure + // surfaces here rather than at construction. + Error::Io(format!( + "reading {} failed: {e} (check the identity has Storage Blob Data Reader \ + on the container — account-level Owner or Contributor does not grant \ + blob read)", + redact(url), + )) + })?; + + let etag: Box = probe + .properties + .etag + .as_ref() + .map(ToString::to_string) + .ok_or_else(|| Error::Protocol("blob response carried no ETag".into()))? + .into(); + + let last_modified: Box = probe + .properties + .last_modified + .map(|t| httpdate::fmt_http_date(t.into())) + .ok_or_else(|| Error::Protocol("blob response carried no Last-Modified".into()))? + .into(); + + let size = probe + .headers + .get_optional_str(&CONTENT_RANGE) + .and_then(parse_content_range_total) + .ok_or_else(|| Error::Protocol("blob response carried no usable Content-Range".into()))?; + + // The probe *is* the first tar block; keep it rather than paying for it twice. + let prefetch = probe + .body + .collect() + .await + .map_err(|e| Error::Io(format!("reading blob probe body: {e}")))?; + + Ok(( + Storage::AzureBlob { + client: Box::new(client), + // Pinned so a mid-flight republish fails loudly instead of serving bytes + // read at stale offsets. See `read_azure_range`. + etag: etag.clone(), + }, + ArchiveSource { + etag, + last_modified, + size, + prefetch: Some(prefetch), + }, + )) +} + +/// Read `length` bytes at `offset` from the blob. +/// +/// Every read is conditional on the ETag captured when the archive was opened. The tile +/// index maps ids to byte offsets in one specific version of the blob, so if the archive +/// is replaced while rati is running those offsets now point into different data. Without +/// the guard the service happily returns whatever occupies that range, and rati serves it +/// under the old ETag and `Cache-Control: immutable` — silently poisoning client and CDN +/// caches. With it, Azure answers 412 and the read fails visibly. +fn ranged_read(etag: &str, offset: u64, length: u64) -> BlobClientDownloadOptions<'static> { + BlobClientDownloadOptions { + range: Some(HttpRange::new(offset, length)), + if_match: Some(Etag::from(etag)), + ..Default::default() + } +} + +pub(super) async fn read_azure_range( + client: &BlobClient, + etag: &str, + offset: u64, + length: u64, +) -> Result { + let response = client + .download(Some(ranged_read(etag, offset, length))) + .await + .map_err(|e| { + Error::Io(format!( + "blob download(offset={offset}, len={length}): {e} \ + (a 412 here means the archive was replaced while rati was running; restart it)" + )) + })?; + + response + .body + .collect() + .await + .map_err(|e| Error::Io(format!("reading blob response body: {e}"))) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn is_azure_url_test() { + // Any http(s) archive is Azure by elimination; rati has no other HTTP backend. + assert!(is_azure_url( + "https://acct.blob.core.windows.net/tiles/planet.tar" + )); + assert!(is_azure_url("https://example.com/tiles/planet.tar")); + assert!(is_azure_url( + "http://127.0.0.1:10000/devstoreaccount1/valhalla/tiles.tar" + )); + + assert!(!is_azure_url("s3://bucket/planet.tar")); + assert!(!is_azure_url("./planet.tar")); + assert!(!is_azure_url("/data/planet.tar")); + } + + #[test] + fn has_blob_host_suffix_test() { + assert!(has_blob_host_suffix("acct.blob.core.windows.net")); + assert!(has_blob_host_suffix("acct.blob.core.usgovcloudapi.net")); + assert!(has_blob_host_suffix("acct.blob.core.chinacloudapi.cn")); + + // The suffix alone, with no account label, is not an endpoint. + assert!(!has_blob_host_suffix("blob.core.windows.net")); + assert!(!has_blob_host_suffix("example.com")); + // A lookalike must not match by prefix. + assert!(!has_blob_host_suffix( + "acct.blob.core.windows.net.evil.test" + )); + } + + #[test] + fn has_sas_token_test() { + assert!(has_sas_token( + "https://acct.blob.core.windows.net/t/p.tar?sv=2024-11-04&sig=abc%3D" + )); + assert!(has_sas_token( + "https://acct.blob.core.windows.net/t/p.tar?sp=r&st=2026-01-01&sig=x" + )); + + assert!(!has_sas_token("https://acct.blob.core.windows.net/t/p.tar")); + // a query string without a signature is not a SAS + assert!(!has_sas_token( + "https://acct.blob.core.windows.net/t/p.tar?snapshot=2026-01-01" + )); + // a parameter merely ending in "sig" must not match + assert!(!has_sas_token( + "https://acct.blob.core.windows.net/t/p.tar?nosig=x" + )); + } + + /// A bearer token must never leave for a host that has not proved it is Azure Blob, + /// and never over plaintext. These are the only inputs that decide it. + #[test] + fn only_an_https_blob_host_is_credentialed() { + let anonymous = |u: &str| credential(u).unwrap().is_none(); + + // Plaintext, whatever the host: a token must not go out in the clear. + assert!(anonymous( + "http://127.0.0.1:10000/devstoreaccount1/valhalla/tiles.tar" + )); + assert!(anonymous("http://acct.blob.core.windows.net/t/p.tar")); + + // An https host that is not Azure Blob — including one claiming the emulator + // account in its path, which is how the round-1 leak reached a foreign host. + assert!(anonymous("https://example.com/t/p.tar")); + assert!(anonymous( + "https://attacker.example/devstoreaccount1/c/p.tar" + )); + + // A SAS already authorizes the read, so no token is needed. + assert!(anonymous( + "https://acct.blob.core.windows.net/t/p.tar?sv=1&sig=abc" + )); + } + + /// Every ranged read must be pinned to the ETag captured at open. Without it a blob + /// replaced mid-flight is served at stale offsets under the old ETag and + /// `Cache-Control: immutable`, poisoning client and CDN caches. + #[test] + fn every_ranged_read_is_pinned_to_the_etag() { + let opts = ranged_read("\"0x8DF12581E17C90B\"", 1536, 44_504_000); + assert_eq!( + opts.if_match, + Some(Etag::from("\"0x8DF12581E17C90B\"")), + "ranged reads must carry If-Match" + ); + // HttpRange keeps its fields private; its Display is the wire value. + let range = opts.range.expect("a ranged read must carry a range"); + assert_eq!(range.to_string(), "bytes=1536-44505535"); + } + + #[test] + fn parse_content_range_total_test() { + assert_eq!( + parse_content_range_total("bytes 0-511/85899345920"), + Some(85_899_345_920) + ); + assert_eq!(parse_content_range_total("bytes 0-0/1"), Some(1)); + + // unsatisfiable / unknown total + assert_eq!(parse_content_range_total("bytes 0-511/*"), None); + assert_eq!(parse_content_range_total("bytes */1234"), Some(1234)); + assert_eq!(parse_content_range_total("garbage"), None); + assert_eq!(parse_content_range_total(""), None); + } +} From d8e2804b8be3b1c6b9dfd99a3c6e68be3bad5b28 Mon Sep 17 00:00:00 2001 From: Kyle Brooks Date: Tue, 29 Sep 2026 09:10:07 -0400 Subject: [PATCH 6/6] chore: tidy leftovers from folding the storage modules Folding `local` and `s3` into plain functions left duplicated `#[cfg(feature = "s3")]` attributes on `parse_s3_url` and `open_s3`, and each old module doc comment attached itself to the first function after it. Also drop the README's Storage Blob Data Reader line. That is Azure RBAC, and the error returned when the first read fails already says it. --- README.md | 2 -- src/storage.rs | 5 ----- 2 files changed, 7 deletions(-) diff --git a/README.md b/README.md index 368430e..5202d8c 100644 --- a/README.md +++ b/README.md @@ -67,8 +67,6 @@ first row that matches wins: | `AZURE_CLIENT_ID` | user-assigned managed identity | | otherwise | system-assigned managed identity | -Whichever identity is used needs **Storage Blob Data Reader** on the account or container. - Three rati-specific notes: - A bearer token is only ever sent to an `https` Azure Blob host. An archive behind a CDN, diff --git a/src/storage.rs b/src/storage.rs index 92323d0..22e7d70 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -130,7 +130,6 @@ impl Storage { } } -/// Filesystem-backed tar archive: `stat` for metadata, positional reads for ranges. /// Open the archive from the local filesystem. ETag is synthesized from `mtime+size` /// (matches a fresh value whenever the archive changes); Last-Modified is the file's /// mtime formatted as an HTTP-date. @@ -190,11 +189,8 @@ async fn read_local_range( }) } -/// S3-backed tar archive: `HeadObject` for metadata, ranged `GetObject` for reads. -#[cfg(feature = "s3")] #[cfg(feature = "s3")] /// Split an `s3://bucket/key` URL into its bucket and key. -#[cfg(feature = "s3")] fn parse_s3_url(url: &str) -> Option<(&str, &str)> { let path = url.strip_prefix("s3://")?; path.split_once('/') @@ -203,7 +199,6 @@ fn parse_s3_url(url: &str) -> Option<(&str, &str)> { #[cfg(feature = "s3")] /// Open the archive from S3: HeadObject for ETag/Last-Modified/size, then hand back the /// pieces `Archive::open` needs to read the rest. -#[cfg(feature = "s3")] async fn open_s3(bucket: &str, key: &str) -> Result<(Storage, ArchiveSource), Error> { let client = aws_sdk_s3::Client::new( &aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await,