diff --git a/Cargo.lock b/Cargo.lock index dd33e5c2af..f6f615ef39 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3889,10 +3889,12 @@ dependencies = [ "iceberg_test_utils", "itertools 0.13.0", "mockito", + "reqsign-core", "reqwest 0.12.28", "serde", "serde_derive", "serde_json", + "sha2 0.10.9", "tokio", "typed-builder", "uuid", diff --git a/crates/catalog/rest/Cargo.toml b/crates/catalog/rest/Cargo.toml index e043c195ef..1154e94b07 100644 --- a/crates/catalog/rest/Cargo.toml +++ b/crates/catalog/rest/Cargo.toml @@ -35,10 +35,12 @@ chrono = { workspace = true } http = { workspace = true } iceberg = { workspace = true } itertools = { workspace = true } +reqsign-core = "3.0.0" reqwest = { workspace = true } serde = { workspace = true } serde_derive = { workspace = true } serde_json = { workspace = true } +sha2 = "0.10" tokio = { workspace = true } typed-builder = { workspace = true } uuid = { workspace = true, features = ["v4"] } diff --git a/crates/catalog/rest/public-api.txt b/crates/catalog/rest/public-api.txt index 776b11c40a..9d7df5dce8 100644 --- a/crates/catalog/rest/public-api.txt +++ b/crates/catalog/rest/public-api.txt @@ -1,4 +1,22 @@ pub mod iceberg_catalog_rest +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::AwsCredentials +pub iceberg_catalog_rest::AwsCredentials::access_key_id: alloc::string::String +pub iceberg_catalog_rest::AwsCredentials::secret_access_key: alloc::string::String +pub iceberg_catalog_rest::AwsCredentials::session_token: core::option::Option +impl core::clone::Clone for iceberg_catalog_rest::AwsCredentials +pub fn iceberg_catalog_rest::AwsCredentials::clone(&self) -> iceberg_catalog_rest::AwsCredentials pub struct iceberg_catalog_rest::CommitTableRequest pub iceberg_catalog_rest::CommitTableRequest::identifier: core::option::Option pub iceberg_catalog_rest::CommitTableRequest::requirements: alloc::vec::Vec @@ -175,6 +193,25 @@ impl serde_core::ser::Serialize for iceberg_catalog_rest::NamespaceResponse pub fn iceberg_catalog_rest::NamespaceResponse::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_catalog_rest::NamespaceResponse pub fn iceberg_catalog_rest::NamespaceResponse::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> +pub struct iceberg_catalog_rest::NoopAuthManager +impl core::fmt::Debug for iceberg_catalog_rest::NoopAuthManager +pub fn iceberg_catalog_rest::NoopAuthManager::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::NoopAuthManager +pub fn iceberg_catalog_rest::NoopAuthManager::catalog_session<'life0, 'life1, 'async_trait>(&'life0 self, _props: &'life1 std::collections::hash::map::HashMap) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub fn iceberg_catalog_rest::NoopAuthManager::init_session<'life0, 'async_trait>(&'life0 self) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait +pub struct iceberg_catalog_rest::OAuth2Manager +impl iceberg_catalog_rest::OAuth2Manager +pub fn iceberg_catalog_rest::OAuth2Manager::new(token_endpoint: impl core::convert::Into) -> Self +pub fn iceberg_catalog_rest::OAuth2Manager::with_client(self, client: reqwest::async_impl::client::Client) -> Self +pub fn iceberg_catalog_rest::OAuth2Manager::with_credential(self, client_id: core::option::Option, client_secret: alloc::string::String) -> Self +pub fn iceberg_catalog_rest::OAuth2Manager::with_extra_headers(self, headers: http::header::map::HeaderMap) -> Self +pub fn iceberg_catalog_rest::OAuth2Manager::with_extra_oauth_params(self, params: std::collections::hash::map::HashMap) -> Self +pub fn iceberg_catalog_rest::OAuth2Manager::with_token(self, token: impl core::convert::Into) -> Self +impl core::fmt::Debug for iceberg_catalog_rest::OAuth2Manager +pub fn iceberg_catalog_rest::OAuth2Manager::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::OAuth2Manager +pub fn iceberg_catalog_rest::OAuth2Manager::catalog_session<'life0, 'life1, 'async_trait>(&'life0 self, props: &'life1 std::collections::hash::map::HashMap) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub fn iceberg_catalog_rest::OAuth2Manager::init_session<'life0, 'async_trait>(&'life0 self) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait 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 @@ -230,6 +267,7 @@ pub fn iceberg_catalog_rest::RestCatalog::update_namespace<'life0, 'life1, 'asyn pub fn iceberg_catalog_rest::RestCatalog::update_table<'life0, 'async_trait>(&'life0 self, commit: iceberg::catalog::TableCommit) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait pub struct iceberg_catalog_rest::RestCatalogBuilder impl iceberg_catalog_rest::RestCatalogBuilder +pub fn iceberg_catalog_rest::RestCatalogBuilder::with_auth_manager(self, auth_manager: alloc::sync::Arc) -> Self pub fn iceberg_catalog_rest::RestCatalogBuilder::with_client(self, client: reqwest::async_impl::client::Client) -> Self impl core::default::Default for iceberg_catalog_rest::RestCatalogBuilder pub fn iceberg_catalog_rest::RestCatalogBuilder::default() -> Self @@ -241,6 +279,22 @@ pub fn iceberg_catalog_rest::RestCatalogBuilder::load(self, name: impl core::con pub fn iceberg_catalog_rest::RestCatalogBuilder::with_kms_client_factory(self, kms_client_factory: alloc::sync::Arc) -> Self pub fn iceberg_catalog_rest::RestCatalogBuilder::with_runtime(self, runtime: iceberg::runtime::Runtime) -> Self pub fn iceberg_catalog_rest::RestCatalogBuilder::with_storage_factory(self, storage_factory: alloc::sync::Arc) -> Self +pub struct iceberg_catalog_rest::SigV4AuthManager +impl iceberg_catalog_rest::SigV4AuthManager +pub fn iceberg_catalog_rest::SigV4AuthManager::new(delegate: alloc::sync::Arc, signer: iceberg_catalog_rest::SigV4Signer) -> 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, 'async_trait>(&'life0 self, props: &'life1 std::collections::hash::map::HashMap) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub fn iceberg_catalog_rest::SigV4AuthManager::init_session<'life0, 'async_trait>(&'life0 self) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait +pub struct iceberg_catalog_rest::SigV4Signer +impl iceberg_catalog_rest::SigV4Signer +pub fn iceberg_catalog_rest::SigV4Signer::new(credentials: iceberg_catalog_rest::AwsCredentials, region: alloc::string::String, service: alloc::string::String, mode: iceberg_catalog_rest::PayloadHashMode) -> Self +pub fn iceberg_catalog_rest::SigV4Signer::sign(&self, request: &mut reqwest::async_impl::request::Request) -> 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 pub iceberg_catalog_rest::StorageCredential::prefix: alloc::string::String @@ -287,6 +341,33 @@ impl serde_core::ser::Serialize for iceberg_catalog_rest::UpdateNamespacePropert pub fn iceberg_catalog_rest::UpdateNamespacePropertiesResponse::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_catalog_rest::UpdateNamespacePropertiesResponse pub fn iceberg_catalog_rest::UpdateNamespacePropertiesResponse::deserialize<__D>(__deserializer: __D) -> core::result::Result::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::AUTH_TYPE_SIGV4: &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_SIGNING_ACCESS_KEY: &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_SIGNING_SECRET_KEY: &str +pub const iceberg_catalog_rest::REST_CATALOG_PROP_SIGNING_SESSION_TOKEN: &str +pub const iceberg_catalog_rest::REST_CATALOG_PROP_SIGV4_DELEGATE_AUTH_TYPE: &str +pub const iceberg_catalog_rest::REST_CATALOG_PROP_SIGV4_ENABLED: &str pub const iceberg_catalog_rest::REST_CATALOG_PROP_URI: &str pub const iceberg_catalog_rest::REST_CATALOG_PROP_WAREHOUSE: &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, 'async_trait>(&'life0 self, props: &'life1 std::collections::hash::map::HashMap) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub fn iceberg_catalog_rest::AuthManager::init_session<'life0, 'async_trait>(&'life0 self) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait +impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::NoopAuthManager +pub fn iceberg_catalog_rest::NoopAuthManager::catalog_session<'life0, 'life1, 'async_trait>(&'life0 self, _props: &'life1 std::collections::hash::map::HashMap) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub fn iceberg_catalog_rest::NoopAuthManager::init_session<'life0, 'async_trait>(&'life0 self) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait +impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::OAuth2Manager +pub fn iceberg_catalog_rest::OAuth2Manager::catalog_session<'life0, 'life1, 'async_trait>(&'life0 self, props: &'life1 std::collections::hash::map::HashMap) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub fn iceberg_catalog_rest::OAuth2Manager::init_session<'life0, 'async_trait>(&'life0 self) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait +impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::SigV4AuthManager +pub fn iceberg_catalog_rest::SigV4AuthManager::catalog_session<'life0, 'life1, 'async_trait>(&'life0 self, props: &'life1 std::collections::hash::map::HashMap) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub fn iceberg_catalog_rest::SigV4AuthManager::init_session<'life0, 'async_trait>(&'life0 self) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait +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 reqwest::async_impl::request::Request) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub fn iceberg_catalog_rest::AuthSession::invalidate<'life0, 'async_trait>(&'life0 self) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait +pub fn iceberg_catalog_rest::AuthSession::refresh<'life0, 'async_trait>(&'life0 self) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait diff --git a/crates/catalog/rest/src/auth/mod.rs b/crates/catalog/rest/src/auth/mod.rs new file mode 100644 index 0000000000..a21ebd3dcd --- /dev/null +++ b/crates/catalog/rest/src/auth/mod.rs @@ -0,0 +1,108 @@ +// 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. + +//! Pluggable authentication for the REST catalog, mirroring Iceberg Java's +//! `AuthManager`/`AuthSession` API. + +mod oauth2; +mod sigv4; + +use std::collections::HashMap; +use std::fmt::Debug; +use std::sync::Arc; + +use async_trait::async_trait; +use iceberg::Result; +pub use oauth2::OAuth2Manager; +use reqwest::Request; +pub use sigv4::SigV4AuthManager; + +/// `rest.auth.type` value disabling authentication. +pub const AUTH_TYPE_NONE: &str = "none"; +/// `rest.auth.type` value selecting OAuth2 token authentication. +pub const AUTH_TYPE_OAUTH2: &str = "oauth2"; +/// `rest.auth.type` value selecting AWS SigV4 request signing. +pub const AUTH_TYPE_SIGV4: &str = "sigv4"; + +/// Creates the [`AuthSession`]s used to authenticate REST catalog requests. +/// +/// A manager is created once per catalog, either from the `rest.auth.type` +/// property or injected through `RestCatalogBuilder::with_auth_manager`, and +/// lives for the lifetime of the catalog. +#[async_trait] +pub trait AuthManager: Debug + Send + Sync { + /// Session used for the initial `/v1/config` handshake, built from the + /// user-supplied configuration. + async fn init_session(&self) -> Result>; + + /// Session used for all subsequent catalog requests, given the properties + /// merged from the user configuration and the server's config response. + /// + /// Implementations may carry state (e.g. a cached token) over from the + /// init session. + async fn catalog_session( + &self, + props: &HashMap, + ) -> Result>; +} + +/// Authenticates outgoing REST catalog requests. +#[async_trait] +pub trait AuthSession: Debug + Send + Sync { + /// Applies authentication to the request (adds headers, signs, ...). + async fn authenticate(&self, request: &mut Request) -> Result<()>; + + /// Drops any cached credentials so the next request re-authenticates. + async fn invalidate(&self) -> Result<()> { + Ok(()) + } + + /// Proactively refreshes cached credentials (e.g. re-exchanges an OAuth2 + /// client credential for a new token), leaving them intact on failure. + async fn refresh(&self) -> Result<()> { + Ok(()) + } +} + +/// [`AuthManager`] that performs no authentication. +#[derive(Debug)] +pub struct NoopAuthManager; + +/// [`AuthSession`] that performs no authentication. +#[derive(Debug)] +struct NoopSession; + +#[async_trait] +impl AuthManager for NoopAuthManager { + async fn init_session(&self) -> Result> { + Ok(Arc::new(NoopSession)) + } + + async fn catalog_session( + &self, + _props: &HashMap, + ) -> Result> { + Ok(Arc::new(NoopSession)) + } +} + +#[async_trait] +impl AuthSession for NoopSession { + async fn authenticate(&self, _request: &mut Request) -> Result<()> { + Ok(()) + } +} diff --git a/crates/catalog/rest/src/auth/oauth2.rs b/crates/catalog/rest/src/auth/oauth2.rs new file mode 100644 index 0000000000..a61a5719f7 --- /dev/null +++ b/crates/catalog/rest/src/auth/oauth2.rs @@ -0,0 +1,319 @@ +// 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. + +use std::collections::HashMap; +use std::fmt::{Debug, Formatter}; +use std::sync::Arc; + +use async_trait::async_trait; +use http::StatusCode; +use iceberg::{Error, ErrorKind, Result}; +use reqwest::header::HeaderMap; +use reqwest::{Client, Method, Request}; +use tokio::sync::Mutex; + +use super::{AuthManager, AuthSession}; +use crate::catalog::{ + RestCatalogConfig, credential_from_props, extra_headers_from_props, oauth_params_from_props, +}; +use crate::types::{ErrorResponse, TokenResponse}; + +/// Per-phase OAuth2 parameters (init vs. post-handshake catalog phase). +#[derive(Clone)] +struct OAuth2Params { + extra_headers: HeaderMap, + token_endpoint: String, + credential: Option<(Option, String)>, + extra_oauth_params: HashMap, +} + +/// [`AuthManager`] implementing the OAuth2 client-credentials flow used by +/// Iceberg REST catalogs. +/// +/// A configured `token` is used directly; otherwise `credential` is exchanged +/// for a token at the token endpoint and cached. The cached token is shared +/// across sessions so it survives the config handshake. +pub struct OAuth2Manager { + client: Client, + token: Arc>>, + init_params: OAuth2Params, +} + +impl OAuth2Manager { + /// Creates a manager exchanging credentials at `token_endpoint`, with no + /// token or credential configured. Combine with the `with_*` methods: + /// + /// ```rust,ignore + /// let manager = OAuth2Manager::new("https://auth.example.com/v1/oauth/tokens") + /// .with_credential(Some("client-id".into()), "client-secret".into()); + /// ``` + pub fn new(token_endpoint: impl Into) -> Self { + Self { + client: Client::default(), + token: Arc::new(Mutex::new(None)), + init_params: OAuth2Params { + extra_headers: HeaderMap::new(), + token_endpoint: token_endpoint.into(), + credential: None, + extra_oauth_params: HashMap::new(), + }, + } + } + + /// Sets a bearer token used directly (takes precedence over `credential`). + pub fn with_token(mut self, token: impl Into) -> Self { + self.token = Arc::new(Mutex::new(Some(token.into()))); + self + } + + /// Sets the client credential exchanged for a token at the token endpoint. + pub fn with_credential(mut self, client_id: Option, client_secret: String) -> Self { + self.init_params.credential = Some((client_id, client_secret)); + self + } + + /// Sets the HTTP client used for token requests. + pub fn with_client(mut self, client: Client) -> Self { + self.client = client; + self + } + + /// Sets extra headers sent with token requests. + pub fn with_extra_headers(mut self, headers: HeaderMap) -> Self { + self.init_params.extra_headers = headers; + self + } + + /// Sets extra OAuth2 form parameters (e.g. `scope`, `audience`). + pub fn with_extra_oauth_params(mut self, params: HashMap) -> Self { + self.init_params.extra_oauth_params = params; + self + } + + pub(crate) fn from_config(cfg: &RestCatalogConfig) -> Result { + Ok(Self { + client: cfg.client().unwrap_or_default(), + token: Arc::new(Mutex::new(cfg.token())), + init_params: OAuth2Params { + extra_headers: cfg.extra_headers()?, + token_endpoint: cfg.get_token_endpoint(), + credential: cfg.credential(), + extra_oauth_params: cfg.extra_oauth_params(), + }, + }) + } +} + +impl Debug for OAuth2Manager { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + f.debug_struct("OAuth2Manager") + .field("token_endpoint", &self.init_params.token_endpoint) + .finish_non_exhaustive() + } +} + +#[async_trait] +impl AuthManager for OAuth2Manager { + async fn init_session(&self) -> Result> { + Ok(Arc::new(OAuth2Session { + client: self.client.clone(), + token: self.token.clone(), + params: self.init_params.clone(), + })) + } + + async fn catalog_session( + &self, + props: &HashMap, + ) -> Result> { + // The server config may carry a new token (or restate the user's). + if let Some(token) = props.get("token") { + *self.token.lock().await = Some(token.clone()); + } + Ok(Arc::new(OAuth2Session { + client: self.client.clone(), + token: self.token.clone(), + params: OAuth2Params { + extra_headers: extra_headers_from_props(props)?, + token_endpoint: props + .get("oauth2-server-uri") + .cloned() + .unwrap_or_else(|| self.init_params.token_endpoint.clone()), + credential: credential_from_props(props) + .or_else(|| self.init_params.credential.clone()), + extra_oauth_params: oauth_params_from_props(props), + }, + })) + } +} + +/// [`AuthSession`] adding a `Authorization: Bearer ` header. +struct OAuth2Session { + client: Client, + /// Cached bearer token, shared with the owning [`OAuth2Manager`]. + token: Arc>>, + params: OAuth2Params, +} + +impl Debug for OAuth2Session { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + f.debug_struct("OAuth2Session") + .field("token_endpoint", &self.params.token_endpoint) + .finish_non_exhaustive() + } +} + +impl OAuth2Session { + async fn exchange_credential_for_token(&self) -> Result { + // Credential must exist here. + let (client_id, client_secret) = self.params.credential.as_ref().ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + "Credential must be provided for authentication", + ) + })?; + + let mut params = HashMap::with_capacity(4); + params.insert("grant_type", "client_credentials"); + if let Some(client_id) = client_id { + params.insert("client_id", client_id); + } + params.insert("client_secret", client_secret); + params.extend( + self.params + .extra_oauth_params + .iter() + .map(|(k, v)| (k.as_str(), v.as_str())), + ); + + let mut auth_req = self + .client + .request(Method::POST, &self.params.token_endpoint) + .headers(self.params.extra_headers.clone()) + .form(¶ms) + .build()?; + // extra headers add content-type application/json header it's necessary to override it with proper type + // note that form call doesn't add content-type header if already present + auth_req.headers_mut().insert( + http::header::CONTENT_TYPE, + http::HeaderValue::from_static("application/x-www-form-urlencoded"), + ); + let auth_url = auth_req.url().clone(); + let auth_resp = self.client.execute(auth_req).await?; + + let auth_res: TokenResponse = if auth_resp.status() == StatusCode::OK { + let text = auth_resp + .bytes() + .await + .map_err(|err| err.with_url(auth_url.clone()))?; + Ok(serde_json::from_slice(&text).map_err(|e| { + Error::new( + ErrorKind::Unexpected, + "Failed to parse response from rest catalog server!", + ) + .with_context("operation", "auth") + .with_context("url", auth_url.to_string()) + .with_context("json", String::from_utf8_lossy(&text)) + .with_source(e) + })?) + } else { + let code = auth_resp.status(); + let text = auth_resp + .bytes() + .await + .map_err(|err| err.with_url(auth_url.clone()))?; + let e: ErrorResponse = serde_json::from_slice(&text).map_err(|e| { + Error::new(ErrorKind::Unexpected, "Received unexpected response") + .with_context("code", code.to_string()) + .with_context("operation", "auth") + .with_context("url", auth_url.to_string()) + .with_context("json", String::from_utf8_lossy(&text)) + .with_source(e) + })?; + Err(Error::from(e)) + }?; + Ok(auth_res.access_token) + } +} + +#[async_trait] +impl AuthSession for OAuth2Session { + /// Adds a bearer token to the authorization header. + /// + /// Three modes: + /// + /// 1. **No authentication** - Skip when both `credential` and `token` are missing. + /// 2. **Token authentication** - Use the provided `token` directly. + /// 3. **OAuth authentication** - Exchange `credential` for a token, cache it, then use it. + /// + /// When both `credential` and `token` are present, `token` takes precedence. + /// + /// # TODO: Support automatic token refreshing. + async fn authenticate(&self, req: &mut Request) -> Result<()> { + // Clone the token from lock without holding the lock for entire function. + let token = self.token.lock().await.clone(); + + if self.params.credential.is_none() && token.is_none() { + return Ok(()); + } + + // Either use the provided token or exchange credential for token, cache and use that + let token = match token { + Some(token) => token, + None => { + let token = self.exchange_credential_for_token().await?; + // Update token so that we use it for next request instead of + // exchanging credential for token from the server again + *self.token.lock().await = Some(token.clone()); + token + } + }; + + // Insert token in request. + req.headers_mut().insert( + http::header::AUTHORIZATION, + format!("Bearer {token}").parse().map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + "Invalid token received from catalog server!", + ) + .with_source(e) + })?, + ); + + Ok(()) + } + + /// Invalidate the current token without generating a new one. On the next + /// request, the session will attempt to generate a new token. + async fn invalidate(&self) -> Result<()> { + *self.token.lock().await = None; + Ok(()) + } + + /// Invalidate the current token and set a new one. Generates a new token + /// before invalidating the current one, meaning the old token will be used + /// until this function acquires the lock and overwrites the token. + /// + /// If credential is invalid, or the request fails, this method will return + /// an error and leave the current token unchanged. + async fn refresh(&self) -> Result<()> { + let new_token = self.exchange_credential_for_token().await?; + *self.token.lock().await = Some(new_token); + Ok(()) + } +} diff --git a/crates/catalog/rest/src/auth/sigv4.rs b/crates/catalog/rest/src/auth/sigv4.rs new file mode 100644 index 0000000000..e203f24646 --- /dev/null +++ b/crates/catalog/rest/src/auth/sigv4.rs @@ -0,0 +1,108 @@ +// 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. + +use std::collections::HashMap; +use std::sync::Arc; + +use async_trait::async_trait; +use iceberg::Result; +use reqwest::Request; +use reqwest::header::{AUTHORIZATION, HeaderName}; + +use super::{AuthManager, AuthSession}; +use crate::catalog::{ + REST_CATALOG_PROP_SIGNING_NAME, REST_CATALOG_PROP_SIGNING_REGION, sigv4_signer_from_props, +}; +use crate::signing::SigV4Signer; + +/// Header the delegate's `Authorization` is relocated to before signing, so +/// token-based auth composes with SigV4 (Iceberg Java parity: AWS needs the +/// `Authorization` header for the signature itself). +const RELOCATED_AUTH_HEADER: HeaderName = HeaderName::from_static("x-iceberg-authorization"); + +/// [`AuthManager`] that SigV4-signs every request, wrapping a delegate +/// manager whose authentication (e.g. an OAuth2 bearer token) is relocated to +/// `X-Iceberg-Authorization` and included in the signature. +#[derive(Debug)] +pub struct SigV4AuthManager { + delegate: Arc, + signer: SigV4Signer, +} + +impl SigV4AuthManager { + /// Creates a SigV4 manager signing with `signer` on top of `delegate`. + pub fn new(delegate: Arc, signer: SigV4Signer) -> Self { + Self { delegate, signer } + } +} + +#[async_trait] +impl AuthManager for SigV4AuthManager { + async fn init_session(&self) -> Result> { + Ok(Arc::new(SigV4Session { + delegate: self.delegate.init_session().await?, + signer: self.signer.clone(), + })) + } + + async fn catalog_session( + &self, + props: &HashMap, + ) -> Result> { + // The merged properties include the server's defaults/overrides, so a + // server-supplied `rest.signing-*` is honored here. When the props + // don't define a signing config (e.g. an injected manager with an + // explicit signer), keep the configured one. + let signer = if props.contains_key(REST_CATALOG_PROP_SIGNING_REGION) + && props.contains_key(REST_CATALOG_PROP_SIGNING_NAME) + { + sigv4_signer_from_props(props)? + } else { + self.signer.clone() + }; + Ok(Arc::new(SigV4Session { + delegate: self.delegate.catalog_session(props).await?, + signer, + })) + } +} + +/// [`AuthSession`] applying the delegate's auth, then SigV4-signing. +#[derive(Debug)] +struct SigV4Session { + delegate: Arc, + signer: SigV4Signer, +} + +#[async_trait] +impl AuthSession for SigV4Session { + async fn authenticate(&self, request: &mut Request) -> Result<()> { + self.delegate.authenticate(request).await?; + if let Some(auth) = request.headers_mut().remove(AUTHORIZATION) { + request.headers_mut().insert(RELOCATED_AUTH_HEADER, auth); + } + self.signer.sign(request) + } + + async fn invalidate(&self) -> Result<()> { + self.delegate.invalidate().await + } + + async fn refresh(&self) -> Result<()> { + self.delegate.refresh().await + } +} diff --git a/crates/catalog/rest/src/catalog.rs b/crates/catalog/rest/src/catalog.rs index da05fc7e3b..e321d280cf 100644 --- a/crates/catalog/rest/src/catalog.rs +++ b/crates/catalog/rest/src/catalog.rs @@ -38,9 +38,14 @@ use reqwest::{Client, Method, StatusCode, Url}; use tokio::sync::OnceCell; use typed_builder::TypedBuilder; +use crate::auth::{ + AUTH_TYPE_NONE, AUTH_TYPE_OAUTH2, AUTH_TYPE_SIGV4, AuthManager, NoopAuthManager, OAuth2Manager, + SigV4AuthManager, +}; use crate::client::{ HttpClient, deserialize_catalog_response, deserialize_unexpected_catalog_error, }; +use crate::signing::{AwsCredentials, PayloadHashMode, SigV4Signer}; use crate::types::{ CatalogConfig, CommitTableRequest, CommitTableResponse, CreateNamespaceRequest, CreateTableRequest, ListNamespaceResponse, ListTablesResponse, LoadTableResult, @@ -53,6 +58,23 @@ pub const REST_CATALOG_PROP_URI: &str = "uri"; pub const REST_CATALOG_PROP_WAREHOUSE: &str = "warehouse"; /// Disable header redaction in error logs (defaults to false for security) pub const REST_CATALOG_PROP_DISABLE_HEADER_REDACTION: &str = "disable-header-redaction"; +/// Authentication scheme: `none`, `oauth2` (default) or `sigv4`. +pub const REST_CATALOG_PROP_AUTH_TYPE: &str = "rest.auth.type"; +/// Enable AWS SigV4 request signing for the REST catalog. +pub const REST_CATALOG_PROP_SIGV4_ENABLED: &str = "rest.sigv4-enabled"; +/// SigV4 signing service name (required when sigv4 is enabled). +pub const REST_CATALOG_PROP_SIGNING_NAME: &str = "rest.signing-name"; +/// SigV4 signing region (required when sigv4 is enabled). +pub const REST_CATALOG_PROP_SIGNING_REGION: &str = "rest.signing-region"; +/// SigV4 access key id (optional; falls back to the `AWS_ACCESS_KEY_ID` env var). +pub const REST_CATALOG_PROP_SIGNING_ACCESS_KEY: &str = "rest.signing-access-key-id"; +/// SigV4 secret access key (optional; falls back to the `AWS_SECRET_ACCESS_KEY` env var). +pub const REST_CATALOG_PROP_SIGNING_SECRET_KEY: &str = "rest.signing-secret-access-key"; +/// SigV4 session token (optional; falls back to the `AWS_SESSION_TOKEN` env var). +pub const REST_CATALOG_PROP_SIGNING_SESSION_TOKEN: &str = "rest.signing-session-token"; +/// Auth scheme SigV4 wraps: `none` or `oauth2`. When unset, `oauth2` is +/// inferred if a `token` or `credential` is configured, `none` otherwise. +pub const REST_CATALOG_PROP_SIGV4_DELEGATE_AUTH_TYPE: &str = "rest.auth.sigv4.delegate-auth-type"; const ICEBERG_REST_SPEC_VERSION: &str = "0.14.1"; const CARGO_PKG_VERSION: &str = env!("CARGO_PKG_VERSION"); @@ -76,6 +98,7 @@ impl Default for RestCatalogBuilder { warehouse: None, props: HashMap::new(), client: None, + custom_auth_manager: None, }, storage_factory: None, kms_client_factory: None, @@ -160,6 +183,12 @@ impl RestCatalogBuilder { self.config.client = Some(client); self } + + /// Injects a custom auth manager, overriding the `rest.auth.type` configuration. + pub fn with_auth_manager(mut self, auth_manager: Arc) -> Self { + self.config.custom_auth_manager = Some(auth_manager); + self + } } /// Rest catalog configuration. @@ -178,6 +207,9 @@ pub(crate) struct RestCatalogConfig { #[builder(default)] client: Option, + + #[builder(default)] + custom_auth_manager: Option>, } impl RestCatalogConfig { @@ -244,89 +276,28 @@ impl RestCatalogConfig { /// Get the credentials from the config. The client can use these credentials to fetch a new /// token. - /// - /// ## Output - /// - /// - `None`: No credential is set. - /// - `Some(None, client_secret)`: No client_id is set, use client_secret directly. - /// - `Some(Some(client_id), client_secret)`: Both client_id and client_secret are set. pub(crate) fn credential(&self) -> Option<(Option, String)> { - let cred = self.props.get("credential")?; - - match cred.split_once(':') { - Some((client_id, client_secret)) => { - Some((Some(client_id.to_string()), client_secret.to_string())) - } - None => Some((None, cred.to_string())), - } + credential_from_props(&self.props) } - /// Get the extra headers from config, which includes: - /// - /// - `content-type` - /// - `x-client-version` - /// - `user-agent` - /// - All headers specified by `header.xxx` in props. + /// Get the extra headers from config, see [`extra_headers_from_props`]. pub(crate) fn extra_headers(&self) -> Result { - let mut headers = HeaderMap::from_iter([ - ( - header::CONTENT_TYPE, - HeaderValue::from_static("application/json"), - ), - ( - HeaderName::from_static("x-client-version"), - HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION), - ), - ( - header::USER_AGENT, - HeaderValue::from_str(&format!("iceberg-rs/{CARGO_PKG_VERSION}")).unwrap(), - ), - ]); - - for (key, value) in self - .props - .iter() - .filter_map(|(k, v)| k.strip_prefix("header.").map(|k| (k, v))) - { - headers.insert( - HeaderName::from_str(key).map_err(|e| { - Error::new( - ErrorKind::DataInvalid, - format!("Invalid header name: {key}"), - ) - .with_source(e) - })?, - HeaderValue::from_str(value).map_err(|e| { - Error::new( - ErrorKind::DataInvalid, - format!("Invalid header value: {value}"), - ) - .with_source(e) - })?, - ); - } - - Ok(headers) + extra_headers_from_props(&self.props) } /// Get the optional OAuth headers from the config. pub(crate) fn extra_oauth_params(&self) -> HashMap { - let mut params = HashMap::new(); - - if let Some(scope) = self.props.get("scope") { - params.insert("scope".to_string(), scope.to_string()); - } else { - params.insert("scope".to_string(), "catalog".to_string()); - } - - let optional_params = ["audience", "resource"]; - for param_name in optional_params { - if let Some(value) = self.props.get(param_name) { - params.insert(param_name.to_string(), value.to_string()); - } - } + oauth_params_from_props(&self.props) + } - params + /// The properties handed to [`AuthManager::catalog_session`], with the + /// resolved token endpoint made explicit. + pub(crate) fn auth_props(&self) -> HashMap { + let mut props = self.props.clone(); + props + .entry("oauth2-server-uri".to_string()) + .or_insert_with(|| self.get_token_endpoint()); + props } /// Check if header redaction is disabled in error logs. @@ -340,6 +311,78 @@ impl RestCatalogConfig { .unwrap_or(false) } + fn sigv4_enabled(&self) -> bool { + self.props + .get(REST_CATALOG_PROP_SIGV4_ENABLED) + .map(|v| v.eq_ignore_ascii_case("true")) + .unwrap_or(false) + } + + /// The configured auth scheme: explicit `rest.auth.type`, the legacy + /// `rest.sigv4-enabled` switch, or the default `oauth2` (which behaves as + /// no auth when neither `token` nor `credential` is set). + fn auth_type(&self) -> String { + self.props + .get(REST_CATALOG_PROP_AUTH_TYPE) + .cloned() + .unwrap_or_else(|| { + if self.sigv4_enabled() { + AUTH_TYPE_SIGV4.to_string() + } else { + AUTH_TYPE_OAUTH2.to_string() + } + }) + } + + /// Resolves the auth manager: a `with_auth_manager` override wins, + /// otherwise one is built from the `rest.auth.type` configuration. + pub(crate) fn resolve_auth_manager(&self) -> Result> { + if let Some(auth_manager) = &self.custom_auth_manager { + return Ok(auth_manager.clone()); + } + match self.auth_type().as_str() { + AUTH_TYPE_NONE => Ok(Arc::new(NoopAuthManager)), + AUTH_TYPE_OAUTH2 => Ok(Arc::new(OAuth2Manager::from_config(self)?)), + AUTH_TYPE_SIGV4 => { + // SigV4 signs on top of a delegate auth (Java parity): explicit + // `rest.auth.sigv4.delegate-auth-type`, or inferred from + // whether token-based auth is configured. + let delegate: Arc = match self + .props + .get(REST_CATALOG_PROP_SIGV4_DELEGATE_AUTH_TYPE) + .map(String::as_str) + { + Some(AUTH_TYPE_NONE) => Arc::new(NoopAuthManager), + Some(AUTH_TYPE_OAUTH2) => Arc::new(OAuth2Manager::from_config(self)?), + Some(other) => { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "unknown '{REST_CATALOG_PROP_SIGV4_DELEGATE_AUTH_TYPE}': {other}" + ), + )); + } + None if self.token().is_some() || self.credential().is_some() => { + Arc::new(OAuth2Manager::from_config(self)?) + } + None => Arc::new(NoopAuthManager), + }; + Ok(Arc::new(SigV4AuthManager::new( + delegate, + self.build_sigv4_signer()?, + ))) + } + other => Err(Error::new( + ErrorKind::DataInvalid, + format!("unknown '{REST_CATALOG_PROP_AUTH_TYPE}': {other}"), + )), + } + } + + fn build_sigv4_signer(&self) -> Result { + sigv4_signer_from_props(&self.props) + } + /// Merge the `RestCatalogConfig` with the a [`CatalogConfig`] (fetched from the REST server). pub(crate) fn merge_with_config(mut self, mut config: CatalogConfig) -> Self { if let Some(uri) = config.overrides.remove("uri") { @@ -355,6 +398,160 @@ impl RestCatalogConfig { } } +/// Builds a [`SigV4Signer`] from `rest.signing-*` properties (credentials fall +/// back to the standard `AWS_*` environment variables). +/// +/// Used both at catalog construction (user properties) and after the config +/// handshake (properties merged with the server's defaults/overrides), so +/// server-supplied `rest.signing-*` values are honored. +pub(crate) fn sigv4_signer_from_props(props: &HashMap) -> Result { + let region = props + .get(REST_CATALOG_PROP_SIGNING_REGION) + .cloned() + .ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "'{REST_CATALOG_PROP_SIGNING_REGION}' is required when '{REST_CATALOG_PROP_SIGV4_ENABLED}' is true" + ), + ) + })?; + let name = props + .get(REST_CATALOG_PROP_SIGNING_NAME) + .cloned() + .ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "'{REST_CATALOG_PROP_SIGNING_NAME}' is required when '{REST_CATALOG_PROP_SIGV4_ENABLED}' is true" + ), + ) + })?; + let access_key_id = props + .get(REST_CATALOG_PROP_SIGNING_ACCESS_KEY) + .cloned() + .or_else(|| std::env::var("AWS_ACCESS_KEY_ID").ok()) + .ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + "missing SigV4 access key id (set 'rest.signing-access-key-id' or AWS_ACCESS_KEY_ID)", + ) + })?; + let secret_access_key = props + .get(REST_CATALOG_PROP_SIGNING_SECRET_KEY) + .cloned() + .or_else(|| std::env::var("AWS_SECRET_ACCESS_KEY").ok()) + .ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + "missing SigV4 secret access key (set 'rest.signing-secret-access-key' or AWS_SECRET_ACCESS_KEY)", + ) + })?; + let session_token = props + .get(REST_CATALOG_PROP_SIGNING_SESSION_TOKEN) + .cloned() + .or_else(|| std::env::var("AWS_SESSION_TOKEN").ok()); + + let credentials = AwsCredentials { + access_key_id, + secret_access_key, + session_token, + }; + Ok(SigV4Signer::new( + credentials, + region, + name, + PayloadHashMode::IcebergRest, + )) +} + +/// Parses the `credential` property. +/// +/// ## Output +/// +/// - `None`: No credential is set. +/// - `Some(None, client_secret)`: No client_id is set, use client_secret directly. +/// - `Some(Some(client_id), client_secret)`: Both client_id and client_secret are set. +pub(crate) fn credential_from_props( + props: &HashMap, +) -> Option<(Option, String)> { + let cred = props.get("credential")?; + + match cred.split_once(':') { + Some((client_id, client_secret)) => { + Some((Some(client_id.to_string()), client_secret.to_string())) + } + None => Some((None, cred.to_string())), + } +} + +/// The extra headers added to each request, which include: +/// +/// - `content-type` +/// - `x-client-version` +/// - `user-agent` +/// - All headers specified by `header.xxx` in props. +pub(crate) fn extra_headers_from_props(props: &HashMap) -> Result { + let mut headers = HeaderMap::from_iter([ + ( + header::CONTENT_TYPE, + HeaderValue::from_static("application/json"), + ), + ( + HeaderName::from_static("x-client-version"), + HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION), + ), + ( + header::USER_AGENT, + HeaderValue::from_str(&format!("iceberg-rs/{CARGO_PKG_VERSION}")).unwrap(), + ), + ]); + + for (key, value) in props + .iter() + .filter_map(|(k, v)| k.strip_prefix("header.").map(|k| (k, v))) + { + headers.insert( + HeaderName::from_str(key).map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid header name: {key}"), + ) + .with_source(e) + })?, + HeaderValue::from_str(value).map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid header value: {value}"), + ) + .with_source(e) + })?, + ); + } + + Ok(headers) +} + +/// The optional OAuth parameters added to each authentication request. +pub(crate) fn oauth_params_from_props(props: &HashMap) -> HashMap { + let mut params = HashMap::new(); + + if let Some(scope) = props.get("scope") { + params.insert("scope".to_string(), scope.to_string()); + } else { + params.insert("scope".to_string(), "catalog".to_string()); + } + + let optional_params = ["audience", "resource"]; + for param_name in optional_params { + if let Some(value) = props.get(param_name) { + params.insert(param_name.to_string(), value.to_string()); + } + } + + params +} + #[derive(Debug)] struct RestContext { client: HttpClient, @@ -429,10 +626,10 @@ impl RestCatalog { async fn context(&self) -> Result<&RestContext> { self.ctx .get_or_try_init(|| async { - let client = HttpClient::new(&self.user_config)?; + let client = HttpClient::new(&self.user_config).await?; let catalog_config = RestCatalog::load_config(&client, &self.user_config).await?; let config = self.user_config.clone().merge_with_config(catalog_config); - let client = client.update_with(&config)?; + let client = client.update_with(&config).await?; Ok(RestContext { config, client }) }) @@ -510,7 +707,7 @@ impl RestCatalog { /// Invalidate the current token without generating a new one. On the next request, the client /// will attempt to generate a new token. pub async fn invalidate_token(&self) -> Result<()> { - self.context().await?.client.invalidate_token().await + self.context().await?.client.session().invalidate().await } /// Invalidate the current token and set a new one. Generates a new token before invalidating @@ -520,7 +717,7 @@ impl RestCatalog { /// If credential is invalid, or the request fails, this method will return an error and leave /// the current token unchanged. pub async fn regenerate_token(&self) -> Result<()> { - self.context().await?.client.regenerate_token().await + self.context().await?.client.session().refresh().await } } @@ -1612,6 +1809,313 @@ mod tests { list_ns_mock.assert_async().await; } + #[tokio::test] + async fn test_sigv4_signs_requests() { + let mut server = Server::new_async().await; + + let config_mock = create_config_mock(&mut server).await; + + // With sigv4 enabled, requests must carry an AWS SigV4 Authorization header. + let list_ns_mock = server + .mock("GET", "/v1/namespaces") + .match_header( + "authorization", + mockito::Matcher::Regex("^AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/".to_string()), + ) + .with_body(r#"{"namespaces": []}"#) + .create_async() + .await; + + let props = HashMap::from([ + ( + REST_CATALOG_PROP_SIGV4_ENABLED.to_string(), + "true".to_string(), + ), + ( + REST_CATALOG_PROP_SIGNING_REGION.to_string(), + "us-east-1".to_string(), + ), + ( + REST_CATALOG_PROP_SIGNING_NAME.to_string(), + "glue".to_string(), + ), + ( + REST_CATALOG_PROP_SIGNING_ACCESS_KEY.to_string(), + "AKIDEXAMPLE".to_string(), + ), + ( + REST_CATALOG_PROP_SIGNING_SECRET_KEY.to_string(), + "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY".to_string(), + ), + ]); + let catalog = RestCatalog::new( + RestCatalogConfig::builder() + .uri(server.url()) + .props(props) + .build(), + None, + Runtime::current(), + None, + ); + + let namespaces = catalog.list_namespaces(None).await.unwrap(); + assert!(namespaces.is_empty()); + + config_mock.assert_async().await; + list_ns_mock.assert_async().await; + } + + #[tokio::test] + async fn test_sigv4_composes_with_token_auth() { + // SigV4 signs on top of token auth: the bearer token is relocated to + // `X-Iceberg-Authorization` and the signature takes `Authorization`. + let props = HashMap::from([ + ( + REST_CATALOG_PROP_SIGV4_ENABLED.to_string(), + "true".to_string(), + ), + ( + REST_CATALOG_PROP_SIGNING_REGION.to_string(), + "us-east-1".to_string(), + ), + ( + REST_CATALOG_PROP_SIGNING_NAME.to_string(), + "glue".to_string(), + ), + ( + REST_CATALOG_PROP_SIGNING_ACCESS_KEY.to_string(), + "AKIDEXAMPLE".to_string(), + ), + ( + REST_CATALOG_PROP_SIGNING_SECRET_KEY.to_string(), + "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY".to_string(), + ), + ("token".to_string(), "some-oauth-token".to_string()), + ]); + let config = RestCatalogConfig::builder() + .uri("http://localhost".to_string()) + .props(props) + .build(); + + let session = config + .resolve_auth_manager() + .unwrap() + .init_session() + .await + .unwrap(); + let mut req = reqwest::Client::new() + .get("https://rest.example.com/v1/config") + .build() + .unwrap(); + session.authenticate(&mut req).await.unwrap(); + + let headers = req.headers(); + assert_eq!( + headers.get("x-iceberg-authorization").unwrap(), + "Bearer some-oauth-token" + ); + let auth = headers.get("authorization").unwrap().to_str().unwrap(); + assert!(auth.starts_with("AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/")); + // The relocated token is part of the signature. + assert!(auth.contains("x-iceberg-authorization")); + } + + #[tokio::test] + async fn test_auth_type_none_disables_auth() { + // An explicit `rest.auth.type=none` wins over a configured token. + let props = HashMap::from([ + (REST_CATALOG_PROP_AUTH_TYPE.to_string(), "none".to_string()), + ("token".to_string(), "some-oauth-token".to_string()), + ]); + let config = RestCatalogConfig::builder() + .uri("http://localhost".to_string()) + .props(props) + .build(); + + let session = config + .resolve_auth_manager() + .unwrap() + .init_session() + .await + .unwrap(); + let mut req = reqwest::Client::new() + .get("https://rest.example.com/v1/config") + .build() + .unwrap(); + session.authenticate(&mut req).await.unwrap(); + assert!(req.headers().get("authorization").is_none()); + } + + #[test] + fn test_unknown_auth_type_is_rejected() { + let props = HashMap::from([( + REST_CATALOG_PROP_AUTH_TYPE.to_string(), + "kerberos".to_string(), + )]); + let config = RestCatalogConfig::builder() + .uri("http://localhost".to_string()) + .props(props) + .build(); + + let err = config.resolve_auth_manager().unwrap_err(); + assert!(err.message().contains(REST_CATALOG_PROP_AUTH_TYPE)); + } + + #[test] + fn test_sigv4_requires_region_and_name() { + // Missing region. + let config = RestCatalogConfig::builder() + .uri("http://localhost".to_string()) + .props(HashMap::from([( + REST_CATALOG_PROP_SIGV4_ENABLED.to_string(), + "true".to_string(), + )])) + .build(); + let err = config.resolve_auth_manager().unwrap_err(); + assert!(err.message().contains(REST_CATALOG_PROP_SIGNING_REGION)); + + // Region present, name missing. + let config = RestCatalogConfig::builder() + .uri("http://localhost".to_string()) + .props(HashMap::from([ + ( + REST_CATALOG_PROP_SIGV4_ENABLED.to_string(), + "true".to_string(), + ), + ( + REST_CATALOG_PROP_SIGNING_REGION.to_string(), + "us-east-1".to_string(), + ), + ])) + .build(); + let err = config.resolve_auth_manager().unwrap_err(); + assert!(err.message().contains(REST_CATALOG_PROP_SIGNING_NAME)); + } + + fn sigv4_props() -> HashMap { + HashMap::from([ + ( + REST_CATALOG_PROP_AUTH_TYPE.to_string(), + AUTH_TYPE_SIGV4.to_string(), + ), + ( + REST_CATALOG_PROP_SIGNING_REGION.to_string(), + "us-east-1".to_string(), + ), + ( + REST_CATALOG_PROP_SIGNING_NAME.to_string(), + "execute-api".to_string(), + ), + ( + REST_CATALOG_PROP_SIGNING_ACCESS_KEY.to_string(), + "ak".to_string(), + ), + ( + REST_CATALOG_PROP_SIGNING_SECRET_KEY.to_string(), + "sk".to_string(), + ), + ]) + } + + #[test] + fn test_sigv4_delegate_auth_type_explicit() { + // `none` forces a Noop delegate even when a token is configured. + let mut props = sigv4_props(); + props.insert("token".to_string(), "tok".to_string()); + props.insert( + REST_CATALOG_PROP_SIGV4_DELEGATE_AUTH_TYPE.to_string(), + AUTH_TYPE_NONE.to_string(), + ); + let config = RestCatalogConfig::builder() + .uri("http://localhost".to_string()) + .props(props) + .build(); + let manager = config.resolve_auth_manager().unwrap(); + assert!(format!("{manager:?}").contains("NoopAuthManager")); + + // `oauth2` forces an OAuth2 delegate even without a token/credential. + let mut props = sigv4_props(); + props.insert( + REST_CATALOG_PROP_SIGV4_DELEGATE_AUTH_TYPE.to_string(), + AUTH_TYPE_OAUTH2.to_string(), + ); + let config = RestCatalogConfig::builder() + .uri("http://localhost".to_string()) + .props(props) + .build(); + let manager = config.resolve_auth_manager().unwrap(); + assert!(format!("{manager:?}").contains("OAuth2Manager")); + + // Unknown delegate types are rejected. + let mut props = sigv4_props(); + props.insert( + REST_CATALOG_PROP_SIGV4_DELEGATE_AUTH_TYPE.to_string(), + "kerberos".to_string(), + ); + let config = RestCatalogConfig::builder() + .uri("http://localhost".to_string()) + .props(props) + .build(); + let err = config.resolve_auth_manager().unwrap_err(); + assert!( + err.message() + .contains(REST_CATALOG_PROP_SIGV4_DELEGATE_AUTH_TYPE) + ); + } + + #[tokio::test] + async fn test_sigv4_catalog_session_rebuilds_signer_from_merged_props() { + let config = RestCatalogConfig::builder() + .uri("http://localhost".to_string()) + .props(sigv4_props()) + .build(); + let manager = config.resolve_auth_manager().unwrap(); + + // Server-supplied `rest.signing-*` in the merged props is honored. + let mut merged = sigv4_props(); + merged.insert( + REST_CATALOG_PROP_SIGNING_REGION.to_string(), + "eu-west-1".to_string(), + ); + let session = manager.catalog_session(&merged).await.unwrap(); + assert!(format!("{session:?}").contains("eu-west-1")); + + // Props without a signing config keep the configured signer. + let session = manager.catalog_session(&HashMap::new()).await.unwrap(); + assert!(format!("{session:?}").contains("us-east-1")); + } + + #[test] + fn test_with_auth_manager_overrides_config() { + // A custom auth manager takes precedence over `rest.auth.type`. + #[derive(Debug)] + struct StubAuthManager; + #[async_trait] + impl AuthManager for StubAuthManager { + async fn init_session(&self) -> Result> { + unimplemented!() + } + async fn catalog_session( + &self, + _props: &HashMap, + ) -> Result> { + unimplemented!() + } + } + + let config = RestCatalogConfig::builder() + .uri("http://localhost".to_string()) + .props(HashMap::from([( + REST_CATALOG_PROP_AUTH_TYPE.to_string(), + "kerberos".to_string(), + )])) + .custom_auth_manager(Some(Arc::new(StubAuthManager))) + .build(); + + // The unknown auth type is never consulted. + assert!(config.resolve_auth_manager().is_ok()); + } + #[tokio::test] async fn test_list_namespace_with_pagination() { let mut server = Server::new_async().await; diff --git a/crates/catalog/rest/src/client.rs b/crates/catalog/rest/src/client.rs index 07dc0620da..086d85bc60 100644 --- a/crates/catalog/rest/src/client.rs +++ b/crates/catalog/rest/src/client.rs @@ -17,34 +17,27 @@ use std::collections::HashMap; use std::fmt::{Debug, Formatter}; +use std::sync::Arc; -use http::StatusCode; use iceberg::{Error, ErrorKind, Result}; use reqwest::header::HeaderMap; use reqwest::{Client, IntoUrl, Method, Request, RequestBuilder, Response}; use serde::de::DeserializeOwned; -use tokio::sync::Mutex; use crate::RestCatalogConfig; -use crate::types::{ErrorResponse, TokenResponse}; +use crate::auth::{AuthManager, AuthSession}; pub(crate) struct HttpClient { client: Client, - /// The token to be used for authentication. - /// - /// It's possible to fetch the token from the server while needed. - token: Mutex>, - /// The token endpoint to be used for authentication. - token_endpoint: String, - /// The credential to be used for authentication. - credential: Option<(Option, String)>, /// Extra headers to be added to each request. extra_headers: HeaderMap, - /// Extra oauth parameters to be added to each authentication request. - extra_oauth_params: HashMap, /// Whether to disable header redaction in error logs (defaults to false for security). disable_header_redaction: bool, + /// The auth manager living for the lifetime of the catalog. + auth_manager: Arc, + /// The session authenticating requests in the current phase. + session: Arc, } impl Debug for HttpClient { @@ -58,16 +51,15 @@ impl Debug for HttpClient { impl HttpClient { /// Create a new http client. - pub fn new(cfg: &RestCatalogConfig) -> Result { - let extra_headers = cfg.extra_headers()?; + pub async fn new(cfg: &RestCatalogConfig) -> Result { + let auth_manager = cfg.resolve_auth_manager()?; + let session = auth_manager.init_session().await?; Ok(HttpClient { client: cfg.client().unwrap_or_default(), - token: Mutex::new(cfg.token()), - token_endpoint: cfg.get_token_endpoint(), - credential: cfg.credential(), - extra_headers, - extra_oauth_params: cfg.extra_oauth_params(), + extra_headers: cfg.extra_headers()?, disable_header_redaction: cfg.disable_header_redaction(), + auth_manager, + session, }) } @@ -75,172 +67,42 @@ impl HttpClient { /// /// If cfg carries new value, we will use cfg instead. /// Otherwise, we will keep the old value. - pub fn update_with(self, cfg: &RestCatalogConfig) -> Result { + /// + /// The auth manager is kept; it derives a new session from the merged + /// properties (carrying over state such as a cached token). + pub async fn update_with(self, cfg: &RestCatalogConfig) -> Result { let extra_headers = (!cfg.extra_headers()?.is_empty()) .then(|| cfg.extra_headers()) .transpose()? .unwrap_or(self.extra_headers); + let session = self.auth_manager.catalog_session(&cfg.auth_props()).await?; Ok(HttpClient { client: cfg.client().unwrap_or(self.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() - } else { - self.token_endpoint - }, - credential: cfg.credential().or(self.credential), extra_headers, - extra_oauth_params: if !cfg.extra_oauth_params().is_empty() { - cfg.extra_oauth_params() - } else { - self.extra_oauth_params - }, disable_header_redaction: cfg.disable_header_redaction(), + auth_manager: self.auth_manager, + session, }) } - /// This API is testing only to assert the token. + /// The session authenticating requests in the current phase. + pub(crate) fn session(&self) -> &Arc { + &self.session + } + + /// This API is testing only to assert the bearer token the session uses. #[cfg(test)] pub(crate) async fn token(&self) -> Option { let mut req = self - .request(Method::GET, &self.token_endpoint) + .request(Method::GET, "http://localhost/unused") .build() .unwrap(); - self.authenticate(&mut req).await.ok(); - self.token.lock().await.clone() - } - - async fn exchange_credential_for_token(&self) -> Result { - // Credential must exist here. - let (client_id, client_secret) = self.credential.as_ref().ok_or_else(|| { - Error::new( - ErrorKind::DataInvalid, - "Credential must be provided for authentication", - ) - })?; - - let mut params = HashMap::with_capacity(4); - params.insert("grant_type", "client_credentials"); - if let Some(client_id) = client_id { - params.insert("client_id", client_id); - } - params.insert("client_secret", client_secret); - params.extend( - self.extra_oauth_params - .iter() - .map(|(k, v)| (k.as_str(), v.as_str())), - ); - - let mut auth_req = self - .request(Method::POST, &self.token_endpoint) - .form(¶ms) - .build()?; - // extra headers add content-type application/json header it's necessary to override it with proper type - // note that form call doesn't add content-type header if already present - auth_req.headers_mut().insert( - http::header::CONTENT_TYPE, - http::HeaderValue::from_static("application/x-www-form-urlencoded"), - ); - let auth_url = auth_req.url().clone(); - let auth_resp = self.client.execute(auth_req).await?; - - let auth_res: TokenResponse = if auth_resp.status() == StatusCode::OK { - let text = auth_resp - .bytes() - .await - .map_err(|err| err.with_url(auth_url.clone()))?; - Ok(serde_json::from_slice(&text).map_err(|e| { - Error::new( - ErrorKind::Unexpected, - "Failed to parse response from rest catalog server!", - ) - .with_context("operation", "auth") - .with_context("url", auth_url.to_string()) - .with_context("json", String::from_utf8_lossy(&text)) - .with_source(e) - })?) - } else { - let code = auth_resp.status(); - let text = auth_resp - .bytes() - .await - .map_err(|err| err.with_url(auth_url.clone()))?; - let e: ErrorResponse = serde_json::from_slice(&text).map_err(|e| { - Error::new(ErrorKind::Unexpected, "Received unexpected response") - .with_context("code", code.to_string()) - .with_context("operation", "auth") - .with_context("url", auth_url.to_string()) - .with_context("json", String::from_utf8_lossy(&text)) - .with_source(e) - })?; - Err(Error::from(e)) - }?; - Ok(auth_res.access_token) - } - - /// Invalidate the current token without generating a new one. On the next request, the client - /// will attempt to generate a new token. - pub(crate) async fn invalidate_token(&self) -> Result<()> { - *self.token.lock().await = None; - Ok(()) - } - - /// Invalidate the current token and set a new one. Generates a new token before invalidating - /// the current token, meaning the old token will be used until this function acquires the lock - /// and overwrites the token. - /// - /// If credential is invalid, or the request fails, this method will return an error and leave - /// the current token unchanged. - pub(crate) async fn regenerate_token(&self) -> Result<()> { - let new_token = self.exchange_credential_for_token().await?; - *self.token.lock().await = Some(new_token.clone()); - Ok(()) - } - - /// Authenticates the request by adding a bearer token to the authorization header. - /// - /// This method supports three authentication modes: - /// - /// 1. **No authentication** - Skip authentication when both `credential` and `token` are missing. - /// 2. **Token authentication** - Use the provided `token` directly for authentication. - /// 3. **OAuth authentication** - Exchange `credential` for a token, cache it, then use it for authentication. - /// - /// When both `credential` and `token` are present, `token` takes precedence. - /// - /// # TODO: Support automatic token refreshing. - async fn authenticate(&self, req: &mut Request) -> Result<()> { - // Clone the token from lock without holding the lock for entire function. - let token = self.token.lock().await.clone(); - - if self.credential.is_none() && token.is_none() { - return Ok(()); - } - - // Either use the provided token or exchange credential for token, cache and use that - let token = match token { - Some(token) => token, - None => { - let token = self.exchange_credential_for_token().await?; - // Update token so that we use it for next request instead of - // exchanging credential for token from the server again - *self.token.lock().await = Some(token.clone()); - token - } - }; - - // Insert token in request. - req.headers_mut().insert( - http::header::AUTHORIZATION, - format!("Bearer {token}").parse().map_err(|e| { - Error::new( - ErrorKind::DataInvalid, - "Invalid token received from catalog server!", - ) - .with_source(e) - })?, - ); - - Ok(()) + self.session.authenticate(&mut req).await.ok(); + req.headers() + .get(http::header::AUTHORIZATION) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.strip_prefix("Bearer ")) + .map(str::to_string) } #[inline] @@ -250,17 +112,12 @@ impl HttpClient { .headers(self.extra_headers.clone()) } - /// Executes the given `Request` and returns a `Response`. - pub async fn execute(&self, mut request: Request) -> Result { - request.headers_mut().extend(self.extra_headers.clone()); - Ok(self.client.execute(request).await?) - } - // Queries the Iceberg REST catalog after authentication with the given `Request` and // returns a `Response`. pub async fn query_catalog(&self, mut request: Request) -> Result { - self.authenticate(&mut request).await?; - self.execute(request).await + request.headers_mut().extend(self.extra_headers.clone()); + self.session.authenticate(&mut request).await?; + Ok(self.client.execute(request).await?) } /// Returns whether header redaction is disabled for this client. diff --git a/crates/catalog/rest/src/lib.rs b/crates/catalog/rest/src/lib.rs index 383728401f..4877107cb9 100644 --- a/crates/catalog/rest/src/lib.rs +++ b/crates/catalog/rest/src/lib.rs @@ -51,11 +51,15 @@ #![deny(missing_docs)] +mod auth; mod catalog; mod client; mod endpoint; +mod signing; mod types; +pub use auth::*; pub use catalog::*; pub use endpoint::Endpoint; +pub use signing::*; pub use types::*; diff --git a/crates/catalog/rest/src/signing.rs b/crates/catalog/rest/src/signing.rs new file mode 100644 index 0000000000..2fa01b832e --- /dev/null +++ b/crates/catalog/rest/src/signing.rs @@ -0,0 +1,577 @@ +// 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. + +use chrono::{DateTime, Utc}; +use iceberg::{Error, ErrorKind, Result}; +use reqsign_core::hash::{base64_encode, hex_hmac_sha256, hex_sha256, hmac_sha256}; +use sha2::{Digest, Sha256}; + +/// Hex SHA-256 of the empty string. +const EMPTY_BODY_HEX_SHA256: &str = + "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"; + +/// How the payload hash is encoded in the `x-amz-content-sha256` header. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum PayloadHashMode { + /// Iceberg Java's RESTSigV4 style: base64 header for non-empty bodies, hex + /// for empty; the canonical request always uses hex. + IcebergRest, + /// Standard AWS SigV4 style: hex everywhere (e.g. AWS Glue). + StandardAws, +} + +/// Derives the AWS SigV4 signing key. +fn signing_key(secret: &str, date: &str, region: &str, service: &str) -> Vec { + let k_date = hmac_sha256(format!("AWS4{secret}").as_bytes(), date.as_bytes()); + let k_region = hmac_sha256(&k_date, region.as_bytes()); + let k_service = hmac_sha256(&k_region, service.as_bytes()); + hmac_sha256(&k_service, b"aws4_request") +} + +/// Computes the value of the `x-amz-content-sha256` header. +fn content_sha256_header(body: &[u8], mode: PayloadHashMode) -> String { + match mode { + PayloadHashMode::StandardAws => hex_sha256(body), + PayloadHashMode::IcebergRest => { + if body.is_empty() { + EMPTY_BODY_HEX_SHA256.to_string() + } else { + base64_encode(&Sha256::digest(body)) + } + } + } +} + +/// Builds the SigV4 canonical request. `headers` are (lowercased, trimmed) +/// pairs; `payload_hash` is always the hex sha256 of the body. +fn canonical_request( + method: &str, + canonical_uri: &str, + canonical_query: &str, + headers: &[(String, String)], + payload_hash: &str, +) -> String { + let mut sorted = headers.to_vec(); + sorted.sort_by(|a, b| a.0.cmp(&b.0)); + + let canonical_headers: String = sorted.iter().map(|(k, v)| format!("{k}:{v}\n")).collect(); + let signed_headers = sorted + .iter() + .map(|(k, _)| k.as_str()) + .collect::>() + .join(";"); + + format!( + "{method}\n{canonical_uri}\n{canonical_query}\n{canonical_headers}\n{signed_headers}\n{payload_hash}" + ) +} + +/// Builds the SigV4 string-to-sign. +fn string_to_sign(amz_date: &str, scope: &str, canonical_request: &str) -> String { + format!( + "AWS4-HMAC-SHA256\n{amz_date}\n{scope}\n{}", + hex_sha256(canonical_request.as_bytes()) + ) +} + +/// Static AWS-style credentials used for SigV4 signing of catalog requests. +#[derive(Clone)] +pub struct AwsCredentials { + /// AWS access key id. + pub access_key_id: String, + /// AWS secret access key. + pub secret_access_key: String, + /// Optional STS session token. + pub session_token: Option, +} + +/// AWS SigV4 signer following Iceberg Java's `RESTSigV4AuthSession`: it adds the +/// required amz headers and signs all request headers except a small blacklist. +#[derive(Clone)] +pub struct SigV4Signer { + credentials: AwsCredentials, + region: String, + service: String, + mode: PayloadHashMode, +} + +impl SigV4Signer { + /// Creates a new SigV4 signer. + pub fn new( + credentials: AwsCredentials, + region: String, + service: String, + mode: PayloadHashMode, + ) -> Self { + Self { + credentials, + region, + service, + mode, + } + } + + /// Signs `request` in place. + pub fn sign(&self, request: &mut reqwest::Request) -> Result<()> { + self.sign_at(request, Utc::now()) + } + + fn sign_at(&self, request: &mut reqwest::Request, now: DateTime) -> Result<()> { + let amz_date = now.format("%Y%m%dT%H%M%SZ").to_string(); + let date = now.format("%Y%m%d").to_string(); + let scope = format!("{date}/{}/{}/aws4_request", self.region, self.service); + + // The signed `host` must match what the HTTP layer sends: reqwest/hyper + // include an explicit non-default port in the Host header (the url crate + // normalizes scheme-default ports away, so `port()` is None for those), + // and the AWS SDK signs `host:port` likewise. + let host = { + let h = request + .url() + .host_str() + .ok_or_else(|| Error::new(ErrorKind::DataInvalid, "request url has no host"))?; + match request.url().port() { + Some(port) => format!("{h}:{port}"), + None => h.to_string(), + } + }; + // AWS SDK v2 parity (Aws4Signer default `doubleUrlEncode=true`): the + // canonical URI is the already-percent-encoded path encoded once more, + // keeping `/`. Unreserved-only paths are unchanged; a literal `,` + // becomes `%2C` and an encoded `%2C` becomes `%252C`, as Java signs. + let canonical_uri = uri_encode(request.url().path(), false); + let mut encoded_pairs: Vec<(String, String)> = request + .url() + .query_pairs() + .map(|(k, v)| (uri_encode(&k, true), uri_encode(&v, true))) + .collect(); + encoded_pairs.sort(); + let canonical_query = encoded_pairs + .iter() + .map(|(k, v)| format!("{k}={v}")) + .collect::>() + .join("&"); + + let body: &[u8] = match request.body() { + None => &[], + Some(b) => b.as_bytes().ok_or_else(|| { + Error::new( + ErrorKind::FeatureUnsupported, + "cannot sign a streaming request body", + ) + })?, + }; + let payload_hex = hex_sha256(body); + let content_header = content_sha256_header(body, self.mode); + + let mut headers = vec![ + ("host".to_string(), host), + ("x-amz-content-sha256".to_string(), content_header.clone()), + ("x-amz-date".to_string(), amz_date.clone()), + ]; + if let Some(tok) = &self.credentials.session_token { + headers.push(("x-amz-security-token".to_string(), tok.clone())); + } + + // Sign every other request header too (except a blacklist), as Iceberg + // Java / botocore do; signing only host/x-amz-* yields a signature mismatch. + const SKIP_HEADERS: &[&str] = &["user-agent", "authorization", "expect", "x-amzn-trace-id"]; + for (name, value) in request.headers().iter() { + let lname = name.as_str().to_ascii_lowercase(); + if SKIP_HEADERS.contains(&lname.as_str()) || headers.iter().any(|(k, _)| *k == lname) { + continue; + } + if let Ok(v) = value.to_str() { + headers.push((lname, v.trim().to_string())); + } + } + + // Canonical payload hash is ALWAYS hex (the IcebergRest split: header may be base64). + let creq = canonical_request( + request.method().as_str(), + &canonical_uri, + &canonical_query, + &headers, + &payload_hex, + ); + let sts = string_to_sign(&amz_date, &scope, &creq); + let key = signing_key( + &self.credentials.secret_access_key, + &date, + &self.region, + &self.service, + ); + let signature = hex_hmac_sha256(&key, sts.as_bytes()); + + let mut signed = headers.iter().map(|(k, _)| k.clone()).collect::>(); + signed.sort(); + let signed_headers = signed.join(";"); + let authorization = format!( + "AWS4-HMAC-SHA256 Credential={}/{scope}, SignedHeaders={signed_headers}, Signature={signature}", + self.credentials.access_key_id + ); + + let h = request.headers_mut(); + h.insert("x-amz-date", amz_date.parse().unwrap()); + h.insert("x-amz-content-sha256", content_header.parse().unwrap()); + if let Some(tok) = &self.credentials.session_token { + h.insert( + "x-amz-security-token", + tok.parse().map_err(|e| { + Error::new(ErrorKind::DataInvalid, "invalid session token").with_source(e) + })?, + ); + } + h.insert( + reqwest::header::AUTHORIZATION, + authorization.parse().map_err(|e| { + Error::new(ErrorKind::Unexpected, "invalid Authorization header").with_source(e) + })?, + ); + Ok(()) + } +} + +impl std::fmt::Debug for SigV4Signer { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SigV4Signer") + .field("region", &self.region) + .field("service", &self.service) + .field("mode", &self.mode) + .field("access_key_id", &self.credentials.access_key_id) + .finish_non_exhaustive() + } +} + +/// RFC 3986 URI encoding (AWS rules). When `encode_slash` is false, `/` is kept. +fn uri_encode(s: &str, encode_slash: bool) -> String { + let mut out = String::with_capacity(s.len()); + for b in s.bytes() { + match b { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => { + out.push(b as char) + } + b'/' if !encode_slash => out.push('/'), + _ => out.push_str(&format!("%{b:02X}")), + } + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + const EMPTY_HEX: &str = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"; + + #[test] + fn content_sha256_header_iceberg_mode() { + let v = content_sha256_header(b"hello", PayloadHashMode::IcebergRest); + assert_eq!(v, "LPJNul+wow4m6DsqxbninhsWHlwfp0JecwQzYpOLmCQ="); + let e = content_sha256_header(b"", PayloadHashMode::IcebergRest); + assert_eq!(e, EMPTY_HEX); + } + + #[test] + fn content_sha256_header_standard_mode() { + let v = content_sha256_header(b"hello", PayloadHashMode::StandardAws); + assert_eq!( + v, + "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824" + ); + } + + #[test] + fn signing_key_and_signature_match_aws_vector() { + let secret = "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY"; + let date = "20150830"; + let region = "us-east-1"; + let service = "service"; + let key = signing_key(secret, date, region, service); + + let string_to_sign = "AWS4-HMAC-SHA256\n\ +20150830T123600Z\n\ +20150830/us-east-1/service/aws4_request\n\ +bb579772317eb040ac9ed261061d46c1f17a8133879d6129b6e1c25292927e63"; + let sig = reqsign_core::hash::hex_hmac_sha256(&key, string_to_sign.as_bytes()); + assert_eq!( + sig, + "5fa00fa31553b73ebf1942676e86291e8372ff2a2260956d9b8aae1d763fbf31" + ); + } + + #[test] + fn canonical_request_get_vanilla() { + let headers = vec![ + ("host".to_string(), "example.amazonaws.com".to_string()), + ("x-amz-date".to_string(), "20150830T123600Z".to_string()), + ]; + let creq = canonical_request("GET", "/", "", &headers, &hex_sha256(b"")); + let expected = "GET\n/\n\n\ +host:example.amazonaws.com\n\ +x-amz-date:20150830T123600Z\n\ +\n\ +host;x-amz-date\n\ +e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"; + assert_eq!(creq, expected); + assert_eq!( + hex_sha256(creq.as_bytes()), + "bb579772317eb040ac9ed261061d46c1f17a8133879d6129b6e1c25292927e63" + ); + } + + #[test] + fn signs_request_iceberg_mode() { + let creds = AwsCredentials { + access_key_id: "AKIDEXAMPLE".into(), + secret_access_key: "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY".into(), + session_token: Some("SESSIONTOKEN".into()), + }; + let signer = SigV4Signer::new( + creds, + "us-east-1".into(), + "glue".into(), + PayloadHashMode::IcebergRest, + ); + let client = reqwest::Client::new(); + let mut req = client + .post("https://rest.example.com/v1/namespaces") + .body("{}") + .build() + .unwrap(); + + signer.sign(&mut req).unwrap(); + + let h = req.headers(); + assert!( + h.get("authorization") + .unwrap() + .to_str() + .unwrap() + .starts_with("AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/") + ); + assert!(h.contains_key("x-amz-date")); + assert_eq!(h.get("x-amz-security-token").unwrap(), "SESSIONTOKEN"); + let csha = h.get("x-amz-content-sha256").unwrap().to_str().unwrap(); + assert_eq!(csha, "RBNvo1WzZ4oRRq0W9+hknpT7T8If536DEMBg9hyq/4o="); + } + + #[test] + fn string_to_sign_get_vanilla() { + let creq_hash = "bb579772317eb040ac9ed261061d46c1f17a8133879d6129b6e1c25292927e63"; + let headers = vec![ + ("host".to_string(), "example.amazonaws.com".to_string()), + ("x-amz-date".to_string(), "20150830T123600Z".to_string()), + ]; + let creq = canonical_request("GET", "/", "", &headers, &hex_sha256(b"")); + assert_eq!(hex_sha256(creq.as_bytes()), creq_hash); + let sts = string_to_sign( + "20150830T123600Z", + "20150830/us-east-1/service/aws4_request", + &creq, + ); + assert!(sts.ends_with(creq_hash)); + assert!(sts.starts_with("AWS4-HMAC-SHA256\n20150830T123600Z\n")); + } + + /// Empty body uses the hex constant and existing headers are signed too + /// (mirrors Java's `TestRESTSigV4AuthSession::authenticateWithoutBody`). + #[test] + fn signs_empty_body_and_all_headers() { + let creds = AwsCredentials { + access_key_id: "AKIDEXAMPLE".into(), + secret_access_key: "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY".into(), + session_token: None, + }; + let signer = SigV4Signer::new( + creds, + "us-east-1".into(), + "glue".into(), + PayloadHashMode::IcebergRest, + ); + let client = reqwest::Client::new(); + let mut req = client + .get("https://rest.example.com/v1/config") + .header("content-type", "application/json") + .header("content-encoding", "gzip") + .build() + .unwrap(); + + signer.sign(&mut req).unwrap(); + + let h = req.headers(); + assert_eq!(h.get("x-amz-content-sha256").unwrap(), EMPTY_HEX); + assert!(!h.contains_key("x-amz-security-token")); + let auth = h.get("authorization").unwrap().to_str().unwrap(); + assert!(auth.starts_with("AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/")); + assert!(auth.contains( + "SignedHeaders=content-encoding;content-type;host;x-amz-content-sha256;x-amz-date" + )); + } + + /// The signed `host` must include an explicit non-default port, matching + /// what reqwest/hyper put on the wire and what the AWS SDK signs. + #[test] + fn signs_host_with_non_default_port() { + use chrono::TimeZone; + + let creds = AwsCredentials { + access_key_id: "AKIDEXAMPLE".into(), + secret_access_key: "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY".into(), + session_token: None, + }; + let signer = SigV4Signer::new( + creds, + "us-east-1".into(), + "glue".into(), + PayloadHashMode::IcebergRest, + ); + let client = reqwest::Client::new(); + let mut req = client + .get("https://rest.example.com:8181/v1/config") + .build() + .unwrap(); + let now = Utc.with_ymd_and_hms(2015, 8, 30, 12, 36, 0).unwrap(); + + signer.sign_at(&mut req, now).unwrap(); + + // Independently recompute the signature with `host:port` in the + // canonical request; a signer that drops the port won't match. + let headers = vec![ + ("host".to_string(), "rest.example.com:8181".to_string()), + ("x-amz-content-sha256".to_string(), EMPTY_HEX.to_string()), + ("x-amz-date".to_string(), "20150830T123600Z".to_string()), + ]; + let creq = canonical_request("GET", "/v1/config", "", &headers, EMPTY_HEX); + let sts = string_to_sign( + "20150830T123600Z", + "20150830/us-east-1/glue/aws4_request", + &creq, + ); + let key = signing_key( + "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY", + "20150830", + "us-east-1", + "glue", + ); + let expected = hex_hmac_sha256(&key, sts.as_bytes()); + + let auth = req + .headers() + .get("authorization") + .unwrap() + .to_str() + .unwrap(); + assert!(auth.ends_with(&format!("Signature={expected}")), "{auth}"); + } + + /// AWS SDK v2 parity (`doubleUrlEncode`): the canonical URI encodes the + /// serialized path once more — literal `,` becomes `%2C`, an encoded + /// `%2C` becomes `%252C` — while plain paths stay byte-identical. + #[test] + fn canonical_uri_is_aws_double_encoded() { + use chrono::TimeZone; + + assert_eq!(uri_encode("/v1/namespaces", false), "/v1/namespaces"); + assert_eq!( + uri_encode("/v1/namespaces/a%2Cb/tables/x,y", false), + "/v1/namespaces/a%252Cb/tables/x%2Cy" + ); + + let creds = AwsCredentials { + access_key_id: "AKIDEXAMPLE".into(), + secret_access_key: "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY".into(), + session_token: None, + }; + let signer = SigV4Signer::new( + creds, + "us-east-1".into(), + "glue".into(), + PayloadHashMode::IcebergRest, + ); + let client = reqwest::Client::new(); + let mut req = client + .get("https://rest.example.com/v1/namespaces/a%2Cb/tables/x,y") + .build() + .unwrap(); + let now = Utc.with_ymd_and_hms(2015, 8, 30, 12, 36, 0).unwrap(); + + signer.sign_at(&mut req, now).unwrap(); + + let headers = vec![ + ("host".to_string(), "rest.example.com".to_string()), + ("x-amz-content-sha256".to_string(), EMPTY_HEX.to_string()), + ("x-amz-date".to_string(), "20150830T123600Z".to_string()), + ]; + let creq = canonical_request( + "GET", + "/v1/namespaces/a%252Cb/tables/x%2Cy", + "", + &headers, + EMPTY_HEX, + ); + let sts = string_to_sign( + "20150830T123600Z", + "20150830/us-east-1/glue/aws4_request", + &creq, + ); + let key = signing_key( + "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY", + "20150830", + "us-east-1", + "glue", + ); + let expected = hex_hmac_sha256(&key, sts.as_bytes()); + + let auth = req + .headers() + .get("authorization") + .unwrap() + .to_str() + .unwrap(); + assert!(auth.ends_with(&format!("Signature={expected}")), "{auth}"); + } + + #[test] + fn signs_request_standard_mode_uses_hex_header() { + let creds = AwsCredentials { + access_key_id: "AKIDEXAMPLE".into(), + secret_access_key: "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY".into(), + session_token: None, + }; + let signer = SigV4Signer::new( + creds, + "us-east-1".into(), + "glue".into(), + PayloadHashMode::StandardAws, + ); + let client = reqwest::Client::new(); + let mut req = client + .post("https://rest.example.com/v1/namespaces") + .body("hello") + .build() + .unwrap(); + + signer.sign(&mut req).unwrap(); + + // StandardAws keeps the header in hex. + assert_eq!( + req.headers().get("x-amz-content-sha256").unwrap(), + "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824" + ); + } +}