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