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
This commit is contained in:
Chris
2026-10-06 19:58:58 -07:00
committed by GitHub
parent 073b4d838e
commit cd0b895ad7
6 changed files with 155 additions and 52 deletions
+26 -16
View File
@@ -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<ServerContextSlot>) -> 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<Option<ShutdownHandle>, Box<dyn std::error::Error + Send + Sync>> {
pub async fn init_ftp_system(
server_ctx: Arc<ServerContextSlot>,
) -> Result<Option<ShutdownHandle>, Box<dyn std::error::Error + Send + Sync>> {
{
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<Option<ShutdownHandle>, Box<dyn std::er
config.validate().await?;
// Create FTP server with protocol storage client
let fs = crate::storage_api::startup::init::ecfs::FS::new();
let storage_client = ProtocolStorageClient::new(fs);
let storage_client = protocol_storage_client_for_server(server_ctx);
let server: FtpsServer<ProtocolStorageClient> = 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<Option<ShutdownHandle>, Box<dyn std::er
/// the server in a background task.
#[cfg(feature = "ftps")]
#[instrument(skip_all)]
pub async fn init_ftps_system() -> Result<Option<ShutdownHandle>, Box<dyn std::error::Error + Send + Sync>> {
pub async fn init_ftps_system(
server_ctx: Arc<ServerContextSlot>,
) -> Result<Option<ShutdownHandle>, Box<dyn std::error::Error + Send + Sync>> {
{
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<Option<ShutdownHandle>, Box<dyn std::e
config.validate().await?;
// Create FTPS server with protocol storage client
let fs = crate::storage_api::startup::init::ecfs::FS::new();
let storage_client = ProtocolStorageClient::new(fs);
let storage_client = protocol_storage_client_for_server(server_ctx);
let server: FtpsServer<ProtocolStorageClient> = 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<Option<ShutdownHandle>, Box<dyn std::e
/// the server in a background task.
#[cfg(feature = "webdav")]
#[instrument(skip_all)]
pub async fn init_webdav_system() -> Result<Option<ShutdownHandle>, Box<dyn std::error::Error + Send + Sync>> {
pub async fn init_webdav_system(
server_ctx: Arc<ServerContextSlot>,
) -> Result<Option<ShutdownHandle>, Box<dyn std::error::Error + Send + Sync>> {
{
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<Option<ShutdownHandle>, Box<dyn std:
};
// Create WebDAV server with protocol storage client
let fs = crate::storage_api::startup::init::ecfs::FS::new();
let storage_client = ProtocolStorageClient::new(fs);
let storage_client = protocol_storage_client_for_server(server_ctx);
let server: WebDavServer<crate::protocols::ProtocolStorageClient> = 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<Option<ShutdownHandle>, Box<dyn std:
/// and spawns the listener task.
#[cfg(feature = "sftp")]
#[instrument(skip_all)]
pub async fn init_sftp_system() -> Result<Option<ShutdownHandle>, Box<dyn std::error::Error + Send + Sync>> {
pub async fn init_sftp_system(
server_ctx: Arc<ServerContextSlot>,
) -> Result<Option<ShutdownHandle>, Box<dyn std::error::Error + Send + Sync>> {
{
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<Option<ShutdownHandle>, Box<dyn std::e
// file has insecure permissions.
let host_keys = SftpConfig::load_host_keys(&config.host_key_dir).await?;
let fs = crate::storage_api::startup::init::ecfs::FS::new();
let storage_client = ProtocolStorageClient::new(fs);
let storage_client = protocol_storage_client_for_server(server_ctx);
let server = SftpServer::new(config.clone(), storage_client, host_keys)?;
+4 -2
View File
@@ -254,9 +254,11 @@ impl ProtocolStorageClient {
object: Option<&str>,
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 {
+101 -13
View File
@@ -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<rustfs_policy::policy::Poli
serde_json::from_str(&value.to_string())
}
async fn protocol_test_context(root: Credentials) -> (tempfile::TempDir, Arc<AppContext>, Arc<IamSys<ObjectStore>>) {
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())
@@ -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<OptionalRuntimeServices> {
let protocols = init_protocol_shutdown_senders().await?;
pub(crate) async fn init_optional_runtime_services(server_ctx: Arc<ServerContextSlot>) -> Result<OptionalRuntimeServices> {
let protocols = init_protocol_shutdown_senders(server_ctx).await?;
Ok(OptionalRuntimeServices::new(protocols))
}
+19 -17
View File
@@ -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<ShutdownHandle>,
}
pub(crate) async fn init_protocol_shutdown_senders() -> Result<ProtocolShutdownSenders> {
pub(crate) async fn init_protocol_shutdown_senders(server_ctx: Arc<ServerContextSlot>) -> Result<ProtocolShutdownSenders> {
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<Option<ShutdownHandle>> {
init_protocol("ftp", init_ftp_system).await
async fn init_ftp_protocol(server_ctx: Arc<ServerContextSlot>) -> Result<Option<ShutdownHandle>> {
init_protocol("ftp", || init_ftp_system(server_ctx)).await
}
#[cfg(not(feature = "ftps"))]
async fn init_ftp_protocol() -> Result<Option<ShutdownHandle>> {
async fn init_ftp_protocol(_server_ctx: Arc<ServerContextSlot>) -> Result<Option<ShutdownHandle>> {
Ok(None)
}
#[cfg(feature = "ftps")]
async fn init_ftps_protocol() -> Result<Option<ShutdownHandle>> {
init_protocol("ftps", init_ftps_system).await
async fn init_ftps_protocol(server_ctx: Arc<ServerContextSlot>) -> Result<Option<ShutdownHandle>> {
init_protocol("ftps", || init_ftps_system(server_ctx)).await
}
#[cfg(not(feature = "ftps"))]
async fn init_ftps_protocol() -> Result<Option<ShutdownHandle>> {
async fn init_ftps_protocol(_server_ctx: Arc<ServerContextSlot>) -> Result<Option<ShutdownHandle>> {
Ok(None)
}
#[cfg(feature = "webdav")]
async fn init_webdav_protocol() -> Result<Option<ShutdownHandle>> {
init_protocol("webdav", init_webdav_system).await
async fn init_webdav_protocol(server_ctx: Arc<ServerContextSlot>) -> Result<Option<ShutdownHandle>> {
init_protocol("webdav", || init_webdav_system(server_ctx)).await
}
#[cfg(not(feature = "webdav"))]
async fn init_webdav_protocol() -> Result<Option<ShutdownHandle>> {
async fn init_webdav_protocol(_server_ctx: Arc<ServerContextSlot>) -> Result<Option<ShutdownHandle>> {
Ok(None)
}
#[cfg(feature = "sftp")]
async fn init_sftp_protocol() -> Result<Option<ShutdownHandle>> {
init_protocol("sftp", init_sftp_system).await
async fn init_sftp_protocol(server_ctx: Arc<ServerContextSlot>) -> Result<Option<ShutdownHandle>> {
init_protocol("sftp", || init_sftp_system(server_ctx)).await
}
#[cfg(not(feature = "sftp"))]
async fn init_sftp_protocol() -> Result<Option<ShutdownHandle>> {
async fn init_sftp_protocol(_server_ctx: Arc<ServerContextSlot>) -> Result<Option<ShutdownHandle>> {
Ok(None)
}
+1 -1
View File
@@ -84,7 +84,7 @@ pub(crate) async fn init_startup_runtime_services(
) -> Result<StartupServiceRuntime> {
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()