1use std::time::Duration;
29
30use percent_encoding::{utf8_percent_encode, AsciiSet, NON_ALPHANUMERIC};
31use reqwest::blocking::Client;
32use reqwest::header::{HeaderValue, ETAG, IF_MATCH, IF_NONE_MATCH};
33use reqwest::StatusCode;
34use sha2::{Digest, Sha256};
35
36use local_driver::s3_sign::{
37 sign_s3_empty_body, sign_s3_get_with_query, sign_s3_no_body, sign_s3_put_object,
38};
39
40use crate::{Error, ObjectStore, Precondition};
41
42const R2_REGION: &str = "auto";
44
45pub const R2_ACCESS_KEY_SLOT: &str = "cloudflare-r2-access-key-id";
47pub const R2_SECRET_KEY_SLOT: &str = "cloudflare-r2-secret-key";
49pub const R2_ACCESS_KEY_ENV: &str = "CF_R2_ACCESS_KEY_ID";
51pub const R2_SECRET_KEY_ENV: &str = "CF_R2_SECRET_KEY";
53
54const QUERY_VALUE: &AsciiSet = &NON_ALPHANUMERIC
58 .remove(b'-')
59 .remove(b'_')
60 .remove(b'.')
61 .remove(b'~');
62
63const DEFAULT_CONTENT_TYPE: &str = "application/octet-stream";
65
66fn content_type_for_key(key: &str) -> &'static str {
75 let ext = match key.rsplit_once('.') {
76 Some((_, e)) if !e.contains('/') => e,
78 _ => "",
79 };
80 match ext.to_ascii_lowercase().as_str() {
81 "html" | "htm" => "text/html; charset=utf-8",
82 "css" => "text/css; charset=utf-8",
83 "js" | "mjs" => "text/javascript; charset=utf-8",
84 "json" | "map" => "application/json",
85 "webmanifest" => "application/manifest+json",
86 "xml" => "application/xml",
87 "txt" => "text/plain; charset=utf-8",
88 "svg" => "image/svg+xml",
89 "webp" => "image/webp",
90 "png" => "image/png",
91 "jpg" | "jpeg" => "image/jpeg",
92 "gif" => "image/gif",
93 "avif" => "image/avif",
94 "ico" => "image/x-icon",
95 "woff2" => "font/woff2",
96 "woff" => "font/woff",
97 "ttf" => "font/ttf",
98 "otf" => "font/otf",
99 "wasm" => "application/wasm",
100 "pdf" => "application/pdf",
101 _ => DEFAULT_CONTENT_TYPE,
102 }
103}
104
105pub struct R2ObjectStore {
111 account_id: String,
112 bucket: String,
113 access_key: String,
114 secret_key: String,
115 client: Option<Client>,
116}
117
118impl Drop for R2ObjectStore {
119 fn drop(&mut self) {
120 let Some(client) = self.client.take() else { return };
129 std::thread::spawn(move || drop(client));
130 }
131}
132
133impl R2ObjectStore {
134 pub fn new(
139 account_id: impl Into<String>,
140 bucket: impl Into<String>,
141 access_key: impl Into<String>,
142 secret_key: impl Into<String>,
143 ) -> Result<Self, Error> {
144 let client = Client::builder()
145 .timeout(Duration::from_secs(300))
146 .build()
147 .map_err(|e| Error::Backend(format!("reqwest client: {e}")))?;
148 Ok(Self {
149 account_id: account_id.into(),
150 bucket: bucket.into(),
151 access_key: access_key.into(),
152 secret_key: secret_key.into(),
153 client: Some(client),
154 })
155 }
156
157 fn client(&self) -> &Client {
158 self.client
159 .as_ref()
160 .expect("client is Some until Drop takes it")
161 }
162
163 pub fn from_vault(
169 account_id: impl Into<String>,
170 bucket: impl Into<String>,
171 ) -> Result<Self, Error> {
172 let access_key = fob::get_or_env(R2_ACCESS_KEY_SLOT, R2_ACCESS_KEY_ENV)
173 .map_err(|e| Error::Auth(format!("vault read {R2_ACCESS_KEY_SLOT}: {e}")))?
174 .ok_or_else(|| {
175 Error::Auth(format!(
176 "missing R2 credential: set vault slot {R2_ACCESS_KEY_SLOT} or env {R2_ACCESS_KEY_ENV}"
177 ))
178 })?;
179 let secret_key = fob::get_or_env(R2_SECRET_KEY_SLOT, R2_SECRET_KEY_ENV)
180 .map_err(|e| Error::Auth(format!("vault read {R2_SECRET_KEY_SLOT}: {e}")))?
181 .ok_or_else(|| {
182 Error::Auth(format!(
183 "missing R2 credential: set vault slot {R2_SECRET_KEY_SLOT} or env {R2_SECRET_KEY_ENV}"
184 ))
185 })?;
186 Self::new(account_id, bucket, access_key, secret_key)
187 }
188
189 fn endpoint(&self) -> String {
190 format!("https://{}.r2.cloudflarestorage.com", self.account_id)
191 }
192
193 fn object_url(&self, key: &str) -> String {
194 format!("{}/{}/{}", self.endpoint(), self.bucket, key)
195 }
196
197 fn bucket_url(&self) -> String {
198 format!("{}/{}", self.endpoint(), self.bucket)
199 }
200}
201
202fn io_err(ctx: &str, e: impl std::fmt::Display) -> Error {
204 Error::Io(format!("{ctx}: {e}"))
205}
206
207impl ObjectStore for R2ObjectStore {
208 fn put(&self, key: &str, data: Vec<u8>) -> Result<(), Error> {
209 let url = self.object_url(key);
210 let body_sha256 = {
211 let mut h = Sha256::new();
212 h.update(&data);
213 hex::encode(h.finalize())
214 };
215 let headers = sign_s3_put_object(
216 &url,
217 &body_sha256,
218 content_type_for_key(key),
219 data.len(),
220 R2_REGION,
221 &self.access_key,
222 &self.secret_key,
223 None,
226 )
227 .map_err(|e| Error::Backend(format!("sign PUT {key}: {e}")))?;
228
229 let resp = self
230 .client()
231 .put(&url)
232 .headers(headers)
233 .body(data)
234 .send()
235 .map_err(|e| io_err(&format!("PUT {key}"), e))?;
236 check_status(resp, "PUT", key)
237 }
238
239 fn get(&self, key: &str) -> Result<Option<Vec<u8>>, Error> {
240 let url = self.object_url(key);
241 let headers = sign_s3_get_with_query(
247 &url,
248 "",
249 R2_REGION,
250 &self.access_key,
251 &self.secret_key,
252 )
253 .map_err(|e| Error::Backend(format!("sign GET {key}: {e}")))?;
254
255 let resp = self
256 .client()
257 .get(&url)
258 .headers(headers)
259 .send()
260 .map_err(|e| io_err(&format!("GET {key}"), e))?;
261
262 match resp.status() {
263 StatusCode::OK => {
264 let bytes = resp
265 .bytes()
266 .map_err(|e| io_err(&format!("read GET {key}"), e))?;
267 Ok(Some(bytes.to_vec()))
268 }
269 StatusCode::NOT_FOUND => Ok(None),
270 s => Err(status_err("GET", key, s, resp.text().ok())),
271 }
272 }
273
274 fn head(&self, key: &str) -> Result<bool, Error> {
275 let url = self.object_url(key);
276 let headers = sign_s3_no_body(
278 "HEAD",
279 &url,
280 "",
281 R2_REGION,
282 &self.access_key,
283 &self.secret_key,
284 )
285 .map_err(|e| Error::Backend(format!("sign HEAD {key}: {e}")))?;
286
287 let resp = self
288 .client()
289 .head(&url)
290 .headers(headers)
291 .send()
292 .map_err(|e| io_err(&format!("HEAD {key}"), e))?;
293
294 match resp.status() {
295 StatusCode::OK => Ok(true),
296 StatusCode::NOT_FOUND => Ok(false),
297 s => Err(status_err("HEAD", key, s, None)),
298 }
299 }
300
301 fn delete(&self, key: &str) -> Result<(), Error> {
302 let url = self.object_url(key);
303 let headers = sign_s3_empty_body(
304 "DELETE",
305 &url,
306 R2_REGION,
307 &self.access_key,
308 &self.secret_key,
309 )
310 .map_err(|e| Error::Backend(format!("sign DELETE {key}: {e}")))?;
311
312 let resp = self
313 .client()
314 .delete(&url)
315 .headers(headers)
316 .send()
317 .map_err(|e| io_err(&format!("DELETE {key}"), e))?;
318
319 match resp.status() {
320 StatusCode::OK | StatusCode::NO_CONTENT | StatusCode::NOT_FOUND => Ok(()),
323 s => Err(status_err("DELETE", key, s, resp.text().ok())),
324 }
325 }
326
327 fn list_prefix(&self, prefix: &str) -> Result<Vec<String>, Error> {
328 Ok(self
329 .list_prefix_detailed(prefix)?
330 .into_iter()
331 .map(|m| m.key)
332 .collect())
333 }
334
335 fn put_if(&self, key: &str, data: Vec<u8>, cond: Precondition) -> Result<String, Error> {
336 let url = self.object_url(key);
337 let body_sha256 = {
338 let mut h = Sha256::new();
339 h.update(&data);
340 hex::encode(h.finalize())
341 };
342 let mut headers = sign_s3_put_object(
348 &url,
349 &body_sha256,
350 content_type_for_key(key),
351 data.len(),
352 R2_REGION,
353 &self.access_key,
354 &self.secret_key,
355 None,
356 )
357 .map_err(|e| Error::Backend(format!("sign PUT {key}: {e}")))?;
358
359 match &cond {
360 Precondition::IfAbsent => {
361 headers.insert(IF_NONE_MATCH, HeaderValue::from_static("*"));
362 }
363 Precondition::IfMatch(etag) => {
364 let v = HeaderValue::from_str(etag)
365 .map_err(|e| Error::Backend(format!("invalid If-Match etag {etag:?}: {e}")))?;
366 headers.insert(IF_MATCH, v);
367 }
368 }
369
370 let resp = self
371 .client()
372 .put(&url)
373 .headers(headers)
374 .body(data)
375 .send()
376 .map_err(|e| io_err(&format!("PUT(if) {key}"), e))?;
377
378 let status = resp.status();
379 if status == StatusCode::PRECONDITION_FAILED {
380 return Err(Error::PreconditionFailed(format!(
381 "put_if {key}: precondition not met ({cond:?})"
382 )));
383 }
384 if !status.is_success() {
385 return Err(status_err("PUT(if)", key, status, resp.text().ok()));
386 }
387 match resp.headers().get(ETAG).and_then(|v| v.to_str().ok()) {
390 Some(e) => Ok(e.to_string()),
391 None => self
392 .etag(key)?
393 .ok_or_else(|| Error::Backend(format!("PUT(if) {key} returned no ETag"))),
394 }
395 }
396
397 fn etag(&self, key: &str) -> Result<Option<String>, Error> {
398 let url = self.object_url(key);
399 let headers = sign_s3_no_body(
401 "HEAD",
402 &url,
403 "",
404 R2_REGION,
405 &self.access_key,
406 &self.secret_key,
407 )
408 .map_err(|e| Error::Backend(format!("sign HEAD {key}: {e}")))?;
409
410 let resp = self
411 .client()
412 .head(&url)
413 .headers(headers)
414 .send()
415 .map_err(|e| io_err(&format!("HEAD(etag) {key}"), e))?;
416
417 match resp.status() {
418 StatusCode::OK => Ok(resp
419 .headers()
420 .get(ETAG)
421 .and_then(|v| v.to_str().ok())
422 .map(|s| s.to_string())),
423 StatusCode::NOT_FOUND => Ok(None),
424 s => Err(status_err("HEAD(etag)", key, s, None)),
425 }
426 }
427}
428
429#[derive(Debug, Clone, PartialEq, Eq)]
431pub struct ObjectMeta {
432 pub key: String,
434 pub size: u64,
436 pub last_modified: String,
438}
439
440impl R2ObjectStore {
441 pub fn list_prefix_detailed(&self, prefix: &str) -> Result<Vec<ObjectMeta>, Error> {
447 let mut entries = Vec::new();
448 let mut continuation_token: Option<String> = None;
449 let bucket_url = self.bucket_url();
450 let encoded_prefix = utf8_percent_encode(prefix, QUERY_VALUE).to_string();
451
452 loop {
453 let mut params: Vec<(String, String)> =
456 vec![("list-type".to_string(), "2".to_string())];
457 if let Some(token) = &continuation_token {
458 let encoded = utf8_percent_encode(token, QUERY_VALUE).to_string();
459 params.push(("continuation-token".to_string(), encoded));
460 }
461 params.push(("prefix".to_string(), encoded_prefix.clone()));
462 params.sort_by(|a, b| a.0.cmp(&b.0));
463 let canonical_query = params
464 .iter()
465 .map(|(k, v)| format!("{k}={v}"))
466 .collect::<Vec<_>>()
467 .join("&");
468
469 let url_with_query = format!("{bucket_url}?{canonical_query}");
470
471 let headers = sign_s3_get_with_query(
472 &bucket_url,
473 &canonical_query,
474 R2_REGION,
475 &self.access_key,
476 &self.secret_key,
477 )
478 .map_err(|e| Error::Backend(format!("sign LIST {prefix}: {e}")))?;
479
480 let resp = self
481 .client()
482 .get(&url_with_query)
483 .headers(headers)
484 .send()
485 .map_err(|e| io_err(&format!("LIST {prefix}"), e))?;
486
487 if !resp.status().is_success() {
488 return Err(status_err("LIST", prefix, resp.status(), resp.text().ok()));
489 }
490 let body = resp
491 .text()
492 .map_err(|e| io_err(&format!("LIST {prefix} body"), e))?;
493 let (page_entries, next_token) = parse_list_v2_detailed(&body);
494 entries.extend(page_entries);
495 if let Some(t) = next_token {
496 continuation_token = Some(t);
497 } else {
498 break;
499 }
500 }
501 Ok(entries)
502 }
503}
504
505fn check_status(resp: reqwest::blocking::Response, verb: &str, key: &str) -> Result<(), Error> {
506 if resp.status().is_success() {
507 Ok(())
508 } else {
509 let status = resp.status();
510 let body = resp.text().ok();
511 Err(status_err(verb, key, status, body))
512 }
513}
514
515fn status_err(verb: &str, key: &str, status: StatusCode, body: Option<String>) -> Error {
516 let snippet = body
517 .as_deref()
518 .map(|s| s.chars().take(200).collect::<String>())
519 .unwrap_or_default();
520 let msg = format!("{verb} {key} → {status} {snippet}");
521 match status {
522 StatusCode::FORBIDDEN | StatusCode::UNAUTHORIZED => Error::Auth(msg),
523 StatusCode::NOT_FOUND => Error::NotFound(msg),
524 _ => Error::Backend(msg),
525 }
526}
527
528fn parse_list_v2(body: &str) -> (Vec<String>, Option<String>) {
535 let keys = extract_all_tags(body, "Key");
536 let next = extract_first_tag(body, "NextContinuationToken");
537 let truncated = extract_first_tag(body, "IsTruncated")
538 .map(|v| v.trim().eq_ignore_ascii_case("true"))
539 .unwrap_or(false);
540 (keys, if truncated { next } else { None })
541}
542
543fn parse_list_v2_detailed(body: &str) -> (Vec<ObjectMeta>, Option<String>) {
550 let blocks = extract_all_tags(body, "Contents");
551 let entries = blocks
552 .into_iter()
553 .filter_map(|block| {
554 let key = extract_first_tag(&block, "Key")?;
555 let size = extract_first_tag(&block, "Size")?.trim().parse::<u64>().ok()?;
556 let last_modified = extract_first_tag(&block, "LastModified")?;
557 Some(ObjectMeta { key, size, last_modified })
558 })
559 .collect();
560 let next = extract_first_tag(body, "NextContinuationToken");
561 let truncated = extract_first_tag(body, "IsTruncated")
562 .map(|v| v.trim().eq_ignore_ascii_case("true"))
563 .unwrap_or(false);
564 (entries, if truncated { next } else { None })
565}
566
567fn extract_all_tags(body: &str, tag: &str) -> Vec<String> {
568 let open = format!("<{tag}>");
569 let close = format!("</{tag}>");
570 let mut out = Vec::new();
571 let mut search = body;
572 while let Some(start) = search.find(&open) {
573 let content_start = start + open.len();
574 if let Some(end) = search[content_start..].find(&close) {
575 out.push(search[content_start..content_start + end].to_string());
576 search = &search[content_start + end + close.len()..];
577 } else {
578 break;
579 }
580 }
581 out
582}
583
584fn extract_first_tag(body: &str, tag: &str) -> Option<String> {
585 extract_all_tags(body, tag).into_iter().next()
586}
587
588#[cfg(test)]
589mod tests {
590 use super::*;
591
592 #[test]
593 fn parse_list_v2_extracts_keys() {
594 let body = r#"<?xml version="1.0" encoding="UTF-8"?>
595 <ListBucketResult>
596 <IsTruncated>false</IsTruncated>
597 <Contents><Key>yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz</Key></Contents>
598 <Contents><Key>yubaba/release-manifest.json</Key></Contents>
599 </ListBucketResult>"#;
600 let (keys, next) = parse_list_v2(body);
601 assert_eq!(
602 keys,
603 vec![
604 "yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz".to_string(),
605 "yubaba/release-manifest.json".to_string(),
606 ]
607 );
608 assert!(next.is_none());
609 }
610
611 #[test]
612 fn parse_list_v2_returns_continuation_when_truncated() {
613 let body = r#"<ListBucketResult>
614 <IsTruncated>true</IsTruncated>
615 <NextContinuationToken>abc123</NextContinuationToken>
616 <Contents><Key>a</Key></Contents>
617 </ListBucketResult>"#;
618 let (keys, next) = parse_list_v2(body);
619 assert_eq!(keys, vec!["a".to_string()]);
620 assert_eq!(next.as_deref(), Some("abc123"));
621 }
622
623 #[test]
624 fn parse_list_v2_ignores_token_when_not_truncated() {
625 let body = r#"<ListBucketResult>
628 <IsTruncated>false</IsTruncated>
629 <NextContinuationToken>stale</NextContinuationToken>
630 <Contents><Key>a</Key></Contents>
631 </ListBucketResult>"#;
632 let (_, next) = parse_list_v2(body);
633 assert!(next.is_none());
634 }
635
636 #[test]
637 fn parse_list_v2_detailed_extracts_size_and_mtime() {
638 let body = r#"<?xml version="1.0" encoding="UTF-8"?>
639 <ListBucketResult>
640 <IsTruncated>false</IsTruncated>
641 <Contents>
642 <Key>yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz</Key>
643 <LastModified>2026-06-08T20:14:32.000Z</LastModified>
644 <ETag>"abc"</ETag>
645 <Size>4823104</Size>
646 <StorageClass>STANDARD</StorageClass>
647 </Contents>
648 <Contents>
649 <Key>yubaba/release-manifest.json</Key>
650 <LastModified>2026-06-08T20:14:35.000Z</LastModified>
651 <Size>412</Size>
652 </Contents>
653 </ListBucketResult>"#;
654 let (entries, next) = parse_list_v2_detailed(body);
655 assert_eq!(entries.len(), 2);
656 assert_eq!(entries[0].key, "yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz");
657 assert_eq!(entries[0].size, 4823104);
658 assert_eq!(entries[0].last_modified, "2026-06-08T20:14:32.000Z");
659 assert_eq!(entries[1].key, "yubaba/release-manifest.json");
660 assert_eq!(entries[1].size, 412);
661 assert!(next.is_none());
662 }
663
664 #[test]
665 fn r2_object_store_constructs_with_explicit_keys() {
666 let s = R2ObjectStore::new("acct", "yah-dev", "AK", "SK").unwrap();
667 assert_eq!(s.object_url("k"), "https://acct.r2.cloudflarestorage.com/yah-dev/k");
668 assert_eq!(s.bucket_url(), "https://acct.r2.cloudflarestorage.com/yah-dev");
669 }
670
671 #[test]
672 fn object_url_preserves_slashes_in_key() {
673 let s = R2ObjectStore::new("acct", "b", "AK", "SK").unwrap();
674 assert_eq!(
675 s.object_url("yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz"),
676 "https://acct.r2.cloudflarestorage.com/b/yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz"
677 );
678 }
679
680 #[test]
681 fn content_type_inferred_from_extension() {
682 assert_eq!(
683 content_type_for_key("yah-marketing/cloud/index.html"),
684 "text/html; charset=utf-8"
685 );
686 assert_eq!(content_type_for_key("app.css"), "text/css; charset=utf-8");
687 assert_eq!(content_type_for_key("bundle.mjs"), "text/javascript; charset=utf-8");
688 assert_eq!(content_type_for_key("illustrations/horse.webp"), "image/webp");
689 assert_eq!(content_type_for_key("manifest.json"), "application/json");
690 assert_eq!(content_type_for_key("pointers/releases"), DEFAULT_CONTENT_TYPE);
692 assert_eq!(content_type_for_key("v1.2/binary"), DEFAULT_CONTENT_TYPE);
693 }
694}