From cd0b895ad7feba274f3240ccc5eb39674b439b2a Mon Sep 17 00:00:00 2001 From: Chris Date: Tue, 6 Oct 2026 19:58:58 -0700 Subject: [PATCH] fix(protocols): share the server context with protocol storage (#8367) * fix(protocols): bind storage authorization to the startup server context * test(protocols): publish the context through the runtime boundary --- rustfs/src/init.rs | 42 ++++--- rustfs/src/protocols/client.rs | 6 +- rustfs/src/protocols/policy_test.rs | 114 ++++++++++++++++-- .../src/startup_optional_runtime_sidecars.rs | 7 +- rustfs/src/startup_protocols.rs | 36 +++--- rustfs/src/startup_services.rs | 2 +- 6 files changed, 155 insertions(+), 52 deletions(-) diff --git a/rustfs/src/init.rs b/rustfs/src/init.rs index 526301bc3..6125a74ed 100644 --- a/rustfs/src/init.rs +++ b/rustfs/src/init.rs @@ -12,6 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. +#[cfg(any(feature = "ftps", feature = "webdav", feature = "sftp"))] +use crate::protocols::ProtocolStorageClient; use crate::runtime_sources::current_region; use crate::server::ShutdownHandle; use crate::server::runtime_sources::current_notify_interface; @@ -21,6 +23,8 @@ use crate::storage_api::startup::init::{ get_bucket_notification_config, process_lambda_configurations, process_queue_configurations, process_topic_configurations, }; use crate::storage_api::startup::services::ECStore; +#[cfg(any(feature = "ftps", feature = "webdav", feature = "sftp"))] +use crate::storage_api::startup::services::ServerContextSlot; use crate::storage_api::startup::sse::log_sse_kms_key_policy_mode; use crate::{admin, config, startup_runtime_sources, version}; use rustfs_config::{ @@ -921,14 +925,21 @@ pub async fn init_auto_tuner(ctx: tokio_util::sync::CancellationToken) { } } +#[cfg(any(feature = "ftps", feature = "webdav", feature = "sftp"))] +pub(crate) fn protocol_storage_client_for_server(server_ctx: Arc) -> ProtocolStorageClient { + let fs = crate::storage_api::startup::init::ecfs::FS::with_server_ctx(server_ctx); + ProtocolStorageClient::new(fs) +} + /// Initialize the FTP system /// /// This function initializes the FTP server (non-encrypted) if enabled in the configuration. #[cfg(feature = "ftps")] #[instrument(skip_all)] -pub async fn init_ftp_system() -> Result, Box> { +pub async fn init_ftp_system( + server_ctx: Arc, +) -> Result, Box> { { - use crate::protocols::ProtocolStorageClient; use rustfs_config::{DEFAULT_FTP_ADDRESS, ENV_FTP_ADDRESS, ENV_FTP_ENABLE, ENV_FTP_EXTERNAL_IP, ENV_FTP_PASSIVE_PORTS}; use rustfs_protocols::constants::defaults::DEFAULT_FTPS_PASSIVE_PORTS; use rustfs_protocols::{FtpsConfig, FtpsServer}; @@ -971,8 +982,7 @@ pub async fn init_ftp_system() -> Result, Box = FtpsServer::new(config, storage_client).await?; let bind_addr = server.config().bind_addr; let passive_ports = server.config().passive_ports.clone(); @@ -1040,9 +1050,10 @@ pub async fn init_ftp_system() -> Result, Box Result, Box> { +pub async fn init_ftps_system( + server_ctx: Arc, +) -> Result, Box> { { - use crate::protocols::ProtocolStorageClient; use rustfs_config::{ DEFAULT_FTPS_ADDRESS, ENV_FTPS_ADDRESS, ENV_FTPS_CA_FILE, ENV_FTPS_CERTS_DIR, ENV_FTPS_ENABLE, ENV_FTPS_EXTERNAL_IP, ENV_FTPS_PASSIVE_PORTS, ENV_FTPS_TLS_ENABLED, @@ -1091,8 +1102,7 @@ pub async fn init_ftps_system() -> Result, Box = FtpsServer::new(config, storage_client).await?; let bind_addr = server.config().bind_addr; let passive_ports = server.config().passive_ports.clone(); @@ -1161,9 +1171,10 @@ pub async fn init_ftps_system() -> Result, Box Result, Box> { +pub async fn init_webdav_system( + server_ctx: Arc, +) -> Result, Box> { { - use crate::protocols::ProtocolStorageClient; use rustfs_config::{ DEFAULT_WEBDAV_ADDRESS, ENV_WEBDAV_ADDRESS, ENV_WEBDAV_CA_FILE, ENV_WEBDAV_CERTS_DIR, ENV_WEBDAV_ENABLE, ENV_WEBDAV_MAX_BODY_SIZE, ENV_WEBDAV_MAX_CONNECTIONS, ENV_WEBDAV_REQUEST_TIMEOUT, ENV_WEBDAV_TLS_ENABLED, @@ -1211,8 +1222,7 @@ pub async fn init_webdav_system() -> Result, Box = WebDavServer::new(config, storage_client).await?; let bind_addr = server.config().bind_addr; let tls_enabled = server.config().tls_enabled; @@ -1283,9 +1293,10 @@ pub async fn init_webdav_system() -> Result, Box Result, Box> { +pub async fn init_sftp_system( + server_ctx: Arc, +) -> Result, Box> { { - use crate::protocols::ProtocolStorageClient; use rustfs_config::{ DEFAULT_SFTP_ADDRESS, DEFAULT_SFTP_BANNER, DEFAULT_SFTP_IDLE_TIMEOUT, DEFAULT_SFTP_PART_SIZE, DEFAULT_SFTP_READ_ONLY, ENV_SFTP_ADDRESS, ENV_SFTP_BACKEND_OP_TIMEOUT_SECS, ENV_SFTP_BANNER, ENV_SFTP_ENABLE, ENV_SFTP_HANDLES_PER_SESSION, @@ -1347,8 +1358,7 @@ pub async fn init_sftp_system() -> Result, Box, listing: Option<&ListObjectsV2Input>, ) -> S3Result<()> { - let context = self.fs.server_ctx().app_context(); + let context = self.fs.server_ctx().installed_app_context().ok_or_else(|| { + s3s::S3Error::with_message(s3s::S3ErrorCode::InternalError, "protocol server application context is not initialized") + })?; let (credentials, is_owner) = - crate::auth::check_key_valid_with_context("", session.access_key(), context.as_deref()).await?; + crate::auth::check_key_valid_with_context("", session.access_key(), Some(context.as_ref())).await?; let uri = if let Some(input) = listing { build_list_objects_uri(input)? } else if let Some(key) = object { diff --git a/rustfs/src/protocols/policy_test.rs b/rustfs/src/protocols/policy_test.rs index 5ebeb1643..7d4bdcb79 100644 --- a/rustfs/src/protocols/policy_test.rs +++ b/rustfs/src/protocols/policy_test.rs @@ -14,9 +14,8 @@ //! Regression coverage for policy enforcement at the protocol/backend boundary. -use super::client::ProtocolStorageClient; use crate::runtime_sources::{AppContext, IamInterface, KmsInterface, ServerContextSlot}; -use crate::storage_api::protocols::client::{BucketOperations, FS, ListObjectsV2Input, MakeBucketOptions}; +use crate::storage_api::protocols::client::{BucketOperations, ListObjectsV2Input, MakeBucketOptions}; use rustfs_credentials::Credentials; use rustfs_iam::{ store::{ @@ -54,29 +53,118 @@ fn decode_policy(value: serde_json::Value) -> Result (tempfile::TempDir, Arc, Arc>) { + let (temp, _paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await; + ObjectStore::new(store.clone()) + .save_iam_config(json!({"version": 1}), format!("{}/format.json", *IAM_CONFIG_PREFIX)) + .await + .expect("seed IAM format"); + let iam = rustfs_iam::init_iam_sys_for_context(store.clone()) + .await + .expect("initialize IAM"); + let context = Arc::new(AppContext::new(store, Arc::new(TestIam(iam.clone())), Arc::new(TestKms))); + assert!(context.publish_action_credentials(root)); + (temp, context, iam) +} + +#[test] +#[serial_test::serial] +fn protocol_startup_backend_tracks_server_context_installation() { + crate::app::gating_test_env::run_large_stack_test("protocol-startup-context", || async { + let root = Credentials { + access_key: "protocol-bound-root".into(), + secret_key: "protocol-bound-root-secret".into(), + status: "on".into(), + ..Default::default() + }; + let foreign_root = Credentials { + access_key: "protocol-foreign-root".into(), + secret_key: "protocol-foreign-root-secret".into(), + status: "on".into(), + ..Default::default() + }; + let (_foreign_temp, foreign_context, _foreign_iam) = protocol_test_context(foreign_root.clone()).await; + let ambient_context = crate::runtime_sources::publish_test_app_context(foreign_context.clone()); + assert!( + ambient_context.iam().is_ready(), + "ambient IAM must be ready for the uninstalled-slot probe" + ); + let ambient_root = ambient_context.action_credentials().get().expect("ambient root credentials"); + assert_ne!(ambient_root.access_key, root.access_key, "server roots must differ"); + let (_temp, context, _iam) = protocol_test_context(root.clone()).await; + let slot = ServerContextSlot::new(); + let backend = crate::init::protocol_storage_client_for_server(slot.clone()); + let bucket = "protocol-startup-context"; + for store in [context.object_store(), foreign_context.object_store()] { + store + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("create bucket"); + } + let session = |credentials: Credentials, protocol| { + SessionContext::new( + ProtocolPrincipal::new(Arc::new(rustfs_policy::auth::UserIdentity { + credentials, + ..Default::default() + })), + protocol, + "127.0.0.1".parse().expect("loopback"), + ) + }; + for protocol in [Protocol::WebDav, Protocol::Ftps, Protocol::Sftp] { + for credentials in [ambient_root.clone(), root.clone()] { + let result = backend + .authorize_operation(&session(credentials, protocol), &S3Action::GetObject, bucket, Some("key")) + .await; + assert!( + matches!(result, Err(AuthorizationError::IamUnavailable)), + "uninstalled server slot must reject before ambient credential lookup: {result:?}" + ); + } + } + assert!(slot.install(context)); + let foreign_slot = ServerContextSlot::new(); + assert!(foreign_slot.install(foreign_context)); + let foreign_backend = crate::init::protocol_storage_client_for_server(foreign_slot); + for protocol in [Protocol::WebDav, Protocol::Ftps, Protocol::Sftp] { + foreign_backend + .authorize_operation(&session(foreign_root.clone(), protocol), &S3Action::GetObject, bucket, Some("key")) + .await + .expect("foreign root uses its own installed server slot"); + backend + .authorize_operation(&session(root.clone(), protocol), &S3Action::GetObject, bucket, Some("key")) + .await + .expect("backend constructed before installation uses the installed server slot"); + for (target, credentials) in [(&backend, foreign_root.clone()), (&foreign_backend, root.clone())] { + assert!( + matches!( + target + .authorize_operation(&session(credentials, protocol), &S3Action::GetObject, bucket, Some("key")) + .await, + Err(AuthorizationError::AccessDenied) + ), + "another server's root is not this server's root" + ); + } + } + }); +} + #[test] #[serial_test::serial] fn protocol_policy_denials_and_request_conditions() { crate::app::gating_test_env::run_large_stack_test("protocol-policy-authorization", || async { - let (_temp, _paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await; - ObjectStore::new(store.clone()) - .save_iam_config(json!({"version": 1}), format!("{}/format.json", *IAM_CONFIG_PREFIX)) - .await - .expect("seed IAM format"); - let iam = rustfs_iam::init_iam_sys_for_context(store.clone()) - .await - .expect("initialize IAM"); let root = Credentials { access_key: "protocol-root".into(), secret_key: "protocol-root-secret".into(), status: "on".into(), ..Default::default() }; - let context = Arc::new(AppContext::new(store.clone(), Arc::new(TestIam(iam.clone())), Arc::new(TestKms))); - assert!(context.publish_action_credentials(root.clone())); + let (_temp, context, iam) = protocol_test_context(root.clone()).await; + let store = context.object_store(); let slot = ServerContextSlot::new(); assert!(slot.install(context)); - let backend = ProtocolStorageClient::new(FS::with_server_ctx(slot)); + let backend = crate::init::protocol_storage_client_for_server(slot); let bucket = "protocol-policy-test"; store .make_bucket(bucket, &MakeBucketOptions::default()) diff --git a/rustfs/src/startup_optional_runtime_sidecars.rs b/rustfs/src/startup_optional_runtime_sidecars.rs index aaeadfc89..6fbfab1a3 100644 --- a/rustfs/src/startup_optional_runtime_sidecars.rs +++ b/rustfs/src/startup_optional_runtime_sidecars.rs @@ -15,9 +15,10 @@ use crate::{ server::ShutdownHandle, startup_protocols::{ProtocolShutdownSenders, init_protocol_shutdown_senders}, + storage_api::startup::services::ServerContextSlot, }; use futures_util::future::join_all; -use std::io::Result; +use std::{io::Result, sync::Arc}; use tracing::info; const LOG_COMPONENT_MAIN: &str = "main"; @@ -34,8 +35,8 @@ impl OptionalRuntimeServices { } } -pub(crate) async fn init_optional_runtime_services() -> Result { - let protocols = init_protocol_shutdown_senders().await?; +pub(crate) async fn init_optional_runtime_services(server_ctx: Arc) -> Result { + let protocols = init_protocol_shutdown_senders(server_ctx).await?; Ok(OptionalRuntimeServices::new(protocols)) } diff --git a/rustfs/src/startup_protocols.rs b/rustfs/src/startup_protocols.rs index dadd6b122..d223764d2 100644 --- a/rustfs/src/startup_protocols.rs +++ b/rustfs/src/startup_protocols.rs @@ -19,8 +19,10 @@ use crate::init::init_webdav_system; #[cfg(feature = "ftps")] use crate::init::{init_ftp_system, init_ftps_system}; use crate::server::ShutdownHandle; +use crate::storage_api::startup::services::ServerContextSlot; use std::future::Future; use std::io::{Error, Result}; +use std::sync::Arc; use tracing::{debug, error, info}; const LOG_COMPONENT_MAIN: &str = "main"; @@ -38,52 +40,52 @@ pub(crate) struct ProtocolShutdownSenders { pub(crate) sftp: Option, } -pub(crate) async fn init_protocol_shutdown_senders() -> Result { +pub(crate) async fn init_protocol_shutdown_senders(server_ctx: Arc) -> Result { Ok(ProtocolShutdownSenders { - ftp: init_ftp_protocol().await?, - ftps: init_ftps_protocol().await?, - webdav: init_webdav_protocol().await?, - sftp: init_sftp_protocol().await?, + ftp: init_ftp_protocol(server_ctx.clone()).await?, + ftps: init_ftps_protocol(server_ctx.clone()).await?, + webdav: init_webdav_protocol(server_ctx.clone()).await?, + sftp: init_sftp_protocol(server_ctx).await?, }) } #[cfg(feature = "ftps")] -async fn init_ftp_protocol() -> Result> { - init_protocol("ftp", init_ftp_system).await +async fn init_ftp_protocol(server_ctx: Arc) -> Result> { + init_protocol("ftp", || init_ftp_system(server_ctx)).await } #[cfg(not(feature = "ftps"))] -async fn init_ftp_protocol() -> Result> { +async fn init_ftp_protocol(_server_ctx: Arc) -> Result> { Ok(None) } #[cfg(feature = "ftps")] -async fn init_ftps_protocol() -> Result> { - init_protocol("ftps", init_ftps_system).await +async fn init_ftps_protocol(server_ctx: Arc) -> Result> { + init_protocol("ftps", || init_ftps_system(server_ctx)).await } #[cfg(not(feature = "ftps"))] -async fn init_ftps_protocol() -> Result> { +async fn init_ftps_protocol(_server_ctx: Arc) -> Result> { Ok(None) } #[cfg(feature = "webdav")] -async fn init_webdav_protocol() -> Result> { - init_protocol("webdav", init_webdav_system).await +async fn init_webdav_protocol(server_ctx: Arc) -> Result> { + init_protocol("webdav", || init_webdav_system(server_ctx)).await } #[cfg(not(feature = "webdav"))] -async fn init_webdav_protocol() -> Result> { +async fn init_webdav_protocol(_server_ctx: Arc) -> Result> { Ok(None) } #[cfg(feature = "sftp")] -async fn init_sftp_protocol() -> Result> { - init_protocol("sftp", init_sftp_system).await +async fn init_sftp_protocol(server_ctx: Arc) -> Result> { + init_protocol("sftp", || init_sftp_system(server_ctx)).await } #[cfg(not(feature = "sftp"))] -async fn init_sftp_protocol() -> Result> { +async fn init_sftp_protocol(_server_ctx: Arc) -> Result> { Ok(None) } diff --git a/rustfs/src/startup_services.rs b/rustfs/src/startup_services.rs index 1dcee12fb..5dd0fe9a7 100644 --- a/rustfs/src/startup_services.rs +++ b/rustfs/src/startup_services.rs @@ -84,7 +84,7 @@ pub(crate) async fn init_startup_runtime_services( ) -> Result { init_kms_system(config, store.clone()).await?; - let optional_runtimes = init_optional_runtime_services().await?; + let optional_runtimes = init_optional_runtime_services(server_ctx.clone()).await?; let heartbeat_config = HeartbeatConfig::from_env().map_err(std::io::Error::other)?; let local_trace_capture = heartbeat_config .as_ref()