From b44e82fef1a61172797d35b92424d84963474747 Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 20 Jul 2026 19:55:07 +0800 Subject: [PATCH] fix(notify): restore webhook HTTPS target initialization (#5060) Restore the workspace reqwest default feature stack for RustFS outbound HTTPS clients, while keeping per-crate extra APIs such as json, stream, and multipart explicit. Lazily initialize the notification runtime from admin target access when RUSTFS_NOTIFY_ENABLE=true is already effective, and add regression coverage for HTTPS webhook custom CA handling and target-list visibility. Fixes #5052. Co-authored-by: heihutu --- Cargo.toml | 2 +- crates/e2e_test/Cargo.toml | 2 +- .../e2e_test/src/notification_webhook_test.rs | 156 +++++++++++++++++- crates/ecstore/Cargo.toml | 2 +- crates/iam/Cargo.toml | 2 +- crates/keystone/Cargo.toml | 2 +- crates/kms/Cargo.toml | 2 +- crates/policy/Cargo.toml | 2 +- crates/rio/Cargo.toml | 2 +- crates/targets/Cargo.toml | 2 +- crates/targets/src/target/webhook.rs | 59 +++++++ crates/trusted-proxies/Cargo.toml | 2 +- rustfs/Cargo.toml | 2 +- rustfs/src/admin/handlers/event.rs | 2 +- .../admin/handlers/notify_runtime_access.rs | 22 ++- 15 files changed, 241 insertions(+), 20 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 51dbad82d..8d077ee40 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -156,7 +156,7 @@ http = "1.4.2" http-body = "1.1.0" http-body-util = "0.1.4" minlz = "1.2.3" -reqwest = { default-features = false, version = "0.13.4" } +reqwest = "0.13.4" rustfs-kafka-async = { version = "1.2.0" } socket2 = { version = "0.6.5" } tokio = { version = "1.53.0" } diff --git a/crates/e2e_test/Cargo.toml b/crates/e2e_test/Cargo.toml index bb4298544..6a91afc8a 100644 --- a/crates/e2e_test/Cargo.toml +++ b/crates/e2e_test/Cargo.toml @@ -57,7 +57,7 @@ http.workspace = true http-body-util.workspace = true hyper = { workspace = true, features = ["http2", "http1", "server"] } hyper-util = { workspace = true, features = ["tokio", "server-auto", "server-graceful", "tracing"] } -reqwest = { workspace = true, default-features = false, features = ["rustls", "charset", "http2", "system-proxy", "stream", "json", "multipart"] } +reqwest = { workspace = true, features = ["json", "multipart", "stream"] } rustfs-signer.workspace = true tracing = { workspace = true } tracing-subscriber = { workspace = true, features = ["env-filter", "time"] } diff --git a/crates/e2e_test/src/notification_webhook_test.rs b/crates/e2e_test/src/notification_webhook_test.rs index e071a8d49..9d07182f9 100644 --- a/crates/e2e_test/src/notification_webhook_test.rs +++ b/crates/e2e_test/src/notification_webhook_test.rs @@ -43,6 +43,12 @@ use s3s::Body; use serde_json::Value; use serial_test::serial; use std::error::Error; +use std::path::Path; +use std::sync::{ + Arc, Once, + atomic::{AtomicBool, Ordering}, +}; +use std::thread; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpListener; use tokio::sync::mpsc; @@ -150,6 +156,83 @@ async fn spawn_event_collector() -> Result<(String, mpsc::UnboundedReceiver Result<(String, Arc, thread::JoinHandle<()>), BoxError> { + use rustls::{ + ServerConfig, + pki_types::{PrivateKeyDer, PrivatePkcs8KeyDer}, + }; + use std::io::ErrorKind; + use std::net::TcpListener as StdTcpListener; + + static INSTALL_CRYPTO_PROVIDER: Once = Once::new(); + INSTALL_CRYPTO_PROVIDER.call_once(|| { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + }); + + let listener = StdTcpListener::bind("0.0.0.0:0")?; + listener.set_nonblocking(true)?; + let addr = listener.local_addr()?; + let endpoint_ip = local_ip()?; + let endpoint_host = format!("{endpoint_ip}.nip.io"); + + let rcgen::CertifiedKey { cert, signing_key } = rcgen::generate_simple_self_signed(vec![endpoint_host.clone()])?; + std::fs::write(ca_path, cert.pem())?; + + let cert_chain = vec![cert.der().clone()]; + let key_der = PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(signing_key.serialize_der())); + let server_config = Arc::new( + ServerConfig::builder() + .with_no_client_auth() + .with_single_cert(cert_chain, key_der)?, + ); + + let running = Arc::new(AtomicBool::new(true)); + let server_running = Arc::clone(&running); + let handle = thread::spawn(move || { + while server_running.load(Ordering::Relaxed) { + match listener.accept() { + Ok((stream, _)) => { + let config = Arc::clone(&server_config); + handle_https_probe(stream, config); + } + Err(err) if err.kind() == ErrorKind::WouldBlock => thread::sleep(Duration::from_millis(20)), + Err(_) => break, + } + } + }); + + Ok((format!("https://{endpoint_host}:{}/events", addr.port()), running, handle)) +} + +fn handle_https_probe(stream: std::net::TcpStream, server_config: Arc) { + use std::io::{Read, Write}; + + let _ = stream.set_read_timeout(Some(Duration::from_secs(5))); + let _ = stream.set_write_timeout(Some(Duration::from_secs(5))); + let Ok(connection) = rustls::ServerConnection::new(server_config) else { + return; + }; + let mut tls_stream = rustls::StreamOwned::new(connection, stream); + let mut buf = [0u8; 1024]; + if tls_stream.read(&mut buf).is_err() { + return; + } + let response = "HTTP/1.1 200 OK\r\ncontent-length: 0\r\nconnection: close\r\n\r\n"; + let _ = tls_stream.write_all(response.as_bytes()); + let _ = tls_stream.flush(); +} + +fn stop_https_event_collector(endpoint: &str, running: Arc, handle: thread::JoinHandle<()>) -> TestResult { + running.store(false, Ordering::Relaxed); + if let Ok(parsed) = endpoint.parse::() + && let Some(port) = parsed.port() + { + let _ = std::net::TcpStream::connect(("127.0.0.1", port)); + } + handle.join().map_err(|_| "https event collector thread panicked")?; + Ok(()) +} + /// Decoded object key of the first record in an event envelope. fn event_key(envelope: &Value) -> Option { let raw = envelope["Records"][0]["s3"]["object"]["key"].as_str()?; @@ -284,13 +367,24 @@ async fn enable_notify_module(env: &RustFSTestEnvironment) -> TestResult { /// Registers a webhook notification target with a persistent queue directory, so /// delivery goes through the durable store-and-forward path. async fn configure_webhook_target(env: &RustFSTestEnvironment, target_name: &str, endpoint: &str) -> TestResult { + configure_webhook_target_with_key_values(env, target_name, vec![("endpoint", endpoint.to_string())]).await +} + +async fn configure_webhook_target_with_key_values( + env: &RustFSTestEnvironment, + target_name: &str, + mut key_values: Vec<(&str, String)>, +) -> TestResult { let queue_dir = format!("{}/notify-queue-{target_name}", env.temp_dir); tokio::fs::create_dir_all(&queue_dir).await?; + if !key_values.iter().any(|(key, _)| *key == "queue_dir") { + key_values.push(("queue_dir", queue_dir)); + } let payload = serde_json::json!({ - "key_values": [ - { "key": "endpoint", "value": endpoint }, - { "key": "queue_dir", "value": queue_dir }, - ] + "key_values": key_values + .into_iter() + .map(|(key, value)| serde_json::json!({ "key": key, "value": value })) + .collect::>(), }); let url = format!("{}/rustfs/admin/v3/target/notify_webhook/{target_name}", env.url); let response = signed_admin_request(env, http::Method::PUT, &url, Some(payload.to_string().into_bytes())).await?; @@ -321,6 +415,28 @@ async fn wait_for_target_registered(env: &RustFSTestEnvironment, target_name: &s Err(format!("target {target_name} was not registered in admin ARNs").into()) } +async fn wait_for_target_listed(env: &RustFSTestEnvironment, target_name: &str) -> TestResult { + let url = format!("{}/rustfs/admin/v3/target/list", env.url); + for _ in 0..40 { + let response = signed_admin_request(env, http::Method::GET, &url, None).await?; + if response.status() == StatusCode::OK { + let body: Value = serde_json::from_slice(&response.bytes().await?)?; + let listed = body["notification_endpoints"].as_array().is_some_and(|endpoints| { + endpoints.iter().any(|endpoint| { + endpoint["account_id"].as_str() == Some(target_name) + && endpoint["service"].as_str() == Some("webhook") + && endpoint["status"].as_str().is_some() + }) + }); + if listed { + return Ok(()); + } + } + tokio::time::sleep(Duration::from_millis(250)).await; + } + Err(format!("target {target_name} was not listed in admin targets").into()) +} + /// Binds a bucket to a webhook target for ObjectCreated:*/ObjectRemoved:* events, /// filtered to `prefix` + `suffix`. async fn put_notification_config(client: &Client, bucket: &str, target_name: &str, prefix: &str, suffix: &str) -> TestResult { @@ -367,6 +483,38 @@ fn trimmed_etag(value: Option<&str>) -> Option { // Tests // --------------------------------------------------------------------------- +/// Regression for rustfs#5052: with the notify module enabled through +/// RUSTFS_NOTIFY_ENABLE, an HTTPS webhook using a configured CA must be accepted +/// and remain visible in the admin target list. +#[tokio::test] +#[serial] +async fn test_https_webhook_target_lists_with_notify_env_enabled() -> TestResult { + 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 (endpoint, running, handle) = spawn_https_event_collector(&ca_path)?; + let target = "peri1https"; + + configure_webhook_target_with_key_values( + &env, + target, + vec![ + ("endpoint", endpoint.clone()), + ("client_ca", ca_path.to_string_lossy().into_owned()), + ], + ) + .await?; + wait_for_target_listed(&env, target).await?; + + env.stop_server(); + stop_https_event_collector(&endpoint, running, handle)?; + Ok(()) +} + /// PUT / multipart-complete / DELETE each deliver one event with correct fields, /// and the prefix/suffix filter drops non-matching keys. #[tokio::test] diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index 0b330a904..c35289f72 100644 --- a/crates/ecstore/Cargo.toml +++ b/crates/ecstore/Cargo.toml @@ -123,7 +123,7 @@ libc.workspace = true # (backlog#1178); "fs" comes from the workspace default. rustix = { workspace = true, features = ["process", "fs"] } rustfs-madmin.workspace = true -reqwest = { workspace = true, default-features = false, features = ["rustls", "http2", "system-proxy"] } +reqwest = { workspace = true } aes-gcm = { workspace = true, features = ["rand_core"] } chacha20poly1305.workspace = true aws-sdk-s3 = { workspace = true, default-features = false, features = ["sigv4a", "default-https-client", "rt-tokio"] } diff --git a/crates/iam/Cargo.toml b/crates/iam/Cargo.toml index 11538c2a1..fa91300cb 100644 --- a/crates/iam/Cargo.toml +++ b/crates/iam/Cargo.toml @@ -51,7 +51,7 @@ rustfs-utils = { workspace = true, features = ["path"] } rustfs-io-metrics.workspace = true tokio-util = { workspace = true, features = ["io", "compat"] } pollster.workspace = true -reqwest = { workspace = true, default-features = false, features = ["rustls", "http2", "system-proxy"] } +reqwest = { workspace = true } moka = { workspace = true, features = ["future"] } openidconnect = { workspace = true, default-features = false, features = ["accept-rfc3339-timestamps"] } http = { workspace = true } diff --git a/crates/keystone/Cargo.toml b/crates/keystone/Cargo.toml index 43014fbbf..0ff3a66ca 100644 --- a/crates/keystone/Cargo.toml +++ b/crates/keystone/Cargo.toml @@ -27,7 +27,7 @@ authors.workspace = true [dependencies] tokio = { workspace = true, features = ["rt", "sync"] } -reqwest = { workspace = true, default-features = false, features = ["rustls", "http2", "system-proxy", "json"] } +reqwest = { workspace = true, features = ["json"] } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true, features = ["raw_value"] } thiserror = { workspace = true } diff --git a/crates/kms/Cargo.toml b/crates/kms/Cargo.toml index c30a7aedf..aada8547b 100644 --- a/crates/kms/Cargo.toml +++ b/crates/kms/Cargo.toml @@ -60,7 +60,7 @@ rustfs-utils = { workspace = true } rustfs-security-governance = { workspace = true } # HTTP client for Vault -reqwest = { workspace = true, default-features = false, features = ["rustls", "http2", "system-proxy"] } +reqwest = { workspace = true } vaultrs = { workspace = true } [dev-dependencies] diff --git a/crates/policy/Cargo.toml b/crates/policy/Cargo.toml index 9a6e5282a..6d10ac82d 100644 --- a/crates/policy/Cargo.toml +++ b/crates/policy/Cargo.toml @@ -42,7 +42,7 @@ ipnetwork = { workspace = true, features = ["serde"] } base64-simd = { workspace = true } jsonwebtoken = { workspace = true, features = ["aws_lc_rs"] } regex = { workspace = true } -reqwest = { workspace = true, default-features = false, features = ["rustls", "http2", "system-proxy", "json"] } +reqwest = { workspace = true, features = ["json"] } chrono = { workspace = true, features = ["serde"] } tracing.workspace = true moka = { workspace = true, features = ["future"] } diff --git a/crates/rio/Cargo.toml b/crates/rio/Cargo.toml index 1ecd1c392..f5745ec04 100644 --- a/crates/rio/Cargo.toml +++ b/crates/rio/Cargo.toml @@ -43,7 +43,7 @@ crc-fast = { workspace = true } pin-project-lite.workspace = true serde = { workspace = true, features = ["derive"] } bytes = { workspace = true, features = ["serde"] } -reqwest = { workspace = true, default-features = false, features = ["rustls", "http2", "system-proxy", "stream"] } +reqwest = { workspace = true, features = ["stream"] } rustls-pki-types.workspace = true tokio-util = { workspace = true, features = ["io", "compat"] } faster-hex.workspace = true diff --git a/crates/targets/Cargo.toml b/crates/targets/Cargo.toml index a26a156e6..027d882e3 100644 --- a/crates/targets/Cargo.toml +++ b/crates/targets/Cargo.toml @@ -26,7 +26,7 @@ lapin = { workspace = true, default-features = false, features = ["tokio", "rust libc = { workspace = true } pulsar = { workspace = true, default-features = false, features = ["tokio-rustls-runtime", "telemetry"] } regex = { workspace = true } -reqwest = { workspace = true, default-features = false, features = ["rustls", "http2", "system-proxy"] } +reqwest = { workspace = true } rumqttc = { workspace = true, features = ["websocket"] } redis = { workspace = true, features = ["connection-manager", "tokio-rustls-comp", "tls-rustls-insecure"] } rustls = { workspace = true, default-features = false, features = ["aws-lc-rs", "logging", "tls12", "prefer-post-quantum", "std"] } diff --git a/crates/targets/src/target/webhook.rs b/crates/targets/src/target/webhook.rs index cfd29866b..7c45006bb 100644 --- a/crates/targets/src/target/webhook.rs +++ b/crates/targets/src/target/webhook.rs @@ -896,4 +896,63 @@ mod tests { handle.join().expect("mock server thread"); } + + #[tokio::test] + async fn test_webhook_client_reaches_https_origin_with_custom_ca() { + use rustls::{ + ServerConfig, ServerConnection, StreamOwned, + pki_types::{PrivateKeyDer, PrivatePkcs8KeyDer}, + }; + use std::io::{Read, Write}; + use std::net::TcpListener; + 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!["localhost".to_string()]).expect("cert should generate"); + let temp_dir = tempfile::tempdir().expect("tempdir"); + let ca_path = temp_dir.path().join("webhook-ca.pem"); + std::fs::write(&ca_path, cert.pem()).expect("write ca pem"); + + let cert_chain = vec![cert.der().clone()]; + let key_der = PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(signing_key.serialize_der())); + let server_config = Arc::new( + ServerConfig::builder() + .with_no_client_auth() + .with_single_cert(cert_chain, key_der) + .expect("server cert should be valid"), + ); + + let listener = TcpListener::bind("127.0.0.1:0").expect("bind tls server"); + let addr = listener.local_addr().expect("local addr"); + let handle = std::thread::spawn(move || { + let (stream, _) = listener.accept().expect("accept tls client"); + let connection = ServerConnection::new(server_config).expect("server connection"); + let mut tls_stream = StreamOwned::new(connection, stream); + let mut buf = [0u8; 1024]; + let _ = tls_stream.read(&mut buf); + let response = "HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok"; + tls_stream.write_all(response.as_bytes()).expect("write response"); + tls_stream.flush().expect("flush response"); + }); + + let args = WebhookArgs { + client_ca: ca_path.to_string_lossy().into_owned(), + ..base_args() + }; + let client = WebhookTarget::::build_http_client(&args).expect("build https client"); + let resp = client + .head(format!("https://localhost:{}/hook", addr.port())) + .send() + .await + .expect("https webhook probe should trust configured ca"); + + assert_eq!(resp.status(), reqwest::StatusCode::OK); + assert_eq!(resp.text_with_charset("utf-8").await.expect("read response body"), ""); + handle.join().expect("tls server thread"); + } } diff --git a/crates/trusted-proxies/Cargo.toml b/crates/trusted-proxies/Cargo.toml index c62442990..0a15e5a7f 100644 --- a/crates/trusted-proxies/Cargo.toml +++ b/crates/trusted-proxies/Cargo.toml @@ -31,7 +31,7 @@ http = { workspace = true } ipnetwork = { workspace = true, features = ["serde"] } metrics = { workspace = true } moka = { workspace = true, features = ["future", "sync"] } -reqwest = { workspace = true, default-features = false, features = ["rustls", "charset", "http2", "system-proxy", "json"] } +reqwest = { workspace = true, features = ["json"] } rustfs-config = { workspace = true } rustfs-utils = { workspace = true, features = ["net"] } serde = { workspace = true, features = ["derive"] } diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 1c662c56b..cd1570eb0 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -118,7 +118,7 @@ hyper-util = { workspace = true, features = ["tokio", "server-auto", "server-gra http.workspace = true http-body.workspace = true http-body-util.workspace = true -reqwest = { workspace = true, default-features = false, features = ["rustls", "charset", "http2", "system-proxy", "stream"] } +reqwest = { workspace = true, features = ["json", "stream"] } socket2 = { workspace = true, features = ["all"] } tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "signal", "process", "io-util", "fs"] } tokio-rustls = { workspace = true, default-features = false, features = ["logging", "tls12", "aws-lc-rs"] } diff --git a/rustfs/src/admin/handlers/event.rs b/rustfs/src/admin/handlers/event.rs index 799337f3e..3e6e2634c 100644 --- a/rustfs/src/admin/handlers/event.rs +++ b/rustfs/src/admin/handlers/event.rs @@ -371,7 +371,7 @@ impl Operation for ListTargetsArns { log_notification_target_operation_blocked!("list_target_arns", None, None, &reason); return Err(s3_error!(InvalidRequest, "{reason}")); } - let ns = get_notification_system()?; + let ns = get_notification_system().await?; let region = req.region.clone().ok_or_else(|| { warn!( diff --git a/rustfs/src/admin/handlers/notify_runtime_access.rs b/rustfs/src/admin/handlers/notify_runtime_access.rs index 77cc57b53..e76de9a70 100644 --- a/rustfs/src/admin/handlers/notify_runtime_access.rs +++ b/rustfs/src/admin/handlers/notify_runtime_access.rs @@ -12,16 +12,30 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::server::init_event_notifier; use rustfs_config::server_config::Config; use s3s::{S3Result, s3_error}; use std::sync::Arc; +use tokio::sync::Mutex; -pub(crate) fn get_notification_system() -> S3Result> { +static NOTIFICATION_SYSTEM_INIT_LOCK: Mutex<()> = Mutex::const_new(()); + +pub(crate) async fn get_notification_system() -> S3Result> { + if let Some(system) = rustfs_notify::notification_system() { + return Ok(system); + } + + let _guard = NOTIFICATION_SYSTEM_INIT_LOCK.lock().await; + if let Some(system) = rustfs_notify::notification_system() { + return Ok(system); + } + + init_event_notifier().await; rustfs_notify::notification_system().ok_or_else(|| s3_error!(InternalError, "notification system not initialized")) } pub(crate) async fn load_notification_config_snapshot() -> S3Result<(Arc, Config)> { - let system = get_notification_system()?; + let system = get_notification_system().await?; let config = system.config.read().await.clone(); Ok((system, config)) } @@ -31,7 +45,7 @@ pub(crate) async fn set_notification_target_config( target_name: &str, kvs: rustfs_config::server_config::KVS, ) -> S3Result<()> { - let system = get_notification_system()?; + let system = get_notification_system().await?; system .set_target_config(subsystem, target_name, kvs) .await @@ -39,7 +53,7 @@ pub(crate) async fn set_notification_target_config( } pub(crate) async fn remove_notification_target_config(subsystem: &str, target_name: &str) -> S3Result<()> { - let system = get_notification_system()?; + let system = get_notification_system().await?; system .remove_target_config(subsystem, target_name) .await