zakariya-s commented on code in PR #2932:
URL: https://github.com/apache/iceberg-rust/pull/2932#discussion_r4376900583
##########
crates/storage/opendal/src/utils.rs:
##########
@@ -27,3 +28,331 @@ pub(crate) fn from_opendal_error(e: opendal::Error) ->
iceberg::Error {
)
.with_source(e)
}
+
+/// The non-empty value of `key` in a vended credential's config.
+#[cfg(any(
+ feature = "opendal-s3",
+ feature = "opendal-gcs",
+ feature = "opendal-azdls"
+))]
+pub(crate) fn required_credential_property<'a>(
+ config: &'a std::collections::HashMap<String, String>,
+ key: &str,
+) -> reqsign_core::Result<&'a str> {
+ config
+ .get(key)
+ .map(String::as_str)
+ .filter(|value| !value.is_empty())
+ .ok_or_else(|| {
+ reqsign_core::Error::unexpected(format!("vended credential is
missing {key}"))
+ })
+}
+
+/// The latest epoch millisecond `reqsign` timestamps can represent,
+/// `9999-12-30T22:00:00Z`.
+#[cfg(any(
+ feature = "opendal-s3",
+ feature = "opendal-gcs",
+ feature = "opendal-azdls"
+))]
+const MAX_TIMESTAMP_MILLIS: i64 = 253_402_207_200_000;
+
+/// The epoch-millisecond expiry under `key` in a vended credential's config,
+/// if present, as the `reqsign` timestamp backend credentials use. Later
+/// expiries, such as `Long.MAX_VALUE` for a credential that never expires,
+/// are clamped to the latest representable timestamp.
+#[cfg(any(
+ feature = "opendal-s3",
+ feature = "opendal-gcs",
+ feature = "opendal-azdls"
+))]
+pub(crate) fn credential_expiry(
+ config: &std::collections::HashMap<String, String>,
+ key: &str,
+) -> reqsign_core::Result<Option<reqsign_core::time::Timestamp>> {
+ config
+ .get(key)
+ .filter(|value| !value.is_empty())
+ .map(|value| {
+ value
+ .parse::<i64>()
+ .ok()
+ .and_then(|millis| {
+ reqsign_core::time::Timestamp::from_millisecond(
+ millis.min(MAX_TIMESTAMP_MILLIS),
+ )
+ .ok()
+ })
+ .ok_or_else(|| {
+ reqsign_core::Error::unexpected(format!(
+ "vended credential has an invalid {key}"
+ ))
+ })
+ })
+ .transpose()
+}
+
+/// Validate that a provider's credential covers the location for which the
+/// backend requested it.
+#[cfg(any(
+ feature = "opendal-s3",
+ feature = "opendal-gcs",
+ feature = "opendal-azdls"
+))]
+pub(crate) fn validate_credential_prefix(
+ location: &str,
+ credential: &iceberg::io::StorageCredential,
+) -> reqsign_core::Result<()> {
+ if credential.covers(location) {
+ Ok(())
+ } else {
+ Err(reqsign_core::Error::unexpected(uncovered_location_message(
+ location, credential,
+ )))
+ }
+}
+
+/// The error message for a vended credential that does not cover `location`.
+pub(crate) fn uncovered_location_message(
+ location: &str,
+ credential: &iceberg::io::StorageCredential,
+) -> String {
+ format!(
+ "vended credential prefix {:?} does not cover storage location
{location:?}",
+ credential.prefix()
+ )
+}
+
+/// The root of the storage location containing `path`, e.g. `s3://bucket/`.
+pub(crate) fn storage_root(path: &str) -> iceberg::Result<String> {
+ let url = url::Url::parse(path)?;
+ Ok(format!("{}://{}/", url.scheme(), url.authority()))
+}
+
+/// Loads vended credentials for one storage location on behalf of a backend's
+/// `reqsign` credential provider.
+#[cfg(any(
+ feature = "opendal-s3",
+ feature = "opendal-gcs",
+ feature = "opendal-azdls"
+))]
+pub(crate) struct VendedCredentialSource {
+ provider: std::sync::Arc<dyn iceberg::io::StorageCredentialProvider>,
+ /// Location handed to the provider: the path an operator serves, or the
+ /// scope shared by a bulk-delete batch.
+ location: String,
+}
+
+#[cfg(any(
+ feature = "opendal-s3",
+ feature = "opendal-gcs",
+ feature = "opendal-azdls"
+))]
+impl VendedCredentialSource {
+ pub(crate) fn new(
+ provider: std::sync::Arc<dyn iceberg::io::StorageCredentialProvider>,
+ location: String,
+ ) -> Self {
+ Self { provider, location }
+ }
+
+ /// Load a credential covering the location and convert its config with
+ /// `extract` into the backend's credential.
+ ///
+ /// Errors are logged here: reqsign's credential chain replaces them with
an
+ /// error that only says no credential could be loaded, and reports the
+ /// cause only through the `log` crate.
+ pub(crate) async fn load<T>(
+ &self,
+ backend: &str,
+ extract: impl FnOnce(&std::collections::HashMap<String, String>) ->
reqsign_core::Result<T>,
+ ) -> reqsign_core::Result<T> {
+ self.try_load(backend, extract).await.inspect_err(|error| {
+ tracing::warn!(
Review Comment:
done in #2976 (7586b8d11e9971f2cf5b3b5055d0732169f2ee48)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
