htsget_storage/
lib.rs

1//! Module providing the abstractions needed to read files from an storage
2//!
3
4pub use htsget_config::resolver::{IdResolver, ResolveResponse, StorageResolver};
5pub use htsget_config::types::{
6  Class, Format, Headers, HtsGetError, JsonResponse, Query, Response, Url,
7};
8
9#[cfg(feature = "experimental")]
10use crate::c4gh::storage::C4GHStorage;
11use crate::error::Result;
12use crate::error::StorageError;
13#[cfg(feature = "experimental")]
14use crate::error::StorageError::InvalidInput;
15#[cfg(feature = "url")]
16use crate::json_path::JsonPathStorage;
17use crate::local::FileStorage;
18#[cfg(feature = "aws")]
19use crate::s3::S3Storage;
20use crate::types::{BytesPositionOptions, DataBlock, GetOptions, HeadOptions, RangeUrlOptions};
21#[cfg(feature = "url")]
22use crate::url::UrlStorage;
23use async_trait::async_trait;
24use base64::Engine;
25use base64::engine::general_purpose;
26use cfg_if::cfg_if;
27#[cfg(feature = "experimental")]
28use htsget_config::encryption_scheme::EncryptionScheme;
29use htsget_config::storage;
30#[cfg(feature = "experimental")]
31use htsget_config::storage::c4gh::C4GHKeys;
32use pin_project_lite::pin_project;
33use std::fmt;
34use std::fmt::{Debug, Formatter};
35use std::pin::Pin;
36use std::task::{Context, Poll};
37use tokio::io::{AsyncRead, ReadBuf};
38#[cfg(feature = "experimental")]
39use tracing::debug;
40
41#[cfg(feature = "experimental")]
42pub mod c4gh;
43pub mod error;
44#[cfg(feature = "url")]
45pub mod json_path;
46pub mod local;
47#[cfg(feature = "aws")]
48pub mod s3;
49pub mod types;
50#[cfg(feature = "url")]
51pub mod url;
52
53pin_project! {
54  /// A Streamable type represents any AsyncRead data used by `StorageTrait`.
55  pub struct Streamable {
56    #[pin]
57    inner: Box<dyn AsyncRead + Send + Sync + Unpin + 'static>,
58  }
59}
60
61impl Streamable {
62  /// Create a new Streamable from an AsyncRead.
63  pub fn from_async_read(inner: impl AsyncRead + Send + Sync + Unpin + 'static) -> Self {
64    Self {
65      inner: Box::new(inner),
66    }
67  }
68}
69
70impl AsyncRead for Streamable {
71  fn poll_read(
72    self: Pin<&mut Self>,
73    cx: &mut Context<'_>,
74    buf: &mut ReadBuf<'_>,
75  ) -> Poll<std::io::Result<()>> {
76    self.project().inner.poll_read(cx, buf)
77  }
78}
79
80/// The top-level storage type is created from any `StorageTrait`.
81pub struct Storage {
82  inner: Box<dyn StorageTrait + Send + Sync + 'static>,
83}
84
85impl Storage {
86  /// Get the inner value.
87  pub fn into_inner(self) -> Box<dyn StorageTrait + Send + Sync> {
88    self.inner
89  }
90}
91
92impl Clone for Storage {
93  fn clone(&self) -> Self {
94    Self {
95      inner: self.inner.clone_box(),
96    }
97  }
98}
99
100impl Debug for Storage {
101  fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
102    write!(f, "Storage")
103  }
104}
105
106#[async_trait]
107impl StorageMiddleware for Storage {
108  async fn preprocess(&mut self, key: &str, options: GetOptions<'_>) -> Result<()> {
109    self.inner.preprocess(key, options).await
110  }
111
112  async fn postprocess(
113    &self,
114    key: &str,
115    positions_options: BytesPositionOptions<'_>,
116  ) -> Result<Vec<DataBlock>> {
117    self.inner.postprocess(key, positions_options).await
118  }
119}
120
121#[async_trait]
122impl StorageTrait for Storage {
123  async fn get(&self, key: &str, options: GetOptions<'_>) -> Result<Streamable> {
124    self.inner.get(key, options).await
125  }
126
127  async fn range_url(&self, key: &str, options: RangeUrlOptions<'_>) -> Result<Url> {
128    self.inner.range_url(key, options).await
129  }
130
131  async fn head(&self, key: &str, options: HeadOptions<'_>) -> Result<u64> {
132    self.inner.head(key, options).await
133  }
134
135  fn data_url(&self, data: Vec<u8>, class: Option<Class>) -> Url {
136    self.inner.data_url(data, class)
137  }
138}
139
140impl Storage {
141  #[cfg(feature = "experimental")]
142  /// Wrap an existing storage with C4GH storage
143  pub async fn from_c4gh_keys(
144    keys: Option<&C4GHKeys>,
145    encryption_scheme: Option<EncryptionScheme>,
146    storage: Storage,
147    query: &Query,
148    forward_public_key: bool,
149  ) -> Result<Storage> {
150    match (keys, encryption_scheme) {
151      (Some(keys), Some(EncryptionScheme::C4GH)) => {
152        let (
153          server_decryption_keys,
154          mut client_encryption_keys,
155          client_using_header,
156          encoded_public_key,
157        ) = keys
158          .clone()
159          .into_inner()
160          .await
161          .map_err(|err| StorageError::InternalError(err.to_string()))?;
162
163        if let Some(client_using_header) = client_using_header {
164          let public_key = client_using_header.get_public_key(query.request().headers())?;
165          client_encryption_keys
166            .iter_mut()
167            .for_each(|key| key.recipient_pubkey = public_key.clone());
168        }
169
170        debug!("attempting to fetch Crypt4GH data");
171
172        let encoded_public_key =
173          String::from_utf8(encoded_public_key).map_err(|err| InvalidInput(err.to_string()))?;
174        Ok(Storage::new(C4GHStorage::new_box(
175          server_decryption_keys,
176          client_encryption_keys,
177          storage.into_inner(),
178          forward_public_key,
179          encoded_public_key,
180        )))
181      }
182      (None, Some(EncryptionScheme::C4GH)) => Err(StorageError::UnsupportedFormat(
183        "C4GH keys have not been configured for this id".to_string(),
184      )),
185      _ => {
186        debug!("attempting to fetch non-encrypted data");
187        Ok(storage)
188      }
189    }
190  }
191
192  /// Create from local storage config.
193  pub async fn from_file(file: &storage::file::File, _query: &Query) -> Result<Storage> {
194    let storage = Storage::new(FileStorage::new(
195      file.local_path(),
196      file.clone(),
197      file.ticket_headers().to_vec(),
198    )?);
199
200    cfg_if! {
201      if #[cfg(feature = "experimental")] {
202        Self::from_c4gh_keys(file.keys(), _query.encryption_scheme(), storage, _query, false).await
203      } else {
204        Ok(storage)
205      }
206    }
207  }
208
209  /// Create from s3 config.
210  #[cfg(feature = "aws")]
211  pub async fn from_s3(s3: &storage::s3::S3, _query: &Query) -> Result<Storage> {
212    let storage = Storage::new(
213      S3Storage::new_with_default_config(
214        s3.bucket().to_string(),
215        s3.endpoint().map(str::to_string),
216        s3.path_style(),
217      )
218      .await,
219    );
220
221    cfg_if! {
222      if #[cfg(feature = "experimental")] {
223        Self::from_c4gh_keys(s3.keys(), _query.encryption_scheme(), storage, _query, false).await
224      } else {
225        Ok(storage)
226      }
227    }
228  }
229
230  /// Create from url config, converting to either a  [`UrlStorage`] or a
231  /// [`JsonPathStorage`] on parsing.
232  #[cfg(feature = "url")]
233  pub async fn from_url(mut url: storage::url::Url, _query: &Query) -> Result<Storage> {
234    use htsget_config::config::advanced::callout::{Parse, TicketSource};
235    use htsget_config::storage::url::JsonPathOrUrl;
236
237    let client = url
238      .client_cloned()
239      .map_err(|err| StorageError::InternalError(err.to_string()))?;
240
241    let allow_headers_backend = url.forward().headers().allow().to_vec();
242    let deny_headers_backend = url.forward().headers().deny().to_vec();
243    let allow_headers_client = url.reflect().headers().allow().to_vec();
244    let deny_headers_client = url.reflect().headers().deny().to_vec();
245
246    let storage = match url.parse().clone() {
247      Parse::Bytes { ticket_url } => {
248        let response_url = ticket_url.unwrap_or_else(|| url.url().clone());
249        Storage::new(UrlStorage::new(
250          client,
251          url.url().clone(),
252          response_url,
253          allow_headers_backend,
254          deny_headers_backend,
255          allow_headers_client,
256          deny_headers_client,
257        ))
258      }
259      Parse::JsonPath {
260        content_path,
261        size_path,
262        ticket,
263      } => {
264        let response_path = ticket.map(|t| match t {
265          TicketSource::JsonPath { path } => JsonPathOrUrl::JsonPath(path),
266          TicketSource::Url { url } => JsonPathOrUrl::Url(url),
267        });
268        let mut json_path_storage = JsonPathStorage::new(
269          client,
270          url.url().clone(),
271          content_path,
272          size_path,
273          response_path,
274          allow_headers_backend,
275          allow_headers_client,
276        );
277        json_path_storage.set_deny_headers_backend(deny_headers_backend);
278        json_path_storage.set_deny_headers_client(deny_headers_client);
279        Storage::new(json_path_storage)
280      }
281    };
282
283    cfg_if! {
284      if #[cfg(feature = "experimental")] {
285        Self::from_c4gh_keys(url.keys(), _query.encryption_scheme(), storage, _query, url.forward_public_key()).await
286      } else {
287        Ok(storage)
288      }
289    }
290  }
291
292  pub fn new(inner: impl StorageTrait + Send + Sync + 'static) -> Self {
293    Self {
294      inner: Box::new(inner),
295    }
296  }
297}
298
299/// A Storage represents some kind of object based storage (either locally or in the cloud)
300/// that can be used to retrieve files for alignments, variants or its respective indexes.
301#[async_trait]
302pub trait StorageTrait: StorageMiddleware + StorageClone {
303  /// Get the object using the key.
304  async fn get(&self, key: &str, options: GetOptions<'_>) -> Result<Streamable>;
305
306  /// Get the url of the object represented by the key using a bytes range. It is not required for
307  /// this function to check for the existent of the key, so this should be ensured beforehand.
308  async fn range_url(&self, key: &str, options: RangeUrlOptions<'_>) -> Result<Url>;
309
310  /// Get the size of the object represented by the key.
311  async fn head(&self, key: &str, options: HeadOptions<'_>) -> Result<u64>;
312
313  /// Get the url of the object using an inline data uri.
314  fn data_url(&self, data: Vec<u8>, class: Option<Class>) -> Url {
315    Url::new(format!(
316      "data:;base64,{}",
317      general_purpose::STANDARD.encode(data)
318    ))
319    .set_class(class)
320  }
321}
322
323/// Allow the `StorageTrait` to be cloned. This allows cloning a dynamic trait inside a Box.
324/// See https://crates.io/crates/dyn-clone for a similar pattern.
325pub trait StorageClone {
326  fn clone_box(&self) -> Box<dyn StorageTrait + Send + Sync>;
327}
328
329impl<T> StorageClone for T
330where
331  T: StorageTrait + Send + Sync + Clone + 'static,
332{
333  fn clone_box(&self) -> Box<dyn StorageTrait + Send + Sync> {
334    Box::new(self.clone())
335  }
336}
337
338/// A middleware trait which related to transforming or processing data returned from `StorageTrait`.
339#[async_trait]
340pub trait StorageMiddleware {
341  /// Preprocess any required state before it is requested by `StorageTrait`.
342  async fn preprocess(&mut self, _key: &str, _options: GetOptions<'_>) -> Result<()> {
343    Ok(())
344  }
345
346  /// Postprocess data blocks before they are returned to the client.
347  async fn postprocess(
348    &self,
349    _key: &str,
350    positions_options: BytesPositionOptions<'_>,
351  ) -> Result<Vec<DataBlock>> {
352    Ok(DataBlock::from_bytes_positions(
353      positions_options.into_inner(),
354    ))
355  }
356}
357
358/// Formats a url for use with storage.
359pub trait UrlFormatter {
360  /// Returns the url with the path.
361  fn format_url<K: AsRef<str>>(&self, key: K) -> Result<String>;
362}
363
364impl UrlFormatter for storage::file::File {
365  fn format_url<K: AsRef<str>>(&self, key: K) -> Result<String> {
366    let mut url = if let Some(origin) = self.ticket_origin() {
367      origin.to_string()
368    } else {
369      format!("{}://{}", self.scheme(), self.authority())
370    };
371    if !url.ends_with('/') {
372      url = format!("{url}/");
373    }
374
375    let url = ::url::Url::parse(&url).map_err(|err| StorageError::InvalidUri(err.to_string()))?;
376    url
377      .join(key.as_ref())
378      .map_err(|err| StorageError::InvalidUri(err.to_string()))
379      .map(|url| url.to_string())
380  }
381}
382
383#[cfg(test)]
384mod tests {
385  use super::*;
386  use crate::local::FileStorage;
387  use htsget_config::types::Scheme;
388  use htsget_test::util::default_dir_data;
389  use http::uri::Authority;
390  #[cfg(feature = "experimental")]
391  use {
392    htsget_config::storage::c4gh::header::C4GHHeader,
393    htsget_config::types::Request,
394    htsget_test::util::default_dir,
395    http::{HeaderMap, HeaderName},
396    tokio::fs,
397  };
398
399  #[test]
400  fn data_url() {
401    let result = FileStorage::<storage::file::File>::new(
402      default_dir_data(),
403      storage::file::File::default(),
404      vec![],
405    )
406    .unwrap()
407    .data_url(b"Hello World!".to_vec(), Some(Class::Header));
408    let url = data_url::DataUrl::process(&result.url);
409    let (result, _) = url.unwrap().decode_to_vec().unwrap();
410    assert_eq!(result, b"Hello World!");
411  }
412
413  #[test]
414  fn http_formatter_authority() {
415    let formatter = storage::file::File::new(
416      Scheme::Http,
417      Authority::from_static("127.0.0.1:8080"),
418      "data".to_string(),
419    );
420    test_formatter_authority(formatter, "http");
421  }
422
423  #[test]
424  fn https_formatter_authority() {
425    let formatter = storage::file::File::new(
426      Scheme::Https,
427      Authority::from_static("127.0.0.1:8080"),
428      "data".to_string(),
429    );
430    test_formatter_authority(formatter, "https");
431  }
432
433  #[cfg(feature = "experimental")]
434  #[tokio::test]
435  async fn from_c4gh_keys() {
436    let server_keys = tokio::spawn(async { Ok(C4GHKeys::from_key_pair(vec![], vec![])) });
437    let client_keys = tokio::spawn(async { Ok(C4GHKeys::from_key_pair(vec![], vec![])) });
438    let encoded_public_key = tokio::spawn(async { Ok(vec![]) });
439    let storage = Storage::new(
440      FileStorage::new(default_dir_data(), storage::file::File::default(), vec![]).unwrap(),
441    );
442
443    let result = Storage::from_c4gh_keys(
444      Some(&C4GHKeys::from_join_handle(
445        server_keys,
446        client_keys,
447        None,
448        encoded_public_key,
449      )),
450      Some(EncryptionScheme::C4GH),
451      storage.clone(),
452      &Default::default(),
453      true,
454    )
455    .await;
456    assert!(result.is_ok());
457
458    let result =
459      Storage::from_c4gh_keys(None, None, storage.clone(), &Default::default(), true).await;
460    assert!(result.is_ok());
461
462    let public_key = fs::read_to_string(default_dir().join("data/c4gh/keys/alice.pub"))
463      .await
464      .unwrap();
465    let encoded_key = general_purpose::STANDARD.encode(public_key);
466
467    let mut headers = HeaderMap::new();
468    headers.insert(
469      C4GHHeader::format_header_name()
470        .parse::<HeaderName>()
471        .unwrap(),
472      encoded_key.parse().unwrap(),
473    );
474    let query = Query::new(
475      "id".to_string(),
476      Format::Bam,
477      Request::new("id".to_string(), Default::default(), headers),
478    );
479
480    let server_keys = tokio::spawn(async { Ok(C4GHKeys::from_key_pair(vec![], vec![])) });
481    let client_keys = tokio::spawn(async { Ok(C4GHKeys::from_key_pair(vec![], vec![])) });
482    let encoded_public_key = tokio::spawn(async { Ok(vec![]) });
483
484    let result = Storage::from_c4gh_keys(
485      Some(&C4GHKeys::from_join_handle(
486        server_keys,
487        client_keys,
488        Some(C4GHHeader),
489        encoded_public_key,
490      )),
491      Some(EncryptionScheme::C4GH),
492      storage.clone(),
493      &query,
494      true,
495    )
496    .await;
497    assert!(result.is_ok());
498
499    let result =
500      Storage::from_c4gh_keys(None, None, storage.clone(), &Default::default(), true).await;
501    assert!(result.is_ok());
502
503    let result = Storage::from_c4gh_keys(
504      None,
505      Some(EncryptionScheme::C4GH),
506      storage,
507      &Default::default(),
508      true,
509    )
510    .await;
511    assert!(matches!(result, Err(StorageError::UnsupportedFormat(_))));
512  }
513
514  fn test_formatter_authority(formatter: storage::file::File, scheme: &str) {
515    assert_eq!(
516      formatter.format_url("path").unwrap(),
517      format!("{scheme}://127.0.0.1:8080/path")
518    )
519  }
520}
htsget_storage/lib.rs

htsget_storage/
lib.rs
