From a7223e8f9cd5dcf03889367280e027174a661e8d Mon Sep 17 00:00:00 2001 From: wan9chi Date: Sun, 4 Oct 2026 15:27:34 +0800 Subject: [PATCH] refactor(cache): let remote cache requests carry auth headers `vt_remote_cache::Client::new` now takes an `Auth`, which supplies the headers for each request by operation (fetch, download, or store). It can use the client's HTTP client to get credentials, and when it fails, the request isn't sent and the operation fails with `Error::Auth`. The plan resolves how requests authenticate into `ResolvedRemoteCacheConfig::auth`, which holds everything needed to build the credentials, so nothing reads envs after planning. `vt` builds the `Auth` for it with `build_auth` and caches clients by endpoint and auth. The only kind is `anonymous`, which adds no headers, so requests are unchanged. Co-Authored-By: Claude Opus 5.5 --- crates/vt/src/session/cache/mod.rs | 36 +++-- crates/vt/src/session/cache/remote.rs | 140 ++++++++++------ crates/vt_plan/src/remote_cache.rs | 33 ++-- .../query_cli_overrides_invalid_env.jsonc | 5 +- .../query_controls_in_env_fingerprinted.jsonc | 5 +- .../snapshots/query_default_read.jsonc | 5 +- ...query_empty_env_endpoint_uses_config.jsonc | 5 +- .../query_nested_cli_overrides_env.jsonc | 5 +- .../query_prefix_controls_fingerprinted.jsonc | 5 +- .../query_sibling_endpoint_isolation.jsonc | 10 +- .../query_sibling_mode_isolation.jsonc | 5 +- .../query_endpoint_not_validated.jsonc | 5 +- .../query_nested_endpoint_enables_read.jsonc | 5 +- crates/vt_remote_cache/README.md | 4 +- crates/vt_remote_cache/src/auth/mod.rs | 42 +++++ crates/vt_remote_cache/src/lib.rs | 151 +++++++++++++++--- 16 files changed, 356 insertions(+), 105 deletions(-) create mode 100644 crates/vt_remote_cache/src/auth/mod.rs diff --git a/crates/vt/src/session/cache/mod.rs b/crates/vt/src/session/cache/mod.rs index 4ff29bbba..2caefd817 100644 --- a/crates/vt/src/session/cache/mod.rs +++ b/crates/vt/src/session/cache/mod.rs @@ -355,16 +355,18 @@ impl ExecutionCache { clippy::manual_let_else, reason = "naming every access mode makes adding one a compile error here" )] - let url = match &cache_metadata.remote_cache { - Some(ResolvedRemoteCacheConfig { - access: RemoteCacheAccess::Read | RemoteCacheAccess::ReadWrite, - url, - }) => url, + let remote_config = match &cache_metadata.remote_cache { + Some( + remote_config @ ResolvedRemoteCacheConfig { + access: RemoteCacheAccess::Read | RemoteCacheAccess::ReadWrite, + .. + }, + ) => remote_config, None => return Ok(Err(local_miss)), }; let remote_miss = match self .try_hit_remote( - url, + remote_config, cache_metadata, &cache_key, globbed_inputs, @@ -421,15 +423,16 @@ impl ExecutionCache { Ok(Err(CacheMiss::NotFound)) } - /// Fetch the entry from the remote cache at `endpoint`. An exact entry - /// that passes validation is a hit once its output archive is downloaded. + /// Fetch the entry from the remote cache that `remote_config` configures. + /// An exact entry that passes validation is a hit once its output archive + /// is downloaded. /// A fallback entry, a failed validation, or a failed read is a miss. An /// error while validating counts as a failed read, so the remote entry /// never fails the task. #[expect(clippy::too_many_arguments, reason = "forwarded from `try_hit`")] async fn try_hit_remote( &self, - endpoint: &Arc, + remote_config: &ResolvedRemoteCacheConfig, cache_metadata: &CacheMetadata, cache_key: &CacheEntryKey, globbed_inputs: &BTreeMap, @@ -439,7 +442,7 @@ impl ExecutionCache { ) -> anyhow::Result> { let fetched = self .remote_clients - .fetch(endpoint, cache_key, &cache_metadata.execution_cache_key, cancel_token) + .fetch(remote_config, cache_key, &cache_metadata.execution_cache_key, cancel_token) .await; let validate = |cache_value: &CacheEntryValue| { cache_value.validate(&cache_metadata.unfiltered_envs, globbed_inputs, workspace_root) @@ -454,7 +457,7 @@ impl ExecutionCache { Some(blob_id) => { match self .remote_clients - .download_archive(endpoint, &blob_id, cache_dir, cancel_token) + .download_archive(remote_config, &blob_id, cache_dir, cancel_token) .await { Ok(archive_name) => Some(archive_name), @@ -517,14 +520,19 @@ impl ExecutionCache { self.record(&cache_key, execution_cache_key, &cache_value, cache_dir).await?; let upload_error = Arc::new(OnceLock::new()); - let url = match &cache_metadata.remote_cache { - Some(ResolvedRemoteCacheConfig { access: RemoteCacheAccess::ReadWrite, url }) => url, + let remote_config = match &cache_metadata.remote_cache { + Some( + remote_config @ ResolvedRemoteCacheConfig { + access: RemoteCacheAccess::ReadWrite, + .. + }, + ) => remote_config, Some(ResolvedRemoteCacheConfig { access: RemoteCacheAccess::Read, .. }) | None => { return Ok(upload_error); } }; match self.remote_clients.prepare_upload( - url, + remote_config, &cache_key, execution_cache_key, &cache_value, diff --git a/crates/vt/src/session/cache/remote.rs b/crates/vt/src/session/cache/remote.rs index fb7a87d32..b5df5fffe 100644 --- a/crates/vt/src/session/cache/remote.rs +++ b/crates/vt/src/session/cache/remote.rs @@ -24,8 +24,14 @@ use rustc_hash::FxHashMap; use tokio::sync::mpsc; use tokio_util::{sync::CancellationToken, task::TaskTracker}; use vt_path::AbsolutePath; -use vt_plan::cache_metadata::ExecutionCacheKey; -use vt_remote_cache::{Client, Download, Fetched}; +use vt_plan::{ + cache_metadata::ExecutionCacheKey, + remote_cache::{RemoteCacheAuth, ResolvedRemoteCacheConfig}, +}; +use vt_remote_cache::{ + Client, Download, Fetched, + auth::{Anonymous, Auth}, +}; use vt_str::Str; use wincode::{ SchemaWrite, @@ -130,35 +136,49 @@ pub(super) fn resolve( } } -/// Remote cache clients, each created when its endpoint is first used. +/// An endpoint and the auth that its requests use. Each has its own client. +#[derive(Debug, PartialEq, Eq, Hash)] +struct ClientKey { + url: Arc, + auth: RemoteCacheAuth, +} + +/// Remote cache clients, each created when its endpoint is first used with +/// its auth. #[derive(Debug, Default)] pub struct RemoteClients { - clients: Mutex, Arc>>, + clients: Mutex>>, } impl RemoteClients { - fn client(&self, endpoint: &Arc) -> Result, vt_remote_cache::Error> { + fn client( + &self, + remote_config: &ResolvedRemoteCacheConfig, + ) -> Result, vt_remote_cache::Error> { + let key = + ClientKey { url: Arc::clone(&remote_config.url), auth: remote_config.auth.clone() }; let mut clients = self.clients.lock().unwrap_or_else(PoisonError::into_inner); - if let Some(client) = clients.get(endpoint) { + if let Some(client) = clients.get(&key) { return Ok(Arc::clone(client)); } - let client = Arc::new(Client::new(endpoint)?); - clients.insert(Arc::clone(endpoint), Arc::clone(&client)); + let client = Arc::new(Client::new(&remote_config.url, build_auth(&remote_config.auth))?); + clients.insert(key, Arc::clone(&client)); drop(clients); Ok(client) } - /// Fetch the entry stored under `cache_key`, falling back to the entry - /// last stored for `execution_cache_key`. Returns `None` if neither key - /// matched. Stops when `cancel_token` is cancelled. + /// Fetch the entry stored under `cache_key` in the remote cache that + /// `remote_config` configures, falling back to the entry last stored for + /// `execution_cache_key`. Returns `None` if neither key matched. Stops + /// when `cancel_token` is cancelled. pub(super) async fn fetch( &self, - endpoint: &Arc, + remote_config: &ResolvedRemoteCacheConfig, cache_key: &CacheEntryKey, execution_cache_key: &ExecutionCacheKey, cancel_token: &CancellationToken, ) -> Result, ReadError> { - let client = self.client(endpoint).map_err(ReadError::Fetch)?; + let client = self.client(remote_config).map_err(ReadError::Fetch)?; let key = encode_key(cache_key)?; let secondary_key = encode_key(execution_cache_key)?; cancel_token @@ -168,18 +188,19 @@ impl RemoteClients { .map_err(ReadError::Fetch) } - /// Download the blob `blob_id` into `cache_dir`, checking that it decodes - /// as an output archive as it arrives. It's downloaded to a `.tmp` file, - /// which is renamed once the check passes and removed otherwise, such as - /// when `cancel_token` is cancelled. Returns the archive's file name. + /// Download the blob `blob_id` from the remote cache that `remote_config` + /// configures into `cache_dir`, checking that it decodes as an output + /// archive as it arrives. It's downloaded to a `.tmp` file, which is + /// renamed once the check passes and removed otherwise, such as when + /// `cancel_token` is cancelled. Returns the archive's file name. pub(super) async fn download_archive( &self, - endpoint: &Arc, + remote_config: &ResolvedRemoteCacheConfig, blob_id: &str, cache_dir: &AbsolutePath, cancel_token: &CancellationToken, ) -> Result { - let client = self.client(endpoint).map_err(ReadError::Download)?; + let client = self.client(remote_config).map_err(ReadError::Download)?; let archive_name = vt_str::format!("{}.tar.zst", uuid::Uuid::new_v4()); let archive_path = cache_dir.join(archive_name.as_str()); let temp_path = cache_dir.join(vt_str::format!("{archive_name}.tmp").as_str()); @@ -196,19 +217,20 @@ impl RemoteClients { } /// Prepare to upload an entry that was just recorded locally, along with - /// its output archive in `cache_dir`: get the endpoint's client and encode - /// the entry. Nothing is sent until the returned future is polled, so an - /// invalid endpoint or an entry that doesn't encode fails here. The future - /// owns everything the upload needs, so it can run in a spawned task. + /// its output archive in `cache_dir`, to the remote cache that + /// `remote_config` configures: get its client and encode the entry. + /// Nothing is sent until the returned future is polled, so an invalid + /// endpoint or an entry that doesn't encode fails here. The future owns + /// everything the upload needs, so it can run in a spawned task. pub(super) fn prepare_upload( &self, - endpoint: &Arc, + remote_config: &ResolvedRemoteCacheConfig, cache_key: &CacheEntryKey, execution_cache_key: &ExecutionCacheKey, cache_value: &CacheEntryValue, cache_dir: &AbsolutePath, ) -> Result> + Send + use<>, UploadError> { - let client = self.client(endpoint)?; + let client = self.client(remote_config)?; let key = encode_key(cache_key)?; let secondary_key = encode_key(execution_cache_key)?; let value = serialize_cache(cache_value)?; @@ -220,6 +242,13 @@ impl RemoteClients { } } +/// The credentials that requests carry for `auth`. +fn build_auth(auth: &RemoteCacheAuth) -> Arc { + match auth { + RemoteCacheAuth::Anonymous => Arc::new(Anonymous), + } +} + /// Uploads running in the background. Each keeps running after its task /// finishes, until [`Self::wait`] waits for all of them. #[derive(Debug, Default)] @@ -395,7 +424,10 @@ mod tests { use tokio::sync::oneshot; use vt_graph::config::ResolvedGlobConfig; use vt_path::{AbsolutePathBuf, RelativePathBuf}; - use vt_plan::cache_metadata::{EnvValueHash, SpawnFingerprint}; + use vt_plan::{ + cache_metadata::{EnvValueHash, SpawnFingerprint}, + remote_cache::RemoteCacheAccess, + }; use super::*; use crate::session::{ @@ -604,14 +636,25 @@ mod tests { } } + /// A `read-write` remote cache at `url` without credentials. + fn anonymous(url: &str) -> ResolvedRemoteCacheConfig { + ResolvedRemoteCacheConfig { + access: RemoteCacheAccess::ReadWrite, + url: Arc::from(url), + auth: RemoteCacheAuth::Anonymous, + } + } + /// Serve one request on a loopback endpoint: once the request head /// arrives, write `response`, then send nothing more and keep the - /// connection open until the client closes it. Returns the endpoint and a - /// receiver that resolves once `response` is written. - fn serve_stalled(response: &'static [u8]) -> (Arc, oneshot::Receiver<()>) { + /// connection open until the client closes it. Returns the remote cache + /// there and a receiver that resolves once `response` is written. + fn serve_stalled( + response: &'static [u8], + ) -> (ResolvedRemoteCacheConfig, oneshot::Receiver<()>) { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); - let endpoint = - Arc::from(vt_str::format!("http://{}/projects/test", listener.local_addr().unwrap())); + let remote_config = + anonymous(&vt_str::format!("http://{}/projects/test", listener.local_addr().unwrap())); let (responded_sender, responded) = oneshot::channel(); std::thread::spawn(move || { let (mut stream, _) = listener.accept().unwrap(); @@ -626,20 +669,20 @@ mod tests { let _ = responded_sender.send(()); while stream.read(&mut buf).is_ok_and(|n| n > 0) {} }); - (endpoint, responded) + (remote_config, responded) } #[tokio::test] async fn failed_archive_check_stops_the_download() { // The response announces more than it sends, so only the failed check // can end the download. - let (endpoint, _) = + let (remote_config, _) = serve_stalled(b"HTTP/1.1 200 OK\r\ncontent-length: 1000\r\n\r\nnot an archive"); let dir = tempfile::tempdir().unwrap(); let cache_dir = AbsolutePathBuf::new(dir.path().to_path_buf()).unwrap(); let error = RemoteClients::default() - .download_archive(&endpoint, "1", &cache_dir, &CancellationToken::new()) + .download_archive(&remote_config, "1", &cache_dir, &CancellationToken::new()) .await .unwrap_err(); assert!(matches!(error, ReadError::CorruptArchive(_)), "{error:?}"); @@ -648,13 +691,13 @@ mod tests { #[tokio::test] async fn cancelling_stops_a_fetch() { - let (endpoint, requested) = serve_stalled(b""); + let (remote_config, requested) = serve_stalled(b""); let cancel_token = CancellationToken::new(); let key = cache_key(ResolvedGlobConfig::default_auto()); let execution_key = ExecutionCacheKey::ExecAPI(Arc::from([])); let clients = RemoteClients::default(); - let fetch = clients.fetch(&endpoint, &key, &execution_key, &cancel_token); + let fetch = clients.fetch(&remote_config, &key, &execution_key, &cancel_token); let (fetched, ()) = tokio::join!(fetch, async { requested.await.unwrap(); cancel_token.cancel(); @@ -666,14 +709,14 @@ mod tests { async fn cancelling_stops_a_download_and_removes_it() { // The response announces a body that never arrives, so only // cancelling can end the download. - let (endpoint, responded) = + let (remote_config, responded) = serve_stalled(b"HTTP/1.1 200 OK\r\ncontent-length: 1000\r\n\r\n"); let cancel_token = CancellationToken::new(); let dir = tempfile::tempdir().unwrap(); let cache_dir = AbsolutePathBuf::new(dir.path().to_path_buf()).unwrap(); let clients = RemoteClients::default(); - let download = clients.download_archive(&endpoint, "1", &cache_dir, &cancel_token); + let download = clients.download_archive(&remote_config, "1", &cache_dir, &cancel_token); let (downloaded, ()) = tokio::join!(download, async { responded.await.unwrap(); // Cancel once the `.tmp` file exists, so the download has started @@ -687,16 +730,17 @@ mod tests { assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 0); } - /// Prepare to upload an entry without an output archive to `endpoint`. + /// Prepare to upload an entry without an output archive to the remote + /// cache that `remote_config` configures. fn prepare_upload( clients: &RemoteClients, - endpoint: &Arc, + remote_config: &ResolvedRemoteCacheConfig, ) -> Result> + use<>, UploadError> { let key = cache_key(ResolvedGlobConfig::default_auto()); let execution_key = ExecutionCacheKey::ExecAPI(Arc::from([])); let value = CacheEntryValue { output_archive: None, ..cache_value() }; let cache_dir = vt_path::current_dir().unwrap(); - clients.prepare_upload(endpoint, &key, &execution_key, &value, &cache_dir) + clients.prepare_upload(remote_config, &key, &execution_key, &value, &cache_dir) } fn upload_error(error: &OnceLock) -> Option { @@ -705,8 +749,8 @@ mod tests { #[test] fn upload_to_an_invalid_endpoint_fails_before_it_starts() { - let endpoint = Arc::from("cache.example/projects/test"); - let Err(error) = prepare_upload(&RemoteClients::default(), &endpoint) else { + let remote_config = anonymous("cache.example/projects/test"); + let Err(error) = prepare_upload(&RemoteClients::default(), &remote_config) else { panic!("an invalid endpoint should fail"); }; assert!( @@ -720,15 +764,15 @@ mod tests { let (error_status, _) = serve_stalled(b"HTTP/1.1 500 Internal Server Error\r\ncontent-length: 0\r\n\r\n"); // Nothing can listen on port 0. - let unreachable = Arc::from("http://127.0.0.1:0/projects/test"); + let unreachable = anonymous("http://127.0.0.1:0/projects/test"); let clients = RemoteClients::default(); let uploads = RemoteUploads::default(); - for (endpoint, message) in + for (remote_config, message) in [(error_status, "HTTP status 500"), (unreachable, "network error")] { let error = Arc::new(OnceLock::new()); - uploads.spawn(prepare_upload(&clients, &endpoint).unwrap(), Arc::clone(&error)); + uploads.spawn(prepare_upload(&clients, &remote_config).unwrap(), Arc::clone(&error)); uploads.wait(&CancellationToken::new()).await; assert_eq!(uploads.pending(), 0); assert_eq!(upload_error(&error).as_deref(), Some(message)); @@ -737,11 +781,11 @@ mod tests { #[tokio::test] async fn interrupting_the_wait_cancels_the_uploads() { - let (endpoint, requested) = serve_stalled(b""); + let (remote_config, requested) = serve_stalled(b""); let clients = RemoteClients::default(); let uploads = RemoteUploads::default(); let error = Arc::new(OnceLock::new()); - uploads.spawn(prepare_upload(&clients, &endpoint).unwrap(), Arc::clone(&error)); + uploads.spawn(prepare_upload(&clients, &remote_config).unwrap(), Arc::clone(&error)); requested.await.unwrap(); assert_eq!(uploads.pending(), 1); diff --git a/crates/vt_plan/src/remote_cache.rs b/crates/vt_plan/src/remote_cache.rs index d02a78808..8dcff2da6 100644 --- a/crates/vt_plan/src/remote_cache.rs +++ b/crates/vt_plan/src/remote_cache.rs @@ -1,5 +1,5 @@ //! Remote cache settings: the mode requested with `--remote-cache` or -//! `VP_REMOTE_CACHE`, and the access resolved from it for each `vp run` level. +//! `VP_REMOTE_CACHE`, and the access and auth resolved for each `vp run` level. use std::{ffi::OsStr, sync::Arc}; @@ -32,13 +32,23 @@ impl RemoteCacheMode { } } -/// Remote cache access and endpoint resolved for a `vp run` level from the -/// requested mode, `VP_REMOTE_CACHE_URL`, and `cache.remote.url`. +/// Remote cache access, endpoint, and auth resolved for a `vp run` level from +/// the requested mode, `VP_REMOTE_CACHE_URL`, and `cache.remote.url`. #[derive(Debug, Clone, Serialize)] pub struct ResolvedRemoteCacheConfig { pub access: RemoteCacheAccess, /// Endpoint as configured. It is validated when the remote cache is used. pub url: Arc, + pub auth: RemoteCacheAuth, +} + +/// How requests to the remote cache authenticate. It holds everything needed +/// to build the credentials, so nothing reads envs after planning. +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)] +#[serde(tag = "kind", rename_all = "kebab-case")] +pub enum RemoteCacheAuth { + /// Requests carry no credentials. + Anonymous, } /// Remote cache access after resolution. `off` resolves to no remote cache. @@ -80,18 +90,15 @@ pub(crate) fn resolve( None => configured_url.filter(|url| !url.is_empty()).cloned(), }; - match (mode, url) { - (Some(RemoteCacheMode::Off), _) | (None, None) => Ok(None), - (Some(RemoteCacheMode::Read) | None, Some(url)) => { - Ok(Some(ResolvedRemoteCacheConfig { access: RemoteCacheAccess::Read, url })) - } - (Some(RemoteCacheMode::ReadWrite), Some(url)) => { - Ok(Some(ResolvedRemoteCacheConfig { access: RemoteCacheAccess::ReadWrite, url })) - } + let (access, url) = match (mode, url) { + (Some(RemoteCacheMode::Off), _) | (None, None) => return Ok(None), + (Some(RemoteCacheMode::Read) | None, Some(url)) => (RemoteCacheAccess::Read, url), + (Some(RemoteCacheMode::ReadWrite), Some(url)) => (RemoteCacheAccess::ReadWrite, url), (Some(RemoteCacheMode::Read | RemoteCacheMode::ReadWrite), None) => { - Err(Error::MissingRemoteCacheEndpoint) + return Err(Error::MissingRemoteCacheEndpoint); } - } + }; + Ok(Some(ResolvedRemoteCacheConfig { access, url, auth: RemoteCacheAuth::Anonymous })) } #[cfg(test)] diff --git a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_cli_overrides_invalid_env.jsonc b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_cli_overrides_invalid_env.jsonc index 52053f1c5..9aba7e0fc 100644 --- a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_cli_overrides_invalid_env.jsonc +++ b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_cli_overrides_invalid_env.jsonc @@ -23,7 +23,10 @@ }, "remote_cache": { "access": "read-write", - "url": "https://cache.example/projects/test" + "url": "https://cache.example/projects/test", + "auth": { + "kind": "anonymous" + } } }, "spawn_command": { diff --git a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_controls_in_env_fingerprinted.jsonc b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_controls_in_env_fingerprinted.jsonc index 4d9a5ef95..503e9501f 100644 --- a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_controls_in_env_fingerprinted.jsonc +++ b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_controls_in_env_fingerprinted.jsonc @@ -26,7 +26,10 @@ }, "remote_cache": { "access": "read-write", - "url": "https://other.example/projects/test" + "url": "https://other.example/projects/test", + "auth": { + "kind": "anonymous" + } } }, "spawn_command": { diff --git a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_default_read.jsonc b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_default_read.jsonc index 4e74f8cde..9d5bbc8ff 100644 --- a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_default_read.jsonc +++ b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_default_read.jsonc @@ -23,7 +23,10 @@ }, "remote_cache": { "access": "read", - "url": "https://cache.example/projects/test" + "url": "https://cache.example/projects/test", + "auth": { + "kind": "anonymous" + } } }, "spawn_command": { diff --git a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_empty_env_endpoint_uses_config.jsonc b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_empty_env_endpoint_uses_config.jsonc index e5e06b568..129a07ffe 100644 --- a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_empty_env_endpoint_uses_config.jsonc +++ b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_empty_env_endpoint_uses_config.jsonc @@ -23,7 +23,10 @@ }, "remote_cache": { "access": "read", - "url": "https://cache.example/projects/test" + "url": "https://cache.example/projects/test", + "auth": { + "kind": "anonymous" + } } }, "spawn_command": { diff --git a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_nested_cli_overrides_env.jsonc b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_nested_cli_overrides_env.jsonc index c4507f82a..ed2a3d6a2 100644 --- a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_nested_cli_overrides_env.jsonc +++ b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_nested_cli_overrides_env.jsonc @@ -34,7 +34,10 @@ }, "remote_cache": { "access": "read", - "url": "https://cache.example/projects/test" + "url": "https://cache.example/projects/test", + "auth": { + "kind": "anonymous" + } } }, "spawn_command": { diff --git a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_prefix_controls_fingerprinted.jsonc b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_prefix_controls_fingerprinted.jsonc index c01626269..d17a6b462 100644 --- a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_prefix_controls_fingerprinted.jsonc +++ b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_prefix_controls_fingerprinted.jsonc @@ -26,7 +26,10 @@ }, "remote_cache": { "access": "read", - "url": "https://cache.example/projects/test" + "url": "https://cache.example/projects/test", + "auth": { + "kind": "anonymous" + } } }, "spawn_command": { diff --git a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_sibling_endpoint_isolation.jsonc b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_sibling_endpoint_isolation.jsonc index 2d495235c..e04f8d187 100644 --- a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_sibling_endpoint_isolation.jsonc +++ b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_sibling_endpoint_isolation.jsonc @@ -34,7 +34,10 @@ }, "remote_cache": { "access": "read", - "url": "https://other.example/projects/nested" + "url": "https://other.example/projects/nested", + "auth": { + "kind": "anonymous" + } } }, "spawn_command": { @@ -79,7 +82,10 @@ }, "remote_cache": { "access": "read", - "url": "https://cache.example/projects/test" + "url": "https://cache.example/projects/test", + "auth": { + "kind": "anonymous" + } } }, "spawn_command": { diff --git a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_sibling_mode_isolation.jsonc b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_sibling_mode_isolation.jsonc index 34b06f5f9..0ec2ee663 100644 --- a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_sibling_mode_isolation.jsonc +++ b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_config/snapshots/query_sibling_mode_isolation.jsonc @@ -76,7 +76,10 @@ }, "remote_cache": { "access": "read", - "url": "https://cache.example/projects/test" + "url": "https://cache.example/projects/test", + "auth": { + "kind": "anonymous" + } } }, "spawn_command": { diff --git a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_without_endpoint/snapshots/query_endpoint_not_validated.jsonc b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_without_endpoint/snapshots/query_endpoint_not_validated.jsonc index 9331bbe09..7a887eecd 100644 --- a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_without_endpoint/snapshots/query_endpoint_not_validated.jsonc +++ b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_without_endpoint/snapshots/query_endpoint_not_validated.jsonc @@ -15,7 +15,10 @@ "cache_metadata": { "remote_cache": { "access": "read", - "url": "cache.example/projects/test" + "url": "cache.example/projects/test", + "auth": { + "kind": "anonymous" + } } } } diff --git a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_without_endpoint/snapshots/query_nested_endpoint_enables_read.jsonc b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_without_endpoint/snapshots/query_nested_endpoint_enables_read.jsonc index b95a400c9..e98507030 100644 --- a/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_without_endpoint/snapshots/query_nested_endpoint_enables_read.jsonc +++ b/crates/vt_plan/tests/plan_snapshots/fixtures/remote_cache_without_endpoint/snapshots/query_nested_endpoint_enables_read.jsonc @@ -26,7 +26,10 @@ "cache_metadata": { "remote_cache": { "access": "read", - "url": "https://cache.example/projects/inner" + "url": "https://cache.example/projects/inner", + "auth": { + "kind": "anonymous" + } } } } diff --git a/crates/vt_remote_cache/README.md b/crates/vt_remote_cache/README.md index 632729b08..60e4d6e3a 100644 --- a/crates/vt_remote_cache/README.md +++ b/crates/vt_remote_cache/README.md @@ -4,6 +4,8 @@ Client for the [remote cache server API](https://github.com/voidzero-dev/vite-ta `Client::new` takes the configured endpoint, which can include a namespace path, such as `https://cache.example.com/projects/my-project`. Each operation appends its route to that path, so a store goes to `https://cache.example.com/projects/my-project/store`. Endpoints that aren't HTTP or HTTPS URLs are rejected. +`Client::new` also takes an `Auth`, which supplies the headers that authenticate each request. It's asked before every request with the request's operation (fetch, download, or store), so it can authenticate some operations and not others. It can use the client's HTTP client to get its credentials, such as a token. If it fails, the request isn't sent, and the operation fails with `Error::Auth`. `Anonymous` adds no headers. + `Client::fetch` sends a key and a secondary key as a CBOR map of byte strings. A 404 response means neither key matched, so `fetch` returns `None` without decoding the body. The body of a 200 response is a CBOR map tagged by `kind`, and `Fetched` decodes only two kinds: `exact`, an exact match with its value and blob ID, and `fallback`, a fallback match with the key it's stored under. Any other kind is a malformed response. An exact match must include `blob_id`, which is null when there's no blob. The fallback's value and blob ID aren't decoded. `Client::download` gets a blob by its ID. After a 200 response, it returns a `Download`, from which the caller reads the blob's chunks as they arrive. @@ -14,4 +16,4 @@ Only HTTP 200 counts as success for every operation, except that a fetch also ac reqwest configures TLS. It uses the process's default rustls crypto provider, which the client installs as ring unless one is already installed, and verifies certificates with the operating system's verifier. HTTPS requests use HTTP/2 if the server accepts it during the TLS handshake and HTTP/1.1 otherwise. HTTP endpoints always use HTTP/1.1. Requests go through the proxy set in `HTTPS_PROXY`, `HTTP_PROXY`, or `ALL_PROXY`, except for hosts in `NO_PROXY`. Without those variables, the system proxy settings are used on macOS and Windows. Connections time out after 10 seconds. Reads time out after 60 seconds, and until the response headers arrive, that limit also covers sending the request. -`Error` names the kind of failure: an invalid endpoint, a client that couldn't be created, a blob file that couldn't be read, a network error (including timeouts and responses that end early), a status other than 200 (or 404, for a fetch), or a malformed fetch response. Its messages contain no OS-specific details, so they can be shown to users as is. The details are in the source: the underlying error, the parse error for an endpoint that isn't a URL, or the message in an error response's body. Network errors leave out the request URL, since the endpoint may contain credentials, such as a token in its query. +`Error` names the kind of failure: an invalid endpoint, a client that couldn't be created, a blob file that couldn't be read, credentials that couldn't be obtained, a network error (including timeouts and responses that end early), a status other than 200 (or 404, for a fetch), or a malformed fetch response. Its messages contain no OS-specific details, so they can be shown to users as is. The details are in the source: the underlying error, the parse error for an endpoint that isn't a URL, or the message in an error response's body. Network errors leave out the request URL, since the endpoint may contain credentials, such as a token in its query. diff --git a/crates/vt_remote_cache/src/auth/mod.rs b/crates/vt_remote_cache/src/auth/mod.rs new file mode 100644 index 000000000..c86ee222b --- /dev/null +++ b/crates/vt_remote_cache/src/auth/mod.rs @@ -0,0 +1,42 @@ +//! Credentials for remote cache requests. An [`Auth`] supplies the headers +//! that each request carries. + +use std::{error::Error as StdError, fmt, pin::Pin, sync::Arc}; + +use reqwest::header::HeaderMap; + +/// The operation a request performs. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Operation { + /// `POST {endpoint}/fetch`. + Fetch, + /// `GET {endpoint}/blob/{blob_id}`. + Download, + /// `POST {endpoint}/store`. + Store, +} + +/// Why an [`Auth`] couldn't supply credentials. +pub type AuthError = Arc; + +/// The headers that an [`Auth`] supplies for a request, once they're ready. +pub type AuthHeaders<'a> = Pin> + Send + 'a>>; + +/// Supplies the headers that authenticate requests to an endpoint. +pub trait Auth: fmt::Debug + Send + Sync { + /// The headers to add to a request for `operation`. The request is sent + /// once they're ready, and isn't sent if this fails. `http` is the + /// client's HTTP client, for any requests needed to get the credentials. + /// It doesn't follow redirects. + fn headers<'a>(&'a self, operation: Operation, http: &'a reqwest::Client) -> AuthHeaders<'a>; +} + +/// Requests carry no credentials. +#[derive(Debug)] +pub struct Anonymous; + +impl Auth for Anonymous { + fn headers<'a>(&'a self, _: Operation, _: &'a reqwest::Client) -> AuthHeaders<'a> { + Box::pin(std::future::ready(Ok(HeaderMap::new()))) + } +} diff --git a/crates/vt_remote_cache/src/lib.rs b/crates/vt_remote_cache/src/lib.rs index be6006fab..90349ab74 100644 --- a/crates/vt_remote_cache/src/lib.rs +++ b/crates/vt_remote_cache/src/lib.rs @@ -1,12 +1,14 @@ //! Client for the remote cache server API. Keys, values, and blobs are opaque //! bytes; the caller decides what they contain. -use std::time::Duration; +pub mod auth; + +use std::{sync::Arc, time::Duration}; use bytes::Bytes; use reqwest::{ Response, StatusCode, - header::CONTENT_TYPE, + header::{CONTENT_TYPE, HeaderMap}, multipart::{Form, Part}, }; use serde::{Deserialize, Serialize}; @@ -14,6 +16,8 @@ use url::{ParseError, Url}; use vt_path::AbsolutePath; use vt_str::Str; +use crate::auth::{Auth, AuthError, Operation}; + /// Time allowed to establish a connection, including the TLS handshake. const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); @@ -37,6 +41,10 @@ pub enum Error { /// The blob file couldn't be opened. #[error("failed to read the blob")] ReadBlob(#[source] std::io::Error), + /// The request's credentials couldn't be obtained, so it wasn't sent. + /// The source says why. + #[error("failed to authenticate")] + Auth(#[source] AuthError), /// No complete response arrived, for example because the connection /// failed, timed out, or closed before the whole body arrived. The source /// leaves out the request URL, because the endpoint may contain @@ -124,6 +132,7 @@ pub enum Fetched { #[derive(Debug)] pub struct Client { http: reqwest::Client, + auth: Arc, fetch_url: Url, store_url: Url, /// `{endpoint}/blob`, to which each download appends a blob ID. @@ -133,12 +142,13 @@ pub struct Client { impl Client { /// Create a client for `endpoint`, a base URL that may include a /// namespace path, such as `https://cache.example.com/projects/my-project`. + /// Each request carries the headers that `auth` supplies for it. /// /// # Errors /// /// Returns [`Error::InvalidEndpoint`] if `endpoint` isn't a usable URL, or /// [`Error::HttpClient`] if the HTTP client can't be created. - pub fn new(endpoint: &str) -> Result { + pub fn new(endpoint: &str, auth: Arc) -> Result { let endpoint = parse_endpoint(endpoint)?; let fetch_url = route_url(&endpoint, "fetch")?; let store_url = route_url(&endpoint, "store")?; @@ -156,7 +166,11 @@ impl Client { .redirect(reqwest::redirect::Policy::none()) .build() .map_err(Error::HttpClient)?; - Ok(Self { http, fetch_url, store_url, blob_url }) + Ok(Self { http, auth, fetch_url, store_url, blob_url }) + } + + async fn auth_headers(&self, operation: Operation) -> Result { + self.auth.headers(operation, &self.http).await.map_err(Error::Auth) } /// Fetch the entry stored under `key` with `POST {endpoint}/fetch`, @@ -165,14 +179,16 @@ impl Client { /// /// # Errors /// - /// Returns an error if the request fails, the server responds with a - /// status other than 200 or 404, or the body of a 200 response isn't an - /// exact or fallback match. + /// Returns an error if the credentials can't be obtained, the request + /// fails, the server responds with a status other than 200 or 404, or the + /// body of a 200 response isn't an exact or fallback match. pub async fn fetch(&self, key: &[u8], secondary_key: &[u8]) -> Result, Error> { + let headers = self.auth_headers(Operation::Fetch).await?; let body = encode_cbor(&FetchRequest { key, secondary_key }); let response = self .http .post(self.fetch_url.clone()) + .headers(headers) .header(CONTENT_TYPE, "application/cbor") .body(body) .send() @@ -194,12 +210,13 @@ impl Client { /// /// # Errors /// - /// Returns an error if the request fails or the server responds with a - /// status other than 200. + /// Returns an error if the credentials can't be obtained, the request + /// fails, or the server responds with a status other than 200. pub async fn download(&self, blob_id: &str) -> Result { let mut url = self.blob_url.clone(); url.path_segments_mut().map_err(|()| Error::InvalidEndpoint(None))?.push(blob_id); - let response = self.http.get(url).send().await.map_err(network_error)?; + let headers = self.auth_headers(Operation::Download).await?; + let response = self.http.get(url).headers(headers).send().await.map_err(network_error)?; Ok(Download { response: check_status(response).await? }) } @@ -209,8 +226,9 @@ impl Client { /// /// # Errors /// - /// Returns an error if the blob file can't be opened, the request fails, - /// or the server responds with a status other than 200. + /// Returns an error if the credentials can't be obtained, the blob file + /// can't be opened, the request fails, or the server responds with a + /// status other than 200. pub async fn store( &self, key: &[u8], @@ -218,6 +236,7 @@ impl Client { value: &[u8], blob: Option<&AbsolutePath>, ) -> Result<(), Error> { + let headers = self.auth_headers(Operation::Store).await?; let metadata = StoreMetadata { key, secondary_key, value }; let mut form = Form::new().part("metadata", metadata_part(&metadata)); if let Some(blob) = blob { @@ -226,6 +245,7 @@ impl Client { let response = self .http .post(self.store_url.clone()) + .headers(headers) .multipart(form) .send() .await @@ -299,9 +319,11 @@ mod tests { net::TcpListener, }; + use reqwest::header::{HeaderName, HeaderValue}; use vt_path::AbsolutePathBuf; use super::*; + use crate::auth::{Anonymous, AuthHeaders}; fn store_url(endpoint: &str) -> Result { route_url(&parse_endpoint(endpoint)?, "store") @@ -323,7 +345,8 @@ mod tests { fn rejects_endpoints_that_are_not_http_urls() { for endpoint in ["cache.example/projects/test", "ftp://cache.example", "mailto:a@b.example"] { - assert!(matches!(Client::new(endpoint), Err(Error::InvalidEndpoint(_))), "{endpoint}"); + let client = Client::new(endpoint, Arc::new(Anonymous)); + assert!(matches!(client, Err(Error::InvalidEndpoint(_))), "{endpoint}"); } } @@ -445,8 +468,12 @@ mod tests { } fn client_for(listener: &TcpListener) -> Client { + client_with_auth(listener, Arc::new(Anonymous)) + } + + fn client_with_auth(listener: &TcpListener, auth: Arc) -> Client { let port = listener.local_addr().unwrap().port(); - Client::new(&vt_str::format!("http://127.0.0.1:{port}/projects/test")).unwrap() + Client::new(&vt_str::format!("http://127.0.0.1:{port}/projects/test"), auth).unwrap() } /// The `metadata` part of a store request for key `k`, secondary key `s`, @@ -630,6 +657,91 @@ mod tests { server.join().unwrap(); } + /// Adds an `x-operation` header that names the request's operation. + #[derive(Debug)] + struct NamesOperation; + + impl Auth for NamesOperation { + fn headers<'a>(&'a self, operation: Operation, _: &'a reqwest::Client) -> AuthHeaders<'a> { + let name = match operation { + Operation::Fetch => "fetch", + Operation::Download => "download", + Operation::Store => "store", + }; + let headers = HeaderMap::from_iter([( + HeaderName::from_static("x-operation"), + HeaderValue::from_static(name), + )]); + Box::pin(std::future::ready(Ok(headers))) + } + } + + /// Run `send` with a client whose requests carry the headers from `auth`, + /// and return the request it sent. The server responds with 404. + async fn request_with_auth>( + auth: Arc, + send: impl FnOnce(Client) -> F, + ) -> Vec { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let client = client_with_auth(&listener, auth); + let server = + std::thread::spawn(move || serve_once(&listener, "HTTP/1.1 404 Not Found", b"")); + send(client).await; + server.join().unwrap() + } + + #[tokio::test] + async fn each_request_carries_the_auth_headers_for_its_operation() { + let auth: Arc = Arc::new(NamesOperation); + let fetch = request_with_auth(Arc::clone(&auth), |client| async move { + assert_eq!(client.fetch(b"k", b"s").await.unwrap(), None); + }) + .await; + assert!(contains(&fetch, b"x-operation: fetch\r\n")); + + let download = request_with_auth(Arc::clone(&auth), |client| async move { + client.download("7").await.unwrap_err(); + }) + .await; + assert!(contains(&download, b"x-operation: download\r\n")); + + let store = request_with_auth(auth, |client| async move { + client.store(b"k", b"s", b"v", None).await.unwrap_err(); + }) + .await; + assert!(contains(&store, b"x-operation: store\r\n")); + } + + /// Fails to supply credentials for every request. + #[derive(Debug)] + struct Unavailable; + + impl Auth for Unavailable { + fn headers<'a>(&'a self, _: Operation, _: &'a reqwest::Client) -> AuthHeaders<'a> { + let error: AuthError = Arc::new(std::io::Error::other("no credentials")); + Box::pin(std::future::ready(Err(error))) + } + } + + #[tokio::test] + async fn requests_without_credentials_are_not_sent() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let client = client_with_auth(&listener, Arc::new(Unavailable)); + + let errors = [ + client.fetch(b"k", b"s").await.unwrap_err(), + client.download("7").await.unwrap_err(), + client.store(b"k", b"s", b"v", None).await.unwrap_err(), + ]; + for error in errors { + assert!(matches!(error, Error::Auth(_)), "{error:?}"); + assert_eq!(error.to_string(), "failed to authenticate"); + assert_eq!(std::error::Error::source(&error).unwrap().to_string(), "no credentials"); + } + listener.set_nonblocking(true).unwrap(); + assert_eq!(listener.accept().unwrap_err().kind(), std::io::ErrorKind::WouldBlock); + } + /// Accept one connection and return the TLS record it starts with, which /// is the `ClientHello`, then close the connection without responding. fn read_client_hello(listener: &TcpListener) -> Vec { @@ -647,8 +759,11 @@ mod tests { async fn offers_http2_over_tls() { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let port = listener.local_addr().unwrap().port(); - let client = - Client::new(&vt_str::format!("https://127.0.0.1:{port}/projects/test")).unwrap(); + let client = Client::new( + &vt_str::format!("https://127.0.0.1:{port}/projects/test"), + Arc::new(Anonymous), + ) + .unwrap(); let server = std::thread::spawn(move || read_client_hello(&listener)); let error = client.fetch(b"k", b"s").await.unwrap_err(); @@ -660,8 +775,8 @@ mod tests { #[tokio::test] async fn network_errors_leave_out_the_url() { - let client = - Client::new("http://user:password@127.0.0.1:0/projects/test?token=secret").unwrap(); + let endpoint = "http://user:password@127.0.0.1:0/projects/test?token=secret"; + let client = Client::new(endpoint, Arc::new(Anonymous)).unwrap(); let error = client.fetch(b"k", b"s").await.unwrap_err(); assert!(matches!(error, Error::Network(_)), "{error:?}");