Implement AWS Sigv4 support for Iceberg REST catalogs on Glue

Original change: #1
Upstream PR: apache#917

Author: phillipleblanc

Cargo.lock

@@ -57,6 +57,9 @@ aws-sdk-glue = "1.39"
 aws-sdk-s3tables = "1.28.0"
 backon = "1.5.1"
 base64 = "0.22.1"
+aws-sdk-sts = "1.63"
+aws-sigv4 = { version = "1.2", features = ["sign-http"] }
+aws-credential-types = "1.2"
 bimap = "0.6"
 bytes = "1.10"
 chrono = "0.4.41"
@@ -101,6 +104,7 @@ pretty_assertions = "1.4"
 rand = "0.8.5"
 regex = "1.10.5"
 reqwest = { version = "0.12.12", default-features = false, features = ["json"] }
+reqwest-middleware = { version = "0.4.0", features = ["json"] }
 roaring = { version = "0.11" }
 rust_decimal = "1.37.1"
 serde = { version = "1.0.219", features = ["rc"] }

Cargo.toml

@@ -29,12 +29,18 @@ license = { workspace = true }
 repository = { workspace = true }
 
 [dependencies]
+anyhow = { workspace = true }
 async-trait = { workspace = true }
+aws-config = { workspace = true, optional = true }
+aws-sigv4 = { workspace = true, optional = true }
+aws-credential-types = { workspace = true, optional = true }
+aws-sdk-sts = { workspace = true, optional = true }
 chrono = { workspace = true }
 http = { workspace = true }
 iceberg = { workspace = true }
 itertools = { workspace = true }
 reqwest = { workspace = true }
+reqwest-middleware = { workspace = true }
 serde = { workspace = true }
 serde_derive = { workspace = true }
 serde_json = { workspace = true }
@@ -49,3 +55,7 @@ iceberg_test_utils = { path = "../../test_utils", features = ["tests"] }
 mockito = { workspace = true }
 port_scanner = { workspace = true }
 tokio = { workspace = true }
+wiremock = "=0.6.4"
+[features]
+default = ["sigv4"]
+sigv4 = ["aws-config", "aws-sigv4", "aws-credential-types", "aws-sdk-sts"]

crates/catalog/rest/Cargo.toml

@@ -153,6 +153,10 @@ impl RestCatalogConfig {
             .join("/")
     }
 
+    pub(crate) fn base_url(&self) -> String {
+        [&self.uri, PATH_V1].join("/")
+    }
+
     fn config_endpoint(&self) -> String {
         [&self.uri, PATH_V1, "config"].join("/")
     }
@@ -308,6 +312,45 @@ impl RestCatalogConfig {
     }
 }
 
+#[cfg(feature = "sigv4")]
+impl RestCatalogConfig {
+    pub(crate) fn sigv4_enabled(&self) -> bool {
+        self.props
+            .get("rest.sigv4-enabled")
+            .is_some_and(|v| v == "true")
+    }
+
+    pub(crate) fn signing_region(&self) -> Option<String> {
+        self.props.get("rest.signing-region").cloned()
+    }
+
+    pub(crate) fn signing_name(&self) -> Option<String> {
+        self.props.get("rest.signing-name").cloned()
+    }
+
+    pub(crate) fn access_key_id(&self) -> Option<String> {
+        self.props.get("rest.access-key-id").cloned()
+    }
+
+    pub(crate) fn secret_access_key(&self) -> Option<String> {
+        self.props.get("rest.secret-access-key").cloned()
+    }
+
+    pub(crate) fn session_token(&self) -> Option<String> {
+        self.props.get("rest.session-token").cloned()
+    }
+
+    pub(crate) fn role_arn(&self) -> Option<String> {
+        self.props.get("rest.client.assume-role.arn").cloned()
+    }
+
+    pub(crate) fn role_session_name(&self) -> Option<String> {
+        self.props
+            .get("rest.client.assume-role.session-name")
+            .cloned()
+    }
+}
+
 #[derive(Debug)]
 struct RestContext {
     client: HttpClient,

crates/catalog/rest/src/catalog.rs

@@ -21,15 +21,16 @@ use std::fmt::{Debug, Formatter};
 use http::StatusCode;
 use iceberg::{Error, ErrorKind, Result};
 use reqwest::header::HeaderMap;
-use reqwest::{Client, IntoUrl, Method, Request, RequestBuilder, Response};
+use reqwest::{IntoUrl, Method, Request, Response};
+use reqwest_middleware::{ClientBuilder, ClientWithMiddleware, RequestBuilder};
 use serde::de::DeserializeOwned;
 use tokio::sync::Mutex;
 
 use crate::RestCatalogConfig;
 use crate::types::{ErrorResponse, TokenResponse};
 
 pub(crate) struct HttpClient {
-    client: Client,
+    client: ClientWithMiddleware,
 
     /// The token to be used for authentication.
     ///
@@ -56,10 +57,39 @@ impl Debug for HttpClient {
 
 impl HttpClient {
     /// Create a new http client.
+    #[allow(unused_mut)]
     pub fn new(cfg: &RestCatalogConfig) -> Result<Self> {
         let extra_headers = cfg.extra_headers()?;
+        let mut client_builder = ClientBuilder::new(cfg.client().unwrap_or_default());
+
+        #[cfg(feature = "sigv4")]
+        if cfg.sigv4_enabled() {
+            let mut sigv4_middleware = crate::middleware::sigv4::SigV4Middleware::new(
+                &cfg.base_url(),
+                cfg.signing_name().as_deref().unwrap_or("glue"),
+                cfg.signing_region().as_deref(),
+            );
+
+            if let (Some(access_key_id), Some(secret_access_key)) =
+                (cfg.access_key_id(), cfg.secret_access_key())
+            {
+                sigv4_middleware = sigv4_middleware.with_credentials(
+                    access_key_id,
+                    secret_access_key,
+                    cfg.session_token(),
+                );
+            }
+
+            if let Some(role_arn) = cfg.role_arn() {
+                sigv4_middleware = sigv4_middleware.with_role(role_arn, cfg.role_session_name());
+            }
+
+            client_builder = client_builder.with(sigv4_middleware);
+        }
+
         Ok(HttpClient {
-            client: cfg.client().unwrap_or_default(),
+            client: client_builder.build(),
+
             token: Mutex::new(cfg.token()),
             token_endpoint: cfg.get_token_endpoint(),
             credential: cfg.credential(),
@@ -77,8 +107,26 @@ impl HttpClient {
             .then(|| cfg.extra_headers())
             .transpose()?
             .unwrap_or(self.extra_headers);
+
+        let client = match cfg.client() {
+            Some(client) => {
+                let mut client_builder = ClientBuilder::new(client);
+                #[cfg(feature = "sigv4")]
+                if cfg.sigv4_enabled() {
+                    client_builder =
+                        client_builder.with(crate::middleware::sigv4::SigV4Middleware::new(
+                            &cfg.base_url(),
+                            cfg.signing_name().as_deref().unwrap_or("glue"),
+                            cfg.signing_region().as_deref(),
+                        ));
+                }
+                client_builder.build()
+            }
+            None => ClientBuilder::from_client(self.client).build(),
+        };
+
         Ok(HttpClient {
-            client: cfg.client().unwrap_or(self.client),
+            client,
             token: Mutex::new(cfg.token().or_else(|| self.token.into_inner())),
             token_endpoint: if !cfg.get_token_endpoint().is_empty() {
                 cfg.get_token_endpoint()

crates/catalog/rest/src/client.rs

@@ -53,6 +53,7 @@
 
 mod catalog;
 mod client;
+mod middleware;
 mod types;
 
 pub use catalog::*;

crates/catalog/rest/src/lib.rs

@@ -0,0 +1,21 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Middleware that is applied on requests to the Rest Catalog API.
+
+#[cfg(feature = "sigv4")]
+pub(crate) mod sigv4;