diff --git a/rustfs/src/admin/router.rs b/rustfs/src/admin/router.rs index 98e4ea9aa..a72858a8a 100644 --- a/rustfs/src/admin/router.rs +++ b/rustfs/src/admin/router.rs @@ -2655,6 +2655,10 @@ fn is_public_health_path(path: &str) -> bool { path == HEALTH_PREFIX || path == HEALTH_READY_PATH } +fn server_context_not_ready_error() -> S3Error { + s3_error!(ServiceUnavailable, "server context is not ready") +} + fn is_object_zip_download_token_path(method: &Method, uri: &Uri) -> bool { if method != Method::GET { return false; @@ -2783,6 +2787,9 @@ where async fn check_access(&self, req: &mut S3Request) -> S3Result<()> { if let Some(server_ctx) = &self.server_ctx { req.extensions.insert(server_ctx.clone()); + if !is_public_health_path(req.uri.path()) && server_ctx.installed_app_context().is_none() { + return Err(server_context_not_ready_error()); + } } if parse_replication_extension_request(&req.method, &req.uri).is_some() || parse_misc_extension_request(&req.method, &req.uri).is_some() @@ -2836,6 +2843,9 @@ where async fn call(&self, mut req: S3Request) -> S3Result> { if let Some(server_ctx) = &self.server_ctx { req.extensions.insert(server_ctx.clone()); + if !is_public_health_path(req.uri.path()) && server_ctx.installed_app_context().is_none() { + return Err(server_context_not_ready_error()); + } } if let Some(ext_req) = parse_replication_extension_request(&req.method, &req.uri) { return handle_replication_extension_request(&mut req, &ext_req).await; diff --git a/rustfs/src/admin/runtime_sources.rs b/rustfs/src/admin/runtime_sources.rs index 20828a43a..0d053bd52 100644 --- a/rustfs/src/admin/runtime_sources.rs +++ b/rustfs/src/admin/runtime_sources.rs @@ -76,19 +76,29 @@ pub(crate) fn object_store_from_req(req: &s3s::S3Request) -> Option(req: &s3s::S3Request) -> Option> { - req.extensions - .get::>() - .and_then(|slot| slot.app_context()) - .or_else(current_app_context) + app_context_from_extensions(&req.extensions) +} + +/// Resolve an application context from request extensions. +/// +/// An injected slot identifies the server that owns the request. If that +/// server has not finished startup yet, returning its ambient global context +/// could select a different server, so this fails closed. Only requests with +/// no slot retain the legacy ambient fallback. +pub(crate) fn app_context_from_extensions(extensions: &http::Extensions) -> Option> { + match extensions.get::>() { + Some(slot) => slot.installed_app_context(), + None => current_app_context(), + } } /// Field-borrow form of [`object_store_from_req`] for handlers that have /// already moved other request fields (body, credentials) out of the request. pub(crate) fn object_store_from_extensions(extensions: &http::Extensions) -> Option> { - extensions - .get::>() - .and_then(|slot| slot.object_store()) - .or_else(current_object_store_handle) + match extensions.get::>() { + Some(slot) => slot.installed_object_store(), + None => current_object_store_handle(), + } } pub(crate) fn current_notification_system() -> Option> { @@ -168,3 +178,151 @@ pub(crate) fn publish_storage_class_config(config: StorageClassConfig) { root_runtime_sources::fallback_storage_class_interface().set(config); } } + +#[cfg(test)] +mod tests { + use super::{ + AppContext, IamInterface, KmsInterface, ServerContextSlot, app_context_from_extensions, app_context_from_req, + current_app_context, object_store_from_extensions, publish_test_app_context, + }; + use crate::admin::router::{Operation, S3Router}; + use hyper::{Method, StatusCode}; + use matchit::Params; + use rustfs_iam::{store::object::ObjectStore, sys::IamSys}; + use rustfs_kms::KmsServiceManager; + use s3s::route::S3Route; + use s3s::{Body, S3ErrorCode, S3Request, S3Response, s3_error}; + use std::sync::Arc; + + struct UnreadyIam; + + impl IamInterface for UnreadyIam { + fn handle(&self) -> Arc> { + panic!("test context does not resolve IAM") + } + + fn is_ready(&self) -> bool { + false + } + } + + struct TestKms; + + impl KmsInterface for TestKms { + fn handle(&self) -> Arc { + Arc::new(KmsServiceManager::new()) + } + } + + async fn ambient_context() -> Arc { + if let Some(context) = current_app_context() { + return context; + } + + let env = rustfs_test_utils::TestECStoreEnv::builder() + .prefix("server_context_slot") + .disk_count(1) + .init_bucket_metadata(false) + .build() + .await; + let context = Arc::new(AppContext::new(env.ecstore, Arc::new(UnreadyIam), Arc::new(TestKms))); + publish_test_app_context(context); + current_app_context().expect("test context must be globally published") + } + + fn distinct_context(context: &AppContext) -> Arc { + Arc::new(AppContext::new(context.object_store(), Arc::new(UnreadyIam), Arc::new(TestKms))) + } + + fn request(extensions: http::Extensions) -> S3Request { + S3Request { + input: Body::empty(), + method: Method::GET, + uri: "/context-probe".parse().expect("test URI"), + headers: http::HeaderMap::new(), + extensions, + credentials: None, + region: None, + service: None, + trailing_headers: None, + } + } + + #[tokio::test] + #[serial_test::serial] + async fn request_with_empty_slot_never_uses_ambient_context() { + let ambient = ambient_context().await; + let slot = ServerContextSlot::new(); + let mut extensions = http::Extensions::new(); + extensions.insert(slot); + + assert!(app_context_from_extensions(&extensions).is_none()); + assert!(object_store_from_extensions(&extensions).is_none()); + assert!(app_context_from_req(&request(extensions)).is_none()); + assert!(Arc::ptr_eq( + ¤t_app_context().expect("ambient context must remain available"), + &ambient + )); + } + + #[tokio::test] + #[serial_test::serial] + async fn request_with_installed_slot_resolves_only_its_context() { + let ambient = ambient_context().await; + let installed = distinct_context(&ambient); + let slot = ServerContextSlot::new(); + assert!(slot.install(installed.clone())); + let mut extensions = http::Extensions::new(); + extensions.insert(slot); + + let resolved = app_context_from_extensions(&extensions).expect("installed slot must resolve"); + assert!(Arc::ptr_eq(&resolved, &installed)); + assert!(!Arc::ptr_eq(&resolved, &ambient)); + } + + #[tokio::test] + #[serial_test::serial] + async fn request_without_slot_keeps_legacy_ambient_resolution() { + let ambient = ambient_context().await; + let extensions = http::Extensions::new(); + + let resolved = app_context_from_extensions(&extensions).expect("legacy request must use ambient context"); + assert!(Arc::ptr_eq(&resolved, &ambient)); + assert!(object_store_from_extensions(&extensions).is_some()); + } + + struct ContextDependentAdminRoute; + + #[async_trait::async_trait] + impl Operation for ContextDependentAdminRoute { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> s3s::S3Result> { + app_context_from_req(&req).ok_or_else(|| s3_error!(ServiceUnavailable, "server context is not ready"))?; + Ok(S3Response::new((StatusCode::NO_CONTENT, Body::empty()))) + } + } + + #[tokio::test] + #[serial_test::serial] + async fn context_dependent_admin_route_fails_closed_until_slot_installation() { + let ambient = ambient_context().await; + let slot = ServerContextSlot::new(); + let mut router = S3Router::new(false); + router.set_server_ctx(slot.clone()); + router + .insert(Method::GET, "/context-probe", ContextDependentAdminRoute) + .expect("register test route"); + + let err = router + .call(request(http::Extensions::new())) + .await + .expect_err("empty slot must fail closed"); + assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable); + + assert!(slot.install(distinct_context(&ambient))); + let response = router + .call(request(http::Extensions::new())) + .await + .expect("installed slot must serve request"); + assert_eq!(response.status, Some(StatusCode::NO_CONTENT)); + } +} diff --git a/rustfs/src/app/context/server_slot.rs b/rustfs/src/app/context/server_slot.rs index 1d204be34..56ccedf57 100644 --- a/rustfs/src/app/context/server_slot.rs +++ b/rustfs/src/app/context/server_slot.rs @@ -20,11 +20,10 @@ //! request-path owners (the `FS` service), and installs the `AppContext` into //! it once IAM bootstrap completes. //! -//! Until every app subsystem is per-server (backlog#1052 S3), resolution falls -//! back to the process-global `AppContext` singleton when the slot has not -//! been installed — byte-for-byte the ambient resolution the request path used -//! before this seam existed. The fallback is removed when multi-instance flips -//! on (backlog#1052 S5). +//! Some legacy, non-request call sites still use the ambient compatibility +//! resolver. Request dispatch must instead use the strict installed-context +//! accessors: once a request carries a server slot, an empty slot means that +//! server is not ready rather than that another server's global context applies. use super::global::{AppContext, get_global_app_context}; use crate::app::storage_api::context::ECStore; @@ -62,10 +61,26 @@ impl ServerContextSlot { self.app_context.set(context).is_ok() } + /// This server's installed application context, if startup has completed. + /// + /// Request-scoped resolution must use this accessor so an empty slot never + /// resolves through another server's ambient context. + pub fn installed_app_context(&self) -> Option> { + self.app_context.get().cloned() + } + + /// This server's installed object store, if startup has completed. + pub fn installed_object_store(&self) -> Option> { + self.installed_app_context().map(|context| context.object_store()) + } + /// This server's application context: the installed one, or the - /// process-global singleton as the single-instance legacy default. + /// process-global singleton as the legacy compatibility default. + /// + /// This accessor is for callers without request-slot semantics. Request + /// dispatch must use [`Self::installed_app_context`] instead. pub fn app_context(&self) -> Option> { - self.app_context.get().cloned().or_else(get_global_app_context) + self.installed_app_context().or_else(get_global_app_context) } /// This server's object store, resolved through [`Self::app_context`]. @@ -88,7 +103,7 @@ mod tests { // nothing: the same "not ready yet" answer the ambient path gives before // IAM bootstrap completes. #[test] - fn empty_slot_resolves_like_the_ambient_path() { + fn empty_slot_retains_ambient_compatibility() { let slot = ServerContextSlot::new(); assert_eq!(slot.app_context().is_some(), get_global_app_context().is_some()); assert_eq!(slot.object_store().is_some(), get_global_app_context().is_some()); diff --git a/rustfs/src/embedded.rs b/rustfs/src/embedded.rs index 5fb67ee08..430fad85a 100644 --- a/rustfs/src/embedded.rs +++ b/rustfs/src/embedded.rs @@ -60,6 +60,94 @@ use std::net::SocketAddr; use std::path::PathBuf; use tokio_util::sync::CancellationToken; +#[cfg(feature = "e2e-test-hooks")] +use std::sync::{Mutex, OnceLock}; +#[cfg(feature = "e2e-test-hooks")] +use tokio::sync::oneshot; + +#[cfg(feature = "e2e-test-hooks")] +struct EmbeddedStartupHook { + port: u16, + http_bound: oneshot::Sender<()>, + release: oneshot::Receiver<()>, +} + +#[cfg(feature = "e2e-test-hooks")] +static EMBEDDED_STARTUP_HOOK: OnceLock>> = OnceLock::new(); + +/// Test-only gate for the point after an embedded HTTP listener binds and +/// before its application context is installed. +#[cfg(feature = "e2e-test-hooks")] +#[doc(hidden)] +pub struct EmbeddedStartupBarrier { + http_bound: oneshot::Receiver<()>, + release: Option>, +} + +#[cfg(feature = "e2e-test-hooks")] +impl EmbeddedStartupBarrier { + /// Wait until the next embedded server has bound its HTTP listener. + pub async fn wait_until_http_bound(&mut self) { + (&mut self.http_bound) + .await + .expect("embedded startup hook must signal after binding HTTP"); + } + + /// Allow the paused embedded startup to install its application context. + pub fn release(mut self) { + if let Some(release) = self.release.take() { + let _ = release.send(()); + } + } +} + +#[cfg(feature = "e2e-test-hooks")] +impl Drop for EmbeddedStartupBarrier { + fn drop(&mut self) { + if let Some(release) = self.release.take() { + let _ = release.send(()); + } + } +} + +/// Pause the embedded startup for `port` after its HTTP listener binds. +#[cfg(feature = "e2e-test-hooks")] +#[doc(hidden)] +pub fn pause_embedded_startup_after_http_bind(port: u16) -> EmbeddedStartupBarrier { + let (http_bound_tx, http_bound) = oneshot::channel(); + let (release, release_rx) = oneshot::channel(); + let hook = EmbeddedStartupHook { + port, + http_bound: http_bound_tx, + release: release_rx, + }; + let mut active_hook = EMBEDDED_STARTUP_HOOK + .get_or_init(|| Mutex::new(None)) + .lock() + .expect("embedded startup hook lock must not be poisoned"); + assert!(active_hook.replace(hook).is_none(), "only one embedded startup hook may be active"); + + EmbeddedStartupBarrier { + http_bound, + release: Some(release), + } +} + +#[cfg(feature = "e2e-test-hooks")] +pub(crate) async fn wait_for_embedded_startup_hook(port: u16) { + let hook = { + let mut active_hook = EMBEDDED_STARTUP_HOOK + .get_or_init(|| Mutex::new(None)) + .lock() + .expect("embedded startup hook lock must not be poisoned"); + active_hook.take_if(|hook| hook.port == port) + }; + if let Some(hook) = hook { + let _ = hook.http_bound.send(()); + let _ = hook.release.await; + } +} + /// Error type for embedded server operations. #[derive(Debug)] pub enum ServerError { diff --git a/rustfs/src/startup_embedded.rs b/rustfs/src/startup_embedded.rs index d9208ff85..e23bf9563 100644 --- a/rustfs/src/startup_embedded.rs +++ b/rustfs/src/startup_embedded.rs @@ -156,6 +156,9 @@ pub(crate) async fn run_embedded_startup(args: EmbeddedStartupArgs) -> Result; fn s3_client(endpoint: &str, access_key: &str, secret_key: &str) -> Client { let creds = Credentials::new(access_key, secret_key, None, None, "test"); @@ -33,6 +48,62 @@ fn s3_client(endpoint: &str, access_key: &str, secret_key: &str) -> Client { Client::from_conf(config) } +#[cfg(feature = "e2e-test-hooks")] +fn hex(bytes: impl AsRef<[u8]>) -> String { + bytes.as_ref().iter().map(|byte| format!("{byte:02x}")).collect() +} + +#[cfg(feature = "e2e-test-hooks")] +fn sha256_hex(bytes: &[u8]) -> String { + hex(Sha256::digest(bytes)) +} + +#[cfg(feature = "e2e-test-hooks")] +fn hmac(key: &[u8], value: &str) -> Vec { + let mut mac = HmacSha256::new_from_slice(key).expect("HMAC accepts arbitrary key lengths"); + mac.update(value.as_bytes()); + mac.finalize().into_bytes().to_vec() +} + +#[cfg(feature = "e2e-test-hooks")] +fn signed_admin_request( + client: &reqwest::Client, + endpoint: &str, + request_path: &str, + access_key: &str, + secret_key: &str, +) -> 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 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 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); + let region_key = hmac(&date_key, "us-east-1"); + let service_key = hmac(®ion_key, "s3"); + let signing_key = hmac(&service_key, "aws4_request"); + let authorization = format!( + "AWS4-HMAC-SHA256 Credential={access_key}/{scope}, SignedHeaders={signed_headers}, Signature={}", + hex(hmac(&signing_key, &string_to_sign)) + ); + + client + .get(format!("{endpoint}{request_path}")) + .header("host", host) + .header("x-amz-content-sha256", payload_hash) + .header("x-amz-date", amz_date) + .header("authorization", authorization) +} + // backlog#1052 acceptance: a second embedded server in the same process no // longer aborts on write-once startup state — before this change, // `RustFSServer::build()` returned AlreadyStarted (guard) or panicked on @@ -246,3 +317,112 @@ async fn two_embedded_servers_isolate_auth_and_data_planes() { server_a.shutdown().await; server_b.shutdown().await; } + +#[cfg(feature = "e2e-test-hooks")] +#[tokio::test] +async fn second_embedded_server_fails_closed_until_its_context_slot_is_installed() { + let port_a = match find_available_port() { + Ok(port) => port, + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return, + Err(err) => panic!("find free port for server A: {err}"), + }; + let server_a = RustFSServerBuilder::new() + .address(format!("127.0.0.1:{port_a}")) + .access_key("startup-window-access-a") + .secret_key("startup-window-secret-a") + .build() + .await + .expect("start embedded server A"); + let client_a = s3_client(&server_a.endpoint(), server_a.access_key(), server_a.secret_key()); + client_a + .create_bucket() + .bucket("startup-window") + .send() + .await + .expect("server A creates the shared-name bucket"); + client_a + .put_object() + .bucket("startup-window") + .key("marker.txt") + .body(ByteStream::from_static(b"from A")) + .send() + .await + .expect("server A writes its marker"); + + let port_b = match find_available_port() { + Ok(port) => port, + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => { + server_a.shutdown().await; + return; + } + Err(err) => { + server_a.shutdown().await; + panic!("find free port for server B: {err}"); + } + }; + let endpoint_b = format!("http://127.0.0.1:{port_b}"); + let b_access_key = "startup-window-access-b"; + let b_secret_key = "startup-window-secret-b"; + let mut barrier = pause_embedded_startup_after_http_bind(port_b); + let startup_b = tokio::spawn(async move { + RustFSServerBuilder::new() + .address(format!("127.0.0.1:{port_b}")) + .access_key(b_access_key) + .secret_key(b_secret_key) + .build() + .await + }); + tokio::time::timeout(Duration::from_secs(10), barrier.wait_until_http_bound()) + .await + .expect("server B must bind HTTP before installing its context slot"); + + let http = reqwest::Client::builder() + .no_proxy() + .timeout(Duration::from_secs(5)) + .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_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}"); + assert!( + before_install_body.contains("server context is not ready"), + "paused request must not resolve server A: {before_install_body}" + ); + + barrier.release(); + let server_b = tokio::time::timeout(Duration::from_secs(20), startup_b) + .await + .expect("server B startup must complete after releasing the barrier") + .expect("server B startup task must not panic") + .expect("start embedded server B"); + let client_b = s3_client(&server_b.endpoint(), server_b.access_key(), server_b.secret_key()); + client_b + .create_bucket() + .bucket("startup-window") + .send() + .await + .expect("server B creates its isolated shared-name bucket"); + client_b + .put_object() + .bucket("startup-window") + .key("marker.txt") + .body(ByteStream::from_static(b"from B")) + .send() + .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"); + assert_eq!(after_install.status(), StatusCode::OK); + assert_eq!(after_install.bytes().await.expect("read server B marker"), b"from B".as_slice()); + + server_b.shutdown().await; + server_a.shutdown().await; +}