refactor: route notify dispatch through app context (#3789)

* refactor: route notify dispatch through app context * refactor: route admin IAM globals through app context (#3791) * refactor: centralize IAM root credential access (#3792)
2026-08-06 05:17:42 +00:00 · 2026-06-23 20:05:28 +08:00
parent da565c11bc
commit 70a2441407
19 changed files with 272 additions and 72 deletions
@@ -13,6 +13,7 @@
 // limitations under the License.

 use crate::admin::handlers::health::{HealthProbe, build_health_response_parts, collect_dependency_readiness};
+use crate::app::context::resolve_oidc_handle;
 use crate::license::has_valid_license;
 use crate::server::has_path_prefix;
 use crate::server::{
@@ -134,7 +135,7 @@ impl Config {
        let http_prefix = rustfs_config::RUSTFS_HTTP_PREFIX;

        // Collect OIDC provider info if available
-        let oidc = rustfs_iam::get_oidc()
+        let oidc = resolve_oidc_handle()
            .map(|sys| {
                sys.list_visible_providers()
                    .into_iter()
@@ -16,7 +16,7 @@ use super::super::{read_admin_config_without_migrate, save_admin_server_config};
 use super::sts::create_oidc_sts_credentials;
 use crate::admin::auth::validate_admin_request;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
-use crate::app::context::{resolve_object_store_handle, resolve_server_config};
+use crate::app::context::{resolve_object_store_handle, resolve_oidc_handle, resolve_server_config};
 use crate::auth::{check_key_valid, get_session_token};
 use crate::server::{ADMIN_PREFIX, MINIO_ADMIN_PREFIX, RemoteAddr};
 use http::StatusCode;
@@ -268,7 +268,7 @@ pub struct ListOidcProvidersHandler {}
 #[async_trait::async_trait]
 impl Operation for ListOidcProvidersHandler {
    async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        let oidc_sys = rustfs_iam::get_oidc().ok_or_else(|| s3_error!(InternalError, "OIDC not initialized"))?;
+        let oidc_sys = resolve_oidc_handle().ok_or_else(|| s3_error!(InternalError, "OIDC not initialized"))?;

        let providers = oidc_sys.list_visible_providers();
        let json_body = serde_json::to_vec(&providers)
@@ -439,7 +439,7 @@ impl Operation for OidcAuthorizeHandler {
            return Err(s3_error!(InvalidRequest, "invalid provider_id"));
        }

-        let oidc_sys = rustfs_iam::get_oidc().ok_or_else(|| s3_error!(InternalError, "OIDC not initialized"))?;
+        let oidc_sys = resolve_oidc_handle().ok_or_else(|| s3_error!(InternalError, "OIDC not initialized"))?;

        // Derive the callback redirect URI from the request
        let redirect_uri = derive_callback_uri(&req, provider_id)?;
@@ -512,7 +512,7 @@ impl Operation for OidcCallbackHandler {
            ));
        }

-        let oidc_sys = rustfs_iam::get_oidc().ok_or_else(|| s3_error!(InternalError, "OIDC not initialized"))?;
+        let oidc_sys = resolve_oidc_handle().ok_or_else(|| s3_error!(InternalError, "OIDC not initialized"))?;

        let redirect_uri = derive_callback_uri(&req, provider_id)?;

@@ -599,7 +599,7 @@ impl Operation for OidcLogoutHandler {
            return redirect_response(&fallback_location);
        };

-        let location = match rustfs_iam::get_oidc() {
+        let location = match resolve_oidc_handle() {
            Some(oidc_sys) => match oidc_sys.build_logout_url(&logout_token, &fallback_location).await {
                Ok(Some(url)) => url,
                Ok(None) => fallback_location.clone(),
@@ -628,7 +628,7 @@ impl Operation for OidcLogoutHandler {
 /// an explicit redirect_uri is recommended to prevent header manipulation.
 fn derive_callback_uri(req: &S3Request<Body>, provider_id: &str) -> S3Result<String> {
    // Use explicitly configured redirect_uri if available
-    if let Some(oidc_sys) = rustfs_iam::get_oidc()
+    if let Some(oidc_sys) = resolve_oidc_handle()
        && let Some(config) = oidc_sys.get_provider_config(provider_id)
    {
        if let Some(ref uri) = config.redirect_uri {
@@ -34,7 +34,7 @@ use crate::admin::utils::{encode_compatible_admin_payload, read_compatible_admin
 use crate::app::context::{
    resolve_deployment_id, resolve_endpoints_handle, resolve_iam_handle, resolve_object_store_handle, resolve_oidc_handle,
    resolve_outbound_tls_generation, resolve_outbound_tls_state, resolve_region, resolve_replication_pool_handle,
-    resolve_replication_stats_handle, resolve_runtime_port, resolve_server_config,
+    resolve_replication_stats_handle, resolve_runtime_port, resolve_server_config, resolve_token_signing_key,
 };
 use crate::auth::{check_key_valid, get_session_token};
 use crate::config::get_config_snapshot;
@@ -3556,7 +3556,7 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> {
            let Some(sts_credential) = item.sts_credential else {
                return Err(s3_error!(InvalidRequest, "stsCredential is required"));
            };
-            let Some(secret) = rustfs_iam::manager::get_token_signing_key() else {
+            let Some(secret) = resolve_token_signing_key() else {
                return Err(s3_error!(InvalidRequest, "token signing key not initialized"));
            };
            let claims = get_claims_from_token_with_secret(&sts_credential.session_token, &secret)
@@ -19,7 +19,7 @@ use crate::{
        handlers::site_replication::site_replication_iam_change_hook,
        router::{AdminOperation, Operation, S3Router},
    },
-    app::context::resolve_action_credentials,
+    app::context::{resolve_action_credentials, resolve_oidc_handle, resolve_token_signing_key},
    auth::{check_key_valid, extract_string_list_claim, get_session_token},
    server::ADMIN_PREFIX,
    server::RemoteAddr,
@@ -29,7 +29,7 @@ use http::header::HeaderValue;
 use hyper::Method;
 use matchit::Params;
 use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
-use rustfs_iam::{manager::get_token_signing_key, oidc::OidcClaims, sys::SESSION_POLICY_NAME};
+use rustfs_iam::{oidc::OidcClaims, sys::SESSION_POLICY_NAME};
 use rustfs_madmin::{SITE_REPL_API_VERSION, SRIAMItem, SRSTSCredential};
 use rustfs_policy::{
    auth::get_new_credentials_with_metadata,
@@ -59,7 +59,7 @@ fn has_identity_authorization_context(policies: &[String], groups: &[String]) ->
 }

 fn configured_roles_claim_key(provider_id: &str) -> Option<String> {
-    rustfs_iam::get_oidc()
+    resolve_oidc_handle()
        .as_ref()
        .and_then(|oidc_sys| oidc_sys.get_provider_config(provider_id))
        .map(|cfg| cfg.roles_claim.trim().to_string())
@@ -239,7 +239,7 @@ async fn handle_assume_role(
        return Err(s3_error!(InvalidArgument, "invalid policy arg"));
    }

-    let Some(secret) = get_token_signing_key() else {
+    let Some(secret) = resolve_token_signing_key() else {
        return Err(s3_error!(InvalidArgument, "global active sk not init"));
    };

@@ -316,7 +316,7 @@ async fn handle_assume_role_with_web_identity(body: AssumeRoleRequest) -> S3Resu
    }

    // Verify the JWT and extract claims
-    let oidc_sys = rustfs_iam::get_oidc().ok_or_else(|| s3_error!(InternalError, "OIDC not initialized"))?;
+    let oidc_sys = resolve_oidc_handle().ok_or_else(|| s3_error!(InternalError, "OIDC not initialized"))?;

    let (claims, provider_id) = oidc_sys
        .verify_web_identity_token(&body.web_identity_token)
@@ -429,7 +429,7 @@ pub async fn create_oidc_sts_credentials(
    }

    // Generate STS temp credentials
-    let secret = get_token_signing_key().ok_or_else(|| s3_error!(InternalError, "token signing key not initialized"))?;
+    let secret = resolve_token_signing_key().ok_or_else(|| s3_error!(InternalError, "token signing key not initialized"))?;

    let mut new_cred = get_new_credentials_with_metadata(&token_claims, &secret)
        .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("credential generation failed: {e}")))?;
@@ -17,7 +17,7 @@ use crate::admin::{
    auth::{AdminResourceScope, validate_admin_request, validate_admin_request_with_bucket_object},
    router::{AdminOperation, Operation, S3Router},
 };
-use crate::app::context::resolve_object_store_handle;
+use crate::app::context::{resolve_object_store_handle, resolve_token_signing_key};
 use crate::auth::{check_key_valid, get_session_token};
 use crate::server::{RemoteAddr, TABLE_CATALOG_COMPAT_PREFIX, TABLE_CATALOG_PREFIX};
 use crate::table_catalog::{DEFAULT_WAREHOUSE_ID, TableCatalogStore};
@@ -26,7 +26,7 @@ use hyper::Method;
 use matchit::Params;
 use metrics::{counter, histogram};
 use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
-use rustfs_iam::{manager::get_token_signing_key, sys::SESSION_POLICY_NAME};
+use rustfs_iam::sys::SESSION_POLICY_NAME;
 use rustfs_policy::{
    auth::get_new_credentials_with_metadata,
    policy::{
@@ -696,7 +696,7 @@ impl TableCredentialIssuer for IamTableCredentialIssuer {
            serde_json::Value::String(request.scope_prefix.clone()),
        );

-        let secret = get_token_signing_key().ok_or_else(|| s3_error!(InternalError, "token signing key not initialized"))?;
+        let secret = resolve_token_signing_key().ok_or_else(|| s3_error!(InternalError, "token signing key not initialized"))?;
        let mut credential = get_new_credentials_with_metadata(&claims, &secret)
            .map_err(|err| s3_error!(InternalError, "failed to generate table credentials: {}", err))?;
        bind_table_credential_parent(&mut credential, principal);
@@ -37,8 +37,7 @@ use super::{BucketBandwidthMonitor, DynReplicationPool, NotificationSys, Replica
 use crate::config::RustFSBufferConfig;
 use rustfs_config::server_config::Config;
 use rustfs_credentials::Credentials;
-use rustfs_iam::oidc::OidcSys;
-use rustfs_iam::{error::Error as IamError, store::object::ObjectStore, sys::IamSys};
+use rustfs_iam::{error::Error as IamError, oidc::OidcSys, store::object::ObjectStore, sys::IamSys};
 use rustfs_io_metrics::{PerformanceMetrics, internode_metrics::InternodeMetrics};
 use rustfs_kms::{KmsServiceManager, ObjectEncryptionService, init_global_kms_service_manager};
 use rustfs_lock::LockClient;
@@ -85,14 +84,19 @@ pub fn resolve_iam_handle() -> Option<Arc<IamSys<ObjectStore>>> {
    resolve_iam_handle_with(get_global_app_context(), rustfs_iam::get_global_iam_sys)
 }

-/// Resolve OIDC handle using AppContext-first precedence.
+/// Resolve a ready IAM system handle using AppContext-first precedence.
+pub fn resolve_ready_iam_handle() -> rustfs_iam::error::Result<Arc<IamSys<ObjectStore>>> {
+    resolve_ready_iam_handle_with(get_global_app_context(), rustfs_iam::get)
+}
+
+/// Resolve OIDC system handle using AppContext-first precedence.
 pub fn resolve_oidc_handle() -> Option<Arc<OidcSys>> {
    resolve_oidc_handle_with(get_global_app_context(), rustfs_iam::get_oidc)
 }

-/// Resolve a ready IAM system handle using AppContext-first precedence.
-pub fn resolve_ready_iam_handle() -> rustfs_iam::error::Result<Arc<IamSys<ObjectStore>>> {
-    resolve_ready_iam_handle_with(get_global_app_context(), rustfs_iam::get)
+/// Resolve token signing key using AppContext-first precedence.
+pub fn resolve_token_signing_key() -> Option<String> {
+    resolve_token_signing_key_with(get_global_app_context(), rustfs_iam::manager::get_token_signing_key)
 }

 /// Resolve bucket metadata handle using AppContext-first precedence.
@@ -111,6 +115,12 @@ pub fn resolve_object_store_handle_for_context(context: Option<&AppContext>) ->
    context.map(|context| context.object_store()).or_else(new_object_layer_fn)
 }

+/// Resolve notify interface using AppContext-first precedence.
+pub fn resolve_notify_interface() -> Arc<dyn NotifyInterface> {
+    let context = get_global_app_context();
+    resolve_notify_interface_for_context(context.as_deref())
+}
+
 /// Resolve notify interface using an explicit AppContext, falling back to the legacy global notifier.
 pub fn resolve_notify_interface_for_context(context: Option<&AppContext>) -> Arc<dyn NotifyInterface> {
    context
@@ -323,6 +333,12 @@ fn resolve_ready_iam_handle_with(
    fallback()
 }

+fn resolve_token_signing_key_with(context: Option<Arc<AppContext>>, fallback: impl FnOnce() -> Option<String>) -> Option<String> {
+    context
+        .and_then(|context| context.iam().token_signing_key())
+        .or_else(fallback)
+}
+
 fn resolve_bucket_metadata_handle_with(
    context: Option<Arc<AppContext>>,
    fallback: impl FnOnce() -> Option<Arc<RwLock<BucketMetadataSys>>>,
@@ -546,7 +562,7 @@ mod tests {
    };
    use crate::config::{RustFSBufferConfig, WorkloadProfile};
    use async_trait::async_trait;
-    use rustfs_iam::{store::object::ObjectStore, sys::IamSys};
+    use rustfs_iam::{oidc::OidcSys, store::object::ObjectStore, sys::IamSys};
    use rustfs_io_metrics::{PerformanceMetrics, internode_metrics::InternodeMetrics};
    use rustfs_lock::{LocalClient, LockClient};
    use rustfs_s3select_api::{
@@ -562,6 +578,8 @@ mod tests {

    struct TestIamInterface {
        ready: bool,
+        oidc: Option<Arc<OidcSys>>,
+        token_signing_key: Option<String>,
    }

    impl IamInterface for TestIamInterface {
@@ -572,13 +590,23 @@ mod tests {
        fn is_ready(&self) -> bool {
            self.ready
        }
+
+        fn oidc(&self) -> Option<Arc<OidcSys>> {
+            self.oidc.clone()
+        }
+
+        fn token_signing_key(&self) -> Option<String> {
+            self.token_signing_key.clone()
+        }
    }

-    struct TestOidcInterface;
+    struct TestOidcInterface {
+        oidc: Option<Arc<OidcSys>>,
+    }

    impl OidcInterface for TestOidcInterface {
        fn handle(&self) -> Option<Arc<rustfs_iam::oidc::OidcSys>> {
-            None
+            self.oidc.clone()
        }
    }

@@ -985,12 +1013,30 @@ mod tests {
        };
        let context_region: s3s::region::Region = "context-region".parse().expect("test region");
        let fallback_region: s3s::region::Region = "fallback-region".parse().expect("test region");
+        let context_oidc_sys = match OidcSys::empty() {
+            Ok(sys) => sys,
+            Err(err) => unreachable!("test OIDC sys should initialize: {err}"),
+        };
+        let fallback_oidc_sys = match OidcSys::empty() {
+            Ok(sys) => sys,
+            Err(err) => unreachable!("test OIDC fallback sys should initialize: {err}"),
+        };
+        let context_oidc = Arc::new(context_oidc_sys);
+        let fallback_oidc = Arc::new(fallback_oidc_sys);
+        let context_token_signing_key = "context-token-signing-key".to_string();
+        let fallback_token_signing_key = "fallback-token-signing-key".to_string();

        let context = Arc::new(AppContext::with_test_interfaces(
            object_store.clone(),
            AppContextTestInterfaces {
-                iam: Arc::new(TestIamInterface { ready: true }),
-                oidc: Arc::new(TestOidcInterface),
+                iam: Arc::new(TestIamInterface {
+                    ready: true,
+                    oidc: None,
+                    token_signing_key: Some(context_token_signing_key.clone()),
+                }),
+                oidc: Arc::new(TestOidcInterface {
+                    oidc: Some(context_oidc.clone()),
+                }),
                kms: Arc::new(TestKmsInterface {
                    kms: context_kms.clone(),
                }),
@@ -1084,6 +1130,12 @@ mod tests {
            context_outbound_tls_state.generation
        );
        assert!(resolve_iam_ready_with(Some(context.clone()), || false));
+        let resolved_oidc = resolve_oidc_handle_with(Some(context.clone()), || Some(fallback_oidc.clone()));
+        assert!(resolved_oidc.as_ref().is_some_and(|oidc| Arc::ptr_eq(oidc, &context_oidc)));
+        assert_eq!(
+            resolve_token_signing_key_with(Some(context.clone()), || Some(fallback_token_signing_key.clone())).as_deref(),
+            Some(context_token_signing_key.as_str())
+        );
        assert!(Arc::ptr_eq(
            &resolve_bucket_metadata_handle_with(Some(context.clone()), || None).expect("context bucket metadata"),
            &bucket_metadata
@@ -74,6 +74,14 @@ impl IamInterface for IamHandle {
    fn is_ready(&self) -> bool {
        rustfs_iam::get().is_ok()
    }
+
+    fn oidc(&self) -> Option<Arc<OidcSys>> {
+        rustfs_iam::get_oidc()
+    }
+
+    fn token_signing_key(&self) -> Option<String> {
+        rustfs_iam::manager::get_token_signing_key()
+    }
 }

 /// Default OIDC interface adapter.
@@ -23,8 +23,7 @@ use crate::config::RustFSBufferConfig;
 use async_trait::async_trait;
 use rustfs_config::server_config::Config;
 use rustfs_credentials::Credentials;
-use rustfs_iam::oidc::OidcSys;
-use rustfs_iam::{store::object::ObjectStore, sys::IamSys};
+use rustfs_iam::{oidc::OidcSys, store::object::ObjectStore, sys::IamSys};
 use rustfs_io_metrics::{PerformanceMetrics, internode_metrics::InternodeMetrics};
 use rustfs_kms::KmsServiceManager;
 use rustfs_lock::LockClient;
@@ -41,6 +40,12 @@ pub trait IamInterface: Send + Sync {
    #[allow(dead_code)]
    fn handle(&self) -> Arc<IamSys<ObjectStore>>;
    fn is_ready(&self) -> bool;
+    fn oidc(&self) -> Option<Arc<OidcSys>> {
+        None
+    }
+    fn token_signing_key(&self) -> Option<String> {
+        None
+    }
 }

 /// OIDC interface for admin and runtime consumers.
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use crate::app::context::resolve_region;
+use crate::app::context::{resolve_notify_interface, resolve_region};
 use crate::server::ShutdownHandle;
 use crate::storage::{
    get_bucket_notification_config, process_lambda_configurations, process_queue_configurations, process_topic_configurations,
@@ -22,7 +22,6 @@ use rustfs_config::{
    DEFAULT_BUFFER_MAX_SIZE, DEFAULT_BUFFER_MIN_SIZE, DEFAULT_BUFFER_PROFILE, DEFAULT_BUFFER_UNKNOWN_SIZE, DEFAULT_UPDATE_CHECK,
    ENV_RUSTFS_BUFFER_DEFAULT_SIZE, ENV_RUSTFS_BUFFER_MAX_SIZE, ENV_RUSTFS_BUFFER_MIN_SIZE, ENV_UPDATE_CHECK, RUSTFS_REGION,
 };
-use rustfs_notify::notifier_global;
 use rustfs_targets::arn::{ARN, TargetIDError};
 use rustfs_utils::get_env_usize;
 use s3s::s3_error;
@@ -245,7 +244,8 @@ pub async fn add_bucket_notification_configuration(buckets: Vec<String>) {
                    );
                }

-                if let Err(e) = notifier_global::add_event_specific_rules(bucket, region, &event_rules)
+                if let Err(e) = resolve_notify_interface()
+                    .add_event_specific_rules(bucket, region, &event_rules)
                    .await
                    .map_err(|e| s3_error!(InternalError, "Failed to add rules: {e}"))
                {
@@ -13,7 +13,7 @@
 //  limitations under the License.

 use super::{module_switch::resolve_notify_module_state, refresh_persisted_module_switches_from_store};
-use crate::app::context::resolve_server_config;
+use crate::app::context::{resolve_notify_interface, resolve_server_config};
 use crate::storage::{EventArgs as EcstoreEventArgs, StorageObjectInfo, register_event_dispatch_hook};
 use chrono::{DateTime, Utc};
 use rustfs_notify::{EventArgs as NotifyEventArgs, NotifyObjectInfo};
@@ -118,7 +118,7 @@ fn install_ecstore_event_dispatch_hook() {
            return;
        };
        spawn(async move {
-            rustfs_notify::notifier_global::notify(notify_args).await;
+            resolve_notify_interface().notify(notify_args).await;
        });
    });

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use crate::app::context::resolve_action_credentials;
+use crate::app::context::{resolve_action_credentials, resolve_notify_interface};
 use crate::server::{convert_ecstore_object_info, is_audit_module_enabled, is_notify_module_enabled};
 use crate::storage::access::{ReqInfo, request_context_from_req};
 use crate::storage::request_context::{RequestContext, extract_request_id_from_headers};
@@ -24,7 +24,7 @@ use rustfs_audit::{
    global::AuditLogger,
 };
 use rustfs_io_metrics::record_s3_op;
-use rustfs_notify::{EventArgsBuilder, notifier_global};
+use rustfs_notify::EventArgsBuilder;
 use rustfs_s3_ops::{S3Operation, operation_matches_event_name};
 use rustfs_s3_types::EventName;
 use rustfs_targets::{
@@ -383,7 +383,7 @@ impl Drop for OperationHelper {
            if !event_args.is_replication_request() {
                let ctx = state.request_context.clone();
                spawn_background_with_context(ctx, async move {
-                    notifier_global::notify(event_args).await;
+                    resolve_notify_interface().notify(event_args).await;
                });
            }
        }