From ec61a5f86d21a0e4fe93e58bef95ba4f041f388e Mon Sep 17 00:00:00 2001 From: Zakariya Stasa Date: Fri, 7 Aug 2026 11:27:51 +0100 Subject: [PATCH] feat: add refreshable storage credential providers --- crates/iceberg/public-api.txt | 55 ++++++ crates/iceberg/src/io/file_io.rs | 52 ++++- crates/iceberg/src/io/storage/config/gcs.rs | 6 + crates/iceberg/src/io/storage/config/s3.rs | 6 + crates/iceberg/src/io/storage/mod.rs | 207 +++++++++++++++++++- 5 files changed, 321 insertions(+), 5 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index acb588dd78..2eb50ee6f7 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -693,6 +693,13 @@ pub fn iceberg::inspect::SnapshotsTable<'a>::new(table: &'a iceberg::table::Tabl pub async fn iceberg::inspect::SnapshotsTable<'a>::scan(&self) -> iceberg::Result pub fn iceberg::inspect::SnapshotsTable<'a>::schema(&self) -> iceberg::spec::Schema pub mod iceberg::io +pub enum iceberg::io::StorageCredentialKind +pub iceberg::io::StorageCredentialKind::Gcs(iceberg::io::GcsCredential) +pub iceberg::io::StorageCredentialKind::S3(iceberg::io::S3Credential) +impl core::clone::Clone for iceberg::io::StorageCredentialKind +pub fn iceberg::io::StorageCredentialKind::clone(&self) -> iceberg::io::StorageCredentialKind +impl core::fmt::Debug for iceberg::io::StorageCredentialKind +pub fn iceberg::io::StorageCredentialKind::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result pub struct iceberg::io::AzdlsConfig pub iceberg::io::AzdlsConfig::account_key: core::option::Option pub iceberg::io::AzdlsConfig::account_name: core::option::Option @@ -743,6 +750,7 @@ impl iceberg::io::FileIOBuilder pub fn iceberg::io::FileIOBuilder::build(self) -> iceberg::io::FileIO pub fn iceberg::io::FileIOBuilder::config(&self) -> &iceberg::io::StorageConfig pub fn iceberg::io::FileIOBuilder::new(factory: alloc::sync::Arc) -> Self +pub fn iceberg::io::FileIOBuilder::with_credential_provider(self, provider: alloc::sync::Arc) -> Self pub fn iceberg::io::FileIOBuilder::with_prop(self, key: impl alloc::string::ToString, value: impl alloc::string::ToString) -> Self pub fn iceberg::io::FileIOBuilder::with_props(self, args: impl core::iter::traits::collect::IntoIterator) -> Self impl core::clone::Clone for iceberg::io::FileIOBuilder @@ -779,6 +787,15 @@ impl serde_core::ser::Serialize for iceberg::io::GcsConfig pub fn iceberg::io::GcsConfig::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer impl<'de> serde_core::de::Deserialize<'de> for iceberg::io::GcsConfig pub fn iceberg::io::GcsConfig::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> +pub struct iceberg::io::GcsCredential +impl iceberg::io::GcsCredential +pub fn iceberg::io::GcsCredential::into_token(self) -> alloc::string::String +pub fn iceberg::io::GcsCredential::new(token: impl core::convert::Into) -> Self +pub fn iceberg::io::GcsCredential::token(&self) -> &str +impl core::clone::Clone for iceberg::io::GcsCredential +pub fn iceberg::io::GcsCredential::clone(&self) -> iceberg::io::GcsCredential +impl core::fmt::Debug for iceberg::io::GcsCredential +pub fn iceberg::io::GcsCredential::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result pub struct iceberg::io::HfConfig pub iceberg::io::HfConfig::endpoint: core::option::Option pub iceberg::io::HfConfig::revision: core::option::Option @@ -846,6 +863,7 @@ impl core::fmt::Debug for iceberg::io::LocalFsStorageFactory pub fn iceberg::io::LocalFsStorageFactory::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl iceberg::io::StorageFactory for iceberg::io::LocalFsStorageFactory pub fn iceberg::io::LocalFsStorageFactory::build(&self, _config: &iceberg::io::StorageConfig) -> iceberg::Result> +pub fn iceberg::io::LocalFsStorageFactory::build_with_credentials(&self, config: &iceberg::io::StorageConfig, credential_provider: core::option::Option>) -> iceberg::Result> impl serde_core::ser::Serialize for iceberg::io::LocalFsStorageFactory pub fn iceberg::io::LocalFsStorageFactory::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer impl<'de> serde_core::de::Deserialize<'de> for iceberg::io::LocalFsStorageFactory @@ -884,6 +902,7 @@ impl core::fmt::Debug for iceberg::io::MemoryStorageFactory pub fn iceberg::io::MemoryStorageFactory::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl iceberg::io::StorageFactory for iceberg::io::MemoryStorageFactory pub fn iceberg::io::MemoryStorageFactory::build(&self, _config: &iceberg::io::StorageConfig) -> iceberg::Result> +pub fn iceberg::io::MemoryStorageFactory::build_with_credentials(&self, config: &iceberg::io::StorageConfig, credential_provider: core::option::Option>) -> iceberg::Result> impl serde_core::ser::Serialize for iceberg::io::MemoryStorageFactory pub fn iceberg::io::MemoryStorageFactory::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer impl<'de> serde_core::de::Deserialize<'de> for iceberg::io::MemoryStorageFactory @@ -959,6 +978,17 @@ impl serde_core::ser::Serialize for iceberg::io::S3Config pub fn iceberg::io::S3Config::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer impl<'de> serde_core::de::Deserialize<'de> for iceberg::io::S3Config pub fn iceberg::io::S3Config::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> +pub struct iceberg::io::S3Credential +impl iceberg::io::S3Credential +pub fn iceberg::io::S3Credential::access_key_id(&self) -> &str +pub fn iceberg::io::S3Credential::into_parts(self) -> (alloc::string::String, alloc::string::String, core::option::Option) +pub fn iceberg::io::S3Credential::new(access_key_id: impl core::convert::Into, secret_access_key: impl core::convert::Into, session_token: core::option::Option) -> Self +pub fn iceberg::io::S3Credential::secret_access_key(&self) -> &str +pub fn iceberg::io::S3Credential::session_token(&self) -> core::option::Option<&str> +impl core::clone::Clone for iceberg::io::S3Credential +pub fn iceberg::io::S3Credential::clone(&self) -> iceberg::io::S3Credential +impl core::fmt::Debug for iceberg::io::S3Credential +pub fn iceberg::io::S3Credential::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result pub struct iceberg::io::StorageConfig impl iceberg::io::StorageConfig pub fn iceberg::io::StorageConfig::from_props(props: std::collections::hash::map::HashMap) -> Self @@ -996,6 +1026,19 @@ impl serde_core::ser::Serialize for iceberg::io::StorageConfig pub fn iceberg::io::StorageConfig::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer impl<'de> serde_core::de::Deserialize<'de> for iceberg::io::StorageConfig pub fn iceberg::io::StorageConfig::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> +pub struct iceberg::io::StorageCredential +impl iceberg::io::StorageCredential +pub fn iceberg::io::StorageCredential::expires_at(&self) -> core::option::Option +pub fn iceberg::io::StorageCredential::into_kind(self) -> iceberg::io::StorageCredentialKind +pub fn iceberg::io::StorageCredential::kind(&self) -> &iceberg::io::StorageCredentialKind +pub fn iceberg::io::StorageCredential::new(kind: iceberg::io::StorageCredentialKind) -> Self +pub fn iceberg::io::StorageCredential::prefix(&self) -> core::option::Option<&str> +pub fn iceberg::io::StorageCredential::with_expiration(self, expires_at: std::time::SystemTime) -> Self +pub fn iceberg::io::StorageCredential::with_prefix(self, prefix: impl core::convert::Into) -> Self +impl core::clone::Clone for iceberg::io::StorageCredential +pub fn iceberg::io::StorageCredential::clone(&self) -> iceberg::io::StorageCredential +impl core::fmt::Debug for iceberg::io::StorageCredential +pub fn iceberg::io::StorageCredential::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result pub const iceberg::io::ADLS_ACCOUNT_KEY: &str pub const iceberg::io::ADLS_ACCOUNT_NAME: &str pub const iceberg::io::ADLS_AUTHORITY_HOST: &str @@ -1004,6 +1047,8 @@ pub const iceberg::io::ADLS_CLIENT_SECRET: &str pub const iceberg::io::ADLS_CONNECTION_STRING: &str pub const iceberg::io::ADLS_SAS_TOKEN: &str pub const iceberg::io::ADLS_TENANT_ID: &str +pub const iceberg::io::AWS_REFRESH_CREDENTIALS_ENABLED: &str +pub const iceberg::io::AWS_REFRESH_CREDENTIALS_ENDPOINT: &str pub const iceberg::io::CLIENT_REGION: &str pub const iceberg::io::GCS_ALLOW_ANONYMOUS: &str pub const iceberg::io::GCS_CREDENTIALS_JSON: &str @@ -1011,8 +1056,11 @@ pub const iceberg::io::GCS_DISABLE_CONFIG_LOAD: &str pub const iceberg::io::GCS_DISABLE_VM_METADATA: &str pub const iceberg::io::GCS_NO_AUTH: &str pub const iceberg::io::GCS_PROJECT_ID: &str +pub const iceberg::io::GCS_REFRESH_CREDENTIALS_ENABLED: &str +pub const iceberg::io::GCS_REFRESH_CREDENTIALS_ENDPOINT: &str pub const iceberg::io::GCS_SERVICE_PATH: &str pub const iceberg::io::GCS_TOKEN: &str +pub const iceberg::io::GCS_TOKEN_EXPIRES_AT: &str pub const iceberg::io::GCS_USER_PROJECT: &str pub const iceberg::io::HF_ENDPOINT: &str pub const iceberg::io::HF_REVISION: &str @@ -1032,6 +1080,7 @@ pub const iceberg::io::S3_PATH_STYLE_ACCESS: &str pub const iceberg::io::S3_REGION: &str pub const iceberg::io::S3_SECRET_ACCESS_KEY: &str pub const iceberg::io::S3_SESSION_TOKEN: &str +pub const iceberg::io::S3_SESSION_TOKEN_EXPIRES_AT_MS: &str pub const iceberg::io::S3_SSE_KEY: &str pub const iceberg::io::S3_SSE_MD5: &str pub const iceberg::io::S3_SSE_TYPE: &str @@ -1083,12 +1132,18 @@ pub fn iceberg::io::MemoryStorage::read<'life0, 'life1, 'async_trait>(&'life0 se pub fn iceberg::io::MemoryStorage::reader<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg::io::MemoryStorage::write<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str, bs: bytes::bytes::Bytes) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg::io::MemoryStorage::writer<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub trait iceberg::io::StorageCredentialProvider: core::fmt::Debug + core::marker::Send + core::marker::Sync +pub fn iceberg::io::StorageCredentialProvider::load_credential<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub fn iceberg::io::StorageCredentialProvider::supports_path(&self, _path: &str) -> bool pub trait iceberg::io::StorageFactory: core::fmt::Debug + core::marker::Send + core::marker::Sync + typetag::Serialize + typetag::Deserialize pub fn iceberg::io::StorageFactory::build(&self, config: &iceberg::io::StorageConfig) -> iceberg::Result> +pub fn iceberg::io::StorageFactory::build_with_credentials(&self, config: &iceberg::io::StorageConfig, credential_provider: core::option::Option>) -> iceberg::Result> impl iceberg::io::StorageFactory for iceberg::io::LocalFsStorageFactory pub fn iceberg::io::LocalFsStorageFactory::build(&self, _config: &iceberg::io::StorageConfig) -> iceberg::Result> +pub fn iceberg::io::LocalFsStorageFactory::build_with_credentials(&self, config: &iceberg::io::StorageConfig, credential_provider: core::option::Option>) -> iceberg::Result> impl iceberg::io::StorageFactory for iceberg::io::MemoryStorageFactory pub fn iceberg::io::MemoryStorageFactory::build(&self, _config: &iceberg::io::StorageConfig) -> iceberg::Result> +pub fn iceberg::io::MemoryStorageFactory::build_with_credentials(&self, config: &iceberg::io::StorageConfig, credential_provider: core::option::Option>) -> iceberg::Result> pub mod iceberg::memory pub struct iceberg::memory::MemoryCatalog impl core::fmt::Debug for iceberg::memory::MemoryCatalog diff --git a/crates/iceberg/src/io/file_io.rs b/crates/iceberg/src/io/file_io.rs index cd0a4434c4..0257e6a5b2 100644 --- a/crates/iceberg/src/io/file_io.rs +++ b/crates/iceberg/src/io/file_io.rs @@ -22,7 +22,8 @@ use bytes::Bytes; use futures::{Stream, StreamExt}; use super::storage::{ - LocalFsStorageFactory, MemoryStorageFactory, Storage, StorageConfig, StorageFactory, + LocalFsStorageFactory, MemoryStorageFactory, Storage, StorageConfig, StorageCredentialProvider, + StorageFactory, }; use crate::Result; @@ -65,6 +66,8 @@ pub struct FileIO { config: StorageConfig, /// Factory for creating storage instances factory: Arc, + /// Optional provider of refreshable, backend-specific credentials + credential_provider: Option>, /// Cached storage instance (lazily initialized) storage: Arc>>, } @@ -77,6 +80,7 @@ impl FileIO { Self { config: StorageConfig::new(), factory: Arc::new(MemoryStorageFactory), + credential_provider: None, storage: Arc::new(OnceLock::new()), } } @@ -88,6 +92,7 @@ impl FileIO { Self { config: StorageConfig::new(), factory: Arc::new(LocalFsStorageFactory), + credential_provider: None, storage: Arc::new(OnceLock::new()), } } @@ -107,8 +112,11 @@ impl FileIO { return Ok(storage.clone()); } - // Build the storage - let storage = self.factory.build(&self.config)?; + // Build the storage, passing any credential provider so backends that + // support refreshable credentials can wire it into their operators. + let storage = self + .factory + .build_with_credentials(&self.config, self.credential_provider.clone())?; // Try to set it (another thread might have set it first) let _ = self.storage.set(storage.clone()); @@ -191,6 +199,8 @@ pub struct FileIOBuilder { factory: Arc, /// Storage configuration config: StorageConfig, + /// Optional provider of refreshable, backend-specific credentials + credential_provider: Option>, } impl FileIOBuilder { @@ -199,6 +209,7 @@ impl FileIOBuilder { Self { factory, config: StorageConfig::new(), + credential_provider: None, } } @@ -224,11 +235,21 @@ impl FileIOBuilder { &self.config } + /// Attach a provider of refreshable, backend-specific credentials. + pub fn with_credential_provider( + mut self, + provider: Arc, + ) -> Self { + self.credential_provider = Some(provider); + self + } + /// Builds [`FileIO`]. pub fn build(self) -> FileIO { FileIO { config: self.config, factory: self.factory, + credential_provider: self.credential_provider, storage: Arc::new(OnceLock::new()), } } @@ -397,7 +418,20 @@ mod tests { use tempfile::TempDir; use super::{FileIO, FileIOBuilder}; - use crate::io::{LocalFsStorageFactory, MemoryStorageFactory}; + use crate::io::{ + LocalFsStorageFactory, MemoryStorageFactory, StorageCredential, StorageCredentialProvider, + }; + use crate::{ErrorKind, Result}; + + #[derive(Debug)] + struct TestCredentialProvider; + + #[async_trait::async_trait] + impl StorageCredentialProvider for TestCredentialProvider { + async fn load_credential(&self, _path: &str) -> Result { + unreachable!("unsupported factories must reject the provider before loading from it") + } + } fn create_local_file_io() -> FileIO { FileIO::new_with_fs() @@ -544,4 +578,14 @@ mod tests { assert_eq!(file_io.config().get("key1"), Some(&"value1".to_string())); assert_eq!(file_io.config().get("key2"), Some(&"value2".to_string())); } + + #[tokio::test] + async fn test_file_io_rejects_credentials_for_unsupported_factory() { + let file_io = FileIOBuilder::new(Arc::new(MemoryStorageFactory)) + .with_credential_provider(Arc::new(TestCredentialProvider)) + .build(); + + let err = file_io.exists("memory://file").await.unwrap_err(); + assert_eq!(err.kind(), ErrorKind::FeatureUnsupported, "{err}"); + } } diff --git a/crates/iceberg/src/io/storage/config/gcs.rs b/crates/iceberg/src/io/storage/config/gcs.rs index 5b11567f4d..1392e35fd1 100644 --- a/crates/iceberg/src/io/storage/config/gcs.rs +++ b/crates/iceberg/src/io/storage/config/gcs.rs @@ -41,6 +41,12 @@ pub const GCS_NO_AUTH: &str = "gcs.no-auth"; pub const GCS_CREDENTIALS_JSON: &str = "gcs.credentials-json"; /// Google Cloud Storage token. pub const GCS_TOKEN: &str = "gcs.oauth2.token"; +/// Epoch-millisecond timestamp at which the vended GCS OAuth2 token expires. +pub const GCS_TOKEN_EXPIRES_AT: &str = "gcs.oauth2.token-expires-at"; +/// Endpoint used to fetch and refresh vended GCS OAuth2 credentials. +pub const GCS_REFRESH_CREDENTIALS_ENDPOINT: &str = "gcs.oauth2.refresh-credentials-endpoint"; +/// Whether vended GCS OAuth2 credentials should be refreshed. Defaults to `true`. +pub const GCS_REFRESH_CREDENTIALS_ENABLED: &str = "gcs.oauth2.refresh-credentials-enabled"; /// Option to skip signing requests (e.g. for public buckets/folders). pub const GCS_ALLOW_ANONYMOUS: &str = "gcs.allow-anonymous"; /// Option to skip loading the credential from GCE metadata server. diff --git a/crates/iceberg/src/io/storage/config/s3.rs b/crates/iceberg/src/io/storage/config/s3.rs index 664e8637b8..eb8c9f54c5 100644 --- a/crates/iceberg/src/io/storage/config/s3.rs +++ b/crates/iceberg/src/io/storage/config/s3.rs @@ -35,10 +35,16 @@ pub const S3_ACCESS_KEY_ID: &str = "s3.access-key-id"; pub const S3_SECRET_ACCESS_KEY: &str = "s3.secret-access-key"; /// S3 session token (required when using temporary credentials). pub const S3_SESSION_TOKEN: &str = "s3.session-token"; +/// Epoch-millisecond timestamp at which the vended S3 session token expires. +pub const S3_SESSION_TOKEN_EXPIRES_AT_MS: &str = "s3.session-token-expires-at-ms"; /// S3 region. pub const S3_REGION: &str = "s3.region"; /// Region to use for the S3 client (takes precedence over [`S3_REGION`]). pub const CLIENT_REGION: &str = "client.region"; +/// Endpoint used to fetch and refresh vended AWS credentials. +pub const AWS_REFRESH_CREDENTIALS_ENDPOINT: &str = "client.refresh-credentials-endpoint"; +/// Whether vended AWS credentials should be refreshed. Defaults to `true`. +pub const AWS_REFRESH_CREDENTIALS_ENABLED: &str = "client.refresh-credentials-enabled"; /// S3 Path Style Access. pub const S3_PATH_STYLE_ACCESS: &str = "s3.path-style-access"; /// S3 Server Side Encryption Type. diff --git a/crates/iceberg/src/io/storage/mod.rs b/crates/iceberg/src/io/storage/mod.rs index 5276c7771f..70542c5fa3 100644 --- a/crates/iceberg/src/io/storage/mod.rs +++ b/crates/iceberg/src/io/storage/mod.rs @@ -23,6 +23,7 @@ mod memory; use std::fmt::Debug; use std::sync::Arc; +use std::time::SystemTime; use async_trait::async_trait; use bytes::Bytes; @@ -32,7 +33,7 @@ pub use local_fs::{LocalFsStorage, LocalFsStorageFactory}; pub use memory::{MemoryStorage, MemoryStorageFactory}; use super::{FileMetadata, FileRead, FileWrite, InputFile, OutputFile}; -use crate::Result; +use crate::{Error, ErrorKind, Result}; /// Trait for storage operations in Iceberg. /// @@ -139,4 +140,208 @@ pub trait StorageFactory: Debug + Send + Sync { /// A `Result` containing an `Arc` on success, or an error /// if the storage could not be created. fn build(&self, config: &StorageConfig) -> Result>; + + /// Build a new Storage instance, optionally supplying a credential provider + /// that the backend can call to obtain and refresh short-lived credentials. + fn build_with_credentials( + &self, + config: &StorageConfig, + credential_provider: Option>, + ) -> Result> { + if credential_provider.is_some() { + return Err(Error::new( + ErrorKind::FeatureUnsupported, + "Storage factory does not support refreshable credential providers", + )); + } + + self.build(config) + } +} + +/// Supplies fresh, backend-specific storage credentials on demand. +/// +/// A catalog that vends temporary credentials implements this trait so that +/// storage backends can re-fetch credentials as they approach expiry instead +/// of failing once the initial token's TTL runs out. +/// +/// # Caching +/// +/// [`load_credential`](Self::load_credential) may be called very frequently. +/// Implementations must cache internally and only re-fetch when the current +/// credential is at or near expiry; otherwise every object-store request could +/// trigger a call back to the catalog. +#[async_trait] +pub trait StorageCredentialProvider: Debug + Send + Sync { + /// Return whether this provider has refresh configuration for `path`. + /// + /// Backends use this before replacing their normal credential chain. The + /// default is `true` for single-backend providers; multi-backend providers + /// should return `false` for schemes they do not configure. + fn supports_path(&self, _path: &str) -> bool { + true + } + + /// Load a fresh credential for the storage location identified by `path`. + /// + /// `path` is the absolute location being accessed (e.g. + /// `s3://bucket/warehouse/db/table/...`). Providers that vend distinct + /// credentials per location prefix use it to select the most specific + /// match. When the selected credential has a declared + /// [`StorageCredential::prefix`], it must cover `path`. + async fn load_credential(&self, path: &str) -> Result; +} + +/// A vended storage credential together with its scope and expiry. +#[derive(Clone, Debug)] +pub struct StorageCredential { + /// Storage-location prefix this credential is scoped to. `None` represents a + /// credential without a declared scope, sourced from flat storage properties. + prefix: Option, + /// The backend-specific credential material. + kind: StorageCredentialKind, + /// When the credential expires, if known. `None` means non-expiring and + /// backends treat such a credential as always valid and never refresh it. + expires_at: Option, +} + +impl StorageCredential { + /// Create a storage credential with no declared scope or expiration. + pub fn new(kind: StorageCredentialKind) -> Self { + Self { + prefix: None, + kind, + expires_at: None, + } + } + + /// Set the storage-location prefix this credential is scoped to. + pub fn with_prefix(mut self, prefix: impl Into) -> Self { + self.prefix = Some(prefix.into()); + self + } + + /// Set when this credential expires. + pub fn with_expiration(mut self, expires_at: SystemTime) -> Self { + self.expires_at = Some(expires_at); + self + } + + /// Return the storage-location prefix this credential is scoped to. + pub fn prefix(&self) -> Option<&str> { + self.prefix.as_deref() + } + + /// Return the backend-specific credential material. + pub fn kind(&self) -> &StorageCredentialKind { + &self.kind + } + + /// Consume this credential and return its backend-specific material. + pub fn into_kind(self) -> StorageCredentialKind { + self.kind + } + + /// Return when this credential expires. + pub fn expires_at(&self) -> Option { + self.expires_at + } +} + +/// Backend-specific credential material. +#[derive(Clone, Debug)] +pub enum StorageCredentialKind { + /// Amazon S3 credentials. + S3(S3Credential), + /// Google Cloud Storage credentials. + Gcs(GcsCredential), +} + +/// Temporary Amazon S3 credentials. +#[derive(Clone)] +pub struct S3Credential { + /// AWS access key ID. + access_key_id: String, + /// AWS secret access key. + secret_access_key: String, + /// AWS session token, set for temporary (STS/vended) credentials. + session_token: Option, +} + +impl S3Credential { + /// Create temporary Amazon S3 credentials. + pub fn new( + access_key_id: impl Into, + secret_access_key: impl Into, + session_token: Option, + ) -> Self { + Self { + access_key_id: access_key_id.into(), + secret_access_key: secret_access_key.into(), + session_token, + } + } + + /// Return the AWS access key ID. + pub fn access_key_id(&self) -> &str { + &self.access_key_id + } + + /// Return the AWS secret access key. + pub fn secret_access_key(&self) -> &str { + &self.secret_access_key + } + + /// Return the AWS session token, if present. + pub fn session_token(&self) -> Option<&str> { + self.session_token.as_deref() + } + + /// Consume these credentials and return their component values. + pub fn into_parts(self) -> (String, String, Option) { + let Self { + access_key_id, + secret_access_key, + session_token, + } = self; + (access_key_id, secret_access_key, session_token) + } +} + +impl Debug for S3Credential { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("S3Credential").finish_non_exhaustive() + } +} + +/// Temporary Google Cloud Storage credentials (an OAuth2 access token). +#[derive(Clone)] +pub struct GcsCredential { + /// OAuth2 bearer token used to access GCS. + token: String, +} + +impl GcsCredential { + /// Create a Google Cloud Storage credential. + pub fn new(token: impl Into) -> Self { + Self { + token: token.into(), + } + } + + /// Return the OAuth2 bearer token used to access GCS. + pub fn token(&self) -> &str { + &self.token + } + + /// Consume this credential and return its OAuth2 bearer token. + pub fn into_token(self) -> String { + self.token + } +} + +impl Debug for GcsCredential { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("GcsCredential").finish_non_exhaustive() + } }