diff --git a/Cargo.lock b/Cargo.lock index ca565d24c..309c09f0d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3685,6 +3685,7 @@ dependencies = [ "rustfs-protos", "rustfs-rio", "rustfs-signer", + "rustfs-utils", "rustls", "s3s", "serde", @@ -10097,6 +10098,7 @@ dependencies = [ "netif", "proptest", "regex", + "reqwest", "rustix", "serde", "sha1 0.11.0", diff --git a/crates/e2e_test/Cargo.toml b/crates/e2e_test/Cargo.toml index 6a91afc8a..3b5e83b83 100644 --- a/crates/e2e_test/Cargo.toml +++ b/crates/e2e_test/Cargo.toml @@ -33,6 +33,7 @@ rustfs-config = { workspace = true, features = ["constants"] } rustfs-ecstore.workspace = true rustfs-data-usage.workspace = true rustfs-rio.workspace = true +rustfs-utils = { workspace = true, features = ["egress"] } flatbuffers.workspace = true futures.workspace = true rustfs-lock.workspace = true diff --git a/crates/e2e_test/src/notification_webhook_test.rs b/crates/e2e_test/src/notification_webhook_test.rs index 74c6fc093..07f4b4f3c 100644 --- a/crates/e2e_test/src/notification_webhook_test.rs +++ b/crates/e2e_test/src/notification_webhook_test.rs @@ -39,6 +39,7 @@ use local_ip_address::local_ip; use reqwest::StatusCode; use rustfs_signer::constants::UNSIGNED_PAYLOAD; use rustfs_signer::sign_v4; +use rustfs_utils::egress::ENV_OUTBOUND_ALLOW_ORIGINS; use s3s::Body; use serde_json::Value; use serial_test::serial; @@ -70,6 +71,11 @@ fn target_arn(target_name: &str) -> String { format!("arn:rustfs:sqs:{NOTIFY_REGION}:{target_name}:webhook") } +fn endpoint_origin(endpoint: &str) -> Result { + let parsed = reqwest::Url::parse(endpoint)?; + Ok(parsed.origin().ascii_serialization()) +} + // --------------------------------------------------------------------------- // In-test HTTP event receiver // --------------------------------------------------------------------------- @@ -153,7 +159,7 @@ async fn spawn_event_collector() -> Result<(String, mpsc::UnboundedReceiver Result Result T init_logging(); let mut env = RustFSTestEnvironment::new().await?; - env.start_rustfs_server_with_env(vec![], &[("RUSTFS_NOTIFY_ENABLE", "true")]) - .await?; - let ca_path = Path::new(&env.temp_dir).join("https-webhook-ca.pem"); let mut collector = spawn_https_event_collector(&ca_path)?; + let allowed_origin = endpoint_origin(collector.endpoint())?; + env.start_rustfs_server_with_env( + vec![], + &[ + ("RUSTFS_NOTIFY_ENABLE", "true"), + (ENV_OUTBOUND_ALLOW_ORIGINS, allowed_origin.as_str()), + ], + ) + .await?; let target = "peri1https"; let bucket = "peri1-https-events"; let key = "uploads/https.dat"; @@ -635,9 +647,11 @@ async fn test_webhook_event_delivery_and_filtering() -> TestResult { init_logging(); let (endpoint, mut rx, handle) = spawn_event_collector().await?; + let allowed_origin = endpoint_origin(&endpoint)?; let mut env = RustFSTestEnvironment::new().await?; - env.start_rustfs_server(vec![]).await?; + env.start_rustfs_server_with_env(vec![], &[(ENV_OUTBOUND_ALLOW_ORIGINS, allowed_origin.as_str())]) + .await?; enable_notify_module(&env).await?; let bucket = "peri1-events"; @@ -789,15 +803,6 @@ async fn test_webhook_event_delivery_and_filtering() -> TestResult { async fn test_webhook_redelivers_event_after_target_recovers() -> TestResult { init_logging(); - let mut env = RustFSTestEnvironment::new().await?; - env.start_rustfs_server(vec![]).await?; - enable_notify_module(&env).await?; - - let bucket = "peri1-redeliver"; - let target = "peri1redeliver"; - let client = env.create_s3_client(); - client.create_bucket().bucket(bucket).send().await?; - // Configure the target while its endpoint is reachable so activation and // ARN registration complete deterministically. Registering against a dead // endpoint stalls behind the reachability probe's timeout and flakes the @@ -806,10 +811,21 @@ async fn test_webhook_redelivers_event_after_target_recovers() -> TestResult { let listener = TcpListener::bind("0.0.0.0:0").await?; let port = listener.local_addr()?.port(); let endpoint_ip = local_ip()?; - let endpoint = format!("http://{endpoint_ip}.nip.io:{port}/events"); + let endpoint = format!("http://{}/events", std::net::SocketAddr::new(endpoint_ip, port)); + let allowed_origin = endpoint_origin(&endpoint)?; let (setup_tx, _setup_rx) = mpsc::unbounded_channel(); let setup_handle = serve_event_collector(listener, setup_tx); + let mut env = RustFSTestEnvironment::new().await?; + env.start_rustfs_server_with_env(vec![], &[(ENV_OUTBOUND_ALLOW_ORIGINS, allowed_origin.as_str())]) + .await?; + enable_notify_module(&env).await?; + + let bucket = "peri1-redeliver"; + let target = "peri1redeliver"; + let client = env.create_s3_client(); + client.create_bucket().bucket(bucket).send().await?; + configure_webhook_target(&env, target, &endpoint).await?; wait_for_target_registered(&env, target).await?; put_notification_config(&client, bucket, target, "uploads/", ".dat").await?; diff --git a/crates/e2e_test/src/object_lambda_test.rs b/crates/e2e_test/src/object_lambda_test.rs index 897a4636b..69d2e46ef 100644 --- a/crates/e2e_test/src/object_lambda_test.rs +++ b/crates/e2e_test/src/object_lambda_test.rs @@ -18,6 +18,7 @@ use http::header::{CONTENT_TYPE, HOST}; use reqwest::StatusCode; use rustfs_signer::constants::UNSIGNED_PAYLOAD; use rustfs_signer::{pre_sign_v4, sign_v4}; +use rustfs_utils::egress::ENV_OUTBOUND_ALLOW_ORIGINS; use s3s::Body; use serial_test::serial; use std::collections::HashMap; @@ -49,7 +50,7 @@ fn find_header_terminator(buf: &[u8]) -> Option { async fn read_http_request( stream: &mut tokio::net::TcpStream, -) -> Result<(HashMap, Vec), Box> { +) -> Result<(String, HashMap, Vec), Box> { let mut buffer = Vec::new(); let mut chunk = [0_u8; 4096]; @@ -67,7 +68,7 @@ async fn read_http_request( let header_bytes = &buffer[..header_end]; let header_text = std::str::from_utf8(header_bytes)?; let mut lines = header_text.split("\r\n"); - let _request_line = lines.next().ok_or("missing request line")?; + let request_line = lines.next().ok_or("missing request line")?.to_string(); let mut headers = HashMap::new(); for line in lines { if line.is_empty() { @@ -79,8 +80,9 @@ async fn read_http_request( let content_length = headers .get("content-length") - .ok_or("missing content-length header")? - .parse::()?; + .map(|value| value.parse::()) + .transpose()? + .unwrap_or_default(); let body_offset = header_end + 4; while buffer.len().saturating_sub(body_offset) < content_length { let read = stream.read(&mut chunk).await?; @@ -90,7 +92,7 @@ async fn read_http_request( buffer.extend_from_slice(&chunk[..read]); } - Ok((headers, buffer[body_offset..body_offset + content_length].to_vec())) + Ok((request_line, headers, buffer[body_offset..body_offset + content_length].to_vec())) } async fn spawn_object_lambda_webhook_server() -> Result< @@ -130,9 +132,17 @@ async fn spawn_object_lambda_webhook_server_with_response( let handle = tokio::spawn(async move { loop { let (mut stream, _) = listener.accept().await?; - let Ok(Ok((headers, body))) = timeout(Duration::from_secs(2), read_http_request(&mut stream)).await else { + let Ok(Ok((request_line, headers, body))) = timeout(Duration::from_secs(2), read_http_request(&mut stream)).await + else { continue; }; + if request_line == "HEAD / HTTP/1.1" { + stream + .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\nconnection: close\r\n\r\n") + .await?; + stream.shutdown().await?; + continue; + } let payload: serde_json::Value = serde_json::from_slice(&body)?; let output_route = payload["getObjectContext"]["outputRoute"] @@ -412,9 +422,33 @@ async fn wait_for_target_absence( Err(format!("target {target_name} remained visible in admin APIs; targets={last_targets}, arns={last_arns:?}").into()) } -async fn restart_rustfs_server(env: &mut RustFSTestEnvironment) -> Result<(), Box> { +fn endpoint_origin(endpoint: &str) -> Result> { + Ok(reqwest::Url::parse(endpoint)?.origin().ascii_serialization()) +} + +async fn start_rustfs_server_for_endpoint( + env: &mut RustFSTestEnvironment, + endpoint: &str, +) -> Result<(), Box> { + let origin = endpoint_origin(endpoint)?; + env.start_rustfs_server_with_env( + vec![], + &[ + (ENV_OUTBOUND_ALLOW_ORIGINS, origin.as_str()), + ("RUSTFS_NOTIFY_ENABLE", "true"), + ], + ) + .await +} + +async fn restart_rustfs_server(env: &mut RustFSTestEnvironment, endpoint: &str) -> Result<(), Box> { + let origin = endpoint_origin(endpoint)?; env.stop_server(); - env.start_rustfs_server_without_cleanup(vec![]).await + env.start_rustfs_server_without_cleanup_with_env(&[ + (ENV_OUTBOUND_ALLOW_ORIGINS, origin.as_str()), + ("RUSTFS_NOTIFY_ENABLE", "true"), + ]) + .await } async fn spawn_http_origin_probe_server() -> Result< @@ -521,7 +555,7 @@ async fn test_notification_target_persists_across_restart_and_delete() -> Result let (webhook_url, _request_rx, webhook_handle) = spawn_object_lambda_webhook_server().await?; let mut env = RustFSTestEnvironment::new().await?; - env.start_rustfs_server(vec![]).await?; + start_rustfs_server_for_endpoint(&mut env, &webhook_url).await?; let target_name = "restart-target"; configure_webhook_target(&env, target_name, &webhook_url, "secret-token").await?; @@ -535,7 +569,7 @@ async fn test_notification_target_persists_across_restart_and_delete() -> Result "target ARN missing after initial configure: {visible_arns:?}" ); - restart_rustfs_server(&mut env).await?; + restart_rustfs_server(&mut env, &webhook_url).await?; let (targets_after_restart, arns_after_restart) = wait_for_target_visibility(&env, target_name).await?; assert!(notification_target_is_listed(&targets_after_restart, target_name)); @@ -556,7 +590,7 @@ async fn test_notification_target_persists_across_restart_and_delete() -> Result "target ARN still visible after delete: {arns_after_delete:?}" ); - restart_rustfs_server(&mut env).await?; + restart_rustfs_server(&mut env, &webhook_url).await?; let (targets_after_delete_restart, arns_after_delete_restart) = wait_for_target_absence(&env, target_name).await?; assert!(!notification_target_is_listed(&targets_after_delete_restart, target_name)); @@ -581,8 +615,7 @@ async fn test_notification_target_with_path_is_online_via_transport_probe() -> R let (webhook_url, mut probe_rx, probe_handle) = spawn_http_origin_probe_server().await?; let mut env = RustFSTestEnvironment::new().await?; - env.start_rustfs_server_with_env(vec![], &[("RUSTFS_NOTIFY_ENABLE", "true")]) - .await?; + start_rustfs_server_for_endpoint(&mut env, &webhook_url).await?; let target_name = "path-probe"; configure_webhook_target(&env, target_name, &webhook_url, "secret-token").await?; @@ -615,7 +648,7 @@ async fn test_get_object_lambda_accepts_presigned_requests() -> Result<(), Box Result<(), let (webhook_url, request_rx, webhook_handle) = spawn_object_lambda_webhook_server().await?; let mut env = RustFSTestEnvironment::new().await?; - env.start_rustfs_server(vec![]).await?; + start_rustfs_server_for_endpoint(&mut env, &webhook_url).await?; let bucket = "object-lambda-e2e-named-target"; let key = "input.txt"; @@ -696,7 +729,7 @@ async fn test_get_object_lambda_invokes_runtime_webhook_target() -> Result<(), B let (webhook_url, request_rx, webhook_handle) = spawn_object_lambda_webhook_server().await?; let mut env = RustFSTestEnvironment::new().await?; - env.start_rustfs_server(vec![]).await?; + start_rustfs_server_for_endpoint(&mut env, &webhook_url).await?; let bucket = "object-lambda-e2e"; let key = "input.txt"; @@ -777,7 +810,7 @@ async fn test_get_object_lambda_passthroughs_non_success_webhook_response() -> R .await?; let mut env = RustFSTestEnvironment::new().await?; - env.start_rustfs_server(vec![]).await?; + start_rustfs_server_for_endpoint(&mut env, &webhook_url).await?; let bucket = "object-lambda-e2e-failure"; let key = "input.txt"; @@ -832,7 +865,7 @@ async fn test_get_object_lambda_rejects_success_response_without_auth_headers() .await?; let mut env = RustFSTestEnvironment::new().await?; - env.start_rustfs_server(vec![]).await?; + start_rustfs_server_for_endpoint(&mut env, &webhook_url).await?; let bucket = "object-lambda-e2e-missing-auth"; let key = "input.txt"; @@ -879,7 +912,7 @@ async fn test_get_object_lambda_rejects_success_response_with_mismatched_auth_he .await?; let mut env = RustFSTestEnvironment::new().await?; - env.start_rustfs_server(vec![]).await?; + start_rustfs_server_for_endpoint(&mut env, &webhook_url).await?; let bucket = "object-lambda-e2e-mismatched-auth"; let key = "input.txt"; diff --git a/crates/targets/src/config/common.rs b/crates/targets/src/config/common.rs index 4bbbf32e6..12e622352 100644 --- a/crates/targets/src/config/common.rs +++ b/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::validate_outbound_url; +use rustfs_utils::egress::OutboundPolicy; use std::collections::HashSet; use std::path::Path; use std::str::FromStr; @@ -215,16 +215,20 @@ pub(super) fn validate_pulsar_broker_config(broker: &str, config: &KVS, default_ } pub(super) fn parse_url(value: &str, field_label: &str) -> Result { - Url::parse(value).map_err(|e| TargetError::Configuration(format!("Invalid {field_label}: {e} (value: '{value}')"))) + Url::parse(value).map_err(|e| TargetError::Configuration(format!("Invalid {field_label}: {e}"))) } 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}"))) + let policy = + 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}"))) } #[cfg(test)] mod tests { - use super::{parse_jetstream_enable, validate_nats_server_config, validate_pulsar_broker_config}; + use super::{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::{ @@ -238,6 +242,12 @@ mod tests { ServerAddr::from_str("nats://127.0.0.1:4222").expect("valid nats address") } + #[test] + fn parse_url_error_does_not_echo_the_configured_value() { + let err = parse_url("not a URL containing secret-token", "endpoint URL").expect_err("invalid URL should fail"); + assert!(!err.to_string().contains("secret-token")); + } + // 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 { diff --git a/crates/targets/src/config/loader.rs b/crates/targets/src/config/loader.rs index 89d24a5dd..add0bc73d 100644 --- a/crates/targets/src/config/loader.rs +++ b/crates/targets/src/config/loader.rs @@ -84,7 +84,23 @@ fn redact_target_field_value(field_name: &str, value: &str) -> String { return crate::target::mysql::redact_mysql_dsn(value); } if is_sensitive_target_field(field_name) { - return "***redacted***".to_string(); + return crate::target::REDACTED_SECRET.to_string(); + } + if field_name.eq_ignore_ascii_case(rustfs_config::WEBHOOK_ENDPOINT) + || field_name.eq_ignore_ascii_case(rustfs_config::AMQP_URL) + { + return url::Url::parse(value) + .ok() + .and_then(|endpoint| { + let host = match endpoint.host()? { + url::Host::Domain(host) => host.to_string(), + url::Host::Ipv4(host) => host.to_string(), + url::Host::Ipv6(host) => format!("[{host}]"), + }; + let port = endpoint.port().map(|port| format!(":{port}")).unwrap_or_default(); + Some(format!("{}://{host}{port}", endpoint.scheme())) + }) + .unwrap_or_else(|| crate::target::REDACTED_SECRET.to_string()); } value.to_string() } @@ -632,6 +648,24 @@ mod tests { assert_eq!(redact_target_field_value("queue_limit", "1000"), "1000"); } + #[test] + fn redact_target_field_value_strips_endpoint_path_and_query() { + assert_eq!( + redact_target_field_value("endpoint", "https://example.com/private/hook?token=secret"), + "https://example.com" + ); + assert_eq!(redact_target_field_value("endpoint", "not a URL with secret"), "***redacted***"); + } + + #[test] + fn redact_target_field_value_strips_url_credentials() { + assert_eq!( + redact_target_field_value("url", "amqps://user:secret@broker.example/vhost"), + "amqps://broker.example" + ); + assert_eq!(redact_target_field_value("url", "not a URL with secret"), "***redacted***"); + } + #[test] fn redact_dsn_string_partial_redaction() { let dsn = "rustfs:secret123@tcp(mysql.example.com:3306)/rustfs_events"; @@ -671,7 +705,7 @@ mod tests { assert_eq!( redacted, vec![ - ("endpoint".to_string(), "https://example.com/hook".to_string()), + ("endpoint".to_string(), "https://example.com".to_string()), ("password".to_string(), "***redacted***".to_string()), ("client_key".to_string(), "***redacted***".to_string()), ("auth_token".to_string(), "***redacted***".to_string()), diff --git a/crates/targets/src/target/webhook.rs b/crates/targets/src/target/webhook.rs index 8516d5550..e39dc9d49 100644 --- a/crates/targets/src/target/webhook.rs +++ b/crates/targets/src/target/webhook.rs @@ -32,7 +32,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 rustfs_utils::egress::OutboundPolicy; use std::{ error::Error as StdError, fmt, @@ -166,7 +166,8 @@ impl WebhookArgs { if self.endpoint.as_str().is_empty() { return Err(TargetError::Configuration("endpoint empty".to_string())); } - validate_outbound_url(&self.endpoint) + outbound_policy()? + .validate_url(&self.endpoint) .map_err(|err| TargetError::Configuration(format!("webhook endpoint is not allowed: {err}")))?; if !self.queue_dir.is_empty() { @@ -247,8 +248,16 @@ where None }; - // Build HTTP client using the helper function - let http_client = Arc::new(Mutex::new(Self::build_http_client(&args)?)); + let http_client = if args.enable { + Self::build_http_client(&args)? + } else { + Client::builder() + .no_proxy() + .redirect(reqwest::redirect::Policy::none()) + .build() + .map_err(|e| TargetError::Configuration(format!("Failed to build disabled webhook HTTP client: {e}")))? + }; + let http_client = Arc::new(Mutex::new(http_client)); let queue_store = open_target_queue_store( &args.queue_dir, @@ -285,7 +294,19 @@ where } fn build_http_client(args: &WebhookArgs) -> Result { + let resolver = outbound_policy()? + .resolver_for(&args.endpoint) + .map_err(|err| TargetError::Configuration(format!("webhook endpoint is not allowed: {err}")))?; + Self::build_http_client_with_resolver(args, resolver) + } + + fn build_http_client_with_resolver( + args: &WebhookArgs, + resolver: impl reqwest::dns::Resolve + 'static, + ) -> Result { let mut client_builder = Client::builder() + .no_proxy() + .dns_resolver(resolver) .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 @@ -294,11 +315,6 @@ where // 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 @@ -307,6 +323,7 @@ where event = EVENT_WEBHOOK_TARGET_STATE, component = LOG_COMPONENT_TARGETS, subsystem = LOG_SUBSYSTEM_WEBHOOK, + endpoint_origin = %args.endpoint.origin().ascii_serialization(), state = "tls_verification_skipped", fallback = "danger_accept_invalid_certs", "webhook target state" @@ -536,6 +553,10 @@ where } } +fn outbound_policy() -> Result<&'static OutboundPolicy, TargetError> { + OutboundPolicy::from_env_cached().map_err(|err| TargetError::Configuration(format!("invalid outbound policy: {err}"))) +} + #[async_trait] impl Target for WebhookTarget where @@ -755,10 +776,61 @@ where mod tests { use super::{WebhookArgs, WebhookTarget, classify_delivery_status, probe_health_url}; use crate::target::{REDACTED_SECRET, Target, TargetHealthReason, TargetHealthState, TargetType, decode_object_name}; + use std::net::{IpAddr, SocketAddr}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; use url::Url; use url::form_urlencoded; + #[derive(Clone)] + struct StaticResolver(IpAddr); + + impl reqwest::dns::Resolve for StaticResolver { + fn resolve(&self, _name: reqwest::dns::Name) -> reqwest::dns::Resolving { + let address = SocketAddr::new(self.0, 0); + Box::pin(async move { Ok(Box::new(std::iter::once(address)) as reqwest::dns::Addrs) }) + } + } + + #[derive(Clone)] + struct FailingResolver; + + impl reqwest::dns::Resolve for FailingResolver { + fn resolve(&self, _name: reqwest::dns::Name) -> reqwest::dns::Resolving { + Box::pin(async { Err(std::io::Error::new(std::io::ErrorKind::NotFound, "test DNS failure").into()) }) + } + } + + #[derive(Clone, Default)] + struct CapturedLog(std::sync::Arc>>); + + struct CapturedLogWriter(std::sync::Arc>>); + + impl std::io::Write for CapturedLogWriter { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.0.lock().expect("captured log lock").extend_from_slice(buf); + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl<'writer> tracing_subscriber::fmt::MakeWriter<'writer> for CapturedLog { + type Writer = CapturedLogWriter; + + fn make_writer(&'writer self) -> Self::Writer { + CapturedLogWriter(self.0.clone()) + } + } + + impl CapturedLog { + fn contents(&self) -> String { + String::from_utf8(self.0.lock().expect("captured log lock").clone()).expect("captured logs must be UTF-8") + } + } + fn base_args() -> WebhookArgs { WebhookArgs { enable: true, @@ -810,6 +882,103 @@ mod tests { assert!(rendered.contains("WebhookArgs")); } + #[tokio::test] + async fn webhook_client_uses_the_supplied_connection_resolver() { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind resolver test listener"); + let address = listener.local_addr().expect("resolver test listener address"); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.expect("accept resolved webhook request"); + let mut request = [0_u8; 1024]; + let read = stream.read(&mut request).await.expect("read webhook request"); + assert!(String::from_utf8_lossy(&request[..read]).starts_with("GET /hook HTTP/1.1")); + stream + .write_all(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + .await + .expect("write webhook response"); + }); + let args = WebhookArgs { + endpoint: Url::parse(&format!("http://webhook.test:{}/hook", address.port())).expect("endpoint should parse"), + ..base_args() + }; + let client = WebhookTarget::::build_http_client_with_resolver(&args, StaticResolver(address.ip())) + .expect("webhook client should build"); + + let response = client + .get(args.endpoint) + .send() + .await + .expect("resolver should route request to test listener"); + assert_eq!(response.status(), reqwest::StatusCode::NO_CONTENT); + server.await.expect("resolver test server should finish"); + } + + #[test] + fn webhook_client_ignores_environment_proxies_before_dns_filtering() { + const CHILD_ENV: &str = "RUSTFS_TEST_WEBHOOK_PROXY_CHILD"; + const TARGET_URL_ENV: &str = "RUSTFS_TEST_WEBHOOK_PROXY_TARGET_URL"; + const TARGET_ADDR_ENV: &str = "RUSTFS_TEST_WEBHOOK_PROXY_TARGET_ADDR"; + + if std::env::var_os(CHILD_ENV).is_some() { + let endpoint = Url::parse(&std::env::var(TARGET_URL_ENV).expect("child target URL")).expect("target URL"); + let address = std::env::var(TARGET_ADDR_ENV) + .expect("child target address") + .parse::() + .expect("target address"); + let args = WebhookArgs { endpoint, ..base_args() }; + let client = WebhookTarget::::build_http_client_with_resolver(&args, StaticResolver(address.ip())) + .expect("webhook client should build"); + tokio::runtime::Runtime::new().expect("child runtime").block_on(async { + let response = client + .get(args.endpoint) + .send() + .await + .expect("webhook request should bypass environment proxy"); + assert_eq!(response.status(), reqwest::StatusCode::NO_CONTENT); + }); + return; + } + + use std::io::{Read, Write}; + use std::net::TcpListener as StdTcpListener; + + let target_listener = StdTcpListener::bind("127.0.0.1:0").expect("bind webhook target listener"); + let target_address = target_listener.local_addr().expect("webhook target address"); + let target = std::thread::spawn(move || { + let (mut stream, _) = target_listener.accept().expect("accept direct webhook request"); + let mut request = [0_u8; 1024]; + let read = stream.read(&mut request).expect("read direct webhook request"); + assert!(String::from_utf8_lossy(&request[..read]).starts_with("GET /hook HTTP/1.1")); + stream + .write_all(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + .expect("write direct webhook response"); + }); + let proxy_listener = StdTcpListener::bind("127.0.0.1:0").expect("reserve refused proxy address"); + let proxy_address = proxy_listener.local_addr().expect("proxy address"); + drop(proxy_listener); + let target_url = format!("http://webhook.test:{}/hook", target_address.port()); + let proxy_url = format!("http://{proxy_address}"); + let output = std::process::Command::new(std::env::current_exe().expect("resolve current test executable")) + .arg("webhook_client_ignores_environment_proxies_before_dns_filtering") + .arg("--nocapture") + .env(CHILD_ENV, "1") + .env(TARGET_URL_ENV, target_url) + .env(TARGET_ADDR_ENV, target_address.to_string()) + .env("HTTP_PROXY", &proxy_url) + .env("HTTPS_PROXY", &proxy_url) + .env("ALL_PROXY", &proxy_url) + .env("NO_PROXY", "") + .output() + .expect("run isolated proxy test child"); + + assert!( + output.status.success(), + "proxy test child failed: stdout={} stderr={}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + target.join().expect("direct webhook target should finish"); + } + #[test] fn test_validate_skip_tls_verify_and_client_ca_mutually_exclusive() { let args = WebhookArgs { @@ -844,6 +1013,36 @@ mod tests { assert!(args.validate().is_ok()); } + #[test] + fn webhook_tls_warning_redacts_endpoint_details() { + let captured = CapturedLog::default(); + let subscriber = tracing_subscriber::fmt() + .with_ansi(false) + .without_time() + .with_max_level(tracing::Level::WARN) + .with_writer(captured.clone()) + .finish(); + let args = WebhookArgs { + endpoint: Url::parse("https://webhook.test/private?token=secret").expect("webhook endpoint"), + skip_tls_verify: true, + ..base_args() + }; + + tracing::subscriber::with_default(subscriber, || { + WebhookTarget::::build_http_client_with_resolver( + &args, + StaticResolver(IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)), + ) + .expect("webhook client should build"); + }); + + let logs = captured.contents(); + assert!(logs.contains("https://webhook.test")); + for secret in ["/private", "token=secret"] { + assert!(!logs.contains(secret), "TLS warning leaked {secret}: {logs}"); + } + } + #[test] fn test_validate_rejects_loopback_endpoint() { let args = WebhookArgs { @@ -930,8 +1129,9 @@ mod tests { #[tokio::test] async fn dns_failure_has_stable_health_reason() { - let url = Url::parse("http://rustfs-health-check.invalid/").expect("invalid test domain URL"); - let client = WebhookTarget::::build_http_client(&base_args()).expect("build client"); + let url = Url::parse("http://unresolvable.test/").expect("invalid test domain URL"); + let client = WebhookTarget::::build_http_client_with_resolver(&base_args(), FailingResolver) + .expect("build client"); let health = probe_health_url(&client, &url).await; @@ -1001,7 +1201,9 @@ mod tests { let _ = tls_stream.read(&mut request); }); let url = Url::parse(&format!("https://localhost:{}/", address.port())).expect("TLS health URL"); - let client = WebhookTarget::::build_http_client(&base_args()).expect("build client"); + let client = + WebhookTarget::::build_http_client_with_resolver(&base_args(), StaticResolver(address.ip())) + .expect("build client"); let health = probe_health_url(&client, &url).await; @@ -1133,12 +1335,14 @@ mod tests { }); let args = WebhookArgs { + endpoint: Url::parse(&format!("https://localhost:{}/hook", addr.port())).expect("endpoint should parse"), client_ca: ca_path.to_string_lossy().into_owned(), ..base_args() }; - let client = WebhookTarget::::build_http_client(&args).expect("build https client"); + let client = WebhookTarget::::build_http_client_with_resolver(&args, StaticResolver(addr.ip())) + .expect("build https client"); let resp = client - .head(format!("https://localhost:{}/hook", addr.port())) + .head(args.endpoint) .send() .await .expect("https webhook probe should trust configured ca"); diff --git a/crates/utils/Cargo.toml b/crates/utils/Cargo.toml index 1fa2633ce..4ee08d7b8 100644 --- a/crates/utils/Cargo.toml +++ b/crates/utils/Cargo.toml @@ -42,6 +42,7 @@ lz4 = { workspace = true, optional = true } md-5 = { workspace = true, optional = true } netif = { workspace = true, optional = true } regex = { workspace = true, optional = true } +reqwest = { workspace = true, optional = true } rustix = { workspace = true, optional = true, features = ["fs"] } serde = { workspace = true, optional = true, features = ["derive"] } sha1 = { workspace = true, optional = true } @@ -50,7 +51,7 @@ convert_case = { workspace = true, optional = true } siphasher = { workspace = true, optional = true } snap = { workspace = true, optional = true } tempfile = { workspace = true, optional = true } -tokio = { workspace = true, optional = true, features = ["io-util", "time"] } +tokio = { workspace = true, optional = true, features = ["io-util", "net", "time"] } tracing = { workspace = true } transform-stream = { workspace = true, optional = true } url = { workspace = true, optional = true } @@ -72,7 +73,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"] +egress = ["ip", "dep:reqwest", "dep:tokio", "dep:url"] io = ["dep:tokio"] path = [] # path manipulation features compress = ["dep:flate2", "dep:brotli", "dep:snap", "dep:lz4", "dep:zstd"] diff --git a/crates/utils/src/egress.rs b/crates/utils/src/egress.rs index a04adbf36..810eb57c3 100644 --- a/crates/utils/src/egress.rs +++ b/crates/utils/src/egress.rs @@ -12,10 +12,24 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::collections::HashSet; +#[cfg(test)] +use std::collections::{HashMap, VecDeque}; use std::fmt; -use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; +use std::sync::OnceLock; +#[cfg(test)] +use std::sync::{Arc, Mutex}; use url::Url; +/// Comma-separated exact HTTP(S) origins that may resolve to otherwise restricted addresses. +/// +/// This is an operator-owned process setting. Target configuration cannot extend it. Link-local, +/// metadata, and unspecified addresses remain forbidden even when their origin appears in the list. +pub const ENV_OUTBOUND_ALLOW_ORIGINS: &str = "RUSTFS_OUTBOUND_ALLOW_ORIGINS"; + +static OUTBOUND_POLICY: OnceLock> = OnceLock::new(); + #[derive(Debug, Clone, PartialEq, Eq)] pub enum OutboundUrlError { MissingHost, @@ -35,6 +49,199 @@ impl fmt::Display for OutboundUrlError { impl std::error::Error for OutboundUrlError {} +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum OutboundPolicyError { + NonUnicodeEnvironment, + InvalidAllowedOrigin { position: usize }, + MissingHost, + UnsupportedScheme { scheme: String }, + UserInfoNotAllowed, + ForbiddenHost { host: String, reason: &'static str }, +} + +impl fmt::Display for OutboundPolicyError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + OutboundPolicyError::NonUnicodeEnvironment => { + write!(f, "{ENV_OUTBOUND_ALLOW_ORIGINS} must contain valid Unicode") + } + OutboundPolicyError::InvalidAllowedOrigin { position } => { + write!(f, "invalid origin at position {position} in {ENV_OUTBOUND_ALLOW_ORIGINS}") + } + OutboundPolicyError::MissingHost => write!(f, "outbound URL is missing a host"), + OutboundPolicyError::UnsupportedScheme { scheme } => { + write!(f, "outbound URL scheme '{scheme}' is not allowed") + } + OutboundPolicyError::UserInfoNotAllowed => write!(f, "outbound URL userinfo is not allowed"), + OutboundPolicyError::ForbiddenHost { host, reason } => { + write!(f, "outbound URL host '{host}' is not allowed: {reason}") + } + } + } +} + +impl std::error::Error for OutboundPolicyError {} + +#[derive(Debug, Default, PartialEq, Eq)] +pub struct OutboundPolicy { + allowed_restricted_origins: HashSet, +} + +impl OutboundPolicy { + pub fn from_env() -> Result { + match std::env::var(ENV_OUTBOUND_ALLOW_ORIGINS) { + Ok(value) => Self::from_allowed_origins(&value), + Err(std::env::VarError::NotPresent) => Ok(Self::default()), + Err(std::env::VarError::NotUnicode(_)) => Err(OutboundPolicyError::NonUnicodeEnvironment), + } + } + + pub fn from_env_cached() -> Result<&'static Self, OutboundPolicyError> { + match OUTBOUND_POLICY.get_or_init(Self::from_env) { + Ok(policy) => Ok(policy), + Err(err) => Err(err.clone()), + } + } + + pub fn from_allowed_origins(value: &str) -> Result { + let mut allowed_restricted_origins = HashSet::new(); + if value.trim().is_empty() { + return Ok(Self::default()); + } + for (position, entry) in value.split(',').map(str::trim).enumerate() { + let position = position + 1; + if entry.is_empty() { + return Err(OutboundPolicyError::InvalidAllowedOrigin { position }); + } + let url = Url::parse(entry).map_err(|_| OutboundPolicyError::InvalidAllowedOrigin { position })?; + let origin = normalized_origin(&url).ok_or(OutboundPolicyError::InvalidAllowedOrigin { position })?; + if url.path() != "/" || url.query().is_some() || url.fragment().is_some() { + return Err(OutboundPolicyError::InvalidAllowedOrigin { position }); + } + allowed_restricted_origins.insert(origin); + } + Ok(Self { + allowed_restricted_origins, + }) + } + + pub fn validate_url(&self, url: &Url) -> Result<(), OutboundPolicyError> { + validate_http_url_shape(url)?; + let raw_host = url.host_str().ok_or(OutboundPolicyError::MissingHost)?; + let normalized_host = raw_host.trim_end_matches('.').trim_matches(['[', ']']); + let result = if normalized_host.eq_ignore_ascii_case("localhost") { + Err("loopback host") + } else if let Ok(ip) = normalized_host.parse::() { + validate_policy_ip(ip) + } else { + Ok(()) + }; + match result { + Ok(()) => Ok(()), + Err(reason) if self.allows_restricted_origin(url) && restricted_reason_can_be_overridden(reason) => Ok(()), + Err(reason) => Err(OutboundPolicyError::ForbiddenHost { + host: raw_host.to_string(), + reason, + }), + } + } + + pub fn resolver_for(&self, url: &Url) -> Result { + self.validate_url(url)?; + let host = normalized_host(url).ok_or(OutboundPolicyError::MissingHost)?; + Ok(OutboundDnsResolver::new(host, self.allows_restricted_origin(url))) + } + + fn allows_restricted_origin(&self, url: &Url) -> bool { + normalized_origin(url).is_some_and(|origin| self.allowed_restricted_origins.contains(&origin)) + } +} + +#[derive(Clone, Debug)] +pub struct OutboundDnsResolver { + allowed_restricted_host: Option, + #[cfg(test)] + overrides: Option, +} + +#[cfg(test)] +type DnsOverrideAnswers = Arc>>>>; + +impl OutboundDnsResolver { + fn new(host: String, allow_restricted: bool) -> Self { + Self { + allowed_restricted_host: allow_restricted.then_some(host), + #[cfg(test)] + overrides: None, + } + } + + #[cfg(test)] + fn with_overrides(mut self, overrides: HashMap>) -> Self { + let overrides = overrides + .into_iter() + .map(|(host, ips)| { + let addresses = ips.into_iter().map(|ip| SocketAddr::new(ip, 0)).collect(); + (host, VecDeque::from([addresses])) + }) + .collect(); + self.overrides = Some(Arc::new(Mutex::new(overrides))); + self + } + + #[cfg(test)] + fn with_override_sequence(mut self, host: &str, answers: Vec>) -> Self { + self.overrides = Some(Arc::new(Mutex::new(HashMap::from([(host.to_string(), VecDeque::from(answers))])))); + self + } +} + +impl reqwest::dns::Resolve for OutboundDnsResolver { + fn resolve(&self, name: reqwest::dns::Name) -> reqwest::dns::Resolving { + let host = name.as_str().trim_end_matches('.').to_ascii_lowercase(); + let allow_restricted = self.allowed_restricted_host.as_deref() == Some(host.as_str()); + #[cfg(test)] + let overrides = self.overrides.clone(); + Box::pin(async move { + #[cfg(test)] + let overridden = overrides.as_ref().and_then(|entries| { + let mut entries = entries.lock().expect("test DNS overrides must not be poisoned"); + let answers = entries.get_mut(&host)?; + if answers.len() > 1 { + answers.pop_front() + } else { + answers.front().cloned() + } + }); + #[cfg(not(test))] + let overridden: Option> = None; + + let addresses = if let Some(addresses) = overridden { + addresses + } else { + // Do not use the cached resolver in `crate::net`: every new connection must + // classify the addresses returned at that boundary so DNS rebinding fails closed. + tokio::net::lookup_host((host.as_str(), 0)) + .await + .map_err(|err| std::io::Error::new(std::io::ErrorKind::NotFound, err))? + .collect() + }; + let addrs = addresses + .into_iter() + .filter(|address| resolved_ip_allowed(address.ip(), allow_restricted)) + .collect::>(); + if addrs.is_empty() { + return Err(std::io::Error::new( + std::io::ErrorKind::PermissionDenied, + format!("outbound DNS resolution for '{host}' returned no allowed addresses"), + ) + .into()); + } + Ok(Box::new(addrs.into_iter()) as reqwest::dns::Addrs) + }) + } +} + pub fn validate_outbound_url(url: &Url) -> Result<(), OutboundUrlError> { let Some(raw_host) = url.host_str() else { return Err(OutboundUrlError::MissingHost); @@ -58,10 +265,168 @@ pub fn validate_outbound_url(url: &Url) -> Result<(), OutboundUrlError> { }) } +fn validate_http_url_shape(url: &Url) -> Result<(), OutboundPolicyError> { + if !matches!(url.scheme(), "http" | "https") { + return Err(OutboundPolicyError::UnsupportedScheme { + scheme: url.scheme().to_string(), + }); + } + if !url.username().is_empty() || url.password().is_some() { + return Err(OutboundPolicyError::UserInfoNotAllowed); + } + Ok(()) +} + +fn normalized_host(url: &Url) -> Option { + url.host_str() + .map(|host| host.trim_end_matches('.').trim_matches(&['[', ']'][..]).to_ascii_lowercase()) +} + +fn normalized_origin(url: &Url) -> Option { + validate_http_url_shape(url).ok()?; + let host = normalized_host(url)?; + let port = url.port_or_known_default()?; + // `Url::origin()` preserves a trailing DNS dot, but the resolver treats dotted and undotted + // names as the same host. Normalize it here so policy matching follows connection semantics. + if host.parse::().is_ok() { + Some(format!("{}://[{host}]:{port}", url.scheme())) + } else { + Some(format!("{}://{host}:{port}", url.scheme())) + } +} + +fn restricted_reason_can_be_overridden(reason: &str) -> bool { + matches!( + reason, + "loopback host" | "loopback address" | "private address" | "shared address" | "reserved address" + ) +} + +fn resolved_ip_allowed(ip: IpAddr, allow_restricted: bool) -> bool { + match validate_policy_ip(ip) { + Ok(()) => true, + Err(reason) => allow_restricted && restricted_reason_can_be_overridden(reason), + } +} + +fn validate_policy_ip(ip: IpAddr) -> Result<(), &'static str> { + if is_metadata_endpoint(ip) { + return Err("metadata endpoint"); + } + + let original_v6 = match ip { + IpAddr::V6(v6) => Some(v6), + IpAddr::V4(_) => None, + }; + let ip = match original_v6 { + Some(v6) => match policy_embedded_ipv4(v6) { + Some(v4) => IpAddr::V4(v4), + None => IpAddr::V6(v6), + }, + None => ip, + }; + + if is_metadata_endpoint(ip) { + return Err("metadata endpoint"); + } + validate_outbound_ip(ip)?; + if original_v6.is_some_and(|v6| v6.segments()[0] == 0x2002) { + return Err("reserved address"); + } + + match ip { + IpAddr::V4(ipv4) => { + let octets = ipv4.octets(); + if octets[0] == 100 && (64..=127).contains(&octets[1]) { + return Err("shared address"); + } + if octets[0] == 0 + || octets[0] >= 224 + || (octets[0] == 192 && ((octets[1] == 0 && matches!(octets[2], 0 | 2)) || (octets[1] == 88 && octets[2] == 99))) + || (octets[0] == 198 && (octets[1] == 18 || octets[1] == 19)) + || (octets[0] == 198 && octets[1] == 51 && octets[2] == 100) + || (octets[0] == 203 && octets[1] == 0 && octets[2] == 113) + { + return Err("reserved address"); + } + } + IpAddr::V6(ipv6) => { + let segments = ipv6.segments(); + if segments[0] == 0x0064 && segments[1] == 0xff9b && segments[2] == 1 { + return Err("translation address"); + } + if is_special_use_ipv6(ipv6) { + return Err("reserved address"); + } + } + } + + Ok(()) +} + +fn is_special_use_ipv6(ip: Ipv6Addr) -> bool { + let segments = ip.segments(); + let bits = u128::from_be_bytes(ip.octets()); + let ietf_protocol_assignment = segments[0] == 0x2001 + && segments[1] < 0x0200 + && bits != 0x2001_0001_0000_0000_0000_0000_0000_0001 + && bits != 0x2001_0001_0000_0000_0000_0000_0000_0002 + && segments[1] != 0x0003 + && !(segments[1] == 0x0004 && segments[2] == 0x0112) + && !(0x0020..=0x003f).contains(&segments[1]); + + (segments[0] & 0xff00) == 0xff00 + || (segments[0] & 0xffc0) == 0xfec0 + || (segments[0] == 0x0100 && segments[1..4] == [0, 0, 0]) + || ietf_protocol_assignment + || segments[0] == 0x2002 + || (segments[0] == 0x2001 && segments[1] == 0x0db8) + || (segments[0] == 0x3fff && segments[1] <= 0x0fff) + || segments[0] == 0x5f00 +} + +fn is_metadata_endpoint(ip: IpAddr) -> bool { + match ip { + IpAddr::V4(ipv4) => matches!( + ipv4.octets(), + [169, 254, 169, 254] + | [169, 254, 170, 2] + | [169, 254, 170, 23] + | [169, 254, 0, 23] + | [100, 100, 100, 200] + | [168, 63, 129, 16] + ), + IpAddr::V6(ipv6) => matches!( + ipv6.segments(), + [0xfd00, 0x0ec2, 0, 0, 0, 0, 0, 0x0254] + | [0xfd00, 0x0ec2, 0, 0, 0, 0, 0, 0x0023] + | [0xfd20, 0x00ce, 0, 0, 0, 0, 0, 0x0254] + ), + } +} + +fn policy_embedded_ipv4(v6: Ipv6Addr) -> Option { + if let Some(v4) = embedded_ipv4(v6) { + return Some(v4); + } + let segments = v6.segments(); + if segments[..6] == [0, 0, 0, 0, 0xffff, 0] { + let hi = segments[6].to_be_bytes(); + let lo = segments[7].to_be_bytes(); + return Some(Ipv4Addr::new(hi[0], hi[1], lo[0], lo[1])); + } + let octets = v6.octets(); + if octets[..12] == [0x00, 0x64, 0xff, 0x9b, 0, 0, 0, 0, 0, 0, 0, 0] { + return Some(Ipv4Addr::new(octets[12], octets[13], octets[14], octets[15])); + } + if octets[0..2] == [0x20, 0x02] { + return Some(Ipv4Addr::new(octets[2], octets[3], octets[4], octets[5])); + } + + None +} + fn validate_outbound_ip(ip: IpAddr) -> Result<(), &'static str> { - // Reject pure-IPv6 special forms first, before any IPv4 normalization. ::1 (loopback) and :: - // (unspecified) are technically IPv4-compatible forms too, so normalizing first would map - // them to a harmless-looking 0.0.0.1 / 0.0.0.0 and let them through. if let IpAddr::V6(v6) = ip { if v6.is_loopback() { return Err("loopback address"); @@ -77,10 +442,6 @@ fn validate_outbound_ip(ip: IpAddr) -> Result<(), &'static str> { } } - // Normalize IPv4-mapped (::ffff:a.b.c.d) AND IPv4-compatible (::a.b.c.d) IPv6 addresses to - // their embedded IPv4 so the IPv4 rules below apply. The std is_* checks on the IPv6 variant - // never inspect the embedded IPv4, so without this an attacker bypasses the guard with e.g. - // ::ffff:127.0.0.1, ::127.0.0.1 (loopback) or ::169.254.169.254 (cloud metadata service). let ip = match ip { IpAddr::V6(v6) => match embedded_ipv4(v6) { Some(v4) => IpAddr::V4(v4), @@ -92,44 +453,32 @@ 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"); - } + if let IpAddr::V4(ipv4) = ip { + 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"); } - // Genuine IPv6 (no embedded IPv4) was already classified above. - IpAddr::V6(_) => {} } - Ok(()) } -/// Extract the embedded IPv4 from an IPv4-mapped (`::ffff:a.b.c.d`) or IPv4-compatible -/// (`::a.b.c.d`) IPv6 address. The pure-IPv6 specials `::` and `::1` are rejected by the caller -/// before this runs, so returning `None` here means a genuine IPv6 host. fn embedded_ipv4(v6: Ipv6Addr) -> Option { if let Some(v4) = v6.to_ipv4_mapped() { return Some(v4); } - // IPv4-compatible: the top 96 bits are zero and the low 32 bits carry the IPv4. - let segs = v6.segments(); - if segs[0..6] == [0, 0, 0, 0, 0, 0] { - let hi = segs[6].to_be_bytes(); - let lo = segs[7].to_be_bytes(); + let segments = v6.segments(); + if segments[0..6] == [0, 0, 0, 0, 0, 0] { + let hi = segments[6].to_be_bytes(); + let lo = segments[7].to_be_bytes(); let v4 = Ipv4Addr::new(hi[0], hi[1], lo[0], lo[1]); - // `::` and `::1` are already handled by the caller; anything else is a real embedded v4. if !v4.is_unspecified() && v4 != Ipv4Addr::new(0, 0, 0, 1) { return Some(v4); } @@ -139,7 +488,9 @@ fn embedded_ipv4(v6: Ipv6Addr) -> Option { #[cfg(test)] mod tests { - use super::{OutboundUrlError, validate_outbound_url}; + use super::{OutboundPolicy, OutboundUrlError, validate_outbound_url}; + use std::collections::HashMap; + use std::net::SocketAddr; use url::Url; #[test] @@ -174,6 +525,17 @@ mod tests { )); } + #[test] + fn validate_outbound_url_rejects_alternate_ipv4_loopback_syntax() { + for endpoint in ["http://2130706433/hook", "http://0177.0.0.1/hook", "http://0x7f000001/hook"] { + let url = Url::parse(endpoint).expect("alternate IPv4 URL should parse"); + assert!( + validate_outbound_url(&url).is_err(), + "alternate loopback syntax must be rejected: {endpoint}" + ); + } + } + #[test] fn validate_outbound_url_rejects_private_ip() { let url = Url::parse("https://10.0.0.5/webhook").expect("private URL should parse"); @@ -187,6 +549,35 @@ mod tests { )); } + #[test] + fn outbound_policy_rejects_special_use_addresses() { + let policy = OutboundPolicy::default(); + for endpoint in [ + "http://0.0.0.1/hook", + "http://192.0.2.1/hook", + "http://198.18.0.1/hook", + "http://198.51.100.1/hook", + "http://203.0.113.1/hook", + "http://224.0.0.1/hook", + "http://[100::1]/hook", + "http://[2001:100::1]/hook", + "http://[2001:db8::1]/hook", + "http://[3fff::1]/hook", + "http://[5f00::1]/hook", + "http://[ff02::1]/hook", + ] { + let url = Url::parse(endpoint).expect("special-use URL should parse"); + assert!(policy.validate_url(&url).is_err(), "special-use address must be rejected: {endpoint}"); + } + } + + #[test] + fn validate_outbound_url_preserves_legacy_shared_address_behavior() { + let url = Url::parse("http://100.64.0.5/hook").expect("shared URL should parse"); + assert!(validate_outbound_url(&url).is_ok()); + assert!(OutboundPolicy::default().validate_url(&url).is_err()); + } + #[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"); @@ -286,6 +677,26 @@ mod tests { )); } + #[test] + fn outbound_policy_rejects_embedded_private_destinations() { + let policy = OutboundPolicy::default(); + for endpoint in [ + "http://[64:ff9b::7f00:1]/hook", + "http://[64:ff9b::a00:1]/hook", + "http://[64:ff9b::a9fe:a9fe]/latest/meta-data", + "http://[2002:7f00:1::]/hook", + "http://[2002:a00:1::]/hook", + "http://[::ffff:0:7f00:1]/hook", + "http://[::ffff:0:a9fe:a9fe]/latest/meta-data", + ] { + let url = Url::parse(endpoint).expect("embedded IPv4 URL should parse"); + assert!( + policy.validate_url(&url).is_err(), + "embedded private destination must be rejected: {endpoint}" + ); + } + } + #[test] fn validate_outbound_url_rejects_ipv6_loopback_and_unspecified() { // ::1 / :: must stay rejected even though they look like IPv4-compatible forms. @@ -306,4 +717,263 @@ mod tests { } )); } + + #[test] + fn outbound_policy_allows_only_the_exact_operator_origin() { + let policy = OutboundPolicy::from_allowed_origins("http://127.0.0.1:9443").expect("operator origin should parse"); + + assert!( + policy + .validate_url(&Url::parse("http://127.0.0.1:9443/hook").expect("allowed URL")) + .is_ok() + ); + assert!( + policy + .validate_url(&Url::parse("http://127.0.0.1:9444/hook").expect("wrong-port URL")) + .is_err(), + "an allowlisted origin must not authorize another port" + ); + assert!( + policy + .validate_url(&Url::parse("http://[::1]:9443/hook").expect("different-host URL")) + .is_err(), + "an allowlisted origin must not authorize a different host" + ); + + let shared_policy = + OutboundPolicy::from_allowed_origins("http://100.64.0.5:9443").expect("shared operator origin should parse"); + assert!( + shared_policy + .validate_url(&Url::parse("http://100.64.0.5:9443/hook").expect("shared URL")) + .is_ok(), + "an exact operator origin may opt into an internal shared address" + ); + } + + #[test] + fn outbound_policy_never_allows_metadata_or_unspecified_addresses() { + let policy = OutboundPolicy::from_allowed_origins( + "http://169.254.1.10,http://169.254.169.254,http://169.254.170.2,http://169.254.170.23,http://169.254.0.23,http://100.100.100.200,http://168.63.129.16,http://[fd00:ec2::254],http://[fd00:ec2::23],http://[fd20:ce::254],http://[64:ff9b:1::1],http://0.0.0.0", + ) + .expect("syntactically valid origins should parse"); + + for endpoint in [ + "http://169.254.1.10/hook", + "http://169.254.169.254/latest", + "http://169.254.170.2/credentials", + "http://169.254.170.23/v1/credentials", + "http://169.254.0.23/latest/meta-data", + "http://100.100.100.200/latest/meta-data", + "http://168.63.129.16/machine?comp=goalstate", + "http://[fd00:ec2::254]/latest", + "http://[fd00:ec2::23]/v1/credentials", + "http://[fd20:ce::254]/computeMetadata/v1", + "http://[64:ff9b:1::1]/hook", + "http://0.0.0.0/hook", + ] { + assert!( + policy + .validate_url(&Url::parse(endpoint).expect("test endpoint should parse")) + .is_err(), + "endpoint must remain forbidden: {endpoint}" + ); + } + } + + #[test] + fn outbound_policy_rejects_non_http_and_userinfo_origins() { + for origin in [ + "ftp://webhook.internal", + "https://user@webhook.internal", + "https://webhook.internal/path", + ] { + assert!( + OutboundPolicy::from_allowed_origins(origin).is_err(), + "invalid operator origin must fail closed: {origin}" + ); + } + assert!(OutboundPolicy::from_allowed_origins("https://one.example,,https://two.example").is_err()); + } + + #[test] + fn outbound_policy_rejects_non_http_and_userinfo_targets() { + let policy = OutboundPolicy::default(); + for endpoint in ["ftp://webhook.test/hook", "https://user:secret@webhook.test/hook"] { + assert!( + policy + .validate_url(&Url::parse(endpoint).expect("target URL should parse")) + .is_err(), + "target must be rejected: {endpoint}" + ); + } + } + + #[tokio::test] + async fn outbound_dns_resolver_filters_rebinding_and_mixed_answers() { + let policy = OutboundPolicy::default(); + let endpoint = Url::parse("https://webhook.test/hook").expect("endpoint should parse"); + let resolver = policy + .resolver_for(&endpoint) + .expect("public hostname should be accepted") + .with_overrides(HashMap::from([ + ( + "webhook.test".to_string(), + vec![ + "10.0.0.5".parse().expect("private IP"), + "100.64.0.5".parse().expect("shared IP"), + "100.100.100.200".parse().expect("metadata IP"), + "8.8.8.8".parse().expect("public IP"), + "::ffff:127.0.0.1".parse().expect("mapped loopback IP"), + ], + ), + ("rebound.test".to_string(), vec!["127.0.0.1".parse().expect("loopback IP")]), + ])); + + let addrs = reqwest::dns::Resolve::resolve(&resolver, "webhook.test".parse().expect("resolver hostname")) + .await + .expect("one public address should remain") + .map(|addr| addr.ip()) + .collect::>(); + assert_eq!(addrs, vec!["8.8.8.8".parse::().expect("public IP")]); + assert!( + reqwest::dns::Resolve::resolve(&resolver, "rebound.test".parse().expect("resolver hostname")) + .await + .is_err(), + "a rebound answer containing only restricted addresses must fail closed" + ); + } + + #[tokio::test] + async fn exact_operator_origin_allows_private_dns_answers_for_that_host_only() { + let policy = OutboundPolicy::from_allowed_origins("https://webhook.test:9443").expect("operator origin should parse"); + let resolver = policy + .resolver_for(&Url::parse("https://webhook.test:9443/hook").expect("endpoint should parse")) + .expect("allowlisted endpoint should be accepted") + .with_overrides(HashMap::from([ + ("webhook.test".to_string(), vec!["10.0.0.5".parse().expect("private IP")]), + ("other.test".to_string(), vec!["10.0.0.6".parse().expect("private IP")]), + ])); + + let allowed = reqwest::dns::Resolve::resolve(&resolver, "webhook.test".parse().expect("resolver hostname")) + .await + .expect("allowlisted host should resolve") + .count(); + assert_eq!(allowed, 1); + assert!( + reqwest::dns::Resolve::resolve(&resolver, "other.test".parse().expect("resolver hostname")) + .await + .is_err(), + "the allowlist must not authorize another private hostname" + ); + } + + #[tokio::test] + async fn exact_operator_origin_never_allows_metadata_dns_answers() { + let policy = OutboundPolicy::from_allowed_origins("https://webhook.test:9443").expect("operator origin should parse"); + let resolver = policy + .resolver_for(&Url::parse("https://webhook.test:9443/hook").expect("endpoint should parse")) + .expect("allowlisted endpoint should be accepted") + .with_overrides(HashMap::from([( + "webhook.test".to_string(), + vec![ + "169.254.170.23".parse().expect("EKS credential IP"), + "169.254.0.23".parse().expect("Tencent metadata IP"), + "fd00:ec2::23".parse().expect("EKS credential IPv6"), + "fd20:ce::254".parse().expect("GCP metadata IPv6"), + "0.0.0.0".parse().expect("unspecified IP"), + ], + )])); + + assert!( + reqwest::dns::Resolve::resolve(&resolver, "webhook.test".parse().expect("resolver hostname")) + .await + .is_err(), + "metadata and unspecified answers must remain forbidden for an allowlisted origin" + ); + } + + #[tokio::test] + async fn outbound_dns_resolver_preserves_ipv6_scope_id() { + use std::net::SocketAddrV6; + + let policy = OutboundPolicy::default(); + let scoped = SocketAddr::V6(SocketAddrV6::new("2001:4860:4860::8888".parse().expect("public IPv6"), 0, 0, 7)); + let resolver = policy + .resolver_for(&Url::parse("https://webhook.test:9443/hook").expect("endpoint should parse")) + .expect("public endpoint should be accepted") + .with_override_sequence("webhook.test", vec![vec![scoped]]); + + let resolved = reqwest::dns::Resolve::resolve(&resolver, "webhook.test".parse().expect("resolver hostname")) + .await + .expect("public scoped address should resolve") + .collect::>(); + assert_eq!(resolved, vec![scoped]); + } + + #[tokio::test] + async fn reqwest_rechecks_dns_policy_for_each_new_connection() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind first connection listener"); + let address = listener.local_addr().expect("first connection address"); + let endpoint = Url::parse(&format!("http://rebind.test:{}/hook", address.port())).expect("endpoint should parse"); + let policy = OutboundPolicy::from_allowed_origins(&endpoint.origin().ascii_serialization()) + .expect("loopback test origin should be explicitly allowed"); + let resolver = policy + .resolver_for(&endpoint) + .expect("allowlisted endpoint should be accepted") + .with_override_sequence( + "rebind.test", + vec![ + vec![SocketAddr::new(address.ip(), 0)], + vec![SocketAddr::new("169.254.169.254".parse().expect("metadata IP"), 0)], + ], + ); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.expect("accept first connection"); + let mut request = [0_u8; 1024]; + let _ = stream.read(&mut request).await.expect("read first request"); + stream + .write_all(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + .await + .expect("write first response"); + }); + let client = reqwest::Client::builder() + .no_proxy() + .dns_resolver(resolver) + .timeout(std::time::Duration::from_secs(2)) + .build() + .expect("build test client"); + + assert_eq!( + client + .get(endpoint.clone()) + .send() + .await + .expect("first request should connect") + .status(), + reqwest::StatusCode::NO_CONTENT + ); + server.await.expect("first connection server should finish"); + let error = client + .get(endpoint) + .send() + .await + .expect_err("the second connection must reject a metadata DNS answer"); + let mut error_chain = vec![error.to_string()]; + let mut source = std::error::Error::source(&error); + while let Some(error) = source { + error_chain.push(error.to_string()); + source = error.source(); + } + assert!( + error_chain + .iter() + .any(|message| message.contains("returned no allowed addresses")), + "request must fail in the DNS policy layer: {error_chain:?}" + ); + } } diff --git a/rustfs/src/admin/router.rs b/rustfs/src/admin/router.rs index 6aa3083af..3c4179613 100644 --- a/rustfs/src/admin/router.rs +++ b/rustfs/src/admin/router.rs @@ -65,7 +65,7 @@ use rustfs_notify::{Event as NotificationEvent, notification_system}; use rustfs_policy::policy::action::{Action, S3Action}; use rustfs_s3_types::EventName; use rustfs_signer::pre_sign_v4; -use rustfs_utils::egress::validate_outbound_url; +use rustfs_utils::egress::{OutboundDnsResolver, OutboundPolicy}; 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, @@ -229,7 +229,7 @@ struct ListenNotificationFilter { suffix: Option, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone)] struct ObjectLambdaWebhookConfig { endpoint: Url, auth_token: String, @@ -238,6 +238,7 @@ struct ObjectLambdaWebhookConfig { client_ca: String, skip_tls_verify: bool, response_header_timeout: Option, + outbound_resolver: OutboundDnsResolver, } const LAMBDA_WEBHOOK_SUB_SYS: &str = "lambda_webhook"; @@ -612,7 +613,8 @@ fn resolve_object_lambda_webhook_config_from_server_config( let parsed_endpoint = Url::parse(&endpoint).map_err(|_| s3_error!(InvalidRequest, "object lambda target endpoint is invalid"))?; - validate_outbound_url(&parsed_endpoint) + let outbound_resolver = outbound_policy()? + .resolver_for(&parsed_endpoint) .map_err(|err| s3_error!(InvalidRequest, "object lambda target endpoint is not allowed: {}", err))?; Ok(ObjectLambdaWebhookConfig { @@ -623,6 +625,7 @@ fn resolve_object_lambda_webhook_config_from_server_config( client_ca: kvs.lookup(WEBHOOK_CLIENT_CA).unwrap_or_default(), skip_tls_verify: config_enable_is_on(&kvs.lookup(WEBHOOK_SKIP_TLS_VERIFY).unwrap_or_default()), response_header_timeout, + outbound_resolver, }) } @@ -658,9 +661,18 @@ 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)); + build_object_lambda_http_client_with_resolver(config, config.outbound_resolver.clone()) +} + +fn build_object_lambda_http_client_with_resolver( + config: &ObjectLambdaWebhookConfig, + resolver: impl reqwest::dns::Resolve + 'static, +) -> S3Result { + let mut builder = reqwest::Client::builder() + .no_proxy() + .dns_resolver(resolver) + .redirect(reqwest::redirect::Policy::none()) + .user_agent(rustfs_targets::get_user_agent(rustfs_targets::ServiceType::Basis)); if let Some(timeout) = config.response_header_timeout { builder = builder.timeout(timeout); @@ -672,7 +684,7 @@ fn build_object_lambda_http_client(config: &ObjectLambdaWebhookConfig) -> S3Resu component = LOG_COMPONENT_ADMIN, subsystem = LOG_SUBSYSTEM_OBJECT_LAMBDA, result = "tls_verification_disabled", - endpoint = %config.endpoint, + endpoint_origin = %config.endpoint.origin().ascii_serialization(), "admin router state" ); builder = builder.danger_accept_invalid_certs(true); @@ -706,6 +718,17 @@ fn build_object_lambda_http_client(config: &ObjectLambdaWebhookConfig) -> S3Resu .map_err(|e| s3_error!(InternalError, "failed to build object lambda http client: {e}")) } +async fn send_object_lambda_request(request: reqwest::RequestBuilder) -> S3Result { + request + .send() + .await + .map_err(|_| s3_error!(InternalError, "object lambda target request failed")) +} + +fn outbound_policy() -> S3Result<&'static OutboundPolicy> { + OutboundPolicy::from_env_cached().map_err(|err| s3_error!(InvalidRequest, "invalid outbound policy: {}", err)) +} + fn extract_request_scheme(headers: &HeaderMap, uri: &Uri) -> String { get_source_scheme(headers) .and_then(|value| { @@ -1098,10 +1121,7 @@ async fn invoke_object_lambda_target( request_builder = request_builder.header("x-rustfs-object-lambda-version-id", version_id); } - let lambda_response = request_builder - .send() - .await - .map_err(|e| s3_error!(InternalError, "object lambda target request failed: {e}"))?; + let lambda_response = send_object_lambda_request(request_builder).await?; let status = lambda_response.status(); let lambda_headers = lambda_response.headers().clone(); @@ -2576,7 +2596,70 @@ mod tests { use http::Method; use http::Uri; use s3s::S3Request; + use std::net::{IpAddr, SocketAddr}; use time::macros::datetime; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + + #[derive(Clone)] + struct StaticResolver(IpAddr); + + impl reqwest::dns::Resolve for StaticResolver { + fn resolve(&self, _name: reqwest::dns::Name) -> reqwest::dns::Resolving { + let address = SocketAddr::new(self.0, 0); + Box::pin(async move { Ok(Box::new(std::iter::once(address)) as reqwest::dns::Addrs) }) + } + } + + fn object_lambda_test_config(endpoint: Url) -> ObjectLambdaWebhookConfig { + object_lambda_test_config_with_policy(endpoint, &OutboundPolicy::default()) + } + + fn object_lambda_test_config_with_policy(endpoint: Url, policy: &OutboundPolicy) -> ObjectLambdaWebhookConfig { + let outbound_resolver = policy + .resolver_for(&endpoint) + .expect("test endpoint should satisfy its outbound policy"); + ObjectLambdaWebhookConfig { + endpoint, + auth_token: String::new(), + client_cert: String::new(), + client_key: String::new(), + client_ca: String::new(), + skip_tls_verify: false, + response_header_timeout: Some(Duration::from_secs(2)), + outbound_resolver, + } + } + + #[derive(Clone, Default)] + struct CapturedLog(std::sync::Arc>>); + + struct CapturedLogWriter(std::sync::Arc>>); + + impl std::io::Write for CapturedLogWriter { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.0.lock().expect("captured log lock").extend_from_slice(buf); + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl<'writer> tracing_subscriber::fmt::MakeWriter<'writer> for CapturedLog { + type Writer = CapturedLogWriter; + + fn make_writer(&'writer self) -> Self::Writer { + CapturedLogWriter(self.0.clone()) + } + } + + impl CapturedLog { + fn contents(&self) -> String { + String::from_utf8(self.0.lock().expect("captured log lock").clone()).expect("captured logs must be UTF-8") + } + } #[test] fn canonicalize_admin_path_maps_compat_prefix_to_rustfs_prefix() { @@ -3778,6 +3861,214 @@ mod tests { assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); } + #[test] + fn object_lambda_client_ignores_proxies_and_does_not_follow_redirects() { + const CHILD_ENV: &str = "RUSTFS_TEST_OBJECT_LAMBDA_PROXY_CHILD"; + const TARGET_URL_ENV: &str = "RUSTFS_TEST_OBJECT_LAMBDA_TARGET_URL"; + const TARGET_ADDR_ENV: &str = "RUSTFS_TEST_OBJECT_LAMBDA_TARGET_ADDR"; + + if std::env::var_os(CHILD_ENV).is_some() { + let endpoint = Url::parse(&std::env::var(TARGET_URL_ENV).expect("child target URL")).expect("target URL"); + let address = std::env::var(TARGET_ADDR_ENV) + .expect("child target address") + .parse::() + .expect("target address"); + let config = object_lambda_test_config(endpoint); + let client = build_object_lambda_http_client_with_resolver(&config, StaticResolver(address.ip())) + .expect("object lambda client should build"); + tokio::runtime::Runtime::new().expect("child runtime").block_on(async { + let response = client + .get(config.endpoint) + .send() + .await + .expect("object lambda request should bypass environment proxy"); + assert_eq!(response.status(), reqwest::StatusCode::FOUND); + }); + return; + } + + use std::io::{Read, Write}; + use std::net::TcpListener as StdTcpListener; + + let target_listener = StdTcpListener::bind("127.0.0.1:0").expect("bind object lambda target"); + let target_address = target_listener.local_addr().expect("object lambda target address"); + let redirect_listener = StdTcpListener::bind("127.0.0.1:0").expect("bind redirect destination"); + let redirect_address = redirect_listener.local_addr().expect("redirect destination address"); + drop(redirect_listener); + let target = std::thread::spawn(move || { + let (mut stream, _) = target_listener.accept().expect("accept object lambda request"); + let mut request = [0_u8; 1024]; + let read = stream.read(&mut request).expect("read object lambda request"); + assert!(String::from_utf8_lossy(&request[..read]).starts_with("GET /transform HTTP/1.1")); + stream + .write_all( + format!( + "HTTP/1.1 302 Found\r\nLocation: http://127.0.0.1:{}/metadata\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + redirect_address.port() + ) + .as_bytes(), + ) + .expect("write object lambda redirect"); + }); + let proxy_listener = StdTcpListener::bind("127.0.0.1:0").expect("reserve refused proxy address"); + let proxy_address = proxy_listener.local_addr().expect("proxy address"); + drop(proxy_listener); + let target_url = format!("http://object-lambda.test:{}/transform", target_address.port()); + let proxy_url = format!("http://{proxy_address}"); + let output = std::process::Command::new(std::env::current_exe().expect("resolve current test executable")) + .arg("object_lambda_client_ignores_proxies_and_does_not_follow_redirects") + .arg("--nocapture") + .env(CHILD_ENV, "1") + .env(TARGET_URL_ENV, target_url) + .env(TARGET_ADDR_ENV, target_address.to_string()) + .env("HTTP_PROXY", &proxy_url) + .env("HTTPS_PROXY", &proxy_url) + .env("ALL_PROXY", &proxy_url) + .env("NO_PROXY", "") + .output() + .expect("run isolated object lambda proxy test child"); + + assert!( + output.status.success(), + "proxy test child failed: stdout={} stderr={}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + target.join().expect("object lambda target should finish"); + } + + #[tokio::test] + async fn object_lambda_client_uses_configured_outbound_policy_resolver() { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind object lambda policy target"); + let address = listener.local_addr().expect("object lambda policy target address"); + let endpoint = Url::parse(&format!("http://{address}/transform")).expect("object lambda endpoint"); + let policy = OutboundPolicy::from_allowed_origins(&endpoint.origin().ascii_serialization()) + .expect("loopback test origin should be explicitly allowed"); + let config = object_lambda_test_config_with_policy(endpoint.clone(), &policy); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.expect("accept object lambda policy request"); + let mut request = [0_u8; 1024]; + let read = stream.read(&mut request).await.expect("read object lambda policy request"); + assert!(String::from_utf8_lossy(&request[..read]).starts_with("GET /transform HTTP/1.1")); + stream + .write_all(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + .await + .expect("write object lambda policy response"); + }); + + let response = build_object_lambda_http_client(&config) + .expect("object lambda policy client should build") + .get(endpoint) + .send() + .await + .expect("configured policy resolver should reach the allowed origin"); + assert_eq!(response.status(), reqwest::StatusCode::NO_CONTENT); + server.await.expect("object lambda policy target should finish"); + } + + #[tokio::test] + async fn object_lambda_client_preserves_hostname_for_tls_sni() { + use rustls::{ + ServerConfig, ServerConnection, StreamOwned, + pki_types::{PrivateKeyDer, PrivatePkcs8KeyDer}, + }; + use std::io::{Read, Write}; + use std::sync::{Arc, Once}; + + static INSTALL_CRYPTO_PROVIDER: Once = Once::new(); + INSTALL_CRYPTO_PROVIDER.call_once(|| { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + }); + let rcgen::CertifiedKey { cert, signing_key } = + rcgen::generate_simple_self_signed(vec!["object-lambda.test".to_string()]).expect("cert should generate"); + let temp_dir = tempfile::tempdir().expect("create temp directory"); + let ca_path = temp_dir.path().join("object-lambda-ca.pem"); + std::fs::write(&ca_path, cert.pem()).expect("write object lambda CA"); + let server_config = Arc::new( + ServerConfig::builder() + .with_no_client_auth() + .with_single_cert( + vec![cert.der().clone()], + PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(signing_key.serialize_der())), + ) + .expect("server certificate should be valid"), + ); + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind object lambda TLS server"); + let address = listener.local_addr().expect("object lambda TLS address"); + let server = std::thread::spawn(move || { + let (stream, _) = listener.accept().expect("accept object lambda TLS request"); + let connection = ServerConnection::new(server_config).expect("server TLS connection"); + let mut stream = StreamOwned::new(connection, stream); + let mut request = [0_u8; 1024]; + let _ = stream.read(&mut request).expect("read object lambda TLS request"); + assert_eq!(stream.conn.server_name(), Some("object-lambda.test")); + stream + .write_all(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + .expect("write object lambda TLS response"); + }); + let endpoint = + Url::parse(&format!("https://object-lambda.test:{}/transform", address.port())).expect("object lambda TLS endpoint"); + let mut config = object_lambda_test_config(endpoint.clone()); + config.client_ca = ca_path.to_string_lossy().into_owned(); + + let response = build_object_lambda_http_client_with_resolver(&config, StaticResolver(address.ip())) + .expect("object lambda TLS client should build") + .get(endpoint) + .send() + .await + .expect("TLS request should keep the configured hostname for SNI"); + assert_eq!(response.status(), reqwest::StatusCode::NO_CONTENT); + server.join().expect("object lambda TLS server should finish"); + } + + #[test] + fn object_lambda_tls_warning_redacts_endpoint_details() { + let captured = CapturedLog::default(); + let subscriber = tracing_subscriber::fmt() + .with_ansi(false) + .without_time() + .with_max_level(tracing::Level::WARN) + .with_writer(captured.clone()) + .finish(); + let mut config = object_lambda_test_config( + Url::parse("https://object-lambda.test/private?token=secret").expect("object lambda endpoint"), + ); + config.skip_tls_verify = true; + + tracing::subscriber::with_default(subscriber, || { + build_object_lambda_http_client_with_resolver(&config, StaticResolver(IpAddr::V4(std::net::Ipv4Addr::LOCALHOST))) + .expect("object lambda client should build"); + }); + + let logs = captured.contents(); + assert!(logs.contains("https://object-lambda.test")); + for secret in ["/private", "token=secret"] { + assert!(!logs.contains(secret), "TLS warning leaked {secret}: {logs}"); + } + } + + #[tokio::test] + async fn object_lambda_request_error_does_not_expose_endpoint_details() { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("reserve refused endpoint"); + let address = listener.local_addr().expect("refused endpoint address"); + drop(listener); + let request = reqwest::Client::builder() + .no_proxy() + .build() + .expect("test client should build") + .get(format!("http://{address}/private?token=secret")); + + let error = send_object_lambda_request(request) + .await + .expect_err("refused object lambda request should fail"); + let rendered = error.to_string(); + assert!(rendered.contains("object lambda target request failed")); + assert!(!rendered.contains("/private")); + assert!(!rendered.contains("token=secret")); + } + #[test] fn build_object_lambda_get_request_removes_lambda_arn_and_preserves_request_inputs() { let mut req = S3Request {