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
83        .send()
84        .await
85        .context("failed to get")?
86        .error_for_status()
87        .with_context(|| format!("{store} returned an error status for {location}"))?;
88    let meta = header_meta(location, response.headers()).context("Failed to get header")?;
89    let stream = response
90        .bytes_stream()
91        .map_err(|source| Error::Generic {
92            store,
93            source: Box::new(source),
94        })
95        .boxed();
96    Ok(GetResult {
97        range: 0..meta.size,
98        payload: GetResultPayload::Stream(stream),
99        meta,
100        attributes: object_store::Attributes::new(),
101    })
102}
103
104async fn exists(url: &str, store: &'static str, location: &Path, client: &Client) -> Result<bool> {
105    let request = client.request(Method::HEAD, url);
106    let response = request
107        .send()
108        .await
109        .with_context(|| format!("failed to send HEAD request for {location} to {store}"))?;
110    let status = response.status();
111    if status.is_success() {
112        Ok(true)
113    } else if status == reqwest::StatusCode::NOT_FOUND {
114        Ok(false)
115    } else {
116        Err(anyhow!(
117            "{store} returned unexpected status {status} for {location}"
118        ))
119    }
120}
121
122async fn size(url: &str, store: &'static str, location: &Path, client: &Client) -> Result<u64> {
123    let request = client.request(Method::HEAD, url);
124    let response = request
125        .send()
126        .await
127        .with_context(|| format!("failed to send HEAD request for {location} to {store}"))?
128        .error_for_status()
129        .with_context(|| format!("{store} returned an error status for {location}"))?;
130    let content_length = response
131        .headers()
132        .get(CONTENT_LENGTH)
133        .with_context(|| format!("{store} returned no content length for {location}"))?;
134    content_length
135        .to_str()
136        .context("bad content length header")?
137        .parse()
138        .context("invalid content length")
139}
140
141fn header_meta(location: &Path, headers: &HeaderMap) -> Result<ObjectMeta> {
142    let last_modified = headers
143        .get(LAST_MODIFIED)
144        .context("Missing last modified")?;
145
146    let content_length = headers
147        .get(CONTENT_LENGTH)
148        .context("Missing content length")?;
149
150    let last_modified = last_modified.to_str().context("bad header")?;
151    let last_modified = DateTime::parse_from_rfc2822(last_modified)
152        .context("invalid last modified")?
153        .with_timezone(&Utc);
154
155    let content_length = content_length.to_str().context("bad header")?;
156    let content_length = content_length.parse().context("invalid content length")?;
157
158    let e_tag = headers.get(ETAG).context("missing etag")?;
159    let e_tag = e_tag.to_str().context("bad header")?;
160
161    Ok(ObjectMeta {
162        location: location.clone(),
163        last_modified,
164        size: content_length,
165        e_tag: Some(e_tag.to_string()),
166        version: None,
167    })
168}
169
170#[cfg(test)]
171mod tests {
172    use std::fs;
173
174    use iota_config::object_storage_config::{ObjectStoreConfig, ObjectStoreType};
175    use object_store::path::Path;
176    use tempfile::TempDir;
177
178    use crate::object_store::{ObjectStoreGetExt, http::HttpDownloaderBuilder};
179
180    #[tokio::test]
181    pub async fn test_local_download() -> anyhow::Result<()> {
182        let input = TempDir::new()?;
183        let input_path = input.path();
184        let child = input_path.join("child");
185        fs::create_dir(&child)?;
186        let file1 = child.join("file1");
187        fs::write(file1, b"Lorem ipsum")?;
188        let grandchild = child.join("grand_child");
189        fs::create_dir(&grandchild)?;
190        let file2 = grandchild.join("file2");
191        fs::write(file2, b"Lorem ipsum")?;
192
193        let input_store = ObjectStoreConfig {
194            object_store: Some(ObjectStoreType::File),
195            directory: Some(input_path.to_path_buf()),
196            ..Default::default()
197        }
198        .make_http()?;
199
200        let downloaded = input_store.get_bytes(&Path::from("child/file1")).await?;
201        assert_eq!(downloaded.to_vec(), b"Lorem ipsum");
202        Ok(())
203    }
204
205    #[tokio::test]
206    pub async fn test_local_exists() -> anyhow::Result<()> {
207        let input = TempDir::new()?;
208        let input_path = input.path();
209        fs::write(input_path.join("file1"), b"Lorem ipsum")?;
210
211        let input_store = ObjectStoreConfig {
212            object_store: Some(ObjectStoreType::File),
213            directory: Some(input_path.to_path_buf()),
214            ..Default::default()
215        }
216        .make_http()?;
217
218        assert!(input_store.exists(&Path::from("file1")).await?);
219        assert!(!input_store.exists(&Path::from("missing")).await?);
220        Ok(())
221    }
222
223    #[tokio::test]
224    pub async fn test_local_object_size() -> anyhow::Result<()> {
225        let input = TempDir::new()?;
226        let input_path = input.path();
227        fs::write(input_path.join("file1"), b"Lorem ipsum")?;
228
229        let input_store = ObjectStoreConfig {
230            object_store: Some(ObjectStoreType::File),
231            directory: Some(input_path.to_path_buf()),
232            ..Default::default()
233        }
234        .make_http()?;
235
236        assert_eq!(input_store.object_size(&Path::from("file1")).await?, 11);
237        assert!(
238            input_store
239                .object_size(&Path::from("missing"))
240                .await
241                .is_err()
242        );
243        Ok(())
244    }
245}
iota_storage/object_store/http/mod.rs

iota_storage/object_store/http/
mod.rs
