iota_storage/object_store/http/
mod.rs

1// Copyright (c) Mysten Labs, Inc.
2// Modifications Copyright (c) 2024 IOTA Stiftung
3// SPDX-License-Identifier: Apache-2.0
4
5mod gcs;
6mod local;
7mod s3;
8
9use std::sync::Arc;
10
11use anyhow::{Context, Result, anyhow};
12use chrono::{DateTime, Utc};
13use futures::{StreamExt, TryStreamExt};
14use iota_config::object_storage_config::{ObjectStoreConfig, ObjectStoreType};
15use object_store::{Error, GetResult, GetResultPayload, ObjectMeta, path::Path};
16use reqwest::{
17    Client, Method,
18    header::{CONTENT_LENGTH, ETAG, HeaderMap, LAST_MODIFIED},
19};
20
21use crate::object_store::{
22    ObjectStoreGetExt,
23    http::{gcs::GoogleCloudStorage, local::LocalStorage, s3::AmazonS3},
24};
25
26// https://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html
27//
28// Do not URI-encode any of the unreserved characters that RFC 3986 defines:
29// A-Z, a-z, 0-9, hyphen ( - ), underscore ( _ ), period ( . ), and tilde ( ~ ).
30pub(crate) const STRICT_ENCODE_SET: percent_encoding::AsciiSet = percent_encoding::NON_ALPHANUMERIC
31    .remove(b'-')
32    .remove(b'.')
33    .remove(b'_')
34    .remove(b'~');
35const STRICT_PATH_ENCODE_SET: percent_encoding::AsciiSet = STRICT_ENCODE_SET.remove(b'/');
36static DEFAULT_USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"),);
37
38pub trait HttpDownloaderBuilder {
39    fn make_http(&self) -> Result<Arc<dyn ObjectStoreGetExt>>;
40}
41
42impl HttpDownloaderBuilder for ObjectStoreConfig {
43    fn make_http(&self) -> Result<Arc<dyn ObjectStoreGetExt>> {
44        match self.object_store {
45            Some(ObjectStoreType::File) => {
46                Ok(LocalStorage::new(self.directory.as_ref().unwrap()).map(Arc::new)?)
47            }
48            Some(ObjectStoreType::S3) => {
49                let bucket_endpoint = if let Some(endpoint) = &self.aws_endpoint {
50                    if self.aws_virtual_hosted_style_request {
51                        endpoint.clone()
52                    } else {
53                        let bucket = self.bucket.as_ref().unwrap();
54                        format!("{endpoint}/{bucket}")
55                    }
56                } else {
57                    let bucket = self.bucket.as_ref().unwrap();
58                    let region = self.aws_region.as_ref().unwrap();
59                    if self.aws_virtual_hosted_style_request {
60                        format!("https://{bucket}.s3.{region}.amazonaws.com")
61                    } else {
62                        format!("https://s3.{region}.amazonaws.com/{bucket}")
63                    }
64                };
65                Ok(AmazonS3::new(&bucket_endpoint).map(Arc::new)?)
66            }
67            Some(ObjectStoreType::GCS) => {
68                Ok(GoogleCloudStorage::new(self.bucket.as_ref().unwrap()).map(Arc::new)?)
69            }
70            _ => Err(anyhow!("At least one storage backend should be provided")),
71        }
72    }
73}
74
75async fn get(
76    url: &str,
77    store: &'static str,
78    location: &Path,
79    client: &Client,
80) -> Result<GetResult> {
81    let request = client.request(Method::GET, url);
82    let response = request.send().await.context("failed to get")?;
83    let meta = header_meta(location, response.headers()).context("Failed to get header")?;
84    let stream = response
85        .bytes_stream()
86        .map_err(|source| Error::Generic {
87            store,
88            source: Box::new(source),
89        })
90        .boxed();
91    Ok(GetResult {
92        range: 0..meta.size,
93        payload: GetResultPayload::Stream(stream),
94        meta,
95        attributes: object_store::Attributes::new(),
96    })
97}
98
99async fn exists(url: &str, store: &'static str, location: &Path, client: &Client) -> Result<bool> {
100    let request = client.request(Method::HEAD, url);
101    let response = request
102        .send()
103        .await
104        .with_context(|| format!("failed to send HEAD request for {location} to {store}"))?;
105    let status = response.status();
106    if status.is_success() {
107        Ok(true)
108    } else if status == reqwest::StatusCode::NOT_FOUND {
109        Ok(false)
110    } else {
111        Err(anyhow!(
112            "{store} returned unexpected status {status} for {location}"
113        ))
114    }
115}
116
117fn header_meta(location: &Path, headers: &HeaderMap) -> Result<ObjectMeta> {
118    let last_modified = headers
119        .get(LAST_MODIFIED)
120        .context("Missing last modified")?;
121
122    let content_length = headers
123        .get(CONTENT_LENGTH)
124        .context("Missing content length")?;
125
126    let last_modified = last_modified.to_str().context("bad header")?;
127    let last_modified = DateTime::parse_from_rfc2822(last_modified)
128        .context("invalid last modified")?
129        .with_timezone(&Utc);
130
131    let content_length = content_length.to_str().context("bad header")?;
132    let content_length = content_length.parse().context("invalid content length")?;
133
134    let e_tag = headers.get(ETAG).context("missing etag")?;
135    let e_tag = e_tag.to_str().context("bad header")?;
136
137    Ok(ObjectMeta {
138        location: location.clone(),
139        last_modified,
140        size: content_length,
141        e_tag: Some(e_tag.to_string()),
142        version: None,
143    })
144}
145
146#[cfg(test)]
147mod tests {
148    use std::fs;
149
150    use iota_config::object_storage_config::{ObjectStoreConfig, ObjectStoreType};
151    use object_store::path::Path;
152    use tempfile::TempDir;
153
154    use crate::object_store::{ObjectStoreGetExt, http::HttpDownloaderBuilder};
155
156    #[tokio::test]
157    pub async fn test_local_download() -> anyhow::Result<()> {
158        let input = TempDir::new()?;
159        let input_path = input.path();
160        let child = input_path.join("child");
161        fs::create_dir(&child)?;
162        let file1 = child.join("file1");
163        fs::write(file1, b"Lorem ipsum")?;
164        let grandchild = child.join("grand_child");
165        fs::create_dir(&grandchild)?;
166        let file2 = grandchild.join("file2");
167        fs::write(file2, b"Lorem ipsum")?;
168
169        let input_store = ObjectStoreConfig {
170            object_store: Some(ObjectStoreType::File),
171            directory: Some(input_path.to_path_buf()),
172            ..Default::default()
173        }
174        .make_http()?;
175
176        let downloaded = input_store.get_bytes(&Path::from("child/file1")).await?;
177        assert_eq!(downloaded.to_vec(), b"Lorem ipsum");
178        Ok(())
179    }
180
181    #[tokio::test]
182    pub async fn test_local_exists() -> anyhow::Result<()> {
183        let input = TempDir::new()?;
184        let input_path = input.path();
185        fs::write(input_path.join("file1"), b"Lorem ipsum")?;
186
187        let input_store = ObjectStoreConfig {
188            object_store: Some(ObjectStoreType::File),
189            directory: Some(input_path.to_path_buf()),
190            ..Default::default()
191        }
192        .make_http()?;
193
194        assert!(input_store.exists(&Path::from("file1")).await?);
195        assert!(!input_store.exists(&Path::from("missing")).await?);
196        Ok(())
197    }
198}
iota_storage/object_store/http/mod.rs

iota_storage/object_store/http/
mod.rs
