gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-21 10:03:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(notify): validate bucket notification config before saving (#7980)

Browse Source
This commit is contained in:
cxymds
2026-09-17 22:58:29 +08:00
committed by GitHub
parent 5829c66f21
commit db4b8290e7
9 changed files with 192 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/notify/src/bucket_config_manager.rs
+51 -2
View File
@@ -53,7 +53,9 @@ impl NotifyBucketConfigManager {
self.rule_engine.has_subscriber(bucket, event).await
}
pub async fn load_bucket_notification_config(
/// Validates that the runtime can accept `cfg` for `bucket` without
/// mutating any state.
pub async fn validate_bucket_notification_config(
&self,
bucket: &str,
cfg: &BucketNotificationConfig,
@@ -98,6 +100,11 @@ impl NotifyBucketConfigManager {
);
}
Ok(())
}
/// Publishes an already validated `cfg` to the runtime rule state.
pub async fn apply_bucket_notification_config(&self, bucket: &str, cfg: &BucketNotificationConfig) {
self.subscriber_view.apply_bucket_config(bucket, cfg);
self.rule_engine.set_bucket_rules(bucket, cfg.get_rules_map().clone()).await;
info!(
@@ -109,6 +116,15 @@ impl NotifyBucketConfigManager {
rule_count = cfg.get_rules_map().inner().len(),
"notify bucket config state"
);
}
pub async fn load_bucket_notification_config(
&self,
bucket: &str,
cfg: &BucketNotificationConfig,
) -> Result<(), NotificationError> {
self.validate_bucket_notification_config(bucket, cfg).await?;
self.apply_bucket_notification_config(bucket, cfg).await;
Ok(())
}
@@ -122,7 +138,7 @@ impl NotifyBucketConfigManager {
mod tests {
use super::NotifyBucketConfigManager;
use crate::{
BucketNotificationConfig, integration::NotificationMetrics,
BucketNotificationConfig, NotificationError, integration::NotificationMetrics,
notification_system_subscriber::NotificationSystemSubscriberView, notifier::EventNotifier, rule_engine::NotifyRuleEngine,
};
use rustfs_s3_types::EventName;
@@ -156,4 +172,37 @@ mod tests {
manager.remove_bucket_notification_config("bucket").await;
assert!(!manager.subscriber_view.has_subscriber("bucket", &EventName::ObjectCreatedPut));
}
#[tokio::test]
async fn validate_bucket_notification_config_rejects_missing_targets_without_applying_rules() {
let manager = build_manager();
let target_id = TargetID::new("primary".to_string(), "webhook".to_string());
let mut cfg = BucketNotificationConfig::new("us-east-1");
cfg.add_rule(&[EventName::ObjectCreatedPut], "*".to_string(), target_id);
let err = manager
.validate_bucket_notification_config("bucket", &cfg)
.await
.expect_err("validation must fail when the runtime has no notify targets");
assert!(matches!(err, NotificationError::Configuration(_)));
assert!(!manager.has_subscriber("bucket", &EventName::ObjectCreatedPut).await);
assert!(!manager.subscriber_view.has_subscriber("bucket", &EventName::ObjectCreatedPut));
}
#[tokio::test]
async fn failed_validation_keeps_previous_bucket_rules() {
let manager = build_manager();
let target_id = TargetID::new("primary".to_string(), "webhook".to_string());
let mut cfg = BucketNotificationConfig::new("us-east-1");
cfg.add_rule(&[EventName::ObjectCreatedPut], "*".to_string(), target_id);
manager.apply_bucket_notification_config("bucket", &cfg).await;
assert!(manager.has_subscriber("bucket", &EventName::ObjectCreatedPut).await);
let err = manager
.validate_bucket_notification_config("bucket", &cfg)
.await
.expect_err("validation must fail when the runtime has no notify targets");
assert!(matches!(err, NotificationError::Configuration(_)));
assert!(manager.has_subscriber("bucket", &EventName::ObjectCreatedPut).await);
}
}
crates/notify/src/global.rs
+37 -10
View File
@@ -239,6 +239,42 @@ pub mod notifier_global {
.await
}
fn bucket_notification_config_for_rules(
region: &str,
event_rules: &[(Vec<EventName>, String, String, Vec<TargetID>)],
) -> BucketNotificationConfig {
let mut bucket_config = BucketNotificationConfig::new(region);
for (event_names, prefix, suffix, target_ids) in event_rules {
// Use `new_pattern` to construct a matching pattern
let pattern = crate::rules::pattern::new_pattern(Some(prefix.as_str()), Some(suffix.as_str()));
for target_id in target_ids {
bucket_config.add_rule(event_names, pattern.clone(), target_id.clone());
}
}
bucket_config
}
/// Checks that the runtime can accept `event_rules` for `region` without
/// publishing them. Callers use this to fail before persisting a bucket
/// notification configuration the notify subsystem would reject.
pub async fn validate_event_specific_rules(
bucket_name: &str,
region: &str,
event_rules: &[(Vec<EventName>, String, String, Vec<TargetID>)],
) -> Result<(), NotificationError> {
let bucket_config = bucket_notification_config_for_rules(region, event_rules);
// Get global NotificationSystem instance
let notification_sys = notification_system().ok_or(NotificationError::Lifecycle(LifecycleError::NotInitialized))?;
notification_sys
.validate_bucket_notification_config(bucket_name, &bucket_config)
.await
}
/// Dynamically add notification rules according to different event types.
///
/// # Parameter
@@ -256,16 +292,7 @@ pub mod notifier_global {
region: &str,
event_rules: &[(Vec<EventName>, String, String, Vec<TargetID>)],
) -> Result<(), NotificationError> {
let mut bucket_config = BucketNotificationConfig::new(region);
for (event_names, prefix, suffix, target_ids) in event_rules {
// Use `new_pattern` to construct a matching pattern
let pattern = crate::rules::pattern::new_pattern(Some(prefix.as_str()), Some(suffix.as_str()));
for target_id in target_ids {
bucket_config.add_rule(event_names, pattern.clone(), target_id.clone());
}
}
let bucket_config = bucket_notification_config_for_rules(region, event_rules);
// Get global NotificationSystem instance
let notification_sys = notification_system().ok_or(NotificationError::Lifecycle(LifecycleError::NotInitialized))?;
crates/notify/src/integration.rs
+13
View File
@@ -372,6 +372,19 @@ impl NotificationSystem {
self.services.config_manager.lifecycle().is_converged()
}
/// Validates a bucket notification configuration against the runtime targets
/// without publishing it, so callers can reject a request before persisting.
pub async fn validate_bucket_notification_config(
&self,
bucket: &str,
cfg: &BucketNotificationConfig,
) -> Result<(), NotificationError> {
self.services
.bucket_config_manager
.validate_bucket_notification_config(bucket, cfg)
.await
}
/// Loads the bucket notification configuration
pub async fn load_bucket_notification_config(
&self,
rustfs/src/app/bucket_usecase.rs
+56 -35
View File
@@ -92,7 +92,7 @@ use rustfs_policy::policy::{
use rustfs_s3_ops::S3Operation;
use rustfs_targets::{
EventName,
arn::{ARN, TargetIDError},
arn::{ARN, TargetID, TargetIDError},
};
use rustfs_trusted_proxies::ClientInfo;
use rustfs_utils::http::{SUFFIX_FORCE_DELETE, get_header};
@@ -508,6 +508,46 @@ fn validate_notification_configuration_filters(notification_configuration: &Noti
Ok(())
}
fn parse_notification_target_id(arn_str: &str) -> Result<TargetID, TargetIDError> {
ARN::parse(arn_str)
.map(|arn| arn.target_id)
.map_err(|e| TargetIDError::InvalidFormat(e.to_string()))
}
type NotificationEventRule = (Vec<EventName>, String, String, Vec<TargetID>);
/// Builds the notify runtime rules for a bucket notification configuration
/// without touching the store or the runtime rule state.
fn build_notification_event_rules(
notification_configuration: &NotificationConfiguration,
) -> S3Result<Vec<NotificationEventRule>> {
let mut event_rules = Vec::new();
let invalid_arn = |e: TargetIDError| {
S3Error::with_message(S3ErrorCode::InvalidArgument, format!("Invalid ARN in notification configuration: {e}"))
};
process_queue_configurations(
&mut event_rules,
notification_configuration.queue_configurations.clone(),
parse_notification_target_id,
)
.map_err(invalid_arn)?;
process_topic_configurations(
&mut event_rules,
notification_configuration.topic_configurations.clone(),
parse_notification_target_id,
)
.map_err(invalid_arn)?;
process_lambda_configurations(
&mut event_rules,
notification_configuration.lambda_function_configurations.clone(),
parse_notification_target_id,
)
.map_err(invalid_arn)?;
Ok(event_rules)
}
fn sr_bucket_meta_item(bucket: String, item_type: &str) -> SRBucketMeta {
SRBucketMeta {
bucket,
@@ -2440,6 +2480,17 @@ impl DefaultBucketUsecase {
.await
.map_err(ApiError::from)?;
let region = resolve_notification_region(self.global_region(), request_region);
let notify = current_notify_interface_for_context(self.context.as_deref());
let event_rules = build_notification_event_rules(&notification_configuration)?;
// Reject the request before the store write so a failure cannot leave a
// persisted configuration that the notify runtime refused to activate.
notify
.validate_event_specific_rules(&bucket, region.as_str(), &event_rules)
.await
.map_err(|e| s3_error!(InternalError, "Failed to add rules: {e}"))?;
let data = serialize_config(&notification_configuration)?;
update_bucket_config_for_incarnation(&bucket, BUCKET_NOTIFICATION_CONFIG, data, expected_incarnation_id)
.await
@@ -2447,40 +2498,10 @@ impl DefaultBucketUsecase {
notify_bucket_metadata_reload(bucket.clone(), "put bucket notification", request_context, false).await;
let region = resolve_notification_region(self.global_region(), request_region);
let notify = current_notify_interface_for_context(self.context.as_deref());
let clear_rules = notify.clear_bucket_notification_rules(&bucket);
let parse_rules = async {
let mut event_rules = Vec::new();
process_queue_configurations(&mut event_rules, notification_configuration.queue_configurations.clone(), |arn_str| {
ARN::parse(arn_str)
.map(|arn| arn.target_id)
.map_err(|e| TargetIDError::InvalidFormat(e.to_string()))
})?;
process_topic_configurations(&mut event_rules, notification_configuration.topic_configurations.clone(), |arn_str| {
ARN::parse(arn_str)
.map(|arn| arn.target_id)
.map_err(|e| TargetIDError::InvalidFormat(e.to_string()))
})?;
process_lambda_configurations(
&mut event_rules,
notification_configuration.lambda_function_configurations.clone(),
|arn_str| {
ARN::parse(arn_str)
.map(|arn| arn.target_id)
.map_err(|e| TargetIDError::InvalidFormat(e.to_string()))
},
)?;
Ok::<_, TargetIDError>(event_rules)
};
let (clear_result, event_rules_result) = tokio::join!(clear_rules, parse_rules);
clear_result.map_err(|e| s3_error!(InternalError, "Failed to clear rules: {e}"))?;
let event_rules =
event_rules_result.map_err(|e| s3_error!(InvalidArgument, "Invalid ARN in notification configuration: {e}"))?;
notify
.clear_bucket_notification_rules(&bucket)
.await
.map_err(|e| s3_error!(InternalError, "Failed to clear rules: {e}"))?;
warn!("notify event rules: {:?}", &event_rules);
notify
.add_event_specific_rules(&bucket, region.as_str(), &event_rules)
rustfs/src/app/context/handles.rs
+9
View File
@@ -167,6 +167,15 @@ impl NotifyInterface for NotifyHandle {
runtime_sources::notify(args).await;
}
async fn validate_event_specific_rules(
&self,
bucket_name: &str,
region: &str,
event_rules: &[(Vec<EventName>, String, String, Vec<TargetID>)],
) -> Result<(), NotificationError> {
runtime_sources::validate_event_specific_rules(bucket_name, region, event_rules).await
}
async fn add_event_specific_rules(
&self,
bucket_name: &str,
rustfs/src/app/context/interfaces.rs
+7
View File
@@ -73,6 +73,13 @@ pub trait OutboundTlsRuntimeInterface: Send + Sync {
pub trait NotifyInterface: Send + Sync {
async fn notify(&self, args: EventArgs);
async fn validate_event_specific_rules(
&self,
bucket_name: &str,
region: &str,
event_rules: &[(Vec<EventName>, String, String, Vec<TargetID>)],
) -> Result<(), NotificationError>;
async fn add_event_specific_rules(
&self,
bucket_name: &str,
rustfs/src/app/context/runtime_sources.rs
+8
View File
@@ -71,6 +71,14 @@ pub async fn notify(args: EventArgs) {
notifier_global::notify(args).await;
}
pub async fn validate_event_specific_rules(
bucket_name: &str,
region: &str,
event_rules: &[(Vec<EventName>, String, String, Vec<TargetID>)],
) -> Result<(), NotificationError> {
notifier_global::validate_event_specific_rules(bucket_name, region, event_rules).await
}
pub async fn add_event_specific_rules(
bucket_name: &str,
region: &str,
rustfs/src/app/object/extract.rs
+9
View File
@@ -2747,6 +2747,15 @@ mod tests {
let _ = self.events.send(args.version_id);
}
async fn validate_event_specific_rules(
&self,
_bucket_name: &str,
_region: &str,
_event_rules: &[(Vec<rustfs_targets::EventName>, String, String, Vec<rustfs_targets::arn::TargetID>)],
) -> Result<(), rustfs_notify::NotificationError> {
Ok(())
}
async fn add_event_specific_rules(
&self,
_bucket_name: &str,
scripts/test_security_workflow.py
+2 -1
View File
@@ -303,11 +303,12 @@ class SecurityWorkflowTests(WorkflowSteps, unittest.TestCase):
source = (ROOT / f".github/workflows/rustfs-{suite}-test.yml").read_text().splitlines()
# Workflow-level concurrency covers every job, including cleanup,
# regardless of trigger or the runner hosting the job.
expected_group = "rustfs-performance-suite" if suite == "performance" else "rustfs-shared-functional-tests-v2"
self.assertEqual([
line.strip() for line in yaml_block(source, "concurrency", 0)
if line.strip() and not line.lstrip().startswith("#")
], [
"group: rustfs-shared-functional-tests-v2", "cancel-in-progress: false",
f"group: {expected_group}", "cancel-in-progress: false",
])
self.assertIsNotNone(yaml_block(source, "workflow_dispatch", 2))
self.assertIsNotNone(yaml_block(source, "repository_dispatch", 2))