diff --git a/Cargo.lock b/Cargo.lock index 645cf919d..fcaf4384b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10100,6 +10100,7 @@ dependencies = [ "rustfs-kafka-async", "rustfs-s3-types", "rustfs-tls-runtime", + "rustfs-utils", "rustls", "rustls-native-certs", "s3s", diff --git a/crates/targets/Cargo.toml b/crates/targets/Cargo.toml index 7cad1657c..0cc2000e8 100644 --- a/crates/targets/Cargo.toml +++ b/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 } diff --git a/crates/targets/src/config/common.rs b/crates/targets/src/config/common.rs index 4de486f92..7f9473e34 100644 --- a/crates/targets/src/config/common.rs +++ b/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 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}; diff --git a/crates/targets/src/config/target_args.rs b/crates/targets/src/config/target_args.rs index 564a766ce..f32741e0e 100644 --- a/crates/targets/src/config/target_args.rs +++ b/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(); diff --git a/crates/targets/src/target/webhook.rs b/crates/targets/src/target/webhook.rs index bd0a67415..e6d1217ca 100644 --- a/crates/targets/src/target/webhook.rs +++ b/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::::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::::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")); } } diff --git a/crates/utils/Cargo.toml b/crates/utils/Cargo.toml index 0d06966a5..e9198d1da 100644 --- a/crates/utils/Cargo.toml +++ b/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 diff --git a/crates/utils/src/egress.rs b/crates/utils/src/egress.rs new file mode 100644 index 000000000..495de1797 --- /dev/null +++ b/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::() 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", + .. + } + )); + } +} diff --git a/crates/utils/src/lib.rs b/crates/utils/src/lib.rs index 88d138efa..2ac6630fd 100644 --- a/crates/utils/src/lib.rs +++ b/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::*; diff --git a/rustfs/src/admin/handlers/oidc.rs b/rustfs/src/admin/handlers/oidc.rs index 847e77421..b7a230985 100644 --- a/rustfs/src/admin/handlers/oidc.rs +++ b/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"), "_"); diff --git a/rustfs/src/admin/router.rs b/rustfs/src/admin/router.rs index 73b8151df..89ffc4886 100644 --- a/rustfs/src/admin/router.rs +++ b/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 S3Result { + 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::() + .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();