diff --git a/crates/iam/src/lib.rs b/crates/iam/src/lib.rs index 316e064c9..e4ac0662a 100644 --- a/crates/iam/src/lib.rs +++ b/crates/iam/src/lib.rs @@ -168,6 +168,15 @@ pub async fn build_iam_sys(ecstore: Arc) -> Result) -> Result>> { + let iam_instance = build_iam_sys(ecstore).await?; + let _ = IAM_SYS.set(iam_instance.clone()); + Ok(iam_instance) +} + #[instrument(skip(ecstore))] pub async fn init_iam_sys(ecstore: Arc) -> Result>> { if let Some(existing) = IAM_SYS.get() { diff --git a/rustfs/src/auth.rs b/rustfs/src/auth.rs index 298c39fa5..73bf72378 100644 --- a/rustfs/src/auth.rs +++ b/rustfs/src/auth.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::runtime_sources::{AppContext, current_action_credentials, current_ready_iam_handle}; +use crate::runtime_sources::{AppContext, ServerContextSlot, current_action_credentials, current_ready_iam_handle}; use http::HeaderMap; use http::Uri; use rustfs_credentials::Credentials; @@ -113,6 +113,7 @@ pub struct IAMAuth { simple_auth: SimpleAuth, access_key: String, secret_key: SecretKey, + server_ctx: Option>, } impl Clone for IAMAuth { @@ -121,6 +122,7 @@ impl Clone for IAMAuth { simple_auth: SimpleAuth::from_single(self.access_key.clone(), self.secret_key.clone()), access_key: self.access_key.clone(), secret_key: self.secret_key.clone(), + server_ctx: self.server_ctx.clone(), } } } @@ -134,8 +136,19 @@ impl IAMAuth { simple_auth, access_key, secret_key, + server_ctx: None, } } + + pub(crate) fn with_server_context( + ak: impl Into, + sk: impl Into, + server_ctx: std::sync::Arc, + ) -> Self { + let mut auth = Self::new(ak, sk); + auth.server_ctx = Some(server_ctx); + auth + } } #[async_trait::async_trait] @@ -186,7 +199,15 @@ impl S3Auth for IAMAuth { return Ok(key); } - if let Ok(iam_store) = current_ready_iam_handle() { + let iam_store = match &self.server_ctx { + Some(server_ctx) => server_ctx + .installed_app_context() + .filter(|context| context.iam().is_ready()) + .map(|context| context.iam().handle()) + .ok_or(()), + None => current_ready_iam_handle().map_err(|_| ()), + }; + if let Ok(iam_store) = iam_store { // Use check_key instead of get_user to ensure user is loaded from disk if not in cache // This is important for newly created users that may not be in cache yet. // check_key will automatically attempt to load the user from disk if not found in cache. diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index 8656ca486..0859d99bb 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -732,7 +732,7 @@ pub async fn start_http_server( .transpose() .map_err(Error::other)?; - b.set_auth(IAMAuth::new(access_key, secret_key)); + b.set_auth(IAMAuth::with_server_context(access_key, secret_key, server_ctx.clone())); b.set_access(store); b.set_route(storage::metadata_route::with_metadata_route( admin::make_admin_route(config.console_enable, admin_server_ctx)?, diff --git a/rustfs/src/startup_iam.rs b/rustfs/src/startup_iam.rs index 4dffb7f71..6220e1c12 100644 --- a/rustfs/src/startup_iam.rs +++ b/rustfs/src/startup_iam.rs @@ -16,7 +16,7 @@ use crate::runtime_sources::{AppContext, ServerContextSlot}; use crate::server::{ServiceStateManager, publish_ready_when_runtime_ready}; use crate::storage_api::startup::iam::ECStore; use rustfs_common::{GlobalReadiness, SystemStage}; -use rustfs_iam::init_iam_sys; +use rustfs_iam::init_iam_sys_for_context; use rustfs_kms::KmsServiceManager; use std::future::Future; use std::io::Result; @@ -86,15 +86,13 @@ where } async fn finalize_iam_recovery( + iam: Arc>, store: Arc, kms_interface: Arc, readiness: Arc, state_manager: Option>, server_ctx: Arc, ) -> Result<()> { - // Resolve the globally published system directly (not context-first — - // the AppContext does not exist yet; creating it is this function's job). - let iam = rustfs_iam::get().map_err(|_| std::io::Error::other("IAM recovered but unavailable"))?; let endpoint_pools = heal_control_endpoint_pools(store.as_ref())?; AppContext::ensure_startup_after_iam(store, kms_interface, &server_ctx, iam)?; let topology_ready = @@ -147,6 +145,8 @@ fn spawn_iam_recovery_task( let finalize_readiness = readiness; let finalize_state_manager = state_manager; let finalize_server_ctx = server_ctx; + let recovered_iam = Arc::new(std::sync::Mutex::new(None)); + let init_recovered_iam = recovered_iam.clone(); tokio::spawn(async move { run_iam_recovery_loop( initial_interval, @@ -154,7 +154,14 @@ fn spawn_iam_recovery_task( shutdown_token, move || { let store = init_store.clone(); - Box::pin(async move { attempt_init_iam_sys(store).await.map(|_| ()) }) + let recovered_iam = init_recovered_iam.clone(); + Box::pin(async move { + let iam = attempt_init_iam_sys(store).await?; + *recovered_iam + .lock() + .map_err(|_| std::io::Error::other("IAM recovery state poisoned"))? = Some(iam); + Ok(()) + }) }, move || { let store = finalize_store.clone(); @@ -162,7 +169,11 @@ fn spawn_iam_recovery_task( let readiness = finalize_readiness.clone(); let state_manager = finalize_state_manager.clone(); let server_ctx = finalize_server_ctx.clone(); - Box::pin(async move { finalize_iam_recovery(store, kms_interface, readiness, state_manager, server_ctx).await }) + let iam = recovered_iam.lock().ok().and_then(|mut recovered| recovered.take()); + Box::pin(async move { + let iam = iam.ok_or_else(|| std::io::Error::other("IAM recovered but unavailable"))?; + finalize_iam_recovery(iam, store, kms_interface, readiness, state_manager, server_ctx).await + }) }, ) .await; @@ -359,7 +370,7 @@ async fn attempt_init_iam_sys( return Err(std::io::Error::other("forced test IAM bootstrap failure")); } - init_iam_sys(store).await.map_err(std::io::Error::other) + init_iam_sys_for_context(store).await.map_err(std::io::Error::other) } /// Attempt IAM bootstrap at startup. If it fails, enter degraded mode and diff --git a/rustfs/src/storage/access.rs b/rustfs/src/storage/access.rs index 7ca56d505..9b567f0e8 100644 --- a/rustfs/src/storage/access.rs +++ b/rustfs/src/storage/access.rs @@ -677,7 +677,15 @@ pub async fn authorize_request(req: &mut S3Request, action: Action) -> S3R let version_id = req_info.version_id.clone(); if let Some(cred) = &cred { - let Ok(iam_store) = runtime_sources::current_ready_iam_handle() else { + let iam_store = match req.extensions.get::>() { + Some(server_ctx) => server_ctx + .installed_app_context() + .filter(|context| context.iam().is_ready()) + .map(|context| context.iam().handle()) + .ok_or(()), + None => runtime_sources::current_ready_iam_handle().map_err(|_| ()), + }; + let Ok(iam_store) = iam_store else { return Err(S3Error::with_message( S3ErrorCode::InternalError, format!("authorize_request {:?}", IamError::IamSysNotInitialized), diff --git a/rustfs/tests/embedded_multi_instance_test.rs b/rustfs/tests/embedded_multi_instance_test.rs index d19e8b01f..f6010f90b 100644 --- a/rustfs/tests/embedded_multi_instance_test.rs +++ b/rustfs/tests/embedded_multi_instance_test.rs @@ -71,22 +71,27 @@ fn hmac(key: &[u8], value: &str) -> Vec { fn signed_admin_request( client: &reqwest::Client, endpoint: &str, + method: reqwest::Method, request_path: &str, access_key: &str, secret_key: &str, + body: &[u8], ) -> reqwest::RequestBuilder { let host = endpoint .strip_prefix("http://") .or_else(|| endpoint.strip_prefix("https://")) .expect("embedded endpoint scheme"); - let payload_hash = sha256_hex(b""); + let payload_hash = sha256_hex(body); let now = Utc::now(); let amz_date = now.format("%Y%m%dT%H%M%SZ").to_string(); let date = now.format("%Y%m%d").to_string(); let canonical_headers = format!("host:{host}\nx-amz-content-sha256:{payload_hash}\nx-amz-date:{amz_date}\n"); let signed_headers = "host;x-amz-content-sha256;x-amz-date"; let (path, query) = request_path.split_once('?').unwrap_or((request_path, "")); - let canonical_request = format!("GET\n{path}\n{query}\n{canonical_headers}\n{signed_headers}\n{payload_hash}"); + let canonical_request = format!( + "{}\n{path}\n{query}\n{canonical_headers}\n{signed_headers}\n{payload_hash}", + method.as_str() + ); let scope = format!("{date}/us-east-1/s3/aws4_request"); let string_to_sign = format!("AWS4-HMAC-SHA256\n{amz_date}\n{scope}\n{}", sha256_hex(canonical_request.as_bytes())); let date_key = hmac(format!("AWS4{secret_key}").as_bytes(), &date); @@ -99,11 +104,12 @@ fn signed_admin_request( ); client - .get(format!("{endpoint}{request_path}")) + .request(method, format!("{endpoint}{request_path}")) .header("host", host) .header("x-amz-content-sha256", payload_hash) .header("x-amz-date", amz_date) .header("authorization", authorization) + .body(body.to_vec()) } // backlog#1052 acceptance: a second embedded server in the same process no @@ -328,6 +334,81 @@ async fn two_embedded_servers_isolate_auth_and_data_planes_body() { server_b.shutdown().await; } +#[cfg(feature = "e2e-test-hooks")] +#[tokio::test] +async fn embedded_servers_isolate_iam_users_and_policies() { + let port_a = find_available_port().expect("find free port for server A"); + let server_a = RustFSServerBuilder::new() + .address(format!("127.0.0.1:{port_a}")) + .access_key("iam-root-a") + .secret_key("iam-root-secret-a") + .build() + .await + .expect("start embedded server A"); + let port_b = find_available_port().expect("find free port for server B"); + let server_b = RustFSServerBuilder::new() + .address(format!("127.0.0.1:{port_b}")) + .access_key("iam-root-b") + .secret_key("iam-root-secret-b") + .build() + .await + .expect("start embedded server B"); + + let http = reqwest::Client::builder() + .no_proxy() + .build() + .expect("build local admin client"); + let user_body = serde_json::json!({"secretKey": "iam-user-secret-a", "status": "enabled"}).to_string(); + for (path, body) in [ + ("/rustfs/admin/v3/add-user?accessKey=iam-user-a", user_body.as_bytes()), + ( + "/rustfs/admin/v3/set-user-or-group-policy?policyName=readwrite&userOrGroup=iam-user-a&isGroup=false", + b"".as_slice(), + ), + ] { + let response = signed_admin_request( + &http, + &server_a.endpoint(), + reqwest::Method::PUT, + path, + server_a.access_key(), + server_a.secret_key(), + body, + ) + .send() + .await + .expect("send server A IAM request"); + assert!(response.status().is_success(), "server A IAM setup failed: {}", response.status()); + } + + let client_b = s3_client(&server_b.endpoint(), server_b.access_key(), server_b.secret_key()); + client_b + .create_bucket() + .bucket("private-to-b") + .send() + .await + .expect("create server B bucket"); + client_b + .put_object() + .bucket("private-to-b") + .key("secret.txt") + .body(ByteStream::from_static(b"server B data")) + .send() + .await + .expect("write server B object"); + + let cross_instance = s3_client(&server_b.endpoint(), "iam-user-a", "iam-user-secret-a") + .get_object() + .bucket("private-to-b") + .key("secret.txt") + .send() + .await; + assert!(cross_instance.is_err(), "server A IAM user must not authorize against server B"); + + server_b.shutdown().await; + server_a.shutdown().await; +} + #[cfg(feature = "e2e-test-hooks")] #[tokio::test] async fn second_embedded_server_fails_closed_until_its_context_slot_is_installed() { @@ -392,10 +473,11 @@ async fn second_embedded_server_fails_closed_until_its_context_slot_is_installed .build() .expect("build local admin client without proxy"); let inspect_path = "/rustfs/admin/v3/inspect-data?file=marker.txt&volume=startup-window"; - let before_install = signed_admin_request(&http, &endpoint_b, inspect_path, b_access_key, b_secret_key) - .send() - .await - .expect("server B HTTP listener must accept the paused request"); + let before_install = + signed_admin_request(&http, &endpoint_b, reqwest::Method::GET, inspect_path, b_access_key, b_secret_key, b"") + .send() + .await + .expect("server B HTTP listener must accept the paused request"); let before_install_status = before_install.status(); let before_install_body = before_install.text().await.expect("read paused response body"); assert_eq!(before_install_status, StatusCode::SERVICE_UNAVAILABLE, "{before_install_body}"); @@ -426,10 +508,18 @@ async fn second_embedded_server_fails_closed_until_its_context_slot_is_installed .await .expect("server B writes its marker"); - let after_install = signed_admin_request(&http, &server_b.endpoint(), inspect_path, b_access_key, b_secret_key) - .send() - .await - .expect("server B admin request after context installation"); + let after_install = signed_admin_request( + &http, + &server_b.endpoint(), + reqwest::Method::GET, + inspect_path, + b_access_key, + b_secret_key, + b"", + ) + .send() + .await + .expect("server B admin request after context installation"); assert_eq!(after_install.status(), StatusCode::OK); assert_eq!(after_install.bytes().await.expect("read server B marker"), b"from B".as_slice());