fix(targets): explain webhook outbound allowlist failures

crates/ecstore/src/bucket/replication/replication_pool.rs

@@ -1823,20 +1823,12 @@ pub(crate) async fn schedule_replication<S: ReplicationStorage>(
     dsc: ReplicateDecision,
     op_type: ReplicationType,
 ) {
-    let (synchronous, asynchronous) = dsc.partition_by_sync();
-    let mut async_oi = oi;
+    let synchronous = dsc.is_synchronous();
+    let ri = replicate_object_info_from_object_info(oi, dsc, op_type);
 
-    if synchronous.replicate_any() {
-        let ri = replicate_object_info_from_object_info(async_oi.clone(), synchronous, op_type);
-        let state = replicate_object(ri, o.clone()).await;
-        async_oi.replication_status_internal = state.replication_status_internal;
-        async_oi.version_purge_status_internal = state.version_purge_status_internal;
-    }
-
-    if asynchronous.replicate_any()
-        && let Some(pool) = runtime_sources::replication_pool()
-    {
-        let ri = replicate_object_info_from_object_info(async_oi, asynchronous, op_type);
+    if synchronous {
+        replicate_object(ri, o).await
+    } else if let Some(pool) = runtime_sources::replication_pool() {
         let _ = pool.queue_replica_task(ri).await;
     }
 }

crates/ecstore/src/bucket/replication/replication_resyncer.rs

@@ -19,7 +19,7 @@ use super::replication_error_boundary::{Result, is_err_object_not_found, is_err_
 use super::replication_event_sink::{EventArgs, send_event, send_local_event};
 use super::replication_filemeta_boundary::{
     NULL_VERSION_ID, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicatedInfos,
-    ReplicatedTargetInfo, ReplicationAction, ReplicationState, ReplicationStatusType, ReplicationType, VersionPurgeStatusType,
+    ReplicatedTargetInfo, ReplicationAction, ReplicationStatusType, ReplicationType, VersionPurgeStatusType,
     get_replication_state, parse_replicate_decision, replication_statuses_map, target_reset_header, version_purge_statuses_map,
 };
 use super::replication_lock_boundary::ReplicationLockTiming;
@@ -2001,14 +2001,61 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli
     rinfo
 }
 
-pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, storage: Arc<S>) -> ReplicationState {
+pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, storage: Arc<S>) {
     let bucket = roi.bucket.clone();
     let object = roi.name.clone();
 
-    // The admission decision is the target-granular contract. Re-evaluating the
-    // live config here could fan a synchronous request out to targets that were
-    // not admitted, or promote an async target after a mixed-mode split.
-    let tgt_arns = roi.dsc.replicate_target_arns();
+    let cfg = match get_replication_config(&bucket).await {
+        Ok(Some(config)) => config,
+        Ok(None) => {
+            debug!(
+                event = EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                bucket = %bucket,
+                reason = "replication_config_missing",
+                "Skipping replication object because replication config is missing"
+            );
+            send_local_event(EventArgs {
+                event_name: EventName::ObjectReplicationNotTracked.to_string(),
+                bucket_name: bucket.clone(),
+                object: roi.to_object_info(),
+                user_agent: "Internal: [Replication]".to_string(),
+                ..Default::default()
+            });
+            return;
+        }
+        Err(err) => {
+            error!(
+                event = EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                bucket = %bucket,
+                reason = "replication_config_lookup_failed",
+                error = %err,
+                "Failed to look up replication config for object replication"
+            );
+            send_local_event(EventArgs {
+                event_name: EventName::ObjectReplicationNotTracked.to_string(),
+                bucket_name: bucket.clone(),
+                object: roi.to_object_info(),
+                user_agent: "Internal: [Replication]".to_string(),
+                ..Default::default()
+            });
+            return;
+        }
+    };
+
+    let tgt_arns = cfg.filter_target_arns(&ObjectOpts {
+        name: object.clone(),
+        user_tags: roi.user_tags.clone(),
+        ssec: roi.ssec,
+        op_type: roi.op_type,
+        // ExistingObject ops must respect per-rule ExistingObjectReplicationStatus.
+        // Heal ops intentionally bypass it (repairing a past failure is not an initial sync).
+        existing_object: roi.op_type == ReplicationType::ExistingObject,
+        ..Default::default()
+    });
 
     // Acquire a per-object namespace lock so that at most one worker (across all cluster
     // nodes and MRF retry goroutines) replicates this object version at a time.
@@ -2033,7 +2080,7 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
                 user_agent: "Internal: [Replication]".to_string(),
                 ..Default::default()
             });
-            return roi.replication_state.unwrap_or_default();
+            return;
         }
     };
     let _obj_lock_guard = match obj_ns_lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await {
@@ -2056,7 +2103,7 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
                 user_agent: "Internal: [Replication]".to_string(),
                 ..Default::default()
             });
-            return roi.replication_state.unwrap_or_default();
+            return;
         }
     };
 
@@ -2132,10 +2179,8 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
         }
     }
 
-    let previous_state = roi.replication_state.clone().unwrap_or_default();
-    let merged_state = get_replication_state(&rinfos, &previous_state, roi.version_id.map(|v| v.to_string()));
-    let replication_status = merged_state.composite_replication_status();
-    let new_replication_internal = merged_state.replication_status_internal.clone();
+    let replication_status = rinfos.replication_status();
+    let new_replication_internal = rinfos.replication_status_internal();
     let mut object_info = roi.to_object_info();
 
     if roi.replication_status_internal != new_replication_internal || rinfos.replication_resynced() {
@@ -2204,8 +2249,6 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
             }
         }
     }
-
-    merged_state
 }
 
 trait ReplicateObjectInfoExt {

crates/kms/src/api_types.rs

@@ -21,7 +21,7 @@ use crate::config::{
     redacted_secret, redacted_secret_option,
 };
 use crate::service_manager::KmsServiceStatus;
-use crate::types::KeyMetadata;
+use crate::types::{KeyMetadata, KeyUsage};
 use serde::{Deserialize, Deserializer, Serialize};
 use std::collections::HashMap;
 use std::fmt;
@@ -1300,6 +1300,49 @@ mod tests {
 // Key Management API Types
 // ========================================
 
+/// Request to create a new key with optional custom name
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct CreateKeyRequest {
+    /// Custom key name (optional, will auto-generate UUID if not provided)
+    pub key_name: Option<String>,
+    /// Key usage type
+    pub key_usage: KeyUsage,
+    /// Key description
+    pub description: Option<String>,
+    /// Key policy JSON string
+    pub policy: Option<String>,
+    /// Tags for the key
+    pub tags: HashMap<String, String>,
+    /// Origin of the key
+    pub origin: Option<String>,
+}
+
+impl Default for CreateKeyRequest {
+    fn default() -> Self {
+        Self {
+            key_name: None,
+            key_usage: KeyUsage::EncryptDecrypt,
+            description: None,
+            policy: None,
+            tags: HashMap::new(),
+            origin: None,
+        }
+    }
+}
+
+/// Response from create key operation
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct CreateKeyResponse {
+    /// Success flag
+    pub success: bool,
+    /// Status message
+    pub message: String,
+    /// Created key ID (either custom name or auto-generated UUID)
+    pub key_id: String,
+    /// Key metadata
+    pub key_metadata: KeyMetadata,
+}
+
 /// JSON shape returned by the admin delete-key endpoint.
 ///
 /// The delete *request* shape lives in [`crate::types::DeleteKeyRequest`] —
@@ -1317,6 +1360,15 @@ pub struct DeleteKeyResponse {
     pub deletion_date: Option<String>,
 }
 
+/// Request to list all keys
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ListKeysRequest {
+    /// Maximum number of keys to return (1-1000)
+    pub limit: Option<u32>,
+    /// Pagination marker
+    pub marker: Option<String>,
+}
+
 /// Response from list keys operation
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ListKeysResponse {
@@ -1332,6 +1384,13 @@ pub struct ListKeysResponse {
     pub next_marker: Option<String>,
 }
 
+/// Request to describe a key
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DescribeKeyRequest {
+    /// Key ID to describe
+    pub key_id: String,
+}
+
 /// Response from describe key operation
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct DescribeKeyResponse {
@@ -1343,6 +1402,13 @@ pub struct DescribeKeyResponse {
     pub key_metadata: Option<KeyMetadata>,
 }
 
+/// Request to cancel key deletion
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct CancelKeyDeletionRequest {
+    /// Key ID to cancel deletion for
+    pub key_id: String,
+}
+
 /// Response from cancel key deletion operation
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct CancelKeyDeletionResponse {

crates/replication/src/filemeta.rs

@@ -660,27 +660,6 @@ impl ReplicateDecision {
         self.targets_map.values().any(|t| t.synchronous)
     }
 
-    /// Split admitted targets by their configured delivery mode.
-    ///
-    /// Non-replicating entries are intentionally omitted from both decisions.
-    /// Callers must not promote an async target merely because another target is
-    /// synchronous, and unsupported operation paths can keep both partitions
-    /// empty or explicitly async.
-    pub fn partition_by_sync(&self) -> (Self, Self) {
-        let mut synchronous = Self::new();
-        let mut asynchronous = Self::new();
-
-        for target in self.targets_map.values().filter(|target| target.replicate) {
-            if target.synchronous {
-                synchronous.set(target.clone());
-            } else {
-                asynchronous.set(target.clone());
-            }
-        }
-
-        (synchronous, asynchronous)
-    }
-
     /// Updates ReplicateDecision with target's replication decision
     pub fn set(&mut self, target: ReplicateTargetDecision) {
         self.targets_map.insert(target.arn.clone(), target);
@@ -1088,32 +1067,6 @@ mod tests {
         assert_eq!(entry.target_arns, vec!["arn:target-a".to_string()]);
     }
 
-    #[test]
-    fn partition_by_sync_keeps_mixed_targets_independent() {
-        let mut decision = ReplicateDecision::new();
-        decision.set(ReplicateTargetDecision::new("arn:sync".to_string(), true, true));
-        decision.set(ReplicateTargetDecision::new("arn:async".to_string(), true, false));
-        decision.set(ReplicateTargetDecision::new("arn:disabled".to_string(), false, true));
-
-        let (synchronous, asynchronous) = decision.partition_by_sync();
-
-        assert_eq!(synchronous.replicate_target_arns(), vec!["arn:sync".to_string()]);
-        assert_eq!(asynchronous.replicate_target_arns(), vec!["arn:async".to_string()]);
-        assert!(synchronous.is_synchronous());
-        assert!(!asynchronous.is_synchronous());
-    }
-
-    #[test]
-    fn partition_by_sync_does_not_promote_async_targets() {
-        let mut decision = ReplicateDecision::new();
-        decision.set(ReplicateTargetDecision::new("arn:async".to_string(), true, false));
-
-        let (synchronous, asynchronous) = decision.partition_by_sync();
-
-        assert!(!synchronous.replicate_any());
-        assert_eq!(asynchronous.replicate_target_arns(), vec!["arn:async".to_string()]);
-    }
-
     #[test]
     fn target_state_reads_resync_timestamp_from_target_reset_header_key() {
         let arn = "arn:rustfs:replication:us-east-1:target:bucket";

crates/targets/src/config/common.rs

@@ -23,7 +23,7 @@ use rustfs_config::{
     NATS_TLS_CLIENT_KEY, NATS_TOKEN, NATS_USERNAME, PULSAR_AUTH_TOKEN, PULSAR_PASSWORD, PULSAR_QUEUE_DIR, PULSAR_TLS_CA,
     PULSAR_TOPIC, PULSAR_USERNAME,
 };
-use rustfs_utils::egress::OutboundPolicy;
+use rustfs_utils::egress::{ENV_OUTBOUND_ALLOW_ORIGINS, OutboundPolicy, OutboundPolicyError};
 use std::collections::HashSet;
 use std::path::Path;
 use std::str::FromStr;
@@ -223,12 +223,26 @@ pub(super) fn validate_outbound_http_url(value: &Url, field_label: &str) -> Resu
         OutboundPolicy::from_env_cached().map_err(|err| TargetError::Configuration(format!("invalid outbound policy: {err}")))?;
     policy
         .validate_url(value)
-        .map_err(|e| TargetError::Configuration(format!("{field_label} is not allowed: {e}")))
+        .map_err(|err| TargetError::Configuration(format_outbound_http_url_error(field_label, value, err)))
+}
+
+fn format_outbound_http_url_error(field_label: &str, value: &Url, err: OutboundPolicyError) -> String {
+    let base = format!("{field_label} is not allowed: {err}");
+    if matches!(err, OutboundPolicyError::ForbiddenHost { .. }) {
+        let origin = value.origin().ascii_serialization();
+        return format!(
+            "{base}; set {ENV_OUTBOUND_ALLOW_ORIGINS}={origin} to allow this operator-owned endpoint (origin only, no path)"
+        );
+    }
+    base
 }
 
 #[cfg(test)]
 mod tests {
-    use super::{parse_jetstream_enable, parse_url, validate_nats_server_config, validate_pulsar_broker_config};
+    use super::{
+        format_outbound_http_url_error, parse_jetstream_enable, parse_url, validate_nats_server_config,
+        validate_pulsar_broker_config,
+    };
     use async_nats::ServerAddr;
     use rustfs_config::server_config::KVS;
     use rustfs_config::{
@@ -236,7 +250,9 @@ mod tests {
         NATS_SUBJECT, NATS_TOKEN, NATS_USERNAME, PULSAR_TLS_ALLOW_INSECURE, PULSAR_TLS_CA, PULSAR_TLS_HOSTNAME_VERIFICATION,
         PULSAR_TOPIC,
     };
+    use rustfs_utils::egress::{ENV_OUTBOUND_ALLOW_ORIGINS, OutboundPolicyError};
     use std::str::FromStr;
+    use url::Url;
 
     fn nats_server() -> ServerAddr {
         ServerAddr::from_str("nats://127.0.0.1:4222").expect("valid nats address")
@@ -248,6 +264,24 @@ mod tests {
         assert!(!err.to_string().contains("secret-token"));
     }
 
+    #[test]
+    fn outbound_http_error_names_allowlist_origin_without_path() {
+        let url = Url::parse("http://192.168.1.2:1880/webhook/rustfs").expect("valid endpoint");
+        let err = format_outbound_http_url_error(
+            "endpoint URL",
+            &url,
+            OutboundPolicyError::ForbiddenHost {
+                host: "192.168.1.2".to_string(),
+                reason: "private address",
+            },
+        );
+
+        assert!(err.contains(ENV_OUTBOUND_ALLOW_ORIGINS));
+        assert!(err.contains("http://192.168.1.2:1880"));
+        assert!(!err.contains("webhook/rustfs"));
+        assert!(err.contains("origin only, no path"));
+    }
+
     // Absolute on Linux, macOS, and Windows. temp_dir needs no filesystem to exist for a
     // validation-only test, and Path::is_absolute stays true across platforms.
     fn nats_queue_dir() -> String {

rustfs/src/app/lifecycle_transition_api_test.rs

@@ -1495,8 +1495,6 @@ async fn cancelled_transition_waiting_for_prepared_reader_cleans_remote() {
 #[serial]
 #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
 async fn delete_transitioned_object_removes_remote_tier_copy_via_usecase() {
-    use super::storage_api::test::ReqInfo;
-
     let (_disk_paths, ecstore) = setup_test_env().await;
     let usecase = DefaultObjectUsecase::from_global();
 
@@ -1533,11 +1531,6 @@ async fn delete_transitioned_object_removes_remote_tier_copy_via_usecase() {
         Method::DELETE,
     );
     insert_header(&mut req.headers, SUFFIX_FORCE_DELETE, "true");
-    req.extensions.insert(ReqInfo {
-        cred: Some(rustfs_credentials::Credentials::default()),
-        is_owner: true,
-        ..Default::default()
-    });
 
     Box::pin(usecase.execute_delete_object(req))
         .await
@@ -1742,8 +1735,6 @@ async fn compensation_driven_complete_multipart_upload_still_transitions() {
 #[serial]
 #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
 async fn compensation_driven_transition_still_cleans_remote_tier_on_delete() {
-    use super::storage_api::test::ReqInfo;
-
     let (_disk_paths, ecstore) = setup_test_env().await;
     let usecase = DefaultObjectUsecase::from_global();
 
@@ -1780,11 +1771,6 @@ async fn compensation_driven_transition_still_cleans_remote_tier_on_delete() {
         Method::DELETE,
     );
     insert_header(&mut req.headers, SUFFIX_FORCE_DELETE, "true");
-    req.extensions.insert(ReqInfo {
-        cred: Some(rustfs_credentials::Credentials::default()),
-        is_owner: true,
-        ..Default::default()
-    });
 
     Box::pin(usecase.execute_delete_object(req))
         .await

rustfs/src/app/object_usecase.rs

@@ -6615,8 +6615,7 @@ impl DefaultObjectUsecase {
             ));
         }
 
-        let is_owner = req_info_ref(&req).map(|info| info.is_owner).unwrap_or(false);
-        if !recursive_force_delete_is_authorized(&req.headers, is_owner, false) {
+        if !recursive_force_delete_is_authorized(&req.headers, req_info_ref(&req)?.is_owner, false) {
             return Err(S3Error::with_message(
                 S3ErrorCode::AccessDenied,
                 "Recursive force-delete is restricted to administrative requests",
@@ -7082,7 +7081,7 @@ impl DefaultObjectUsecase {
             authorize_request(&mut req, Action::S3Action(S3Action::ReplicateDeleteAction)).await?;
         }
 
-        let is_owner = req_info_ref(&req).map(|info| info.is_owner).unwrap_or(false);
+        let is_owner = req_info_ref(&req)?.is_owner;
         if !recursive_force_delete_is_authorized(&req.headers, is_owner, replica) {
             return Err(S3Error::with_message(
                 S3ErrorCode::AccessDenied,
@@ -13387,23 +13386,6 @@ mod tests {
 
         let err = usecase.execute_delete_objects(req).await.unwrap_err();
         assert_eq!(err.code(), &S3ErrorCode::InternalError);
-        assert_eq!(err.message(), Some("Not init"));
-    }
-
-    #[tokio::test]
-    async fn execute_delete_object_allows_non_force_request_without_req_info_until_store_lookup() {
-        let input = DeleteObjectInput::builder()
-            .bucket("test-bucket".to_string())
-            .key("test-key".to_string())
-            .build()
-            .unwrap();
-
-        let err = DefaultObjectUsecase::without_context()
-            .execute_delete_object(build_request(input, Method::DELETE))
-            .await
-            .expect_err("an uninitialized store should be reported after non-force admission");
-        assert_eq!(err.code(), &S3ErrorCode::InternalError);
-        assert_eq!(err.message(), Some("Not init"));
     }
 
     #[test]