feat(rest): add a SigV4 auth manager over a delegate session

Cargo.toml

@@ -60,6 +60,7 @@ arrow-string = "59.2"
 as-any = "0.3.2"
 async-trait = "0.1.89"
 aws-config = "1.8.7"
+aws-credential-types = "1.2"
 aws-sdk-glue = { version = "1.85", default-features = false, features = [
   "default-https-client",
   "rt-tokio",
@@ -68,6 +69,10 @@ aws-sdk-s3tables = { version = "1.28", default-features = false, features = [
   "default-https-client",
   "rt-tokio",
 ] }
+aws-sigv4 = { version = "1.4", default-features = false, features = [
+  "sign-http",
+  "http1",
+] }
 backon = "1.5.1"
 base64 = "0.22.1"
 bimap = "0.6"
@@ -87,6 +92,7 @@ fnv = "1.0.7"
 form_urlencoded = "1.2.2"
 futures = "0.3"
 hive_metastore = "0.2.0"
+hmac = "0.12"
 http = "1.2"
 iceberg = { version = "0.10.0", path = "./crates/iceberg" }
 iceberg-catalog-glue = { version = "0.10.0", path = "./crates/catalog/glue" }
@@ -124,6 +130,7 @@ serde_derive = "1.0.219"
 serde_json = "1.0.142"
 serde_repr = "0.1.16"
 serde_with = "3.4"
+sha2 = "0.10"
 sqlx = { version = "0.8.1", default-features = false }
 strum = "0.27.2"
 syn = "2"

crates/catalog/rest/Cargo.toml

@@ -29,8 +29,17 @@ keywords = ["iceberg", "rest", "catalog"]
 license = { workspace = true }
 repository = { workspace = true }
 
+[features]
+default = []
+# SigV4 request signing, for catalogs that authenticate the AWS way. Off by
+# default so the AWS crates are not pulled in unless they are used.
+sigv4 = ["dep:aws-credential-types", "dep:aws-sigv4", "dep:base64", "dep:sha2"]
+
 [dependencies]
 async-trait = { workspace = true }
+aws-credential-types = { workspace = true, optional = true }
+aws-sigv4 = { workspace = true, optional = true }
+base64 = { workspace = true, optional = true }
 chrono = { workspace = true }
 http = { workspace = true }
 iceberg = { workspace = true }
@@ -39,6 +48,7 @@ reqwest = { workspace = true }
 serde = { workspace = true }
 serde_derive = { workspace = true }
 serde_json = { workspace = true }
+sha2 = { workspace = true, optional = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
 typed-builder = { workspace = true }
@@ -47,6 +57,7 @@ uuid = { workspace = true, features = ["v4"] }
 [dev-dependencies]
 bytes = { workspace = true }
 futures = { workspace = true }
+hmac = { workspace = true }
 iceberg_test_utils = { path = "../../test_utils", features = ["tests"] }
 mockito = { workspace = true }
 # `stream` lets tests build a streaming body to exercise HttpRequestBody::Streaming.

crates/catalog/rest/public-api.txt

@@ -14,6 +14,18 @@ impl<'a> core::fmt::Debug for iceberg_catalog_rest::HttpRequestBody<'a>
 pub fn iceberg_catalog_rest::HttpRequestBody<'a>::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 impl<'a> core::marker::Copy for iceberg_catalog_rest::HttpRequestBody<'a>
 impl<'a> core::marker::StructuralPartialEq for iceberg_catalog_rest::HttpRequestBody<'a>
+pub enum iceberg_catalog_rest::PayloadHashMode
+pub iceberg_catalog_rest::PayloadHashMode::IcebergRest
+pub iceberg_catalog_rest::PayloadHashMode::StandardAws
+impl core::clone::Clone for iceberg_catalog_rest::PayloadHashMode
+pub fn iceberg_catalog_rest::PayloadHashMode::clone(&self) -> iceberg_catalog_rest::PayloadHashMode
+impl core::cmp::Eq for iceberg_catalog_rest::PayloadHashMode
+impl core::cmp::PartialEq for iceberg_catalog_rest::PayloadHashMode
+pub fn iceberg_catalog_rest::PayloadHashMode::eq(&self, other: &iceberg_catalog_rest::PayloadHashMode) -> bool
+impl core::fmt::Debug for iceberg_catalog_rest::PayloadHashMode
+pub fn iceberg_catalog_rest::PayloadHashMode::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+impl core::marker::Copy for iceberg_catalog_rest::PayloadHashMode
+impl core::marker::StructuralPartialEq for iceberg_catalog_rest::PayloadHashMode
 pub struct iceberg_catalog_rest::CommitTableRequest
 pub iceberg_catalog_rest::CommitTableRequest::identifier: core::option::Option<iceberg::catalog::TableIdent>
 pub iceberg_catalog_rest::CommitTableRequest::requirements: alloc::vec::Vec<iceberg::catalog::TableRequirement>
@@ -223,6 +235,7 @@ pub fn iceberg_catalog_rest::NoopAuthManager::fmt(&self, f: &mut core::fmt::Form
 impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::NoopAuthManager
 pub fn iceberg_catalog_rest::NoopAuthManager::catalog_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _client: &'life1 iceberg_catalog_rest::HttpClient, _props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
 pub fn iceberg_catalog_rest::NoopAuthManager::init_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _client: &'life1 iceberg_catalog_rest::HttpClient, _props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::boxed::Box<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+pub fn iceberg_catalog_rest::NoopAuthManager::signs_requests(&self) -> bool
 pub struct iceberg_catalog_rest::OAuth2Manager
 impl iceberg_catalog_rest::OAuth2Manager
 pub fn iceberg_catalog_rest::OAuth2Manager::new(token_endpoint: impl core::convert::Into<alloc::string::String>) -> Self
@@ -235,6 +248,7 @@ pub fn iceberg_catalog_rest::OAuth2Manager::fmt(&self, f: &mut core::fmt::Format
 impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::OAuth2Manager
 pub fn iceberg_catalog_rest::OAuth2Manager::catalog_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
 pub fn iceberg_catalog_rest::OAuth2Manager::init_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::boxed::Box<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+pub fn iceberg_catalog_rest::OAuth2Manager::signs_requests(&self) -> bool
 pub struct iceberg_catalog_rest::RegisterTableRequest
 pub iceberg_catalog_rest::RegisterTableRequest::metadata_location: alloc::string::String
 pub iceberg_catalog_rest::RegisterTableRequest::name: alloc::string::String
@@ -331,6 +345,24 @@ impl core::default::Default for iceberg_catalog_rest::RestSessionCatalogBuilder
 pub fn iceberg_catalog_rest::RestSessionCatalogBuilder::default() -> Self
 impl core::fmt::Debug for iceberg_catalog_rest::RestSessionCatalogBuilder
 pub fn iceberg_catalog_rest::RestSessionCatalogBuilder::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+pub struct iceberg_catalog_rest::SigV4AuthManager
+impl iceberg_catalog_rest::SigV4AuthManager
+pub async fn iceberg_catalog_rest::SigV4AuthManager::from_properties(delegate: alloc::sync::Arc<dyn iceberg_catalog_rest::AuthManager>, props: &std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> iceberg::error::Result<Self>
+pub fn iceberg_catalog_rest::SigV4AuthManager::new(delegate: alloc::sync::Arc<dyn iceberg_catalog_rest::AuthManager>, signer: iceberg_catalog_rest::SigV4Signer, credentials: aws_credential_types::provider::credentials::SharedCredentialsProvider) -> Self
+impl core::fmt::Debug for iceberg_catalog_rest::SigV4AuthManager
+pub fn iceberg_catalog_rest::SigV4AuthManager::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::SigV4AuthManager
+pub fn iceberg_catalog_rest::SigV4AuthManager::catalog_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+pub fn iceberg_catalog_rest::SigV4AuthManager::init_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::boxed::Box<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+pub fn iceberg_catalog_rest::SigV4AuthManager::signs_requests(&self) -> bool
+pub struct iceberg_catalog_rest::SigV4Signer
+impl iceberg_catalog_rest::SigV4Signer
+pub fn iceberg_catalog_rest::SigV4Signer::new(region: alloc::string::String, service: alloc::string::String, mode: iceberg_catalog_rest::PayloadHashMode) -> Self
+pub fn iceberg_catalog_rest::SigV4Signer::sign(&self, request: &mut iceberg_catalog_rest::HttpRequest, credentials: &aws_credential_types::credentials_impl::Credentials) -> iceberg::error::Result<()>
+impl core::clone::Clone for iceberg_catalog_rest::SigV4Signer
+pub fn iceberg_catalog_rest::SigV4Signer::clone(&self) -> iceberg_catalog_rest::SigV4Signer
+impl core::fmt::Debug for iceberg_catalog_rest::SigV4Signer
+pub fn iceberg_catalog_rest::SigV4Signer::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 pub struct iceberg_catalog_rest::StorageCredential
 pub iceberg_catalog_rest::StorageCredential::config: std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>
 pub iceberg_catalog_rest::StorageCredential::prefix: alloc::string::String
@@ -379,18 +411,31 @@ impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::UpdateNames
 pub fn iceberg_catalog_rest::UpdateNamespacePropertiesResponse::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
 pub const iceberg_catalog_rest::AUTH_TYPE_NONE: &str
 pub const iceberg_catalog_rest::AUTH_TYPE_OAUTH2: &str
+pub const iceberg_catalog_rest::REST_CATALOG_PROP_ACCESS_KEY_ID: &str
 pub const iceberg_catalog_rest::REST_CATALOG_PROP_AUTH_TYPE: &str
 pub const iceberg_catalog_rest::REST_CATALOG_PROP_DISABLE_HEADER_REDACTION: &str
+pub const iceberg_catalog_rest::REST_CATALOG_PROP_SECRET_ACCESS_KEY: &str
+pub const iceberg_catalog_rest::REST_CATALOG_PROP_SESSION_TOKEN: &str
+pub const iceberg_catalog_rest::REST_CATALOG_PROP_SIGNING_NAME: &str
+pub const iceberg_catalog_rest::REST_CATALOG_PROP_SIGNING_REGION: &str
 pub const iceberg_catalog_rest::REST_CATALOG_PROP_URI: &str
 pub const iceberg_catalog_rest::REST_CATALOG_PROP_WAREHOUSE: &str
+pub const iceberg_catalog_rest::SIGNING_NAME_DEFAULT: &str
 pub trait iceberg_catalog_rest::AuthManager: core::fmt::Debug + core::marker::Send + core::marker::Sync
 pub fn iceberg_catalog_rest::AuthManager::catalog_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
 pub fn iceberg_catalog_rest::AuthManager::init_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::boxed::Box<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+pub fn iceberg_catalog_rest::AuthManager::signs_requests(&self) -> bool
 impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::NoopAuthManager
 pub fn iceberg_catalog_rest::NoopAuthManager::catalog_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _client: &'life1 iceberg_catalog_rest::HttpClient, _props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
 pub fn iceberg_catalog_rest::NoopAuthManager::init_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _client: &'life1 iceberg_catalog_rest::HttpClient, _props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::boxed::Box<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+pub fn iceberg_catalog_rest::NoopAuthManager::signs_requests(&self) -> bool
 impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::OAuth2Manager
 pub fn iceberg_catalog_rest::OAuth2Manager::catalog_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
 pub fn iceberg_catalog_rest::OAuth2Manager::init_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::boxed::Box<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+pub fn iceberg_catalog_rest::OAuth2Manager::signs_requests(&self) -> bool
+impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::SigV4AuthManager
+pub fn iceberg_catalog_rest::SigV4AuthManager::catalog_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+pub fn iceberg_catalog_rest::SigV4AuthManager::init_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::boxed::Box<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+pub fn iceberg_catalog_rest::SigV4AuthManager::signs_requests(&self) -> bool
 pub trait iceberg_catalog_rest::AuthSession: core::fmt::Debug + core::marker::Send + core::marker::Sync
 pub fn iceberg_catalog_rest::AuthSession::authenticate<'life0, 'life1, 'async_trait>(&'life0 self, request: &'life1 mut iceberg_catalog_rest::HttpRequest) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait

crates/catalog/rest/src/auth/mod.rs

@@ -19,6 +19,8 @@
 //! `AuthManager`/`AuthSession` API.
 
 mod oauth2;
+#[cfg(feature = "sigv4")]
+mod sigv4;
 
 use std::collections::HashMap;
 use std::fmt::Debug;
@@ -27,6 +29,12 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use iceberg::Result;
 pub use oauth2::OAuth2Manager;
+#[cfg(feature = "sigv4")]
+pub use sigv4::{
+    PayloadHashMode, REST_CATALOG_PROP_ACCESS_KEY_ID, REST_CATALOG_PROP_SECRET_ACCESS_KEY,
+    REST_CATALOG_PROP_SESSION_TOKEN, REST_CATALOG_PROP_SIGNING_NAME,
+    REST_CATALOG_PROP_SIGNING_REGION, SIGNING_NAME_DEFAULT, SigV4AuthManager, SigV4Signer,
+};
 
 use crate::client::HttpClient;
 use crate::request::HttpRequest;
@@ -73,6 +81,14 @@ pub trait AuthManager: Debug + Send + Sync {
         client: &HttpClient,
         props: &HashMap<String, String>,
     ) -> Result<Arc<dyn AuthSession>>;
+
+    /// Whether the sessions this manager builds sign requests, so the catalog
+    /// must not let its HTTP client follow redirects: a redirect would replay a
+    /// signature made for another URL, and across hosts reqwest drops
+    /// `Authorization` while keeping any relocated copy of it.
+    fn signs_requests(&self) -> bool {
+        false
+    }
 }
 
 /// Authenticates outgoing REST catalog requests.

crates/catalog/rest/src/auth/oauth2.rs

@@ -219,7 +219,11 @@ fn attach_bearer(req: &mut HttpRequest, token: &SensitiveString) -> Result<()> {
             .with_source(e)
         })?;
     value.set_sensitive(true);
-    req.headers_mut().insert(http::header::AUTHORIZATION, value);
+    // Java's `OAuth2Util.AuthSession` puts the header only if absent, leaving a
+    // configured `header.authorization` in place.
+    if !req.headers().contains_key(http::header::AUTHORIZATION) {
+        req.headers_mut().insert(http::header::AUTHORIZATION, value);
+    }
     Ok(())
 }