//! Cloudflare R2 implementation of [`ObjectStore`] over S3-compat SigV4.
//!
//! Uses `local_driver::s3_sign` helpers for signature computation and
//! `reqwest::blocking` for HTTP so the [`ObjectStore`] trait stays synchronous.
//! Async consumers wrap calls in `tokio::task::spawn_blocking`.
//!
//! ## Endpoint
//!
//! R2's S3-compat endpoint is `https://<account_id>.r2.cloudflarestorage.com`.
//! Region is always `"auto"`. Bucket lives in the URL path:
//! `https://<account_id>.r2.cloudflarestorage.com/<bucket>/<key>`.
//!
//! ## list_prefix
//!
//! Issues `GET /<bucket>?list-type=2&prefix=<encoded>` (ListObjectsV2) and
//! parses the `<Key>…</Key>` elements out of the XML body. Continues with
//! `&continuation-token=<…>` while `<IsTruncated>true</IsTruncated>` so
//! prefixes larger than the 1000-key page size return complete.
//!
//! @yah:relay(R630, "Object-store correctness + tooling gaps surfaced by standing up the cr.yah.dev registry")
//! @yah:at(2026-07-23T03:06:57Z)
//! @yah:status(open)
//! @yah:next("Both children were found while building yah-cr (the R2-backed OCI registry) on 2026-07-22 and are independent of that work — they are latent defects in shared object-store code that any caller can hit.")
//! @yah:next("Start with the SigV4 child: it is a correctness bug that fails closed but silently constrains every key namespace we can use. The bucket-delete child is additive and can follow.")
//! @yah:gotcha("The SigV4 defect is why cr.yah.dev stores OCI digests as sha256/<hex> instead of the natural sha256:<hex>. That workaround is load-bearing in two files that must stay in lockstep (app/yah/cli/src/cr.rs digest_key, app/yah/workers/yah-cr/src/index.ts digestKey). If the signing bug is fixed, those can be simplified — but only together, and only with a migration for keys already written.")
//! @arch:see(.yah/docs/working/W175-per-publisher-prefix.md)
//! @yah:handoff("Both children fixed and in review. Along the way two further latent defects in the same shared object-store code were found and fixed in-pass, both invisible to every existing test: (1) ObjectStore::delete signed DELETE with sign_s3_empty_body and so 403'd against R2 from the day it landed — which also means `yah cloud service prune` (static_asset_prune.rs) has never reclaimed anything; (2) the ListObjectsV2 parser returned raw XML text, so a key holding & came back as & — harmless before, because such keys couldn't be written, and a broken round-trip the moment B1 made them writable.")
//! @yah:gotcha("2026-08-25: the SigV4 defect is FIXED (R630-B1, in review), so the sha256/<hex> constraint this gotcha describes is lifted — but the workaround was deliberately left in place. Simplifying digest_key / digestKey to the natural sha256:<hex> means rewriting every blob key already in yah-cr-cache: a live-data migration on a running registry, operator-gated, not a drive-by. The two files still must stay in lockstep.")
use std::time::Duration;
use percent_encoding::{utf8_percent_encode, AsciiSet, NON_ALPHANUMERIC};
use reqwest::blocking::Client;
use reqwest::header::{HeaderValue, ETAG, IF_MATCH, IF_NONE_MATCH};
use reqwest::StatusCode;
use sha2::{Digest, Sha256};
use local_driver::s3_sign::{
sign_s3_get_with_query, sign_s3_no_body, sign_s3_put_object, sign_s3_put_object_with,
uri_encode_key, S3PutOptions,
};
use crate::{Error, ObjectStore, Precondition};
/// R2's S3-compat region. The endpoint always accepts `"auto"`.
const R2_REGION: &str = "auto";
/// Keystore slot for the R2 S3 access key id.
pub const R2_ACCESS_KEY_SLOT: &str = "cloudflare-r2-access-key-id";
/// Keystore slot for the R2 S3 secret key.
pub const R2_SECRET_KEY_SLOT: &str = "cloudflare-r2-secret-key";
/// Env var fallback for the R2 access key id.
pub const R2_ACCESS_KEY_ENV: &str = "CF_R2_ACCESS_KEY_ID";
/// Env var fallback for the R2 secret key.
pub const R2_SECRET_KEY_ENV: &str = "CF_R2_SECRET_KEY";
/// Percent-encoding set for query-string values. SigV4 requires
/// unreserved characters (A-Z a-z 0-9 - _ . ~) to remain literal;
/// everything else gets percent-encoded.
const QUERY_VALUE: &AsciiSet = &NON_ALPHANUMERIC
.remove(b'-')
.remove(b'_')
.remove(b'.')
.remove(b'~');
/// Default content-type for keys with no recognized extension.
const DEFAULT_CONTENT_TYPE: &str = "application/octet-stream";
/// Content-type for an object key, inferred from its file extension.
///
/// R2 stores whatever Content-Type we send on PUT and serves it back verbatim
/// (the CDN custom domain does no extension sniffing). An octet-stream default
/// makes browsers *download* html shells instead of rendering them, so we set
/// an explicit type for the extensions a static site actually ships. Unknown
/// or extensionless keys (pointers, the `_yah-manifest.json` sidecar is `.json`
/// and handled) fall back to [`DEFAULT_CONTENT_TYPE`].
fn content_type_for_key(key: &str) -> &'static str {
let ext = match key.rsplit_once('.') {
// A `.` in a directory segment is not an extension.
Some((_, e)) if !e.contains('/') => e,
_ => "",
};
match ext.to_ascii_lowercase().as_str() {
"html" | "htm" => "text/html; charset=utf-8",
"css" => "text/css; charset=utf-8",
"js" | "mjs" => "text/javascript; charset=utf-8",
"json" | "map" => "application/json",
"webmanifest" => "application/manifest+json",
"xml" => "application/xml",
"txt" => "text/plain; charset=utf-8",
"svg" => "image/svg+xml",
"webp" => "image/webp",
"png" => "image/png",
"jpg" | "jpeg" => "image/jpeg",
"gif" => "image/gif",
"avif" => "image/avif",
"ico" => "image/x-icon",
"woff2" => "font/woff2",
"woff" => "font/woff",
"ttf" => "font/ttf",
"otf" => "font/otf",
"wasm" => "application/wasm",
"pdf" => "application/pdf",
_ => DEFAULT_CONTENT_TYPE,
}
}
/// R2-backed [`ObjectStore`].
///
/// Construct with [`R2ObjectStore::new`] when keys are already in hand,
/// or [`R2ObjectStore::from_vault`] to pull them from the yah keystore
/// (with env-var fallback).
pub struct R2ObjectStore {
account_id: String,
bucket: String,
access_key: String,
secret_key: String,
/// Overrides the derived `https://<account_id>.r2.cloudflarestorage.com`.
/// See [`R2ObjectStore::with_endpoint`].
endpoint: Option<String>,
client: Option<Client>,
}
impl Drop for R2ObjectStore {
fn drop(&mut self) {
// `reqwest::blocking::Client` owns a background tokio runtime whose
// Drop panics with "Cannot drop a runtime in a context where blocking
// is not allowed" when the drop happens inside an async context. This
// fires when an `Arc<R2ObjectStore>` reaches zero from inside an
// awaited future (e.g. publish_to_r2). Detach the shutdown onto a
// fresh OS thread which has no tokio runtime context, so the client's
// Drop can shut its internal runtime down cleanly. Dep-neutral — this
// crate keeps its sync/tokio-free profile.
let Some(client) = self.client.take() else { return };
std::thread::spawn(move || drop(client));
}
}
impl R2ObjectStore {
/// Construct with explicit keys.
///
/// `account_id` is the Cloudflare account id (the subdomain in
/// `<account_id>.r2.cloudflarestorage.com`).
pub fn new(
account_id: impl Into<String>,
bucket: impl Into<String>,
access_key: impl Into<String>,
secret_key: impl Into<String>,
) -> Result<Self, Error> {
let client = Client::builder()
.timeout(Duration::from_secs(300))
.build()
.map_err(|e| Error::Backend(format!("reqwest client: {e}")))?;
Ok(Self {
account_id: account_id.into(),
bucket: bucket.into(),
access_key: access_key.into(),
secret_key: secret_key.into(),
endpoint: None,
client: Some(client),
})
}
/// Point this store at an S3-compatible endpoint other than R2 — in
/// practice, the pond tier's local MinIO (`http://127.0.0.1:9000`).
///
/// Everything else about the store is already endpoint-agnostic: the bucket
/// lives in the URL path (path-style addressing, which MinIO also speaks)
/// and SigV4 is signed against whatever host the URL names.
///
/// This exists because without it the pond rehearsal could not exercise the
/// *read* side of a publish at all. `publish_to_pond` uploads a directory
/// tree and offers no way to read an object back, so the one part of a
/// release that is a read-modify-write — the accumulating `index.json` that
/// https://yah.dev/releases renders from — was the one part a green local
/// rehearsal proved nothing about (R330-T32). A conditional-write loop that
/// has never run is a conditional-write loop you do not have.
///
/// The region stays `"auto"`; MinIO accepts it.
pub fn with_endpoint(mut self, endpoint: impl Into<String>) -> Self {
let endpoint = endpoint.into();
let trimmed = endpoint.trim_end_matches('/');
self.endpoint = (!trimmed.is_empty()).then(|| trimmed.to_string());
self
}
fn client(&self) -> &Client {
self.client
.as_ref()
.expect("client is Some until Drop takes it")
}
/// Construct from the yah keystore (vault), falling back to env vars.
///
/// Reads `cloudflare-r2-access-key-id` / `cloudflare-r2-secret-key` slots
/// (env fallback `CF_R2_ACCESS_KEY_ID` / `CF_R2_SECRET_KEY`). Returns
/// [`Error::Auth`] if either is missing.
pub fn from_vault(
account_id: impl Into<String>,
bucket: impl Into<String>,
) -> Result<Self, Error> {
let access_key = fob::get_or_env(R2_ACCESS_KEY_SLOT, R2_ACCESS_KEY_ENV)
.map_err(|e| Error::Auth(format!("vault read {R2_ACCESS_KEY_SLOT}: {e}")))?
.ok_or_else(|| {
Error::Auth(format!(
"missing R2 credential: set vault slot {R2_ACCESS_KEY_SLOT} or env {R2_ACCESS_KEY_ENV}"
))
})?;
let secret_key = fob::get_or_env(R2_SECRET_KEY_SLOT, R2_SECRET_KEY_ENV)
.map_err(|e| Error::Auth(format!("vault read {R2_SECRET_KEY_SLOT}: {e}")))?
.ok_or_else(|| {
Error::Auth(format!(
"missing R2 credential: set vault slot {R2_SECRET_KEY_SLOT} or env {R2_SECRET_KEY_ENV}"
))
})?;
Self::new(account_id, bucket, access_key, secret_key)
}
fn endpoint(&self) -> String {
match &self.endpoint {
Some(e) => e.clone(),
None => format!("https://{}.r2.cloudflarestorage.com", self.account_id),
}
}
/// R630-B1: the key is AWS-`UriEncode`d here, at the one place a key becomes
/// a URL. Doing it here rather than in the signer is what makes the wire
/// path and the signed path the same bytes — see [`uri_encode_key`]. Before
/// this, a key holding `:` (an OCI digest `sha256:<hex>`, a timestamp) went
/// out raw, R2 canonicalized it per spec, and every request 403'd
/// `SignatureDoesNotMatch`.
fn object_url(&self, key: &str) -> String {
format!("{}/{}/{}", self.endpoint(), self.bucket, uri_encode_key(key))
}
fn bucket_url(&self) -> String {
format!("{}/{}", self.endpoint(), self.bucket)
}
/// The one PUT path, with `Cache-Control` optional (R703-B8).
///
/// `put` and `put_cached` differ only in that header, so they share this
/// rather than each carrying their own signing + status handling — the
/// shape where one of two copies quietly stops matching the other.
fn put_inner(
&self,
key: &str,
data: Vec<u8>,
cache_control: Option<&str>,
) -> Result<(), Error> {
let url = self.object_url(key);
let body_sha256 = {
let mut h = Sha256::new();
h.update(&data);
hex::encode(h.finalize())
};
let headers = sign_s3_put_object_with(
&url,
&body_sha256,
data.len(),
R2_REGION,
&self.access_key,
&self.secret_key,
&S3PutOptions {
content_type: content_type_for_key(key),
// Generic object-store put — the BLAKE3 stamp is a static-asset
// catalog concern, not a property of every object (R546-B10).
blake3_meta: None,
cache_control,
},
)
.map_err(|e| Error::Backend(format!("sign PUT {key}: {e}")))?;
let resp = self
.client()
.put(&url)
.headers(headers)
.body(data)
.send()
.map_err(|e| io_err(&format!("PUT {key}"), e))?;
check_status(resp, "PUT", key)
}
}
/// Convert a reqwest error into our generic [`Error`].
fn io_err(ctx: &str, e: impl std::fmt::Display) -> Error {
Error::Io(format!("{ctx}: {e}"))
}
impl ObjectStore for R2ObjectStore {
fn locate(&self, key: &str) -> String {
self.object_url(key)
}
fn put(&self, key: &str, data: Vec<u8>) -> Result<(), Error> {
self.put_inner(key, data, None)
}
fn put_cached(&self, key: &str, data: Vec<u8>, cache_control: &str) -> Result<(), Error> {
self.put_inner(key, data, Some(cache_control))
}
fn get(&self, key: &str) -> Result<Option<Vec<u8>>, Error> {
let url = self.object_url(key);
// GET has no body: reqwest drops the `content-length: 0` header on the
// wire, so signing it (as `sign_s3_empty_body` does) yields a signature
// the server can't reproduce → 403 SignatureDoesNotMatch. Sign with the
// content-length-free helper instead, exactly like ListObjectsV2. The
// empty query string is correct for a plain object GET.
let headers = sign_s3_get_with_query(
&url,
"",
R2_REGION,
&self.access_key,
&self.secret_key,
)
.map_err(|e| Error::Backend(format!("sign GET {key}: {e}")))?;
let resp = self
.client()
.get(&url)
.headers(headers)
.send()
.map_err(|e| io_err(&format!("GET {key}"), e))?;
match resp.status() {
StatusCode::OK => {
let bytes = resp
.bytes()
.map_err(|e| io_err(&format!("read GET {key}"), e))?;
Ok(Some(bytes.to_vec()))
}
StatusCode::NOT_FOUND => Ok(None),
s => Err(status_err("GET", key, s, resp.text().ok())),
}
}
fn head(&self, key: &str) -> Result<bool, Error> {
let url = self.object_url(key);
// HEAD is body-less like GET: sign without content-length (see `get`).
let headers = sign_s3_no_body(
"HEAD",
&url,
"",
R2_REGION,
&self.access_key,
&self.secret_key,
)
.map_err(|e| Error::Backend(format!("sign HEAD {key}: {e}")))?;
let resp = self
.client()
.head(&url)
.headers(headers)
.send()
.map_err(|e| io_err(&format!("HEAD {key}"), e))?;
match resp.status() {
StatusCode::OK => Ok(true),
StatusCode::NOT_FOUND => Ok(false),
s => Err(status_err("HEAD", key, s, None)),
}
}
fn delete(&self, key: &str) -> Result<(), Error> {
let url = self.object_url(key);
// R630-F2: DELETE is body-less, so it signs like GET/HEAD and NOT with
// `sign_s3_empty_body` — reqwest strips the `content-length: 0` that
// signer puts in the canonical headers, R2 recomputes a different
// signature, and every DELETE comes back 403 SignatureDoesNotMatch.
// This code had never been run against live R2 (the trait tests use the
// in-memory store), so `ObjectStore::delete` was 100% broken on R2 from
// the day it landed. Confirmed live 2026-08-25, before and after.
let headers = sign_s3_no_body(
"DELETE",
&url,
"",
R2_REGION,
&self.access_key,
&self.secret_key,
)
.map_err(|e| Error::Backend(format!("sign DELETE {key}: {e}")))?;
let resp = self
.client()
.delete(&url)
.headers(headers)
.send()
.map_err(|e| io_err(&format!("DELETE {key}"), e))?;
match resp.status() {
// S3 DELETE on a missing key returns 204 too — both are success
// semantics for an idempotent delete.
StatusCode::OK | StatusCode::NO_CONTENT | StatusCode::NOT_FOUND => Ok(()),
s => Err(status_err("DELETE", key, s, resp.text().ok())),
}
}
fn list_prefix(&self, prefix: &str) -> Result<Vec<String>, Error> {
Ok(self
.list_prefix_detailed(prefix)?
.into_iter()
.map(|m| m.key)
.collect())
}
fn put_if(&self, key: &str, data: Vec<u8>, cond: Precondition) -> Result<String, Error> {
let url = self.object_url(key);
let body_sha256 = {
let mut h = Sha256::new();
h.update(&data);
hex::encode(h.finalize())
};
// Sign the same fixed header set as an unconditional PUT. The conditional
// header (If-Match / If-None-Match) is added *unsigned* afterwards: SigV4
// only covers the headers in `SignedHeaders`, and S3/R2 honor extra
// unsigned headers — so the precondition is enforced server-side without
// touching the signer.
let mut headers = sign_s3_put_object(
&url,
&body_sha256,
content_type_for_key(key),
data.len(),
R2_REGION,
&self.access_key,
&self.secret_key,
None,
)
.map_err(|e| Error::Backend(format!("sign PUT {key}: {e}")))?;
match &cond {
Precondition::IfAbsent => {
headers.insert(IF_NONE_MATCH, HeaderValue::from_static("*"));
}
Precondition::IfMatch(etag) => {
let v = HeaderValue::from_str(etag)
.map_err(|e| Error::Backend(format!("invalid If-Match etag {etag:?}: {e}")))?;
headers.insert(IF_MATCH, v);
}
}
let resp = self
.client()
.put(&url)
.headers(headers)
.body(data)
.send()
.map_err(|e| io_err(&format!("PUT(if) {key}"), e))?;
let status = resp.status();
if status == StatusCode::PRECONDITION_FAILED {
return Err(Error::PreconditionFailed(format!(
"put_if {key}: precondition not met ({cond:?})"
)));
}
if !status.is_success() {
return Err(status_err("PUT(if)", key, status, resp.text().ok()));
}
// Prefer the ETag echoed in the PUT response; fall back to a HEAD if a
// backend ever omits it (R2 always returns it).
match resp.headers().get(ETAG).and_then(|v| v.to_str().ok()) {
Some(e) => Ok(e.to_string()),
None => self
.etag(key)?
.ok_or_else(|| Error::Backend(format!("PUT(if) {key} returned no ETag"))),
}
}
fn etag(&self, key: &str) -> Result<Option<String>, Error> {
let url = self.object_url(key);
// HEAD is body-less: sign without content-length (see `head`).
let headers = sign_s3_no_body(
"HEAD",
&url,
"",
R2_REGION,
&self.access_key,
&self.secret_key,
)
.map_err(|e| Error::Backend(format!("sign HEAD {key}: {e}")))?;
let resp = self
.client()
.head(&url)
.headers(headers)
.send()
.map_err(|e| io_err(&format!("HEAD(etag) {key}"), e))?;
match resp.status() {
StatusCode::OK => Ok(resp
.headers()
.get(ETAG)
.and_then(|v| v.to_str().ok())
.map(|s| s.to_string())),
StatusCode::NOT_FOUND => Ok(None),
s => Err(status_err("HEAD(etag)", key, s, None)),
}
}
}
/// One `<Contents>` entry from an R2 `ListObjectsV2` response.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ObjectMeta {
/// Object key (full path including any prefix).
pub key: String,
/// Object size in bytes.
pub size: u64,
/// Last-modified timestamp in ISO-8601 / RFC-3339 (R2's `<LastModified>` value).
pub last_modified: String,
}
impl R2ObjectStore {
/// List objects under `prefix` returning key + size + last-modified.
///
/// Same paginated request as [`ObjectStore::list_prefix`] but parses the
/// `<Size>` and `<LastModified>` siblings of each `<Key>` element. Used by
/// the data-tab bucket viewer to render a directory-style listing.
pub fn list_prefix_detailed(&self, prefix: &str) -> Result<Vec<ObjectMeta>, Error> {
let mut entries = Vec::new();
let mut continuation_token: Option<String> = None;
let bucket_url = self.bucket_url();
let encoded_prefix = utf8_percent_encode(prefix, QUERY_VALUE).to_string();
loop {
// Canonical query MUST be sorted by parameter name (SigV4).
// Parameters: continuation-token (optional), list-type, prefix.
let mut params: Vec<(String, String)> =
vec![("list-type".to_string(), "2".to_string())];
if let Some(token) = &continuation_token {
let encoded = utf8_percent_encode(token, QUERY_VALUE).to_string();
params.push(("continuation-token".to_string(), encoded));
}
params.push(("prefix".to_string(), encoded_prefix.clone()));
params.sort_by(|a, b| a.0.cmp(&b.0));
let canonical_query = params
.iter()
.map(|(k, v)| format!("{k}={v}"))
.collect::<Vec<_>>()
.join("&");
let url_with_query = format!("{bucket_url}?{canonical_query}");
let headers = sign_s3_get_with_query(
&bucket_url,
&canonical_query,
R2_REGION,
&self.access_key,
&self.secret_key,
)
.map_err(|e| Error::Backend(format!("sign LIST {prefix}: {e}")))?;
let resp = self
.client()
.get(&url_with_query)
.headers(headers)
.send()
.map_err(|e| io_err(&format!("LIST {prefix}"), e))?;
if !resp.status().is_success() {
return Err(status_err("LIST", prefix, resp.status(), resp.text().ok()));
}
let body = resp
.text()
.map_err(|e| io_err(&format!("LIST {prefix} body"), e))?;
let (page_entries, next_token) = parse_list_v2_detailed(&body);
entries.extend(page_entries);
if let Some(t) = next_token {
continuation_token = Some(t);
} else {
break;
}
}
Ok(entries)
}
}
fn check_status(resp: reqwest::blocking::Response, verb: &str, key: &str) -> Result<(), Error> {
if resp.status().is_success() {
Ok(())
} else {
let status = resp.status();
let body = resp.text().ok();
Err(status_err(verb, key, status, body))
}
}
fn status_err(verb: &str, key: &str, status: StatusCode, body: Option<String>) -> Error {
let snippet = body
.as_deref()
.map(|s| s.chars().take(200).collect::<String>())
.unwrap_or_default();
let msg = format!("{verb} {key} → {status} {snippet}");
match status {
StatusCode::FORBIDDEN | StatusCode::UNAUTHORIZED => Error::Auth(msg),
StatusCode::NOT_FOUND => Error::NotFound(msg),
_ => Error::Backend(msg),
}
}
/// Parse a `ListObjectsV2` XML response for keys + next continuation token.
///
/// Deliberately tiny — full XML parsing is overkill for the two elements we
/// care about. Looks for `<Key>...</Key>` and `<NextContinuationToken>...`
/// inside the body. If R2 ever changes the element shape (it won't — it's
/// S3-compat), the integration test catches it.
fn parse_list_v2(body: &str) -> (Vec<String>, Option<String>) {
let keys = extract_all_tags(body, "Key")
.iter()
.map(|k| decode_xml_entities(k))
.collect();
let next = extract_first_text(body, "NextContinuationToken");
let truncated = extract_first_tag(body, "IsTruncated")
.map(|v| v.trim().eq_ignore_ascii_case("true"))
.unwrap_or(false);
(keys, if truncated { next } else { None })
}
/// Parse `<Contents>` blocks for key + size + last-modified.
///
/// R2's `<Contents>` always has `<Key>` followed by `<LastModified>` and
/// `<Size>` siblings. We walk `<Contents>...</Contents>` blocks and pull the
/// three tags from each — order-insensitive within the block. Entries missing
/// any of the three are skipped (defensive — R2 always emits all three).
fn parse_list_v2_detailed(body: &str) -> (Vec<ObjectMeta>, Option<String>) {
let blocks = extract_all_tags(body, "Contents");
let entries = blocks
.into_iter()
.filter_map(|block| {
let key = extract_first_text(&block, "Key")?;
let size = extract_first_tag(&block, "Size")?.trim().parse::<u64>().ok()?;
let last_modified = extract_first_text(&block, "LastModified")?;
Some(ObjectMeta { key, size, last_modified })
})
.collect();
let next = extract_first_text(body, "NextContinuationToken");
let truncated = extract_first_tag(body, "IsTruncated")
.map(|v| v.trim().eq_ignore_ascii_case("true"))
.unwrap_or(false);
(entries, if truncated { next } else { None })
}
fn extract_all_tags(body: &str, tag: &str) -> Vec<String> {
let open = format!("<{tag}>");
let close = format!("</{tag}>");
let mut out = Vec::new();
let mut search = body;
while let Some(start) = search.find(&open) {
let content_start = start + open.len();
if let Some(end) = search[content_start..].find(&close) {
out.push(search[content_start..content_start + end].to_string());
search = &search[content_start + end + close.len()..];
} else {
break;
}
}
out
}
fn extract_first_tag(body: &str, tag: &str) -> Option<String> {
extract_all_tags(body, tag).into_iter().next()
}
/// Like [`extract_first_tag`] but for a LEAF element, whose text is XML-escaped.
///
/// Never use this on a container like `<Contents>` — decoding a block that
/// still holds child tags would corrupt any `&` that belongs to a nested
/// value before the child tag is even extracted.
fn extract_first_text(body: &str, tag: &str) -> Option<String> {
extract_first_tag(body, tag).map(|raw| decode_xml_entities(&raw))
}
/// Decode the XML 1.0 predefined entities plus numeric character references.
///
/// R630-B1 fallout: object keys go out on the wire percent-encoded but come
/// back from `ListObjectsV2` as XML *text*, so a key holding `&` arrives as
/// `&` and one holding `<` as `<`. Before the SigV4 fix those keys could
/// not be written at all (they 403'd), so the raw-text read was never wrong in
/// practice; now that they can be written, `list_prefix` would hand back a key
/// that no subsequent `get`/`head`/`delete` can resolve. S3 also emits numeric
/// references (` `) for control characters, which are legal in a key.
///
/// An unrecognized `&…` sequence is passed through verbatim rather than
/// dropped — a literal `&` in a document that isn't escaping anything is not
/// something to silently eat.
fn decode_xml_entities(raw: &str) -> String {
if !raw.contains('&') {
return raw.to_string();
}
let mut out = String::with_capacity(raw.len());
let mut rest = raw;
while let Some(amp) = rest.find('&') {
out.push_str(&rest[..amp]);
let tail = &rest[amp..];
let Some(semi) = tail.find(';').filter(|&i| i <= 10) else {
out.push('&');
rest = &tail[1..];
continue;
};
let entity = &tail[1..semi];
let decoded = match entity {
"amp" => Some('&'),
"lt" => Some('<'),
"gt" => Some('>'),
"quot" => Some('"'),
"apos" => Some('\''),
_ => entity
.strip_prefix('#')
.and_then(|n| match n.strip_prefix(['x', 'X']) {
Some(hex) => u32::from_str_radix(hex, 16).ok(),
None => n.parse::<u32>().ok(),
})
.and_then(char::from_u32),
};
match decoded {
Some(c) => {
out.push(c);
rest = &tail[semi + 1..];
}
None => {
out.push('&');
rest = &tail[1..];
}
}
}
out.push_str(rest);
out
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn with_endpoint_redirects_every_url_and_leaves_r2_alone() {
let store = R2ObjectStore::new("acct", "yah-dev", "k", "s").unwrap();
assert_eq!(
store.object_url("yah/index.json"),
"https://acct.r2.cloudflarestorage.com/yah-dev/yah/index.json"
);
// Pond tier: same store, MinIO endpoint, path-style bucket preserved.
let pond = R2ObjectStore::new("pond", "yah-dev", "k", "s")
.unwrap()
.with_endpoint("http://127.0.0.1:9000");
assert_eq!(
pond.object_url("yah/index.json"),
"http://127.0.0.1:9000/yah-dev/yah/index.json"
);
assert_eq!(pond.bucket_url(), "http://127.0.0.1:9000/yah-dev");
}
#[test]
fn with_endpoint_normalizes_trailing_slash_and_ignores_empty() {
let s = R2ObjectStore::new("acct", "b", "k", "s")
.unwrap()
.with_endpoint("http://127.0.0.1:9000/");
assert_eq!(s.object_url("k1"), "http://127.0.0.1:9000/b/k1");
// An empty override is a config mistake, not an instruction to sign
// against the empty host — fall back to the derived R2 endpoint.
let s = R2ObjectStore::new("acct", "b", "k", "s")
.unwrap()
.with_endpoint("");
assert_eq!(
s.object_url("k1"),
"https://acct.r2.cloudflarestorage.com/b/k1"
);
}
/// Accept exactly one HTTP request on an ephemeral loopback port, answer
/// `200`, and hand the raw request head back. Enough of a server to prove
/// what went onto the wire, and no more — the point is the headers, and a
/// mock at the `reqwest` layer would only re-assert what the signer already
/// returned rather than what the client actually sent.
fn one_shot_http() -> (String, std::thread::JoinHandle<String>) {
use std::io::{BufRead, BufReader, Read, Write};
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let url = format!("http://{}", listener.local_addr().unwrap());
let handle = std::thread::spawn(move || {
let (stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream);
let mut head = String::new();
loop {
let mut line = String::new();
if reader.read_line(&mut line).unwrap() == 0 {
break;
}
let done = line == "\r\n";
head.push_str(&line);
if done {
break;
}
}
// Drain the body, else the client sees the connection close
// mid-write and reports a broken pipe instead of our 200.
let len: usize = head
.lines()
.find_map(|l| {
l.strip_prefix("content-length: ")
.or_else(|| l.strip_prefix("Content-Length: "))
})
.and_then(|v| v.trim().parse().ok())
.unwrap_or(0);
let mut body = vec![0u8; len];
reader.read_exact(&mut body).unwrap();
reader
.into_inner()
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")
.unwrap();
head
});
(url, handle)
}
/// R703-B8, the end of the chain: the signer builds a `Cache-Control` and
/// `reqwest` has to actually put it on the socket. Every other test here
/// stops at the `HeaderMap`, which is one `.headers()` call away from being
/// a test that passes while R2 stores an object with no directive.
#[test]
fn put_cached_sends_the_cache_control_header_on_the_wire() {
let (endpoint, server) = one_shot_http();
let store = R2ObjectStore::new("acct", "yah-dev", "AK", "SK")
.unwrap()
.with_endpoint(endpoint);
store
.put_cached(
"yah-desktop/latest.json",
b"{\"version\":\"0.8.22\"}".to_vec(),
crate::CACHE_CONTROL_NO_CACHE,
)
.unwrap();
let head = server.join().unwrap().to_lowercase();
assert!(
head.starts_with("put /yah-dev/yah-desktop/latest.json "),
"{head}"
);
assert!(head.contains("cache-control: no-cache, max-age=0\r\n"), "{head}");
// Sent AND signed — an unsigned header R2 would reject the request over.
assert!(
head.contains("signedheaders=cache-control;content-length;content-type;host;"),
"{head}"
);
}
/// R630-B1, the property the whole fix rests on: the bytes on the wire and
/// the bytes in the SigV4 canonical request are the SAME bytes. Every other
/// test here stops at a `String` or a `HeaderMap`; only this one can catch
/// reqwest re-encoding (or decoding) the path between `object_url` and the
/// socket, which would put the signature back out of sync with the request
/// and give exactly the 403 this ticket is about.
#[test]
fn a_colon_key_goes_on_the_wire_percent_encoded() {
let (endpoint, server) = one_shot_http();
let store = R2ObjectStore::new("acct", "yah-cr", "AK", "SK")
.unwrap()
.with_endpoint(endpoint);
store
.put("blobs/sha256:deadbeef", b"layer".to_vec())
.unwrap();
let head = server.join().unwrap();
let request_line = head.lines().next().unwrap();
assert_eq!(
request_line, "PUT /yah-cr/blobs/sha256%3Adeadbeef HTTP/1.1",
"full head: {head}"
);
// Uppercase hex specifically — R2 canonicalizes with uppercase, so a
// lowercase `%3a` on the wire signs differently from what it computes.
assert!(!request_line.contains("%3a"), "{request_line}");
}
/// And the regression half on the wire: an ordinary key must reach the
/// socket byte-identical to before the encoding was introduced.
#[test]
fn an_unreserved_key_reaches_the_wire_unchanged() {
let (endpoint, server) = one_shot_http();
let store = R2ObjectStore::new("acct", "yah-dev", "AK", "SK")
.unwrap()
.with_endpoint(endpoint);
store
.put("yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz", b"x".to_vec())
.unwrap();
let head = server.join().unwrap();
assert_eq!(
head.lines().next().unwrap(),
"PUT /yah-dev/yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz HTTP/1.1",
"full head: {head}"
);
}
/// R630-F2. `delete` signed with `sign_s3_empty_body` from the day it
/// landed, which puts `content-length: 0` inside the signature — and
/// reqwest strips that header off a body-less request, so R2 recomputed a
/// different canonical request and answered 403 for EVERY delete. The
/// in-memory `delete_is_idempotent` trait test passed throughout, because
/// it never touches this code. Assert against the socket instead.
#[test]
fn delete_signs_without_content_length_and_sends_none() {
let (endpoint, server) = one_shot_http();
let store = R2ObjectStore::new("acct", "yah-dev", "AK", "SK")
.unwrap()
.with_endpoint(endpoint);
store.delete("some/blob.bin").unwrap();
let head = server.join().unwrap().to_lowercase();
assert!(head.starts_with("delete /yah-dev/some/blob.bin "), "{head}");
assert!(
head.contains("signedheaders=host;x-amz-content-sha256;x-amz-date"),
"content-length must NOT be signed on a body-less DELETE: {head}"
);
// And the header genuinely isn't on the wire — which is the whole
// reason signing it was fatal.
assert!(!head.contains("content-length"), "{head}");
}
/// The other half: a plain `put` must still send no directive at all. If it
/// quietly gained a default, versioned release bytes would start carrying
/// whatever that default was.
#[test]
fn a_plain_put_sends_no_cache_control_header() {
let (endpoint, server) = one_shot_http();
let store = R2ObjectStore::new("acct", "yah-dev", "AK", "SK")
.unwrap()
.with_endpoint(endpoint);
store.put("some/blob.bin", b"bytes".to_vec()).unwrap();
let head = server.join().unwrap().to_lowercase();
assert!(!head.contains("cache-control"), "{head}");
}
#[test]
fn parse_list_v2_extracts_keys() {
let body = r#"<?xml version="1.0" encoding="UTF-8"?>
<ListBucketResult>
<IsTruncated>false</IsTruncated>
<Contents><Key>yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz</Key></Contents>
<Contents><Key>yubaba/release-manifest.json</Key></Contents>
</ListBucketResult>"#;
let (keys, next) = parse_list_v2(body);
assert_eq!(
keys,
vec![
"yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz".to_string(),
"yubaba/release-manifest.json".to_string(),
]
);
assert!(next.is_none());
}
#[test]
fn parse_list_v2_returns_continuation_when_truncated() {
let body = r#"<ListBucketResult>
<IsTruncated>true</IsTruncated>
<NextContinuationToken>abc123</NextContinuationToken>
<Contents><Key>a</Key></Contents>
</ListBucketResult>"#;
let (keys, next) = parse_list_v2(body);
assert_eq!(keys, vec!["a".to_string()]);
assert_eq!(next.as_deref(), Some("abc123"));
}
#[test]
fn parse_list_v2_ignores_token_when_not_truncated() {
// Some S3-compat impls emit NextContinuationToken with IsTruncated=false.
// We treat IsTruncated as load-bearing.
let body = r#"<ListBucketResult>
<IsTruncated>false</IsTruncated>
<NextContinuationToken>stale</NextContinuationToken>
<Contents><Key>a</Key></Contents>
</ListBucketResult>"#;
let (_, next) = parse_list_v2(body);
assert!(next.is_none());
}
#[test]
fn parse_list_v2_detailed_extracts_size_and_mtime() {
let body = r#"<?xml version="1.0" encoding="UTF-8"?>
<ListBucketResult>
<IsTruncated>false</IsTruncated>
<Contents>
<Key>yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz</Key>
<LastModified>2026-06-08T20:14:32.000Z</LastModified>
<ETag>"abc"</ETag>
<Size>4823104</Size>
<StorageClass>STANDARD</StorageClass>
</Contents>
<Contents>
<Key>yubaba/release-manifest.json</Key>
<LastModified>2026-06-08T20:14:35.000Z</LastModified>
<Size>412</Size>
</Contents>
</ListBucketResult>"#;
let (entries, next) = parse_list_v2_detailed(body);
assert_eq!(entries.len(), 2);
assert_eq!(entries[0].key, "yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz");
assert_eq!(entries[0].size, 4823104);
assert_eq!(entries[0].last_modified, "2026-06-08T20:14:32.000Z");
assert_eq!(entries[1].key, "yubaba/release-manifest.json");
assert_eq!(entries[1].size, 412);
assert!(next.is_none());
}
#[test]
fn r2_object_store_constructs_with_explicit_keys() {
let s = R2ObjectStore::new("acct", "yah-dev", "AK", "SK").unwrap();
assert_eq!(s.object_url("k"), "https://acct.r2.cloudflarestorage.com/yah-dev/k");
assert_eq!(s.bucket_url(), "https://acct.r2.cloudflarestorage.com/yah-dev");
}
#[test]
fn object_url_preserves_slashes_in_key() {
let s = R2ObjectStore::new("acct", "b", "AK", "SK").unwrap();
assert_eq!(
s.object_url("yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz"),
"https://acct.r2.cloudflarestorage.com/b/yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz"
);
}
/// R630-B1 fallout. A key holding `&` could not be written before the SigV4
/// fix (it 403'd like every other non-unreserved key), so the raw-XML read
/// was never observably wrong. Now that it CAN be written, a `list_prefix`
/// that hands back `a&b` names an object no `get`/`head`/`delete` can
/// resolve — the round-trip would break at the far end instead of the near
/// one.
#[test]
fn list_keys_are_xml_decoded() {
let body = "<ListBucketResult>\
<Contents><Key>a&b.json</Key><Size>1</Size><LastModified>t</LastModified></Contents>\
<Contents><Key>x<y>z</Key><Size>2</Size><LastModified>t</LastModified></Contents>\
<Contents><Key>q"r's</Key><Size>3</Size><LastModified>t</LastModified></Contents>\
<Contents><Key>n mA</Key><Size>4</Size><LastModified>t</LastModified></Contents>\
<IsTruncated>false</IsTruncated></ListBucketResult>";
let (keys, next) = parse_list_v2(body);
assert_eq!(keys, vec!["a&b.json", "x<y>z", "q\"r's", "n\rmA"]);
assert!(next.is_none());
// The detailed form shares the decode, and its `<Contents>` block walk
// must still see the child tags — decoding the block first would break it.
let (entries, _) = parse_list_v2_detailed(body);
assert_eq!(entries.len(), 4);
assert_eq!(entries[0].key, "a&b.json");
assert_eq!(entries[0].size, 1);
}
/// A lone `&` that isn't opening an entity is passed through, not eaten.
#[test]
fn xml_decode_passes_through_a_non_entity_ampersand() {
assert_eq!(decode_xml_entities("a & b"), "a & b");
assert_eq!(decode_xml_entities("¬anentity;"), "¬anentity;");
assert_eq!(decode_xml_entities("plain/key.json"), "plain/key.json");
assert_eq!(decode_xml_entities("&&"), "&&");
}
/// R630-B1. The key becomes a URL exactly here, so this is the one place
/// the AWS encoding can be applied such that the wire path and the SigV4
/// canonical path agree. A `sha256:<hex>` OCI digest is the key shape that
/// forced the issue — cr.yah.dev writes `sha256/<hex>` today purely to dodge
/// it (see the R630 relay gotcha).
#[test]
fn object_url_encodes_a_colon_in_the_key() {
let s = R2ObjectStore::new("acct", "yah-cr", "AK", "SK").unwrap();
assert_eq!(
s.object_url("blobs/sha256:deadbeef"),
"https://acct.r2.cloudflarestorage.com/yah-cr/blobs/sha256%3Adeadbeef"
);
// `locate` shares the choke point, so a caller handed the URL for a
// direct fetch gets the encoded form too.
assert_eq!(s.locate("blobs/sha256:deadbeef"), s.object_url("blobs/sha256:deadbeef"));
}
/// Regression guard: today's keys must produce byte-identical URLs, or the
/// fix for the colon case silently 403s everything that already works.
#[test]
fn object_url_leaves_unreserved_keys_untouched() {
let s = R2ObjectStore::new("acct", "b", "AK", "SK").unwrap();
for key in [
"k",
"yah/index.json",
"yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz",
"releases/v1.2.3-rc.1/yah_1.2.3_aarch64.dmg",
] {
assert_eq!(
s.object_url(key),
format!("https://acct.r2.cloudflarestorage.com/b/{key}")
);
}
}
#[test]
fn content_type_inferred_from_extension() {
assert_eq!(
content_type_for_key("yah-marketing/cloud/index.html"),
"text/html; charset=utf-8"
);
assert_eq!(content_type_for_key("app.css"), "text/css; charset=utf-8");
assert_eq!(content_type_for_key("bundle.mjs"), "text/javascript; charset=utf-8");
assert_eq!(content_type_for_key("illustrations/horse.webp"), "image/webp");
assert_eq!(content_type_for_key("manifest.json"), "application/json");
// Extensionless keys (pointers) and dotted directory segments fall back.
assert_eq!(content_type_for_key("pointers/releases"), DEFAULT_CONTENT_TYPE);
assert_eq!(content_type_for_key("v1.2/binary"), DEFAULT_CONTENT_TYPE);
}
}