feat(rest): support AWS SIGv4

Signed-off-by: xxchan <[email protected]>

Author: xxchan

Cargo.lock

@@ -64,8 +64,8 @@ datafusion-cli = "45"
 datafusion-sqllogictest = "45"
 derive_builder = "0.20"
 dirs = "6"
-expect-test = "1"
 enum-ordinalize = "4.3.0"
+expect-test = "1"
 faststr = "0.2.31"
 fnv = "1.0.7"
 fs-err = "3.1.0"
@@ -93,6 +93,7 @@ port_scanner = "0.1.5"
 pretty_assertions = "1.4"
 rand = "0.8.5"
 regex = "1.10.5"
+reqsign = { version = "0.16.3" }
 reqwest = { version = "0.12.12", default-features = false, features = ["json"] }
 roaring = { version = "0.10", git = "https://github.com/RoaringBitmap/roaring-rs.git" }
 rust_decimal = "1.36"

Cargo.toml

@@ -43,6 +43,7 @@ tokio = { workspace = true, features = ["sync"] }
 tracing = { workspace = true }
 typed-builder = { workspace = true }
 uuid = { workspace = true, features = ["v4"] }
+reqsign = { workspace = true }
 
 [dev-dependencies]
 ctor = { workspace = true }

crates/catalog/rest/Cargo.toml

@@ -28,6 +28,7 @@ use iceberg::{
     TableIdent,
 };
 use itertools::Itertools;
+use reqsign::{AwsConfig, AwsDefaultLoader, AwsV4Signer};
 use reqwest::header::{
     HeaderMap, HeaderName, HeaderValue, {self},
 };
@@ -84,6 +85,30 @@ impl RestCatalogConfig {
         }
     }
 
+    pub(crate) fn get_signer(&self) -> Result<Option<(AwsDefaultLoader, AwsV4Signer)>> {
+        if let Some("true") = self.props.get("rest.sigv4-enabled").map(|s| s.as_str()) {
+            let Some(signing_region) = self.props.get("rest.signing-region") else {
+                return Err(Error::new(
+                    ErrorKind::Unexpected,
+                    "rest.signing-region is not set when rest.sigv4-enabled is true",
+                ));
+            };
+            let Some(signing_name) = self.props.get("rest.signing-name") else {
+                return Err(Error::new(
+                    ErrorKind::Unexpected,
+                    "rest.signing-name is not set when rest.sigv4-enabled is true",
+                ));
+            };
+
+            let config = AwsConfig::default().from_profile().from_env();
+            println!("access_key_id {:?}", config.access_key_id);
+            let loader = AwsDefaultLoader::new(self.client().unwrap_or_default(), config);
+            let signer = AwsV4Signer::new(signing_name, signing_region);
+            Ok(Some((loader, signer)))
+        } else {
+            Ok(None)
+        }
+    }
     fn namespaces_endpoint(&self) -> String {
         self.url_prefixed(&["namespaces"])
     }
@@ -306,6 +331,13 @@ impl RestCatalog {
             None => None,
         };
 
+        if let Some(warehouse_path) = warehouse_path {
+            if warehouse_path.starts_with("arn:aws:") {
+                let file_io = FileIOBuilder::new("s3").with_props(&props).build()?;
+                return Ok(file_io);
+            }
+        }
+
         let file_io = match warehouse_path.or(metadata_location) {
             Some(url) => FileIO::from_path(url)?.with_props(props).build()?,
             None => {
@@ -612,7 +644,10 @@ impl Catalog for RestCatalog {
     /// provided locally to the `RestCatalog` will take precedence.
     async fn load_table(&self, table_ident: &TableIdent) -> Result<Table> {
         let context = self.context().await?;
-
+        println!(
+            "table_endpoint: {:?}",
+            context.config.table_endpoint(table_ident)
+        );
         let request = context
             .client
             .request(Method::GET, context.config.table_endpoint(table_ident))

crates/catalog/rest/src/catalog.rs

@@ -18,8 +18,9 @@
 use std::collections::HashMap;
 use std::fmt::{Debug, Formatter};
 
-use http::StatusCode;
+use http::{HeaderValue, StatusCode};
 use iceberg::{Error, ErrorKind, Result};
+use reqsign::{AwsDefaultLoader, AwsV4Signer};
 use reqwest::header::HeaderMap;
 use reqwest::{Client, IntoUrl, Method, Request, RequestBuilder, Response};
 use serde::de::DeserializeOwned;
@@ -43,6 +44,8 @@ pub(crate) struct HttpClient {
     extra_headers: HeaderMap,
     /// Extra oauth parameters to be added to each authentication request.
     extra_oauth_params: HashMap<String, String>,
+
+    signer: Option<(AwsDefaultLoader, AwsV4Signer)>,
 }
 
 impl Debug for HttpClient {
@@ -65,6 +68,7 @@ impl HttpClient {
             credential: cfg.credential(),
             extra_headers,
             extra_oauth_params: cfg.extra_oauth_params(),
+            signer: cfg.get_signer()?,
         })
     }
 
@@ -88,6 +92,7 @@ impl HttpClient {
             extra_oauth_params: (!cfg.extra_oauth_params().is_empty())
                 .then(|| cfg.extra_oauth_params())
                 .unwrap_or(self.extra_oauth_params),
+            signer: cfg.get_signer()?,
         })
     }
 
@@ -220,6 +225,39 @@ impl HttpClient {
     /// Executes the given `Request` and returns a `Response`.
     pub async fn execute(&self, mut request: Request) -> Result<Response> {
         request.headers_mut().extend(self.extra_headers.clone());
+
+        if let Some((loader, signer)) = &self.signer {
+            match loader.load().await {
+                Ok(Some(credential)) => {
+                    const EMPTY_STRING_SHA256: &str =
+                        "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
+                    request.headers_mut().insert(
+                        "x-amz-content-sha256",
+                        HeaderValue::from_str(EMPTY_STRING_SHA256).unwrap(),
+                    );
+                    if let Err(e) = signer.sign(&mut request, &credential) {
+                        return Err(Error::new(
+                            ErrorKind::Unexpected,
+                            "Failed to sign request for sigv4 signing",
+                        )
+                        .with_source(e));
+                    }
+                }
+                Ok(None) => {
+                    return Err(Error::new(
+                        ErrorKind::Unexpected,
+                        "Credential not found for sigv4 signing",
+                    ));
+                }
+                Err(e) => {
+                    return Err(Error::new(
+                        ErrorKind::Unexpected,
+                        "Failed to load credential for sigv4 signing",
+                    )
+                    .with_source(e));
+                }
+            }
+        }
         Ok(self.client.execute(request).await?)
     }
 
@@ -255,6 +293,7 @@ pub(crate) async fn deserialize_catalog_response<R: DeserializeOwned>(
 /// codes that all endpoints share (400, 404, etc.).
 pub(crate) async fn deserialize_unexpected_catalog_error(response: Response) -> Error {
     let (status, headers) = (response.status(), response.headers().clone());
+    let url = response.url().to_string();
     let bytes = match response.bytes().await {
         Ok(bytes) => bytes,
         Err(err) => return err.into(),
@@ -264,4 +303,5 @@ pub(crate) async fn deserialize_unexpected_catalog_error(response: Response) ->
         .with_context("status", status.to_string())
         .with_context("headers", format!("{:?}", headers))
         .with_context("json", String::from_utf8_lossy(&bytes))
+        .with_context("url", url)
 }