fix(security): add shared outbound egress guard (#3567)

* fix(security): block unsafe outbound webhook targets

* chore: keep issue-3557 plan local only

Cargo.lock

@@ -10100,6 +10100,7 @@ dependencies = [
  "rustfs-kafka-async",
  "rustfs-s3-types",
  "rustfs-tls-runtime",
+ "rustfs-utils",
  "rustls",
  "rustls-native-certs",
  "s3s",

crates/targets/Cargo.toml

@@ -16,6 +16,7 @@ rustfs-config = { workspace = true, features = ["notify", "constants", "audit",
 rustfs-extension-schema = { workspace = true }
 rustfs-tls-runtime = { workspace = true }
 rustfs-s3-types = { workspace = true }
+rustfs-utils = { workspace = true, features = ["egress"] }
 async-trait = { workspace = true }
 async-nats = { workspace = true }
 deadpool-postgres = { workspace = true }

crates/targets/src/config/common.rs

@@ -21,6 +21,7 @@ use rustfs_config::{
     NATS_TLS_CLIENT_CERT, NATS_TLS_CLIENT_KEY, NATS_TOKEN, NATS_USERNAME, PULSAR_AUTH_TOKEN, PULSAR_PASSWORD, PULSAR_QUEUE_DIR,
     PULSAR_TLS_ALLOW_INSECURE, PULSAR_TLS_CA, PULSAR_TLS_HOSTNAME_VERIFICATION, PULSAR_TOPIC, PULSAR_USERNAME,
 };
+use rustfs_utils::egress::validate_outbound_url;
 use std::collections::HashSet;
 use std::path::Path;
 use std::str::FromStr;
@@ -181,6 +182,10 @@ pub(super) fn parse_url(value: &str, field_label: &str) -> Result<Url, TargetErr
     Url::parse(value).map_err(|e| TargetError::Configuration(format!("Invalid {field_label}: {e} (value: '{value}')")))
 }
 
+pub(super) fn validate_outbound_http_url(value: &Url, field_label: &str) -> Result<(), TargetError> {
+    validate_outbound_url(value).map_err(|e| TargetError::Configuration(format!("{field_label} is not allowed: {e}")))
+}
+
 #[cfg(test)]
 mod tests {
     use super::{validate_nats_server_config, validate_pulsar_broker_config};

crates/targets/src/config/target_args.rs

@@ -12,7 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use super::common::{parse_target_bool, parse_url, validate_nats_server_config, validate_pulsar_broker_config};
+use super::common::{
+    parse_target_bool, parse_url, validate_nats_server_config, validate_outbound_http_url, validate_pulsar_broker_config,
+};
 use crate::error::TargetError;
 use crate::target::{
     TargetType,
@@ -137,6 +139,7 @@ pub fn build_webhook_args(config: &KVS, default_queue_dir: &str, target_type: Ta
         .ok_or_else(|| TargetError::Configuration("Missing webhook endpoint".to_string()))?;
     let parsed_endpoint = endpoint.trim();
     let endpoint_url = parse_url(parsed_endpoint, "endpoint URL")?;
+    validate_outbound_http_url(&endpoint_url, "endpoint URL")?;
 
     Ok(WebhookArgs {
         enable: true,
@@ -165,7 +168,8 @@ pub fn validate_webhook_config(config: &KVS, default_queue_dir: &str) -> Result<
         .lookup(WEBHOOK_ENDPOINT)
         .ok_or_else(|| TargetError::Configuration("Missing webhook endpoint".to_string()))?;
     let parsed_endpoint = endpoint.trim();
-    let _ = parse_url(parsed_endpoint, "endpoint URL")?;
+    let endpoint_url = parse_url(parsed_endpoint, "endpoint URL")?;
+    validate_outbound_http_url(&endpoint_url, "endpoint URL")?;
 
     let client_cert = config.lookup(WEBHOOK_CLIENT_CERT).unwrap_or_default();
     let client_key = config.lookup(WEBHOOK_CLIENT_KEY).unwrap_or_default();
@@ -594,8 +598,9 @@ pub fn validate_mysql_config(config: &KVS, default_queue_dir: &str) -> Result<()
 #[cfg(test)]
 mod tests {
     use super::{
-        build_amqp_args, build_kafka_args, build_mysql_args, build_postgres_args, build_redis_args, validate_amqp_config,
-        validate_kafka_config, validate_mysql_config, validate_postgres_config, validate_redis_config,
+        build_amqp_args, build_kafka_args, build_mysql_args, build_postgres_args, build_redis_args, build_webhook_args,
+        validate_amqp_config, validate_kafka_config, validate_mysql_config, validate_postgres_config, validate_redis_config,
+        validate_webhook_config,
     };
     use crate::target::{
         TargetType,
@@ -610,7 +615,7 @@ mod tests {
         MYSQL_QUEUE_DIR, MYSQL_TABLE, MYSQL_TLS_CA, MYSQL_TLS_CLIENT_CERT, MYSQL_TLS_CLIENT_KEY, POSTGRES_DSN_STRING,
         POSTGRES_FORMAT, POSTGRES_QUEUE_DIR, POSTGRES_TABLE, POSTGRES_TLS_CA, POSTGRES_TLS_CLIENT_CERT, POSTGRES_TLS_CLIENT_KEY,
         REDIS_CHANNEL, REDIS_CONNECTION_TIMEOUT, REDIS_MAX_RETRY_DELAY, REDIS_MIN_RETRY_DELAY, REDIS_PIPELINE_BUFFER_SIZE,
-        REDIS_RECONNECT_RETRY_ATTEMPTS, REDIS_RESPONSE_TIMEOUT, REDIS_TLS_ALLOW_INSECURE, REDIS_URL,
+        REDIS_RECONNECT_RETRY_ATTEMPTS, REDIS_RESPONSE_TIMEOUT, REDIS_TLS_ALLOW_INSECURE, REDIS_URL, WEBHOOK_ENDPOINT,
     };
 
     fn absolute_test_path(path: &str) -> String {
@@ -625,6 +630,12 @@ mod tests {
         config
     }
 
+    fn webhook_base_config() -> KVS {
+        let mut config = KVS::new();
+        config.insert(WEBHOOK_ENDPOINT.to_string(), "https://example.com/hook".to_string());
+        config
+    }
+
     fn kafka_base_config() -> KVS {
         let mut config = KVS::new();
         config.insert(KAFKA_BROKERS.to_string(), "127.0.0.1:9092".to_string());
@@ -759,6 +770,25 @@ mod tests {
         assert!(err.to_string().contains("either in url or username/password"));
     }
 
+    #[test]
+    fn build_webhook_args_rejects_loopback_endpoint() {
+        let mut config = webhook_base_config();
+        config.insert(WEBHOOK_ENDPOINT.to_string(), "https://127.0.0.1/hook".to_string());
+
+        let err = build_webhook_args(&config, "/tmp/webhook-queue", TargetType::NotifyEvent)
+            .expect_err("loopback endpoint should be rejected");
+        assert!(err.to_string().contains("not allowed"));
+    }
+
+    #[test]
+    fn validate_webhook_config_rejects_loopback_endpoint() {
+        let mut config = webhook_base_config();
+        config.insert(WEBHOOK_ENDPOINT.to_string(), "https://127.0.0.1/hook".to_string());
+
+        let err = validate_webhook_config(&config, "/tmp/webhook-queue").expect_err("loopback endpoint should be rejected");
+        assert!(err.to_string().contains("not allowed"));
+    }
+
     #[test]
     fn build_kafka_args_accepts_all_ack_alias() {
         let mut config = kafka_base_config();

crates/targets/src/target/webhook.rs

@@ -31,6 +31,7 @@ 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 serde::Serialize;
 use serde::de::DeserializeOwned;
 use std::{
@@ -102,6 +103,8 @@ impl WebhookArgs {
         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);
@@ -679,7 +682,6 @@ where
 mod tests {
     use super::{WebhookArgs, WebhookTarget};
     use crate::target::{REDACTED_SECRET, Target, TargetType, decode_object_name};
-    use tokio::net::TcpListener;
     use url::Url;
     use url::form_urlencoded;
 
@@ -748,6 +750,16 @@ mod tests {
         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"
@@ -810,50 +822,16 @@ mod tests {
         assert!(!target.is_active().await.unwrap());
     }
 
-    #[tokio::test]
-    async fn test_is_active_uses_origin_reachability_for_path_endpoints() {
-        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
-        let address = listener.local_addr().unwrap();
-        let server = async move {
-            use tokio::io::{AsyncReadExt, AsyncWriteExt};
-
-            let (mut stream, _) = listener.accept().await.unwrap();
-            let mut request = Vec::new();
-            let mut buf = [0u8; 1024];
-            loop {
-                let read = stream.read(&mut buf).await.unwrap();
-                if read == 0 {
-                    break;
-                }
-                request.extend_from_slice(&buf[..read]);
-                if request.windows(4).any(|window| window == b"\r\n\r\n") {
-                    break;
-                }
-            }
-
-            let request_line = request
-                .split(|byte| *byte == b'\n')
-                .next()
-                .and_then(|line| std::str::from_utf8(line).ok())
-                .unwrap_or_default()
-                .trim();
-            let path = request_line.split_whitespace().nth(1).unwrap_or_default().to_string();
-
-            if path == "/" {
-                let response = b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
-                let _ = stream.write_all(response).await;
-            }
-            path
-        };
-
+    #[test]
+    fn test_origin_reachability_probe_requires_non_local_endpoint() {
         let args = WebhookArgs {
-            endpoint: Url::parse(&format!("http://{address}/hook")).unwrap(),
+            endpoint: Url::parse("http://127.0.0.1/hook").unwrap(),
             ..base_args()
         };
-        let target = WebhookTarget::<serde_json::Value>::new("path-probe".to_string(), args).unwrap();
-
-        let (is_active, path) = tokio::join!(target.is_active(), server);
-        assert!(is_active.unwrap());
-        assert_eq!(path, "/");
+        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"));
     }
 }

crates/utils/Cargo.toml

@@ -72,6 +72,7 @@ workspace = true
 default = ["ip"]  # features that are enabled by default
 ip = ["dep:local-ip-address"]  # ip characteristics and their dependencies
 net = ["ip", "dep:url", "dep:netif", "dep:futures", "dep:transform-stream", "dep:bytes", "dep:hyper", "dep:tokio"]  # network features with DNS resolver
+egress = ["ip", "dep:url"]
 io = ["dep:tokio"]
 path = []  # path manipulation features
 compress = ["dep:flate2", "dep:brotli", "dep:snap", "dep:lz4", "dep:zstd"]
@@ -82,4 +83,4 @@ os = ["dep:rustix", "dep:tempfile", "dep:windows"]  # operating system utilities
 integration = []  # integration test features
 http = ["dep:convert_case", "dep:http", "dep:regex"]
 obj = ["http"] # object storage features
-full = ["ip", "net", "io", "hash", "os", "integration", "path", "crypto", "string", "compress", "http", "obj"]  # all features
+full = ["ip", "net", "egress", "io", "hash", "os", "integration", "path", "crypto", "string", "compress", "http", "obj"]  # all features

crates/utils/src/egress.rs

@@ -0,0 +1,173 @@
+// 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 std::fmt;
+use std::net::{IpAddr, Ipv4Addr};
+use url::Url;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum OutboundUrlError {
+    MissingHost,
+    ForbiddenHost { host: String, reason: &'static str },
+}
+
+impl fmt::Display for OutboundUrlError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            OutboundUrlError::MissingHost => write!(f, "outbound URL is missing a host"),
+            OutboundUrlError::ForbiddenHost { host, reason } => {
+                write!(f, "outbound URL host '{host}' is not allowed: {reason}")
+            }
+        }
+    }
+}
+
+impl std::error::Error for OutboundUrlError {}
+
+pub fn validate_outbound_url(url: &Url) -> Result<(), OutboundUrlError> {
+    let Some(raw_host) = url.host_str() else {
+        return Err(OutboundUrlError::MissingHost);
+    };
+    let normalized_host = raw_host.trim_end_matches('.').trim_matches(['[', ']']);
+
+    if normalized_host.eq_ignore_ascii_case("localhost") {
+        return Err(OutboundUrlError::ForbiddenHost {
+            host: raw_host.to_string(),
+            reason: "loopback host",
+        });
+    }
+
+    let Ok(ip) = normalized_host.parse::<IpAddr>() else {
+        return Ok(());
+    };
+
+    validate_outbound_ip(ip).map_err(|reason| OutboundUrlError::ForbiddenHost {
+        host: raw_host.to_string(),
+        reason,
+    })
+}
+
+fn validate_outbound_ip(ip: IpAddr) -> Result<(), &'static str> {
+    if ip.is_unspecified() {
+        return Err("unspecified address");
+    }
+
+    if ip == IpAddr::V4(Ipv4Addr::new(169, 254, 169, 254)) {
+        return Err("metadata endpoint");
+    }
+
+    match ip {
+        IpAddr::V4(ipv4) => {
+            if ipv4.is_loopback() {
+                return Err("loopback address");
+            }
+            if ipv4.is_link_local() {
+                return Err("link-local address");
+            }
+            if ipv4.is_private() {
+                return Err("private address");
+            }
+        }
+        IpAddr::V6(ipv6) => {
+            if ipv6.is_loopback() {
+                return Err("loopback address");
+            }
+            if ipv6.is_unicast_link_local() {
+                return Err("link-local address");
+            }
+            if ipv6.is_unique_local() {
+                return Err("private address");
+            }
+        }
+    }
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{OutboundUrlError, validate_outbound_url};
+    use url::Url;
+
+    #[test]
+    fn validate_outbound_url_allows_public_hostname() {
+        let url = Url::parse("https://example.com/webhook").expect("public URL should parse");
+        assert!(validate_outbound_url(&url).is_ok());
+    }
+
+    #[test]
+    fn validate_outbound_url_rejects_localhost() {
+        let url = Url::parse("https://localhost/webhook").expect("localhost URL should parse");
+        let err = validate_outbound_url(&url).expect_err("localhost should be rejected");
+        assert!(matches!(
+            err,
+            OutboundUrlError::ForbiddenHost {
+                reason: "loopback host",
+                ..
+            }
+        ));
+    }
+
+    #[test]
+    fn validate_outbound_url_rejects_loopback_ip() {
+        let url = Url::parse("https://127.0.0.1/webhook").expect("loopback URL should parse");
+        let err = validate_outbound_url(&url).expect_err("loopback IP should be rejected");
+        assert!(matches!(
+            err,
+            OutboundUrlError::ForbiddenHost {
+                reason: "loopback address",
+                ..
+            }
+        ));
+    }
+
+    #[test]
+    fn validate_outbound_url_rejects_private_ip() {
+        let url = Url::parse("https://10.0.0.5/webhook").expect("private URL should parse");
+        let err = validate_outbound_url(&url).expect_err("private IP should be rejected");
+        assert!(matches!(
+            err,
+            OutboundUrlError::ForbiddenHost {
+                reason: "private address",
+                ..
+            }
+        ));
+    }
+
+    #[test]
+    fn validate_outbound_url_rejects_metadata_endpoint() {
+        let url = Url::parse("http://169.254.169.254/latest/meta-data").expect("metadata URL should parse");
+        let err = validate_outbound_url(&url).expect_err("metadata endpoint should be rejected");
+        assert!(matches!(
+            err,
+            OutboundUrlError::ForbiddenHost {
+                reason: "metadata endpoint",
+                ..
+            }
+        ));
+    }
+
+    #[test]
+    fn validate_outbound_url_rejects_link_local_ipv6() {
+        let url = Url::parse("https://[fe80::1]/hook").expect("IPv6 URL should parse");
+        let err = validate_outbound_url(&url).expect_err("link-local IPv6 should be rejected");
+        assert!(matches!(
+            err,
+            OutboundUrlError::ForbiddenHost {
+                reason: "link-local address",
+                ..
+            }
+        ));
+    }
+}

crates/utils/src/lib.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#[cfg(feature = "egress")]
+pub mod egress;
 #[cfg(feature = "ip")]
 pub mod ip;
 #[cfg(feature = "net")]
@@ -20,6 +22,8 @@ pub mod net;
 #[cfg(feature = "http")]
 pub mod http;
 
+#[cfg(feature = "egress")]
+pub use egress::*;
 #[cfg(feature = "net")]
 pub use net::*;
 

rustfs/src/admin/handlers/oidc.rs

@@ -32,6 +32,7 @@ use rustfs_config::server_config::get_global_server_config;
 use rustfs_config::{DEFAULT_DELIMITER, ENABLE_KEY, EnableState, MAX_ADMIN_REQUEST_BODY_SIZE};
 use rustfs_ecstore::config::com::{read_config_without_migrate, save_server_config};
 use rustfs_policy::policy::action::{Action, AdminAction};
+use rustfs_utils::egress::validate_outbound_url;
 use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
 use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
@@ -869,6 +870,8 @@ fn validate_absolute_http_url(value: &str, field_name: &str) -> S3Result<()> {
         return Err(s3_error!(InvalidRequest, "{} must be an absolute http/https URL", field_name));
     }
 
+    validate_outbound_url(&parsed).map_err(|err| s3_error!(InvalidRequest, "{} is not allowed: {}", field_name, err))?;
+
     Ok(())
 }
 
@@ -1253,6 +1256,14 @@ mod tests {
         assert!(!is_valid_scheme(""));
     }
 
+    #[test]
+    fn test_validate_absolute_http_url_rejects_loopback_targets() {
+        let err = validate_absolute_http_url("https://127.0.0.1/.well-known/openid-configuration", "config_url")
+            .expect_err("loopback config URL should be rejected");
+        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
+        assert!(err.message().unwrap_or_default().contains("not allowed"));
+    }
+
     #[test]
     fn test_provider_instance_key() {
         assert_eq!(provider_instance_key("default"), "_");

rustfs/src/admin/router.rs

@@ -69,6 +69,7 @@ use rustfs_policy::policy::action::{Action, S3Action};
 use rustfs_s3_types::EventName;
 use rustfs_signer::pre_sign_v4;
 use rustfs_storage_api::{BucketOperations, BucketOptions};
+use rustfs_utils::egress::validate_outbound_url;
 use rustfs_utils::http::{
     SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_CHECK, SUFFIX_SOURCE_REPLICATION_REQUEST,
     SUFFIX_SOURCE_VERSION_ID, get_source_scheme, insert_header,
@@ -613,8 +614,13 @@ fn resolve_object_lambda_webhook_config_from_server_config(
         None => None,
     };
 
+    let parsed_endpoint =
+        Url::parse(&endpoint).map_err(|_| s3_error!(InvalidRequest, "object lambda target endpoint is invalid"))?;
+    validate_outbound_url(&parsed_endpoint)
+        .map_err(|err| s3_error!(InvalidRequest, "object lambda target endpoint is not allowed: {}", err))?;
+
     Ok(ObjectLambdaWebhookConfig {
-        endpoint: Url::parse(&endpoint).map_err(|_| s3_error!(InvalidRequest, "object lambda target endpoint is invalid"))?,
+        endpoint: parsed_endpoint,
         auth_token: kvs.lookup(WEBHOOK_AUTH_TOKEN).unwrap_or_default(),
         client_cert: kvs.lookup(WEBHOOK_CLIENT_CERT).unwrap_or_default(),
         client_key: kvs.lookup(WEBHOOK_CLIENT_KEY).unwrap_or_default(),
@@ -656,6 +662,8 @@ async fn resolve_object_lambda_webhook_config(uri: &Uri) -> S3Result<ObjectLambd
 }
 
 fn build_object_lambda_http_client(config: &ObjectLambdaWebhookConfig) -> S3Result<reqwest::Client> {
+    validate_outbound_url(&config.endpoint)
+        .map_err(|err| s3_error!(InvalidRequest, "object lambda target endpoint is not allowed: {}", err))?;
     let mut builder = reqwest::Client::builder().user_agent(rustfs_targets::get_user_agent(rustfs_targets::ServiceType::Basis));
 
     if let Some(timeout) = config.response_header_timeout {
@@ -3553,6 +3561,36 @@ mod tests {
         assert_eq!(disabled_err.code(), &S3ErrorCode::InvalidRequest);
     }
 
+    #[test]
+    fn resolve_object_lambda_webhook_config_from_server_config_rejects_loopback_endpoint() {
+        let arn = "arn:acme:s3-object-lambda::transformer:webhook"
+            .parse::<rustfs_targets::arn::ARN>()
+            .expect("arn should parse");
+        let config = rustfs_config::server_config::Config(std::collections::HashMap::from([(
+            LAMBDA_WEBHOOK_SUB_SYS.to_string(),
+            std::collections::HashMap::from([(
+                "transformer".to_string(),
+                rustfs_config::server_config::KVS(vec![
+                    rustfs_config::server_config::KV {
+                        key: ENABLE_KEY.to_string(),
+                        value: "on".to_string(),
+                        hidden_if_empty: false,
+                    },
+                    rustfs_config::server_config::KV {
+                        key: WEBHOOK_ENDPOINT.to_string(),
+                        value: "https://127.0.0.1/transform".to_string(),
+                        hidden_if_empty: false,
+                    },
+                ]),
+            )]),
+        )]));
+
+        let err = resolve_object_lambda_webhook_config_from_server_config(&config, &arn)
+            .expect_err("loopback endpoint should be rejected");
+        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
+        assert!(err.message().unwrap_or_default().contains("not allowed"));
+    }
+
     #[test]
     fn clear_object_lambda_variant_headers_removes_original_object_payload_headers() {
         let mut headers = HeaderMap::new();