diff --git a/crates/e2e_test/src/common.rs b/crates/e2e_test/src/common.rs index 42fe5f92a..7fad7ea76 100644 --- a/crates/e2e_test/src/common.rs +++ b/crates/e2e_test/src/common.rs @@ -30,6 +30,7 @@ use reqwest::StatusCode; use rustfs_signer::constants::UNSIGNED_PAYLOAD; use rustfs_signer::sign_v4; use s3s::Body; +use serde_json; use std::ffi::OsStr; use std::fs as stdfs; use std::io::ErrorKind; @@ -1583,6 +1584,156 @@ impl Drop for RustFSTestClusterEnvironment { } } +/// Send a SigV4-signed HTTP request and return the raw `reqwest::Response`. +/// +/// Unlike [`signed_s3_request`], this variant accepts `body: Option>` +/// (binary-safe) and reorders parameters so that `access_key`/`secret_key` +/// appear before the body — matching the convention used by the replication +/// extension and object-lambda e2e suites. +pub(crate) async fn signed_request( + method: http::Method, + url: &str, + access_key: &str, + secret_key: &str, + body: Option>, + content_type: Option<&str>, +) -> Result> { + let uri = url.parse::()?; + let authority = uri.authority().ok_or("request URL missing authority")?.to_string(); + let mut request = http::Request::builder().method(method.clone()).uri(uri); + request = request.header(HOST, authority); + request = request.header("x-amz-content-sha256", UNSIGNED_PAYLOAD); + if let Some(content_type) = content_type { + request = request.header(CONTENT_TYPE, content_type); + } + + let content_len = body.as_ref().map(|body| body.len() as i64).unwrap_or_default(); + let signed = sign_v4(request.body(Body::empty())?, content_len, access_key, secret_key, "", "us-east-1"); + + let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())?; + let client = local_http_client(); + let mut request_builder = client.request(reqwest_method, url); + for (name, value) in signed.headers() { + request_builder = request_builder.header(name, value); + } + if let Some(body) = body { + request_builder = request_builder.body(body); + } + + Ok(request_builder.send().await?) +} + +/// Like [`signed_request`], but uses a caller-supplied `reqwest::Client` +/// instead of the shared [`local_http_client`]. +pub(crate) async fn signed_request_with_client( + client: &reqwest::Client, + method: http::Method, + url: &str, + access_key: &str, + secret_key: &str, + body: Option>, + content_type: Option<&str>, +) -> Result> { + let uri = url.parse::()?; + let authority = uri.authority().ok_or("request URL missing authority")?.to_string(); + let mut request = http::Request::builder().method(method.clone()).uri(uri); + request = request.header(HOST, authority); + request = request.header("x-amz-content-sha256", UNSIGNED_PAYLOAD); + if let Some(content_type) = content_type { + request = request.header(CONTENT_TYPE, content_type); + } + + let content_len = body.as_ref().map(|body| body.len() as i64).unwrap_or_default(); + let signed = sign_v4(request.body(Body::empty())?, content_len, access_key, secret_key, "", "us-east-1"); + + let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())?; + let mut request_builder = client.request(reqwest_method, url); + for (name, value) in signed.headers() { + request_builder = request_builder.header(name, value); + } + if let Some(body) = body { + request_builder = request_builder.body(body); + } + + Ok(request_builder.send().await?) +} + +/// Like [`signed_request`], but includes a `session_token` in the +/// `x-amz-security-token` header and passes it to the SigV4 signer. +pub(crate) async fn signed_request_with_session_token( + method: http::Method, + url: &str, + access_key: &str, + secret_key: &str, + session_token: &str, + body: Option>, + content_type: Option<&str>, +) -> Result> { + let uri = url.parse::()?; + let authority = uri.authority().ok_or("request URL missing authority")?.to_string(); + let mut request = http::Request::builder().method(method.clone()).uri(uri); + request = request.header(HOST, authority); + request = request.header("x-amz-content-sha256", UNSIGNED_PAYLOAD); + if !session_token.is_empty() { + request = request.header("x-amz-security-token", session_token); + } + if let Some(content_type) = content_type { + request = request.header(CONTENT_TYPE, content_type); + } + + let content_len = body.as_ref().map(|body| body.len() as i64).unwrap_or_default(); + let signed = sign_v4( + request.body(Body::empty())?, + content_len, + access_key, + secret_key, + session_token, + "us-east-1", + ); + + let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())?; + let client = local_http_client(); + let mut request_builder = client.request(reqwest_method, url); + for (name, value) in signed.headers() { + request_builder = request_builder.header(name, value); + } + if let Some(body) = body { + request_builder = request_builder.body(body); + } + + Ok(request_builder.send().await?) +} + +/// Create a new user via the admin API. +pub(crate) async fn admin_create_user( + env: &RustFSTestEnvironment, + username: &str, + secret_key: &str, +) -> Result<(), Box> { + let url = format!("{}/rustfs/admin/v3/add-user?accessKey={}", env.url, username); + let body = serde_json::json!({ + "secretKey": secret_key, + "status": "enabled" + }); + let response = signed_request( + http::Method::PUT, + &url, + &env.access_key, + &env.secret_key, + Some(body.to_string().into_bytes()), + Some("application/json"), + ) + .await?; + + if response.status() != reqwest::StatusCode::OK { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(format!("create user failed: {status} {body}").into()); + } + + Ok(()) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/e2e_test/src/kms/common.rs b/crates/e2e_test/src/kms/common.rs index 8d81cbc18..290cb937d 100644 --- a/crates/e2e_test/src/kms/common.rs +++ b/crates/e2e_test/src/kms/common.rs @@ -40,7 +40,7 @@ use std::time::Duration; use tokio::fs; use tokio::net::TcpStream; use tokio::time::sleep; -use tracing::{debug, error, info}; +use tracing::{debug, error, info, warn}; // KMS-specific constants pub const TEST_BUCKET: &str = "kms-test-bucket"; @@ -177,6 +177,49 @@ pub async fn get_kms_status( Ok(status) } +/// Poll the KMS status endpoint until the backend reports ready or the timeout +/// expires. Replaces hard-coded `sleep(Duration::from_secs(3))` startup waits +/// with an active readiness probe so tests start as soon as KMS is usable +/// (typically < 1 s) instead of always waiting the full 3 s. +/// +/// Uses exponential back-off starting at 200 ms (doubling each attempt, capped +/// at 1 s) up to a total wall-clock budget of 5 s. +pub async fn wait_for_kms_ready( + base_url: &str, + access_key: &str, + secret_key: &str, +) -> Result<(), Box> { + let total_deadline = Duration::from_secs(5); + let start = tokio::time::Instant::now(); + let mut backoff = Duration::from_millis(200); + let max_backoff = Duration::from_secs(1); + let mut first_attempt = true; + + loop { + if !first_attempt { + if start.elapsed() >= total_deadline { + return Err("KMS failed to become ready within 5 seconds".into()); + } + sleep(backoff).await; + backoff = (backoff * 2).min(max_backoff); + } + first_attempt = false; + + match get_kms_status(base_url, access_key, secret_key).await { + Ok(status) => { + info!("KMS is ready (status: {})", status); + return Ok(()); + } + Err(e) => { + if start.elapsed() >= total_deadline { + return Err(format!("KMS did not become ready within 5 s: last error: {e}").into()); + } + warn!(error = %e, elapsed_ms = start.elapsed().as_millis() as u64, "KMS not ready yet, retrying…"); + } + } + } +} + /// Create a default KMS key for testing and return the created key ID pub async fn create_default_key( base_url: &str, @@ -861,6 +904,13 @@ impl LocalKMSTestEnvironment { Ok(default_key_id.to_string()) } + /// Poll the KMS status endpoint until the backend reports ready. + /// + /// Prefer this over a fixed `sleep` after calling `start_rustfs_for_local_kms`. + pub async fn wait_for_kms_ready(&self) -> Result<(), Box> { + wait_for_kms_ready(&self.base_env.url, &self.base_env.access_key, &self.base_env.secret_key).await + } + /// Configure Local KMS backend with a predefined default key pub async fn configure_local_kms(&self) -> Result> { // Use a fixed, predictable default key ID diff --git a/crates/e2e_test/src/object_lambda_test.rs b/crates/e2e_test/src/object_lambda_test.rs index 66f259e6d..aa6f7d4a1 100644 --- a/crates/e2e_test/src/object_lambda_test.rs +++ b/crates/e2e_test/src/object_lambda_test.rs @@ -12,12 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, init_logging, local_http_client}; +use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, init_logging, local_http_client, signed_request}; use aws_sdk_s3::primitives::ByteStream; use http::header::{CONTENT_TYPE, HOST}; use reqwest::StatusCode; -use rustfs_signer::constants::UNSIGNED_PAYLOAD; -use rustfs_signer::{pre_sign_v4, sign_v4}; +use rustfs_signer::pre_sign_v4; use rustfs_utils::egress::ENV_OUTBOUND_ALLOW_ORIGINS; use s3s::Body; use std::collections::HashMap; @@ -227,39 +226,6 @@ async fn presigned_get_request( Ok(local_http_client().get(signed.uri().to_string()).send().await?) } -async fn signed_request( - method: http::Method, - url: &str, - access_key: &str, - secret_key: &str, - body: Option>, - content_type: Option<&str>, -) -> Result> { - let uri = url.parse::()?; - let authority = uri.authority().ok_or("request URL missing authority")?.to_string(); - let mut request = http::Request::builder().method(method.clone()).uri(uri); - request = request.header(HOST, authority); - request = request.header("x-amz-content-sha256", UNSIGNED_PAYLOAD); - if let Some(content_type) = content_type { - request = request.header(CONTENT_TYPE, content_type); - } - - let content_len = body.as_ref().map(|body| body.len() as i64).unwrap_or_default(); - let signed = sign_v4(request.body(Body::empty())?, content_len, access_key, secret_key, "", "us-east-1"); - - let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())?; - let client = local_http_client(); - let mut request_builder = client.request(reqwest_method, url); - for (name, value) in signed.headers() { - request_builder = request_builder.header(name, value); - } - if let Some(body) = body { - request_builder = request_builder.body(body); - } - - Ok(request_builder.send().await?) -} - async fn configure_webhook_target( env: &RustFSTestEnvironment, target_name: &str, diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index d0282c83a..3d51d139b 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -13,8 +13,9 @@ // limitations under the License. use crate::common::{ - RustFSTestEnvironment, awscurl_available, awscurl_post_sts_form_urlencoded, init_logging, local_http_client, - replication_fast_env, rustfs_binary_path, + RustFSTestEnvironment, admin_create_user, awscurl_available, awscurl_post_sts_form_urlencoded, init_logging, + local_http_client, replication_fast_env, rustfs_binary_path, signed_request, signed_request_with_client, + signed_request_with_session_token, }; use crate::fake_s3_target::{ FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target, FaultAction as FakeTargetFault, Operation as FakeTargetOperation, @@ -35,7 +36,7 @@ use base64::{Engine, engine::general_purpose::STANDARD as BASE64_STANDARD}; use bytes::Bytes; use flate2::read::GzDecoder; use futures::{Stream, StreamExt}; -use http::header::{CONTENT_ENCODING, CONTENT_TYPE, HOST}; +use http::header::CONTENT_ENCODING; use http_body_util::{BodyExt, Full}; use hyper::body::Incoming; use hyper::server::conn::http1; @@ -56,9 +57,6 @@ use rustfs_madmin::{ AddServiceAccountReq, ListServiceAccountsResp, PeerInfo, PeerSite, ReplicateAddStatus, ReplicateEditStatus, ReplicateRemoveStatus, SRRemoveReq, SRResyncOpStatus, SRStatusInfo, SiteReplicationInfo, SyncStatus, }; -use rustfs_signer::constants::UNSIGNED_PAYLOAD; -use rustfs_signer::sign_v4; -use s3s::Body; use s3s::header::X_AMZ_REPLICATION_STATUS; use sha2::{Digest, Sha256}; use std::collections::BTreeMap; @@ -387,116 +385,6 @@ struct ReplicationResetStatusTarget { object: String, } -async fn signed_request( - method: http::Method, - url: &str, - access_key: &str, - secret_key: &str, - body: Option>, - content_type: Option<&str>, -) -> Result> { - let uri = url.parse::()?; - let authority = uri.authority().ok_or("request URL missing authority")?.to_string(); - let mut request = http::Request::builder().method(method.clone()).uri(uri); - request = request.header(HOST, authority); - request = request.header("x-amz-content-sha256", UNSIGNED_PAYLOAD); - if let Some(content_type) = content_type { - request = request.header(CONTENT_TYPE, content_type); - } - - let content_len = body.as_ref().map(|body| body.len() as i64).unwrap_or_default(); - let signed = sign_v4(request.body(Body::empty())?, content_len, access_key, secret_key, "", "us-east-1"); - - let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())?; - let client = local_http_client(); - let mut request_builder = client.request(reqwest_method, url); - for (name, value) in signed.headers() { - request_builder = request_builder.header(name, value); - } - if let Some(body) = body { - request_builder = request_builder.body(body); - } - - Ok(request_builder.send().await?) -} - -async fn signed_request_with_client( - client: &reqwest::Client, - method: http::Method, - url: &str, - access_key: &str, - secret_key: &str, - body: Option>, - content_type: Option<&str>, -) -> Result> { - let uri = url.parse::()?; - let authority = uri.authority().ok_or("request URL missing authority")?.to_string(); - let mut request = http::Request::builder().method(method.clone()).uri(uri); - request = request.header(HOST, authority); - request = request.header("x-amz-content-sha256", UNSIGNED_PAYLOAD); - if let Some(content_type) = content_type { - request = request.header(CONTENT_TYPE, content_type); - } - - let content_len = body.as_ref().map(|body| body.len() as i64).unwrap_or_default(); - let signed = sign_v4(request.body(Body::empty())?, content_len, access_key, secret_key, "", "us-east-1"); - - let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())?; - let mut request_builder = client.request(reqwest_method, url); - for (name, value) in signed.headers() { - request_builder = request_builder.header(name, value); - } - if let Some(body) = body { - request_builder = request_builder.body(body); - } - - Ok(request_builder.send().await?) -} - -async fn signed_request_with_session_token( - method: http::Method, - url: &str, - access_key: &str, - secret_key: &str, - session_token: &str, - body: Option>, - content_type: Option<&str>, -) -> Result> { - let uri = url.parse::()?; - let authority = uri.authority().ok_or("request URL missing authority")?.to_string(); - let mut request = http::Request::builder().method(method.clone()).uri(uri); - request = request.header(HOST, authority); - request = request.header("x-amz-content-sha256", UNSIGNED_PAYLOAD); - if !session_token.is_empty() { - request = request.header("x-amz-security-token", session_token); - } - if let Some(content_type) = content_type { - request = request.header(CONTENT_TYPE, content_type); - } - - let content_len = body.as_ref().map(|body| body.len() as i64).unwrap_or_default(); - let signed = sign_v4( - request.body(Body::empty())?, - content_len, - access_key, - secret_key, - session_token, - "us-east-1", - ); - - let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())?; - let client = local_http_client(); - let mut request_builder = client.request(reqwest_method, url); - for (name, value) in signed.headers() { - request_builder = request_builder.header(name, value); - } - if let Some(body) = body { - request_builder = request_builder.body(body); - } - - Ok(request_builder.send().await?) -} - fn extract_xml_tag(xml: &str, tag: &str) -> Option { let open = format!("<{tag}>"); let close = format!(""); @@ -1016,35 +904,6 @@ fn create_user_s3_client(env: &RustFSTestEnvironment, access_key: &str, secret_k Client::from_conf(config) } -async fn admin_create_user( - env: &RustFSTestEnvironment, - username: &str, - secret_key: &str, -) -> Result<(), Box> { - let url = format!("{}/rustfs/admin/v3/add-user?accessKey={}", env.url, username); - let body = serde_json::json!({ - "secretKey": secret_key, - "status": "enabled" - }); - let response = signed_request( - http::Method::PUT, - &url, - &env.access_key, - &env.secret_key, - Some(body.to_string().into_bytes()), - Some("application/json"), - ) - .await?; - - if response.status() != StatusCode::OK { - let status = response.status(); - let body = response.text().await.unwrap_or_default(); - return Err(format!("create user failed: {status} {body}").into()); - } - - Ok(()) -} - async fn admin_add_canned_policy( env: &RustFSTestEnvironment, policy_name: &str,