rustfs/crates/targets/src/target/webhook.rs

// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use crate::plugin::PluginEvent;
use crate::{
    StoreError, Target,
    arn::TargetID,
    error::TargetError,
    runtime::tls::{
        ReloadableTargetTls, TargetTlsInputSet, TlsReloadAdapter, config::ReloadApplyMode, fingerprint::TargetTlsGeneration,
        validate::validate_tls_material,
    },
    store::{Key, Store},
    target::{
        ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliveryCounters, TargetDeliverySnapshot,
        TargetTlsState, TargetType, build_queued_payload, build_target_tls_fingerprint, open_target_queue_store,
        persist_queued_payload_to_store, redacted_secret,
    },
};
use async_trait::async_trait;
use parking_lot::Mutex;
use reqwest::{Client, StatusCode, Url};
use rustfs_tls_runtime::load_cert_bundle_der_bytes;
use rustfs_utils::egress::validate_outbound_url;
use std::{
    fmt,
    marker::PhantomData,
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    time::Duration,
};
use tokio::sync::mpsc;
use tracing::{debug, error, info, instrument, warn};

const LOG_COMPONENT_TARGETS: &str = "targets";
const LOG_SUBSYSTEM_WEBHOOK: &str = "webhook";
const EVENT_WEBHOOK_TARGET_STATE: &str = "webhook_target_state";
const EVENT_WEBHOOK_DELIVERY_STATE: &str = "webhook_delivery_state";

/// Arguments for configuring a Webhook target
#[derive(Clone)]
pub struct WebhookArgs {
    /// Whether the target is enabled
    pub enable: bool,
    /// The endpoint URL to send events to
    pub endpoint: Url,
    /// The authorization token for the endpoint
    pub auth_token: String,
    /// The directory to store events in case of failure
    pub queue_dir: String,
    /// The maximum number of events to store
    pub queue_limit: u64,
    /// The client certificate for TLS (PEM format)
    pub client_cert: String,
    /// The client key for TLS (PEM format)
    pub client_key: String,
    /// The path to a custom client root CA certificate file (PEM format) to trust the server.
    pub client_ca: String,
    /// Skip TLS certificate verification. DANGEROUS: for testing only.
    pub skip_tls_verify: bool,
    /// the target type
    pub target_type: TargetType,
}

impl fmt::Debug for WebhookArgs {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("WebhookArgs")
            .field("enable", &self.enable)
            .field("endpoint", &self.endpoint)
            .field("auth_token", &redacted_secret(&self.auth_token))
            .field("queue_dir", &self.queue_dir)
            .field("queue_limit", &self.queue_limit)
            .field("client_cert", &self.client_cert)
            .field("client_key", &redacted_secret(&self.client_key))
            .field("client_ca", &self.client_ca)
            .field("skip_tls_verify", &self.skip_tls_verify)
            .field("target_type", &self.target_type)
            .finish()
    }
}

impl WebhookArgs {
    /// WebhookArgs verification method
    pub fn validate(&self) -> Result<(), TargetError> {
        if !self.enable {
            return Ok(());
        }

        if self.endpoint.as_str().is_empty() {
            return Err(TargetError::Configuration("endpoint empty".to_string()));
        }
        validate_outbound_url(&self.endpoint)
            .map_err(|err| TargetError::Configuration(format!("webhook endpoint is not allowed: {err}")))?;

        if !self.queue_dir.is_empty() {
            let path = std::path::Path::new(&self.queue_dir);
            if !path.is_absolute() {
                return Err(TargetError::Configuration("webhook queue_dir path should be absolute".to_string()));
            }
        }

        if !self.client_cert.is_empty() && self.client_key.is_empty()
            || self.client_cert.is_empty() && !self.client_key.is_empty()
        {
            return Err(TargetError::Configuration("cert and key must be specified as a pair".to_string()));
        }

        if self.skip_tls_verify && !self.client_ca.is_empty() {
            return Err(TargetError::Configuration(
                "skip_tls_verify and client_ca are mutually exclusive; remove client_ca or disable skip_tls_verify".to_string(),
            ));
        }

        Ok(())
    }
}

/// A target that sends events to a webhook
pub struct WebhookTarget<E>
where
    E: PluginEvent,
{
    id: TargetID,
    args: WebhookArgs,
    health_check_url: Option<Url>,
    http_client: Arc<Mutex<Client>>,
    tls_state: Arc<Mutex<TargetTlsState>>,
    /// When present, the adapter provides coordinator-managed TLS material;
    /// otherwise the inline fingerprint path is used as a fallback.
    tls_adapter: Option<TlsReloadAdapter<Client>>,
    // Add Send + Sync constraints to ensure thread safety
    store: Option<Box<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync>>,
    initialized: AtomicBool,
    cancel_sender: mpsc::Sender<()>,
    delivery_counters: Arc<TargetDeliveryCounters>,
    _phantom: PhantomData<E>,
}

impl<E> WebhookTarget<E>
where
    E: PluginEvent,
{
    /// Clones the WebhookTarget, creating a new instance with the same configuration
    pub fn clone_box(&self) -> Box<dyn Target<E> + Send + Sync> {
        Box::new(WebhookTarget::<E> {
            id: self.id.clone(),
            args: self.args.clone(),
            health_check_url: self.health_check_url.clone(),
            http_client: Arc::clone(&self.http_client),
            tls_state: Arc::clone(&self.tls_state),
            tls_adapter: self.tls_adapter.clone(),
            store: self.store.as_ref().map(|s| s.boxed_clone()),
            initialized: AtomicBool::new(self.initialized.load(Ordering::SeqCst)),
            cancel_sender: self.cancel_sender.clone(),
            delivery_counters: Arc::clone(&self.delivery_counters),
            _phantom: PhantomData,
        })
    }

    /// Creates a new WebhookTarget
    #[instrument(skip(args), fields(target_id = %id))]
    pub fn new(id: String, args: WebhookArgs) -> Result<Self, TargetError> {
        // First verify the parameters
        args.validate()?;
        // Create a TargetID
        let target_id = TargetID::new(id, ChannelTargetType::Webhook.as_str().to_string());
        let health_check_url = if args.enable {
            Some(Self::health_check_url(&args.endpoint)?)
        } else {
            None
        };

        // Build HTTP client using the helper function
        let http_client = Arc::new(Mutex::new(Self::build_http_client(&args)?));

        let queue_store = open_target_queue_store(
            &args.queue_dir,
            args.queue_limit,
            args.target_type,
            ChannelTargetType::Webhook.as_str(),
            &target_id,
            "Failed to open store for Webhook target",
        )?;

        // Create a cancel channel
        let (cancel_sender, _) = mpsc::channel(1);
        info!(
            event = EVENT_WEBHOOK_TARGET_STATE,
            component = LOG_COMPONENT_TARGETS,
            subsystem = LOG_SUBSYSTEM_WEBHOOK,
            target_id = %target_id.id,
            state = "created",
            "webhook target state"
        );
        Ok(WebhookTarget::<E> {
            id: target_id,
            args,
            health_check_url,
            http_client,
            tls_state: Arc::new(Mutex::new(TargetTlsState::default())),
            tls_adapter: None,
            store: queue_store,
            initialized: AtomicBool::new(false),
            cancel_sender,
            delivery_counters: Arc::new(TargetDeliveryCounters::default()),
            _phantom: PhantomData,
        })
    }

    fn build_http_client(args: &WebhookArgs) -> Result<Client, TargetError> {
        let mut client_builder = Client::builder()
            .timeout(Duration::from_secs(30))
            // SSRF hardening (backlog#974): never follow HTTP redirects on webhook delivery.
            // reqwest follows up to 10 redirects by default, which lets a malicious or
            // compromised endpoint use a 3xx response to bounce the outbound request to an
            // internal address (e.g. the cloud metadata service at 169.254.169.254),
            // bypassing the outbound-endpoint validation performed on the configured URL.
            .redirect(reqwest::redirect::Policy::none())
            .user_agent(crate::get_user_agent(crate::ServiceType::Basis));
        #[cfg(test)]
        {
            client_builder = client_builder.no_proxy();
        }

        // 1. Configure server certificate verification
        if args.skip_tls_verify {
            // DANGEROUS: For testing only, skip all certificate verification
            client_builder = client_builder.danger_accept_invalid_certs(true);
            warn!(
                event = EVENT_WEBHOOK_TARGET_STATE,
                component = LOG_COMPONENT_TARGETS,
                subsystem = LOG_SUBSYSTEM_WEBHOOK,
                endpoint = %args.endpoint,
                state = "tls_verification_skipped",
                fallback = "danger_accept_invalid_certs",
                "webhook target state"
            );
        } else if !args.client_ca.is_empty() {
            // Use user-provided custom CA certificate
            let certs_der = load_cert_bundle_der_bytes(&args.client_ca)
                .map_err(|e| TargetError::Configuration(format!("Failed to parse root CA cert: {e}")))?;
            if certs_der.is_empty() {
                return Err(TargetError::Configuration(
                    "Webhook client_ca did not contain any parsable certificates".to_string(),
                ));
            }
            for cert_der in certs_der {
                let ca_cert = reqwest::Certificate::from_der(&cert_der)
                    .map_err(|e| TargetError::Configuration(format!("Failed to load root CA cert: {e}")))?;
                client_builder = client_builder.add_root_certificate(ca_cert);
            }
        }
        // If neither is set, use the system's default trust store

        // 2. Configure client certificate (mTLS)
        if !args.client_cert.is_empty() && !args.client_key.is_empty() {
            let cert = std::fs::read(&args.client_cert)
                .map_err(|e| TargetError::Configuration(format!("Failed to read client cert: {e}")))?;
            let key = std::fs::read(&args.client_key)
                .map_err(|e| TargetError::Configuration(format!("Failed to read client key: {e}")))?;

            let identity = reqwest::Identity::from_pem(&[cert, key].concat())
                .map_err(|e| TargetError::Configuration(format!("Failed to create identity for mTLS: {e}")))?;
            client_builder = client_builder.identity(identity);
        }

        client_builder
            .build()
            .map_err(|e| TargetError::Configuration(format!("Failed to build HTTP client: {e}")))
    }

    async fn refresh_tls(&self) -> Result<(), TargetError> {
        let next_fingerprint =
            build_target_tls_fingerprint(&self.args.client_ca, &self.args.client_cert, &self.args.client_key).await?;
        let tls_changed = {
            let tls_state_guard = self.tls_state.lock();
            tls_state_guard.fingerprint.as_ref() != Some(&next_fingerprint)
        };
        if !tls_changed {
            return Ok(());
        }

        let new_client = Self::build_http_client(&self.args)?;
        {
            let mut tls_state_guard = self.tls_state.lock();
            if tls_state_guard.fingerprint.as_ref() == Some(&next_fingerprint) {
                return Ok(());
            }
            *self.http_client.lock() = new_client;
            tls_state_guard.refresh(next_fingerprint);
        }
        Ok(())
    }

    fn health_check_url(endpoint: &Url) -> Result<Url, TargetError> {
        endpoint
            .host()
            .ok_or_else(|| TargetError::Configuration(format!("Webhook endpoint '{}' is missing a host", endpoint)))?;
        let mut health_check_url = endpoint.clone();
        health_check_url.set_path("/");
        health_check_url.set_query(None);
        health_check_url.set_fragment(None);

        Ok(health_check_url)
    }

    async fn probe_reachability(&self) -> Result<bool, TargetError> {
        let Some(health_check_url) = self.health_check_url.as_ref() else {
            return Ok(false);
        };

        let client = self.http_client.lock().clone();
        match tokio::time::timeout(Duration::from_secs(5), client.head(health_check_url.as_str()).send()).await {
            Ok(Ok(resp)) => {
                debug!(
                    event = EVENT_WEBHOOK_TARGET_STATE,
                    component = LOG_COMPONENT_TARGETS,
                    subsystem = LOG_SUBSYSTEM_WEBHOOK,
                    target_id = %self.id,
                    status = %resp.status(),
                    health_check_url = %health_check_url,
                    state = "reachability_probe_succeeded",
                    "webhook target state"
                );
                Ok(true)
            }
            Ok(Err(err)) if err.is_timeout() => Err(TargetError::Timeout(format!(
                "Webhook health check request to {} timed out",
                health_check_url
            ))),
            Ok(Err(err)) if err.is_connect() => Ok(false),
            Ok(Err(err)) => Err(TargetError::Network(format!(
                "Webhook health check request to {} failed: {}",
                health_check_url, err
            ))),
            Err(_) => Err(TargetError::Timeout(format!(
                "Webhook health check request to {} timed out",
                health_check_url
            ))),
        }
    }

    async fn init_inner(&self) -> Result<(), TargetError> {
        if self.initialized.load(Ordering::SeqCst) {
            return Ok(());
        }

        if !self.args.enable {
            return Ok(());
        }

        // Use the configured reqwest client against the origin URL so proxy and TLS
        // behavior matches real delivery while avoiding path-specific false negatives.
        match self.probe_reachability().await {
            Ok(true) => {
                debug!(
                    event = EVENT_WEBHOOK_TARGET_STATE,
                    component = LOG_COMPONENT_TARGETS,
                    subsystem = LOG_SUBSYSTEM_WEBHOOK,
                    target_id = %self.id,
                    health_check_url = ?self.health_check_url,
                    state = "reachable",
                    "webhook target state"
                );
            }
            Ok(false) => {
                return Err(TargetError::NotConnected);
            }
            Err(err) => {
                return Err(err);
            }
        }

        self.initialized.store(true, Ordering::SeqCst);
        info!(
            event = EVENT_WEBHOOK_TARGET_STATE,
            component = LOG_COMPONENT_TARGETS,
            subsystem = LOG_SUBSYSTEM_WEBHOOK,
            target_id = %self.id,
            state = "initialized",
            "webhook target state"
        );
        Ok(())
    }

    fn build_queued_payload(&self, event: &EntityTarget<E>) -> Result<QueuedPayload, TargetError> {
        build_queued_payload(event)
    }

    async fn send_body(&self, body: Vec<u8>, meta: &QueuedPayloadMeta) -> Result<(), TargetError> {
        debug!(
            event = EVENT_WEBHOOK_DELIVERY_STATE,
            component = LOG_COMPONENT_TARGETS,
            subsystem = LOG_SUBSYSTEM_WEBHOOK,
            target_id = %self.id,
            bucket = %meta.bucket_name,
            object = %meta.object_name,
            payload_event = %meta.event_name,
            payload_len = body.len(),
            state = "sending",
            "webhook delivery state"
        );

        // When a TLS reload adapter is attached, it drives client rebuilds in
        // the background. The inline per-send fingerprint check is skipped.
        if self.tls_adapter.is_none() {
            self.refresh_tls().await?;
        }

        let client = self.http_client.lock().clone();
        let mut req_builder = client
            .post(self.args.endpoint.as_str())
            .header("Content-Type", meta.content_type.as_str());

        if !self.args.auth_token.is_empty() {
            // Split auth_token string to check if the authentication type is included
            match self.args.auth_token.split_whitespace().count() {
                2 => {
                    // Already include authentication type and token, such as "Bearer token123"
                    req_builder = req_builder.header("Authorization", &self.args.auth_token);
                }
                1 => {
                    // Only tokens, need to add "Bearer" prefix
                    req_builder = req_builder.header("Authorization", format!("Bearer {}", self.args.auth_token));
                }
                _ => {
                    // Empty string or other situations, no authentication header is added
                }
            }
        }

        // Send a request
        let resp = req_builder.body(body).send().await.map_err(|e| {
            if e.is_timeout() || e.is_connect() {
                TargetError::NotConnected
            } else {
                TargetError::Request(format!("Failed to send request: {e}"))
            }
        })?;

        let status = resp.status();
        // Drain the response body so the underlying connection is returned to the
        // pool and can be reused (keep-alive) instead of being closed mid-stream
        // (backlog#983). The body content is not needed for delivery accounting.
        let _ = resp.bytes().await;
        if status.is_success() {
            debug!(
                event = EVENT_WEBHOOK_DELIVERY_STATE,
                component = LOG_COMPONENT_TARGETS,
                subsystem = LOG_SUBSYSTEM_WEBHOOK,
                target_id = %self.id,
                status = %status,
                state = "sent",
                "webhook delivery state"
            );
            self.delivery_counters.record_success();
            Ok(())
        } else if status.is_redirection() {
            // SSRF hardening (backlog#974): redirects are intentionally not followed
            // (see build_http_client). Treat a 3xx response as a delivery failure rather
            // than silently chasing the redirect to a potentially internal target.
            Err(TargetError::Request(format!(
                "{} returned redirect '{}'; redirects are not followed for webhook delivery",
                self.args.endpoint, status
            )))
        } else if status == StatusCode::FORBIDDEN {
            Err(TargetError::Authentication(format!(
                "{} returned '{}', please check if your auth token is correctly set",
                self.args.endpoint, status
            )))
        } else {
            Err(TargetError::Request(format!(
                "{} returned '{}', please check your endpoint configuration",
                self.args.endpoint, status
            )))
        }
    }
}

#[async_trait]
impl<E> Target<E> for WebhookTarget<E>
where
    E: PluginEvent,
{
    fn id(&self) -> TargetID {
        self.id.clone()
    }

    async fn is_active(&self) -> Result<bool, TargetError> {
        if !self.args.enable {
            return Ok(false);
        }

        self.probe_reachability().await
    }

    async fn save(&self, event: Arc<EntityTarget<E>>) -> Result<(), TargetError> {
        let queued = match self.build_queued_payload(&event) {
            Ok(queued) => queued,
            Err(err) => {
                self.delivery_counters.record_final_failure();
                return Err(err);
            }
        };

        if let Some(store) = &self.store {
            if let Err(e) = persist_queued_payload_to_store(store.as_ref(), &queued) {
                self.delivery_counters.record_final_failure();
                return Err(e);
            }
            debug!(
                event = EVENT_WEBHOOK_DELIVERY_STATE,
                component = LOG_COMPONENT_TARGETS,
                subsystem = LOG_SUBSYSTEM_WEBHOOK,
                target_id = %self.id,
                state = "store_enqueued",
                "webhook delivery state"
            );
            Ok(())
        } else {
            match self.init().await {
                Ok(_) => (),
                Err(e) => {
                    error!(
                        event = EVENT_WEBHOOK_TARGET_STATE,
                        component = LOG_COMPONENT_TARGETS,
                        subsystem = LOG_SUBSYSTEM_WEBHOOK,
                        target_id = %self.id.id,
                        state = "init_failed",
                        error = %e,
                        "webhook target state"
                    );
                    self.delivery_counters.record_final_failure();
                    return Err(TargetError::NotConnected);
                }
            }
            if let Err(err) = self.send_body(queued.body, &queued.meta).await {
                self.delivery_counters.record_final_failure();
                return Err(err);
            }
            Ok(())
        }
    }

    async fn send_raw_from_store(&self, key: Key, body: Vec<u8>, meta: QueuedPayloadMeta) -> Result<(), TargetError> {
        debug!(
            event = EVENT_WEBHOOK_DELIVERY_STATE,
            component = LOG_COMPONENT_TARGETS,
            subsystem = LOG_SUBSYSTEM_WEBHOOK,
            target_id = %self.id,
            key = %key,
            state = "store_replay_started",
            "webhook delivery state"
        );
        match self.init().await {
            Ok(_) => {}
            Err(e) => {
                error!(
                    event = EVENT_WEBHOOK_TARGET_STATE,
                    component = LOG_COMPONENT_TARGETS,
                    subsystem = LOG_SUBSYSTEM_WEBHOOK,
                    target_id = %self.id.id,
                    state = "init_failed",
                    error = %e,
                    "webhook target state"
                );
                return Err(TargetError::NotConnected);
            }
        }

        if let Err(e) = self.send_body(body, &meta).await {
            if let TargetError::NotConnected = e {
                return Err(TargetError::NotConnected);
            }
            return Err(e);
        }

        debug!(
            event = EVENT_WEBHOOK_DELIVERY_STATE,
            component = LOG_COMPONENT_TARGETS,
            subsystem = LOG_SUBSYSTEM_WEBHOOK,
            target_id = %self.id,
            key = %key,
            state = "store_replay_sent",
            "webhook delivery state"
        );
        Ok(())
    }

    async fn close(&self) -> Result<(), TargetError> {
        // Send cancel signal to background tasks
        let _ = self.cancel_sender.try_send(());
        // Adapter cleanup is done by the coordinator; no local state to reset.
        info!(
            event = EVENT_WEBHOOK_TARGET_STATE,
            component = LOG_COMPONENT_TARGETS,
            subsystem = LOG_SUBSYSTEM_WEBHOOK,
            target_id = %self.id,
            state = "closed",
            "webhook target state"
        );
        Ok(())
    }

    fn store(&self) -> Option<&(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
        // Returns the reference to the internal store
        self.store.as_deref()
    }

    fn clone_dyn(&self) -> Box<dyn Target<E> + Send + Sync> {
        self.clone_box()
    }

    async fn init(&self) -> Result<(), TargetError> {
        if !self.is_enabled() {
            debug!(
                event = EVENT_WEBHOOK_TARGET_STATE,
                component = LOG_COMPONENT_TARGETS,
                subsystem = LOG_SUBSYSTEM_WEBHOOK,
                target_id = %self.id,
                state = "disabled",
                "webhook target state"
            );
            return Ok(());
        }
        self.init_inner().await
    }

    fn is_enabled(&self) -> bool {
        self.args.enable
    }

    fn delivery_snapshot(&self) -> TargetDeliverySnapshot {
        self.delivery_counters.snapshot(
            self.store.as_deref().map_or(0, |store| store.len() as u64),
            // Webhook targets record no terminal failures and keep no failed store.
            0,
        )
    }

    fn record_final_failure(&self) {
        self.delivery_counters.record_final_failure();
    }
}

/// Coordinated TLS hot-reload implementation for Webhook targets.
///
/// The coordinator calls these methods on a background poll loop to detect
/// TLS file changes and rebuild the HTTP client without restarting.
#[async_trait]
impl<E> ReloadableTargetTls for WebhookTarget<E>
where
    E: PluginEvent,
{
    type Material = Client;

    fn tls_input_set(&self) -> TargetTlsInputSet {
        TargetTlsInputSet {
            ca_path: self.args.client_ca.clone(),
            client_cert_path: self.args.client_cert.clone(),
            client_key_path: self.args.client_key.clone(),
            target_label: format!("webhook:{}", self.id.id),
        }
    }

    async fn build_tls_material(&self) -> Result<Self::Material, TargetError> {
        // build_http_client is synchronous (reads files + configures reqwest).
        // The coordinator already runs this in a background task, so the
        // synchronous file I/O does not block the send path.
        Self::build_http_client(&self.args)
    }

    async fn apply_tls_material(
        &self,
        _generation: TargetTlsGeneration,
        material: Arc<Self::Material>,
        _mode: ReloadApplyMode,
    ) -> Result<(), TargetError> {
        *self.http_client.lock() = (*material).clone();
        Ok(())
    }

    async fn validate_tls_files(&self) -> Result<(), TargetError> {
        validate_tls_material(&self.args.client_ca, &self.args.client_cert, &self.args.client_key)
    }
}

#[cfg(test)]
mod tests {
    use super::{WebhookArgs, WebhookTarget};
    use crate::target::{REDACTED_SECRET, Target, TargetType, decode_object_name};
    use url::Url;
    use url::form_urlencoded;

    fn base_args() -> WebhookArgs {
        WebhookArgs {
            enable: true,
            endpoint: Url::parse("https://example.com/hook").unwrap(),
            auth_token: String::new(),
            queue_dir: String::new(),
            queue_limit: 0,
            client_cert: String::new(),
            client_key: String::new(),
            client_ca: String::new(),
            skip_tls_verify: false,
            target_type: TargetType::NotifyEvent,
        }
    }

    #[test]
    fn debug_redacts_webhook_secret_fields() {
        let args = WebhookArgs {
            auth_token: "webhook-token".to_string(),
            client_key: "/etc/rustfs/webhook.key".to_string(),
            ..base_args()
        };

        let rendered = format!("{args:?}");

        assert!(!rendered.contains("webhook-token"));
        assert!(!rendered.contains("/etc/rustfs/webhook.key"));
        assert!(rendered.contains(REDACTED_SECRET));
        assert!(rendered.contains("WebhookArgs"));
    }

    #[test]
    fn test_validate_skip_tls_verify_and_client_ca_mutually_exclusive() {
        let args = WebhookArgs {
            skip_tls_verify: true,
            client_ca: "/path/to/ca.pem".to_string(),
            ..base_args()
        };
        let result = args.validate();
        assert!(result.is_err());
        let err_msg = result.unwrap_err().to_string();
        assert!(
            err_msg.contains("skip_tls_verify") && err_msg.contains("client_ca"),
            "Error message should mention both fields, got: {err_msg}"
        );
    }

    #[test]
    fn test_validate_skip_tls_verify_without_client_ca_is_ok() {
        let args = WebhookArgs {
            skip_tls_verify: true,
            ..base_args()
        };
        assert!(args.validate().is_ok());
    }

    #[test]
    fn test_validate_client_ca_without_skip_tls_verify_is_ok() {
        let args = WebhookArgs {
            client_ca: "/path/to/ca.pem".to_string(),
            ..base_args()
        };
        assert!(args.validate().is_ok());
    }

    #[test]
    fn test_validate_rejects_loopback_endpoint() {
        let args = WebhookArgs {
            endpoint: Url::parse("https://127.0.0.1/hook").expect("loopback endpoint should parse"),
            ..base_args()
        };
        let err = args.validate().expect_err("loopback endpoint should be rejected");
        assert!(err.to_string().contains("not allowed"));
    }

    #[test]
    fn test_decode_object_name_with_spaces() {
        // Test case from the issue: "greeting file (2).csv"
        let object_name = "greeting file (2).csv";

        // Simulate what event.rs does: form-urlencoded encoding (spaces become +)
        let form_encoded = form_urlencoded::byte_serialize(object_name.as_bytes()).collect::<String>();
        assert_eq!(form_encoded, "greeting+file+%282%29.csv");

        // Test the decode_object_name helper function
        let decoded = decode_object_name(&form_encoded).unwrap();
        assert_eq!(decoded, object_name);
        assert!(!decoded.contains('+'), "Decoded string should not contain + symbols");
    }

    #[test]
    fn test_decode_object_name_with_special_chars() {
        // Test with various special characters
        let test_cases = vec![
            ("folder/greeting file (2).csv", "folder%2Fgreeting+file+%282%29.csv"),
            ("test file.txt", "test+file.txt"),
            ("my file (copy).pdf", "my+file+%28copy%29.pdf"),
            ("file with spaces and (parentheses).doc", "file+with+spaces+and+%28parentheses%29.doc"),
        ];

        for (original, form_encoded) in test_cases {
            // Test the decode_object_name helper function
            let decoded = decode_object_name(form_encoded).unwrap();
            assert_eq!(decoded, original, "Failed to decode: {}", form_encoded);
        }
    }

    #[test]
    fn test_decode_object_name_without_spaces() {
        // Test that files without spaces still work correctly
        let object_name = "simple-file.txt";
        let form_encoded = form_urlencoded::byte_serialize(object_name.as_bytes()).collect::<String>();

        let decoded = decode_object_name(&form_encoded).unwrap();
        assert_eq!(decoded, object_name);
    }

    #[test]
    fn test_health_check_url_ignores_endpoint_path() {
        let endpoint = Url::parse("https://example.com:9443/hook/path").unwrap();
        let health_check_url = WebhookTarget::<serde_json::Value>::health_check_url(&endpoint).unwrap();

        assert_eq!(health_check_url.as_str(), "https://example.com:9443/");
    }

    #[tokio::test]
    async fn test_disabled_target_can_be_constructed_without_origin_probe() {
        let args = WebhookArgs {
            enable: false,
            endpoint: Url::parse("about:blank").unwrap(),
            ..base_args()
        };
        let target = WebhookTarget::<serde_json::Value>::new("disabled-target".to_string(), args).unwrap();

        assert!(!target.is_active().await.unwrap());
    }

    #[test]
    fn test_origin_reachability_probe_requires_non_local_endpoint() {
        let args = WebhookArgs {
            endpoint: Url::parse("http://127.0.0.1/hook").unwrap(),
            ..base_args()
        };
        let err = match WebhookTarget::<serde_json::Value>::new("path-probe".to_string(), args) {
            Ok(_) => panic!("loopback origin probes should now be rejected at construction time"),
            Err(err) => err,
        };
        assert!(err.to_string().contains("not allowed"));
    }

    // SSRF hardening regression (backlog#974): the delivery client must not follow HTTP
    // redirects, otherwise a 3xx from the endpoint could bounce the outbound request to an
    // internal address (e.g. the cloud metadata service) and bypass endpoint validation.
    #[tokio::test]
    async fn test_webhook_client_does_not_follow_redirects() {
        use std::io::{Read, Write};
        use std::net::TcpListener;

        // Minimal HTTP server on an ephemeral loopback port.
        let listener = TcpListener::bind("127.0.0.1:0").expect("bind mock server");
        let addr = listener.local_addr().expect("local addr");

        // Serve exactly one request with a 302 pointing at an internal metadata address.
        // If the client followed the redirect it would issue a second request to that
        // (unreachable) target instead of returning the 3xx status.
        let handle = std::thread::spawn(move || {
            if let Ok((mut stream, _)) = listener.accept() {
                let mut buf = [0u8; 1024];
                let _ = stream.read(&mut buf);
                let response =
                    "HTTP/1.1 302 Found\r\nLocation: http://169.254.169.254/latest/meta-data/\r\nContent-Length: 0\r\n\r\n";
                let _ = stream.write_all(response.as_bytes());
                let _ = stream.flush();
            }
        });

        let client = WebhookTarget::<serde_json::Value>::build_http_client(&base_args()).expect("build client");

        let resp = client
            .post(format!("http://{addr}/hook"))
            .body("{}")
            .send()
            .await
            .expect("request should complete without following the redirect");

        // Redirects are disabled, so the 3xx is surfaced as-is rather than chased to the
        // internal Location target.
        assert_eq!(resp.status().as_u16(), 302, "webhook client must not follow redirects");

        handle.join().expect("mock server thread");
    }
}