1use quarb::{AstAdapter, NodeId, Value};
39use std::cell::RefCell;
40
41#[derive(Debug, thiserror::Error)]
43pub enum ObjstoreError {
44 #[error("objstore: {0}")]
45 Http(String),
46 #[error("objstore target: {0} (expected gs://BUCKET[/PREFIX], s3://BUCKET[/PREFIX], or az://ACCOUNT/CONTAINER[/PREFIX])")]
47 Target(String),
48}
49
50enum Backend {
51 Gcs {
52 token: Option<String>,
53 },
54 S3 {
55 region: String,
56 creds: Option<quarb_aws::Credentials>,
57 endpoint: Option<String>,
59 },
60 Azure {
61 account: String,
62 sas: Option<String>,
64 key: Option<Vec<u8>>,
66 endpoint: Option<String>,
68 },
69}
70
71struct Node {
72 key: String,
74 name: Option<String>,
75 parent: Option<NodeId>,
76 is_object: bool,
77 size: Option<i64>,
78 updated: Option<String>,
79 children: RefCell<Option<Vec<NodeId>>>,
80 content: RefCell<Option<String>>,
81}
82
83pub struct ObjstoreAdapter {
85 backend: Backend,
86 bucket: String,
87 base: String,
89 nodes: RefCell<Vec<Node>>,
90}
91
92fn gcp_token(account: Option<&str>) -> Option<String> {
93 if let Ok(t) = std::env::var("QUARB_GCP_TOKEN")
94 && !t.trim().is_empty()
95 {
96 return Some(t.trim().to_string());
97 }
98 let mut cmd = std::process::Command::new("gcloud");
99 cmd.args(["auth", "print-access-token"]);
100 if let Some(a) = account {
101 cmd.arg(a);
102 }
103 let out = cmd.output().ok()?;
104 out.status
105 .success()
106 .then(|| String::from_utf8_lossy(&out.stdout).trim().to_string())
107}
108
109fn urlencode(s: &str) -> String {
110 let mut out = String::new();
111 for b in s.bytes() {
112 match b {
113 b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
114 out.push(b as char)
115 }
116 other => out.push_str(&format!("%{other:02X}")),
117 }
118 }
119 out
120}
121
122type Page = (Vec<String>, Vec<(String, Option<i64>, Option<String>)>);
124
125impl ObjstoreAdapter {
126 pub fn connect(target: &str) -> Result<Self, ObjstoreError> {
129 let (backend, rest) = if let Some(r) = target.strip_prefix("gs://") {
130 ("gs", r)
131 } else if let Some(r) = target.strip_prefix("s3://") {
132 ("s3", r)
133 } else if let Some(r) = target.strip_prefix("az://") {
134 ("az", r)
135 } else {
136 return Err(ObjstoreError::Target(target.to_string()));
137 };
138 let (path, query) = match rest.split_once('?') {
139 Some((p, q)) => (p, Some(q)),
140 None => (rest, None),
141 };
142 let (azure_account, path) = if backend == "az" {
144 match path.split_once('/') {
145 Some((a, rest)) if !a.is_empty() => (Some(a.to_string()), rest),
146 _ => return Err(ObjstoreError::Target(target.to_string())),
147 }
148 } else {
149 (None, path)
150 };
151 let (bucket, prefix) = match path.split_once('/') {
152 Some((b, p)) => (b.to_string(), p.trim_end_matches('/').to_string()),
153 None => (path.to_string(), String::new()),
154 };
155 if bucket.is_empty() {
156 return Err(ObjstoreError::Target(target.to_string()));
157 }
158 let param = |k: &str| {
159 query.and_then(|q| {
160 q.split('&')
161 .find_map(|kv| kv.strip_prefix(&format!("{k}=")).map(str::to_string))
162 })
163 };
164 let backend = if backend == "gs" {
165 let token = if param("auth").is_some() || param("account").is_some() {
166 gcp_token(param("account").as_deref())
167 } else {
168 None
169 };
170 Backend::Gcs { token }
171 } else if backend == "az" {
172 Backend::Azure {
173 account: azure_account.expect("parsed above"),
174 sas: param("sas").map(|s| s.trim_start_matches('?').to_string()),
175 key: std::env::var("AZURE_STORAGE_KEY")
176 .ok()
177 .filter(|k| !k.trim().is_empty())
178 .and_then(|k| quarb::base64_decode(k.trim())),
179 endpoint: param("endpoint").map(|e| e.trim_end_matches('/').to_string()),
180 }
181 } else {
182 Backend::S3 {
183 region: quarb_aws::region(param("region").as_deref()),
184 creds: if param("anon").is_some() {
185 None
186 } else {
187 quarb_aws::load_credentials()
188 },
189 endpoint: param("endpoint").map(|e| e.trim_end_matches('/').to_string()),
190 }
191 };
192 let adapter = ObjstoreAdapter {
193 backend,
194 bucket,
195 base: prefix.clone(),
196 nodes: RefCell::new(vec![Node {
197 key: prefix,
198 name: None,
199 parent: None,
200 is_object: false,
201 size: None,
202 updated: None,
203 children: RefCell::new(None),
204 content: RefCell::new(None),
205 }]),
206 };
207 adapter
208 .list(&adapter.nodes.borrow()[0].key.clone())
209 .map_err(|e| ObjstoreError::Http(format!("probing the bucket: {e}")))?;
210 Ok(adapter)
211 }
212
213 pub fn locator(&self, node: NodeId) -> String {
215 let key = &self.nodes.borrow()[node.0 as usize].key;
216 let rel = key.strip_prefix(&self.base).unwrap_or(key);
217 format!("/{}", rel.trim_start_matches('/'))
218 }
219
220 fn get(&self, url: &str) -> Result<String, String> {
221 let mut req = ureq::get(url);
222 match &self.backend {
223 Backend::Gcs { token: Some(t) } => {
224 req = req.set("Authorization", &format!("Bearer {t}"));
225 }
226 Backend::S3 {
227 region,
228 creds: Some(c),
229 ..
230 } => {
231 for (k, v) in quarb_aws::sign(c, "GET", url, region, "s3", b"", &[]) {
232 if k != "host" {
233 req = req.set(&k, &v);
234 }
235 }
236 }
237 Backend::Azure { account, key: Some(k), .. } => {
238 let date = rfc1123_now();
239 for (name, value) in azure_shared_key_headers(url, account, k, &date) {
240 req = req.set(&name, &value);
241 }
242 }
243 Backend::Azure { .. } => {}
244 _ => {}
245 }
246 req.call()
247 .map_err(|e| e.to_string())?
248 .into_string()
249 .map_err(|e| e.to_string())
250 }
251
252 fn azure_root(&self) -> String {
255 let Backend::Azure { account, endpoint, .. } = &self.backend else {
256 unreachable!("azure_root on a non-Azure backend");
257 };
258 match endpoint {
259 Some(e) => format!("{e}/{account}/{}", self.bucket),
260 None => format!("https://{account}.blob.core.windows.net/{}", self.bucket),
261 }
262 }
263
264 fn azure_url(&self, base: String) -> String {
266 let Backend::Azure { sas: Some(sas), .. } = &self.backend else {
267 return base;
268 };
269 if base.contains('?') {
270 format!("{base}&{sas}")
271 } else {
272 format!("{base}?{sas}")
273 }
274 }
275
276 fn s3_root(&self) -> String {
279 let Backend::S3 {
280 region, endpoint, ..
281 } = &self.backend
282 else {
283 unreachable!("s3_root on a non-S3 backend");
284 };
285 match endpoint {
286 Some(e) => format!("{e}/{}", self.bucket),
287 None if region == "us-east-1" => {
288 format!("https://{}.s3.amazonaws.com", self.bucket)
289 }
290 None => format!("https://{}.s3.{region}.amazonaws.com", self.bucket),
291 }
292 }
293
294 fn list(&self, prefix: &str) -> Result<Page, String> {
296 let dir = if prefix.is_empty() {
297 String::new()
298 } else {
299 format!("{prefix}/")
300 };
301 let mut prefixes = Vec::new();
302 let mut objects = Vec::new();
303 let mut page: Option<String> = None;
304 loop {
305 match &self.backend {
306 Backend::Gcs { .. } => {
307 let mut url = format!(
308 "https://storage.googleapis.com/storage/v1/b/{}/o?delimiter=/&prefix={}",
309 self.bucket,
310 urlencode(&dir)
311 );
312 if let Some(p) = &page {
313 url.push_str(&format!("&pageToken={}", urlencode(p)));
314 }
315 let resp: serde_json::Value = serde_json::from_str(&self.get(&url)?)
316 .map_err(|e| format!("listing: {e}"))?;
317 if let Some(err) = resp.pointer("/error/message").and_then(|v| v.as_str()) {
318 return Err(err.to_string());
319 }
320 if let Some(ps) = resp.pointer("/prefixes").and_then(|v| v.as_array()) {
321 prefixes.extend(
322 ps.iter()
323 .filter_map(|p| p.as_str())
324 .map(|p| p.trim_end_matches('/').to_string()),
325 );
326 }
327 if let Some(items) = resp.pointer("/items").and_then(|v| v.as_array()) {
328 for i in items {
329 let Some(key) = i.pointer("/name").and_then(|v| v.as_str()) else {
330 continue;
331 };
332 if key.ends_with('/') {
333 continue; }
335 objects.push((
336 key.to_string(),
337 i.pointer("/size")
338 .and_then(|v| v.as_str())
339 .and_then(|s| s.parse().ok()),
340 i.pointer("/updated")
341 .and_then(|v| v.as_str())
342 .map(str::to_string),
343 ));
344 }
345 }
346 page = resp
347 .pointer("/nextPageToken")
348 .and_then(|v| v.as_str())
349 .map(str::to_string);
350 }
351 Backend::S3 { .. } => {
352 let mut url = format!(
353 "{}/?list-type=2&delimiter=/&prefix={}",
354 self.s3_root(),
355 urlencode(&dir)
356 );
357 if let Some(p) = &page {
358 url.push_str(&format!("&continuation-token={}", urlencode(p)));
359 }
360 let xml = self.get(&url)?;
361 let (ps, os, next) = parse_s3_listing(&xml)?;
362 prefixes.extend(ps);
363 objects.extend(os);
364 page = next;
365 }
366 Backend::Azure { .. } => {
367 let mut url = format!(
368 "{}?restype=container&comp=list&delimiter=/&prefix={}",
369 self.azure_root(),
370 urlencode(&dir)
371 );
372 if let Some(p) = &page {
373 url.push_str(&format!("&marker={}", urlencode(p)));
374 }
375 let xml = self.get(&self.azure_url(url))?;
376 let (ps, os, next) = parse_azure_listing(&xml)?;
377 prefixes.extend(ps);
378 objects.extend(os);
379 page = next;
380 }
381 }
382 if page.is_none() {
383 break;
384 }
385 }
386 Ok((prefixes, objects))
387 }
388
389 fn push(&self, node: Node) -> NodeId {
390 let mut nodes = self.nodes.borrow_mut();
391 let id = NodeId(nodes.len() as u64);
392 nodes.push(node);
393 id
394 }
395
396 fn content_of(&self, node: NodeId) -> Option<String> {
398 if let Some(c) = &*self.nodes.borrow()[node.0 as usize].content.borrow() {
399 return Some(c.clone());
400 }
401 let (key, is_object) = {
402 let nodes = self.nodes.borrow();
403 let n = &nodes[node.0 as usize];
404 (n.key.clone(), n.is_object)
405 };
406 if !is_object {
407 return None;
408 }
409 let url = match &self.backend {
410 Backend::Gcs { .. } => format!(
411 "https://storage.googleapis.com/storage/v1/b/{}/o/{}?alt=media",
412 self.bucket,
413 urlencode(&key)
414 ),
415 Backend::S3 { .. } => format!(
416 "{}/{}",
417 self.s3_root(),
418 urlencode(&key).replace("%2F", "/")
419 ),
420 Backend::Azure { .. } => self.azure_url(format!(
421 "{}/{}",
422 self.azure_root(),
423 urlencode(&key).replace("%2F", "/")
424 )),
425 };
426 let text = self.get(&url).ok()?;
427 *self.nodes.borrow()[node.0 as usize].content.borrow_mut() = Some(text.clone());
428 Some(text)
429 }
430}
431
432#[allow(clippy::type_complexity)]
434fn parse_s3_listing(
435 xml: &str,
436) -> Result<
437 (
438 Vec<String>,
439 Vec<(String, Option<i64>, Option<String>)>,
440 Option<String>,
441 ),
442 String,
443> {
444 use quick_xml::events::Event;
445 let mut reader = quick_xml::Reader::from_str(xml);
446 let mut prefixes = Vec::new();
447 let mut objects = Vec::new();
448 let mut next = None;
449 let mut path: Vec<String> = Vec::new();
450 let mut cur: (Option<String>, Option<i64>, Option<String>) = (None, None, None);
451 loop {
452 match reader.read_event().map_err(|e| format!("listing: {e}"))? {
453 Event::Start(e) => path.push(String::from_utf8_lossy(e.name().as_ref()).into_owned()),
454 Event::End(e) => {
455 let name = String::from_utf8_lossy(e.name().as_ref()).into_owned();
456 if name == "Contents"
457 && let (Some(k), size, updated) = std::mem::take(&mut cur)
458 && !k.ends_with('/')
459 {
460 objects.push((k, size, updated));
461 }
462 path.pop();
463 }
464 Event::Text(t) => {
465 let text = t.xml_content().map_err(|e| e.to_string())?.into_owned();
466 match path.as_slice() {
467 [.., a, b] if a == "CommonPrefixes" && b == "Prefix" => {
468 prefixes.push(text.trim_end_matches('/').to_string());
469 }
470 [.., a, b] if a == "Contents" && b == "Key" => cur.0 = Some(text),
471 [.., a, b] if a == "Contents" && b == "Size" => {
472 cur.1 = text.parse().ok();
473 }
474 [.., a, b] if a == "Contents" && b == "LastModified" => {
475 cur.2 = Some(text);
476 }
477 [.., b] if b == "NextContinuationToken" => next = Some(text),
478 [.., a, b] if a == "ListBucketResult" && b == "NextContinuationToken" => {
479 next = Some(text)
480 }
481 _ => {}
482 }
483 }
484 Event::Eof => break,
485 _ => {}
486 }
487 }
488 Ok((prefixes, objects, next))
489}
490
491impl AstAdapter for ObjstoreAdapter {
492 fn root(&self) -> NodeId {
493 NodeId(0)
494 }
495
496 fn children(&self, node: NodeId) -> Vec<NodeId> {
497 if let Some(c) = self.nodes.borrow()[node.0 as usize]
498 .children
499 .borrow()
500 .as_ref()
501 {
502 return c.clone();
503 }
504 let (key, is_object) = {
505 let nodes = self.nodes.borrow();
506 let n = &nodes[node.0 as usize];
507 (n.key.clone(), n.is_object)
508 };
509 if is_object {
510 return Vec::new();
511 }
512 let (prefixes, objects) = self.list(&key).unwrap_or_default();
513 let mut ids = Vec::new();
514 for p in prefixes {
515 let name = p.rsplit('/').next().unwrap_or(&p).to_string();
516 ids.push(self.push(Node {
517 key: p,
518 name: Some(name),
519 parent: Some(node),
520 is_object: false,
521 size: None,
522 updated: None,
523 children: RefCell::new(None),
524 content: RefCell::new(None),
525 }));
526 }
527 for (k, size, updated) in objects {
528 let name = k.rsplit('/').next().unwrap_or(&k).to_string();
529 ids.push(self.push(Node {
530 key: k,
531 name: Some(name),
532 parent: Some(node),
533 is_object: true,
534 size,
535 updated,
536 children: RefCell::new(None),
537 content: RefCell::new(None),
538 }));
539 }
540 *self.nodes.borrow()[node.0 as usize].children.borrow_mut() = Some(ids.clone());
541 ids
542 }
543
544 fn name(&self, node: NodeId) -> Option<String> {
545 self.nodes.borrow()[node.0 as usize].name.clone()
546 }
547
548 fn parent(&self, node: NodeId) -> Option<NodeId> {
549 self.nodes.borrow()[node.0 as usize].parent
550 }
551
552 fn traits(&self, node: NodeId) -> Vec<String> {
554 let nodes = self.nodes.borrow();
555 let n = &nodes[node.0 as usize];
556 if n.parent.is_none() {
557 return Vec::new();
558 }
559 vec![if n.is_object { "object" } else { "prefix" }.to_string()]
560 }
561
562 fn default_value(&self, node: NodeId) -> Option<Value> {
564 self.content_of(node).map(Value::Str)
565 }
566
567 fn metadata(&self, node: NodeId, key: &str) -> Option<Value> {
569 let nodes = self.nodes.borrow();
570 let n = &nodes[node.0 as usize];
571 match key {
572 "size" => n.size.map(Value::bytes),
573 "updated" => n.updated.clone().map(Value::Str),
574 "key" => Some(Value::Str(n.key.clone())),
575 _ => None,
576 }
577 }
578}
579
580fn rfc1123_now() -> String {
583 let secs = std::time::SystemTime::now()
584 .duration_since(std::time::UNIX_EPOCH)
585 .expect("clock before 1970")
586 .as_secs();
587 let days = (secs / 86400) as i64;
588 let (h, mi, sec) = ((secs / 3600) % 24, (secs / 60) % 60, secs % 60);
589 let z = days + 719_468;
591 let era = z.div_euclid(146_097);
592 let doe = z.rem_euclid(146_097);
593 let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146_096) / 365;
594 let y = yoe + era * 400;
595 let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
596 let mp = (5 * doy + 2) / 153;
597 let d = doy - (153 * mp + 2) / 5 + 1;
598 let mo = if mp < 10 { mp + 3 } else { mp - 9 };
599 let y = if mo <= 2 { y + 1 } else { y };
600 let weekday = ["Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"]
601 [((days + 4).rem_euclid(7)) as usize];
602 let month = [
603 "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec",
604 ][(mo - 1) as usize];
605 format!("{weekday}, {d:02} {month} {y} {h:02}:{mi:02}:{sec:02} GMT")
606}
607
608fn azure_shared_key_headers(
613 url: &str,
614 account: &str,
615 key: &[u8],
616 date: &str,
617) -> Vec<(String, String)> {
618 let version = "2020-10-02";
619 let rest = url
620 .strip_prefix("https://")
621 .or_else(|| url.strip_prefix("http://"))
622 .unwrap_or(url);
623 let (_, path_query) = rest.split_once('/').unwrap_or((rest, ""));
624 let (path, query) = match path_query.split_once('?') {
625 Some((p, q)) => (p, q),
626 None => (path_query, ""),
627 };
628 let decoded_path = String::from_utf8_lossy(&percent_decode_bytes(path)).into_owned();
635 let mut params: Vec<(String, String)> = query
636 .split('&')
637 .filter(|s| !s.is_empty())
638 .map(|kv| match kv.split_once('=') {
639 Some((k, v)) => (
640 k.to_lowercase(),
641 String::from_utf8_lossy(&percent_decode_bytes(v)).into_owned(),
642 ),
643 None => (kv.to_lowercase(), String::new()),
644 })
645 .collect();
646 params.sort();
647 let mut resource = format!("/{account}/{decoded_path}");
648 for (k, v) in ¶ms {
649 resource.push_str(&format!("\n{k}:{v}"));
650 }
651 let headers = format!("x-ms-date:{date}\nx-ms-version:{version}\n");
652 let string_to_sign =
653 format!("GET\n\n\n\n\n\n\n\n\n\n\n\n{headers}{resource}");
654 let sig = quarb::base64(&hmac_sha256(key, string_to_sign.as_bytes()));
655 vec![
656 ("x-ms-date".to_string(), date.to_string()),
657 ("x-ms-version".to_string(), version.to_string()),
658 (
659 "authorization".to_string(),
660 format!("SharedKey {account}:{sig}"),
661 ),
662 ]
663}
664
665fn hmac_sha256(key: &[u8], msg: &[u8]) -> [u8; 32] {
666 let mut k = [0u8; 64];
667 if key.len() > 64 {
668 k[..32].copy_from_slice(&quarb::sha256(key));
669 } else {
670 k[..key.len()].copy_from_slice(key);
671 }
672 let ipad: Vec<u8> = k.iter().map(|b| b ^ 0x36).collect();
673 let opad: Vec<u8> = k.iter().map(|b| b ^ 0x5c).collect();
674 let inner = quarb::sha256(&[ipad.as_slice(), msg].concat());
675 quarb::sha256(&[opad.as_slice(), &inner].concat())
676}
677
678fn percent_decode_bytes(s: &str) -> Vec<u8> {
679 let b = s.as_bytes();
680 let mut out = Vec::with_capacity(b.len());
681 let mut i = 0;
682 while i < b.len() {
683 if b[i] == b'%'
684 && i + 2 < b.len()
685 && let (Some(h), Some(l)) = (
686 (b[i + 1] as char).to_digit(16),
687 (b[i + 2] as char).to_digit(16),
688 )
689 {
690 out.push((h * 16 + l) as u8);
691 i += 3;
692 } else {
693 out.push(b[i]);
694 i += 1;
695 }
696 }
697 out
698}
699
700fn parse_azure_listing(
703 xml: &str,
704) -> Result<
705 (
706 Vec<String>,
707 Vec<(String, Option<i64>, Option<String>)>,
708 Option<String>,
709 ),
710 String,
711> {
712 use quick_xml::events::Event;
713 let mut reader = quick_xml::Reader::from_str(xml);
714 let mut prefixes = Vec::new();
715 let mut objects: Vec<(String, Option<i64>, Option<String>)> = Vec::new();
716 let mut next = None;
717 let mut stack: Vec<String> = Vec::new();
718 let mut cur_name = String::new();
719 let mut cur_size = None;
720 let mut cur_updated = None;
721 loop {
722 match reader.read_event() {
723 Ok(Event::Start(e)) => {
724 stack.push(String::from_utf8_lossy(e.name().as_ref()).into_owned());
725 }
726 Ok(Event::End(e)) => {
727 let name = String::from_utf8_lossy(e.name().as_ref()).into_owned();
728 if name == "Blob" {
729 objects.push((
730 std::mem::take(&mut cur_name),
731 cur_size.take(),
732 cur_updated.take(),
733 ));
734 }
735 while stack.pop().is_some_and(|s| s != name) {}
736 }
737 Ok(Event::Text(t)) => {
738 let text = t.decode().map_err(|e| e.to_string())?.into_owned();
739 match stack.as_slice() {
740 [.., a, b] if a == "BlobPrefix" && b == "Name" => {
741 prefixes.push(text.trim_end_matches('/').to_string());
742 }
743 [.., a, b] if a == "Blob" && b == "Name" => cur_name = text,
744 [.., a, b] if a == "Properties" && b == "Content-Length" => {
745 cur_size = text.parse().ok();
746 }
747 [.., a, b] if a == "Properties" && b == "Last-Modified" => {
748 cur_updated = Some(text);
749 }
750 [.., b] if b == "NextMarker" => {
751 if !text.is_empty() {
752 next = Some(text);
753 }
754 }
755 _ => {}
756 }
757 }
758 Ok(Event::Eof) => break,
759 Err(e) => return Err(format!("listing XML: {e}")),
760 _ => {}
761 }
762 }
763 Ok((prefixes, objects, next))
764}