gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 08:18:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(security): enforce outbound connection policy (#5135)

Browse Source 
Co-authored-by: cxymds <cxymds@gmail.com>
This commit is contained in:
Zhengchao An
2026-07-23 06:11:28 +08:00
committed by GitHub
parent 6e88ab2a25
commit c9848a6096
10 changed files with 1366 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+2
View File
@@ -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",
crates/e2e_test/Cargo.toml
+1
View File
@@ -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
crates/e2e_test/src/notification_webhook_test.rs
+34 -18
View File
@@ -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<String, BoxError> {
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<Valu
let endpoint_ip = local_ip()?;
let (tx, rx) = mpsc::unbounded_channel();
let handle = serve_event_collector(listener, tx);
Ok((format!("http://{endpoint_ip}.nip.io:{port}/events"), rx, handle))
Ok((format!("http://{}/events", std::net::SocketAddr::new(endpoint_ip, port)), rx, handle))
}
struct HttpsEventCollector {
@@ -209,9 +215,9 @@ fn spawn_https_event_collector(ca_path: &Path) -> Result<HttpsEventCollector, Bo
listener.set_nonblocking(true)?;
let addr = listener.local_addr()?;
let endpoint_ip = local_ip()?;
let endpoint_host = format!("{endpoint_ip}.nip.io");
let endpoint_host = endpoint_ip.to_string();
let rcgen::CertifiedKey { cert, signing_key } = rcgen::generate_simple_self_signed(vec![endpoint_host.clone()])?;
let rcgen::CertifiedKey { cert, signing_key } = rcgen::generate_simple_self_signed(vec![endpoint_host])?;
std::fs::write(ca_path, cert.pem())?;
let cert_chain = vec![cert.der().clone()];
@@ -246,7 +252,7 @@ fn spawn_https_event_collector(ca_path: &Path) -> Result<HttpsEventCollector, Bo
});
Ok(HttpsEventCollector {
endpoint: format!("https://{endpoint_host}:{}/events", addr.port()),
endpoint: format!("https://{}/events", std::net::SocketAddr::new(endpoint_ip, addr.port())),
running,
handle: Some(handle),
events,
@@ -586,11 +592,17 @@ async fn test_https_webhook_target_delivers_event_with_notify_env_enabled() -> 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?;
crates/e2e_test/src/object_lambda_test.rs
+52 -19
View File
@@ -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<usize> {
async fn read_http_request(
stream: &mut tokio::net::TcpStream,
) -> Result<(HashMap<String, String>, Vec<u8>), Box<dyn Error + Send + Sync>> {
) -> Result<(String, HashMap<String, String>, Vec<u8>), Box<dyn Error + Send + Sync>> {
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::<usize>()?;
.map(|value| value.parse::<usize>())
.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<dyn Error + Send + Sync>> {
fn endpoint_origin(endpoint: &str) -> Result<String, Box<dyn Error + Send + Sync>> {
Ok(reqwest::Url::parse(endpoint)?.origin().ascii_serialization())
}
async fn start_rustfs_server_for_endpoint(
env: &mut RustFSTestEnvironment,
endpoint: &str,
) -> Result<(), Box<dyn Error + Send + Sync>> {
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<dyn Error + Send + Sync>> {
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<d
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-presigned";
let key = "input.txt";
@@ -656,7 +689,7 @@ async fn test_get_object_lambda_accepts_named_webhook_target_arn() -> 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";
crates/targets/src/config/common.rs
+14 -4
View File
@@ -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, TargetError> {
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 {
crates/targets/src/config/loader.rs
+36 -2
View File
@@ -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()),
crates/targets/src/target/webhook.rs
+218 -14
View File
@@ -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<Client, TargetError> {
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<Client, TargetError> {
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<E> Target<E> for WebhookTarget<E>
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<std::sync::Mutex<Vec<u8>>>);
struct CapturedLogWriter(std::sync::Arc<std::sync::Mutex<Vec<u8>>>);
impl std::io::Write for CapturedLogWriter {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
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::<serde_json::Value>::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::<SocketAddr>()
.expect("target address");
let args = WebhookArgs { endpoint, ..base_args() };
let client = WebhookTarget::<serde_json::Value>::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::<serde_json::Value>::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::<serde_json::Value>::build_http_client(&base_args()).expect("build client");
let url = Url::parse("http://unresolvable.test/").expect("invalid test domain URL");
let client = WebhookTarget::<serde_json::Value>::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::<serde_json::Value>::build_http_client(&base_args()).expect("build client");
let client =
WebhookTarget::<serde_json::Value>::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::<serde_json::Value>::build_http_client(&args).expect("build https client");
let client = WebhookTarget::<serde_json::Value>::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");
crates/utils/Cargo.toml
+3 -2
View File
@@ -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"]
crates/utils/src/egress.rs
+704 -34
View File
@@ -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<Result<OutboundPolicy, OutboundPolicyError>> = 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<String>,
}
impl OutboundPolicy {
pub fn from_env() -> Result<Self, OutboundPolicyError> {
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<Self, OutboundPolicyError> {
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::<IpAddr>() {
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<OutboundDnsResolver, OutboundPolicyError> {
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<String>,
#[cfg(test)]
overrides: Option<DnsOverrideAnswers>,
}
#[cfg(test)]
type DnsOverrideAnswers = Arc<Mutex<HashMap<String, VecDeque<Vec<SocketAddr>>>>>;
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<String, Vec<IpAddr>>) -> 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<Vec<SocketAddr>>) -> 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<Vec<SocketAddr>> = 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::<Vec<_>>();
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<String> {
url.host_str()
.map(|host| host.trim_end_matches('.').trim_matches(&['[', ']'][..]).to_ascii_lowercase())
}
fn normalized_origin(url: &Url) -> Option<String> {
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::<Ipv6Addr>().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<Ipv4Addr> {
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<Ipv4Addr> {
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<Ipv4Addr> {
#[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::<Vec<_>>();
assert_eq!(addrs, vec!["8.8.8.8".parse::<std::net::IpAddr>().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::<Vec<_>>();
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:?}"
);
}
}
rustfs/src/admin/router.rs
+302 -11
View File
@@ -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<String>,
}
#[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<Duration>,
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<ObjectLambd
}
fn build_object_lambda_http_client(config: &ObjectLambdaWebhookConfig) -> S3Result<reqwest::Client> {
validate_outbound_url(&config.endpoint)
.map_err(|err| s3_error!(InvalidRequest, "object lambda target endpoint is not allowed: {}", err))?;
let mut builder = reqwest::Client::builder().user_agent(rustfs_targets::get_user_agent(rustfs_targets::ServiceType::Basis));
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<reqwest::Client> {
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<reqwest::Response> {
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<std::sync::Mutex<Vec<u8>>>);
struct CapturedLogWriter(std::sync::Arc<std::sync::Mutex<Vec<u8>>>);
impl std::io::Write for CapturedLogWriter {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
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::<SocketAddr>()
.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 {