diff --git a/rustfs/src/connect/client.rs b/rustfs/src/connect/client.rs index 9d283460f..69a4071bb 100644 --- a/rustfs/src/connect/client.rs +++ b/rustfs/src/connect/client.rs @@ -22,6 +22,7 @@ use serde::Deserialize; use uuid::Uuid; use zeroize::Zeroizing; +use super::config::{ProxyConfig, ProxyConfigError}; use super::credential_store::{ CompletedRegistration, CredentialLock, CredentialStore, CredentialStoreError, DeviceCredential, PendingRegistration, PendingRotation, @@ -46,6 +47,7 @@ pub struct ConnectConfig<'a> { pub endpoint: &'a str, pub root_ca_pem: &'a [u8], pub timeout: Duration, + pub proxy: Option<&'a ProxyConfig>, } pub struct ConnectClient { @@ -54,6 +56,7 @@ pub struct ConnectClient { root_certificates: Vec>, client: Client, timeout: Duration, + proxy: Option, } pub(crate) enum RotationAttempt { @@ -106,13 +109,14 @@ impl ConnectClient { return Err(ClientError::RootCertificate); } - let client = build_client(&root_certificates, config.timeout, None)?; + let client = build_client(&root_certificates, config.timeout, None, config.proxy)?; Ok(Self { endpoint, roots, root_certificates, client, timeout: config.timeout, + proxy: config.proxy.cloned(), }) } @@ -327,7 +331,7 @@ impl ConnectClient { identity_pem.push(b'\n'); identity_pem.extend_from_slice(private_key.as_bytes()); let tls_identity = reqwest::Identity::from_pem(&identity_pem).map_err(|_| ClientError::IdentityCertificate)?; - let client = build_client(&self.root_certificates, self.timeout, Some(tls_identity))?; + let client = build_client(&self.root_certificates, self.timeout, Some(tls_identity), self.proxy.as_ref())?; let path = format!("clusterDevices/{}:rotateCredential", credential.uid); let url = self.url(&path)?; let response = match self.send_once(StatusCode::OK, client.post(url).json(&body)).await? { @@ -474,6 +478,7 @@ impl ConnectClient { F: FnMut() -> reqwest::RequestBuilder, { let mut last_status = None; + let mut last_transport_failure = None; for attempt in 0..MAX_ATTEMPTS { match request().send().await { Ok(response) if response.status() == success => return decode_response(response).await, @@ -499,13 +504,16 @@ impl ConnectClient { return Err(ClientError::Rejected { status, reason }); } Err(error) if !error.is_timeout() && !error.is_connect() => return Err(ClientError::Transport(error)), - Err(_) => {} + Err(error) => last_transport_failure = classify_transport_failure(&error, self.proxy.is_some()), } if attempt + 1 < MAX_ATTEMPTS { tokio::time::sleep(Duration::from_millis(50 * (attempt as u64 + 1))).await; } } + if let Some(failure) = last_transport_failure { + return Err(failure.into()); + } Err(ClientError::Unavailable { status: last_status }) } @@ -513,6 +521,9 @@ impl ConnectClient { let response = match request.send().await { Ok(response) => response, Err(error) if error.is_timeout() || error.is_connect() => { + if let Some(failure) = classify_transport_failure(&error, self.proxy.is_some()) { + return Err(failure.into()); + } return Ok(SingleRequest::Unavailable { status: None, retry_after: None, @@ -600,26 +611,67 @@ fn retry_after(headers: &header::HeaderMap, now: DateTime) -> Option], timeout: Duration, identity: Option, + proxy: Option<&ProxyConfig>, ) -> Result { let certificates = roots .iter() .map(|root| reqwest::Certificate::from_der(root.as_ref())) .collect::, _>>()?; let mut builder = Client::builder() + .no_proxy() .https_only(true) .redirect(reqwest::redirect::Policy::none()) .timeout(timeout) .tls_certs_only(certificates); + if let Some(proxy) = proxy { + builder = proxy.apply(builder)?; + } if let Some(identity) = identity { builder = builder.identity(identity); } builder.build().map_err(ClientError::Transport) } +#[derive(Clone, Copy)] +pub(crate) enum TransportFailure { + ProxyAuthentication, + ProxyRejected, + TlsPeer, +} + +pub(crate) fn classify_transport_failure(error: &reqwest::Error, proxy_configured: bool) -> Option { + let mut source: Option<&(dyn std::error::Error + 'static)> = Some(error); + while let Some(error) = source { + let message = error.to_string().to_ascii_lowercase(); + if message.contains("407") || message.contains("proxy authentication") { + return Some(TransportFailure::ProxyAuthentication); + } + if message.contains("certificate") + || message.contains("unknown issuer") + || message.contains("invalid peer") + || message.contains("not valid for") + { + return Some(TransportFailure::TlsPeer); + } + source = error.source(); + } + proxy_configured.then_some(TransportFailure::ProxyRejected) +} + +impl From for ClientError { + fn from(failure: TransportFailure) -> Self { + match failure { + TransportFailure::ProxyAuthentication => Self::ProxyAuthentication, + TransportFailure::ProxyRejected => Self::ProxyRejected, + TransportFailure::TlsPeer => Self::TlsPeer, + } + } +} + async fn decode_response(mut response: reqwest::Response) -> Result { let body = read_body(&mut response).await?; serde_json::from_slice(&body).map_err(|_| ClientError::Response) @@ -663,6 +715,16 @@ pub enum ClientError { Endpoint, #[error("Connect root CA configuration is invalid")] RootCertificate, + #[error("Connect proxy configuration is invalid")] + ProxyConfiguration(#[from] ProxyConfigError), + #[error("Connect proxy authentication failed; verify the configured proxy credential files")] + ProxyAuthentication, + #[error( + "Connect proxy connection failed; verify proxy availability, credentials, the proxy allow-list, and the Connect endpoint" + )] + ProxyRejected, + #[error("Connect TLS peer certificate validation failed; verify the endpoint and configured root CA")] + TlsPeer, #[error( "Connect registration has a pending attempt for a different token; restore the original protected token configuration" )] diff --git a/rustfs/src/connect/config.rs b/rustfs/src/connect/config.rs index 8ecc6f04b..a47f622b8 100644 --- a/rustfs/src/connect/config.rs +++ b/rustfs/src/connect/config.rs @@ -14,16 +14,273 @@ use std::env; use std::ffi::OsString; -#[cfg(target_os = "linux")] -use std::fs; +use std::fmt; +#[cfg(unix)] +use std::fs::{self, OpenOptions}; +#[cfg(unix)] +use std::io::Read as _; +#[cfg(unix)] +use std::os::unix::fs::{MetadataExt as _, OpenOptionsExt as _, PermissionsExt as _}; use std::path::PathBuf; use std::time::Duration; +use reqwest::{ClientBuilder, NoProxy, Proxy, Url}; +use zeroize::Zeroizing; + use super::{CredentialStore, IdentityStore}; pub const ENV_CONNECT_ENDPOINT: &str = "RUSTFS_CONNECT_ENDPOINT"; pub const ENV_CONNECT_ROOT_CA_FILE: &str = "RUSTFS_CONNECT_ROOT_CA_FILE"; pub const ENV_CONNECT_STATE_DIR: &str = "RUSTFS_CONNECT_STATE_DIR"; +pub const ENV_CONNECT_PROXY_URL: &str = "RUSTFS_CONNECT_PROXY_URL"; +pub const ENV_CONNECT_PROXY_BYPASS: &str = "RUSTFS_CONNECT_PROXY_BYPASS"; +pub const ENV_CONNECT_PROXY_USERNAME_FILE: &str = "RUSTFS_CONNECT_PROXY_USERNAME_FILE"; +pub const ENV_CONNECT_PROXY_PASSWORD_FILE: &str = "RUSTFS_CONNECT_PROXY_PASSWORD_FILE"; + +const MAX_PROXY_BYPASS_BYTES: usize = 2048; +const MAX_PROXY_USERNAME_BYTES: usize = 256; +const MAX_PROXY_PASSWORD_BYTES: usize = 4096; + +/// Explicit HTTP CONNECT proxy configuration for RustFS Connect traffic. +#[derive(Clone)] +pub struct ProxyConfig { + url: Url, + bypass: Option, + username: Option>, + password: Option>, +} + +impl ProxyConfig { + /// Creates an unauthenticated proxy configuration. + pub fn new(url: &str, bypass: Option<&str>) -> Result { + let url = proxy_url(url)?; + let bypass = proxy_bypass(bypass)?; + Ok(Self { + url, + bypass, + username: None, + password: None, + }) + } + + /// Adds HTTP Basic authentication without placing credentials in the proxy URL. + pub fn with_basic_auth(mut self, username: &str, password: &str) -> Result { + validate_proxy_secret(username, MAX_PROXY_USERNAME_BYTES)?; + validate_proxy_secret(password, MAX_PROXY_PASSWORD_BYTES)?; + self.username = Some(Zeroizing::new(username.to_owned())); + self.password = Some(Zeroizing::new(password.to_owned())); + Ok(self) + } + + /// Loads an explicit proxy and optional protected Basic-auth files from RustFS-specific environment variables. + pub fn from_env() -> Result, ProxyConfigError> { + Self::from_env_values( + env::var_os(ENV_CONNECT_PROXY_URL), + env::var_os(ENV_CONNECT_PROXY_BYPASS), + env::var_os(ENV_CONNECT_PROXY_USERNAME_FILE), + env::var_os(ENV_CONNECT_PROXY_PASSWORD_FILE), + ) + } + + pub(crate) fn apply(&self, builder: ClientBuilder) -> Result { + let mut proxy = Proxy::https(self.url.clone()).map_err(|_| ProxyConfigError::Url)?; + if let Some(bypass) = self.bypass.as_deref() { + proxy = proxy.no_proxy(NoProxy::from_string(bypass)); + } + if let (Some(username), Some(password)) = (&self.username, &self.password) { + proxy = proxy.basic_auth(username, password); + } + Ok(builder.proxy(proxy)) + } + + #[cfg(unix)] + fn from_env_values( + url: Option, + bypass: Option, + username_file: Option, + password_file: Option, + ) -> Result, ProxyConfigError> { + let configured = url.is_some() || bypass.is_some() || username_file.is_some() || password_file.is_some(); + if !configured { + return Ok(None); + } + let Some(url) = url else { + return Err(ProxyConfigError::Partial); + }; + if username_file.is_some() != password_file.is_some() { + return Err(ProxyConfigError::Partial); + } + let url = url.into_string().map_err(|_| ProxyConfigError::Encoding)?; + let bypass = bypass + .map(|value| value.into_string().map_err(|_| ProxyConfigError::Encoding)) + .transpose()?; + let mut config = Self::new(&url, bypass.as_deref())?; + if let (Some(username_file), Some(password_file)) = (username_file, password_file) { + let username = read_proxy_secret(PathBuf::from(username_file), MAX_PROXY_USERNAME_BYTES)?; + let password = read_proxy_secret(PathBuf::from(password_file), MAX_PROXY_PASSWORD_BYTES)?; + config = config.with_basic_auth(&username, &password)?; + } + Ok(Some(config)) + } + + #[cfg(not(unix))] + fn from_env_values( + url: Option, + bypass: Option, + username_file: Option, + password_file: Option, + ) -> Result, ProxyConfigError> { + if url.is_none() && bypass.is_none() && username_file.is_none() && password_file.is_none() { + Ok(None) + } else { + Err(ProxyConfigError::PlatformSecurity) + } + } +} + +impl fmt::Debug for ProxyConfig { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("ProxyConfig") + .field("configured", &true) + .field("bypass_configured", &self.bypass.is_some()) + .field("authentication_configured", &self.username.is_some()) + .finish() + } +} + +fn proxy_url(value: &str) -> Result { + let url = Url::parse(value).map_err(|_| ProxyConfigError::Url)?; + if url.scheme() != "http" + || url.host_str().is_none() + || url.cannot_be_a_base() + || !url.username().is_empty() + || url.password().is_some() + || !matches!(url.path(), "" | "/") + || url.query().is_some() + || url.fragment().is_some() + { + return Err(ProxyConfigError::Url); + } + Ok(url) +} + +fn proxy_bypass(value: Option<&str>) -> Result, ProxyConfigError> { + let Some(value) = value else { + return Ok(None); + }; + if value.is_empty() + || value.len() > MAX_PROXY_BYPASS_BYTES + || !value.is_ascii() + || value.split(',').any(|entry| !valid_bypass_entry(entry.trim())) + { + return Err(ProxyConfigError::Bypass); + } + Ok(Some(value.to_owned())) +} + +fn valid_bypass_entry(value: &str) -> bool { + if value == "*" || value.parse::().is_ok() { + return true; + } + if let Some((network, prefix)) = value.split_once('/') { + let Ok(address) = network.parse::() else { + return false; + }; + let Ok(prefix) = prefix.parse::() else { + return false; + }; + return prefix <= if address.is_ipv4() { 32 } else { 128 }; + } + let domain = value.strip_prefix('.').unwrap_or(value); + !domain.is_empty() + && domain.len() <= 253 + && domain.split('.').all(|label| { + !label.is_empty() + && label.len() <= 63 + && !label.starts_with('-') + && !label.ends_with('-') + && label.bytes().all(|byte| byte.is_ascii_alphanumeric() || byte == b'-') + }) +} + +fn validate_proxy_secret(value: &str, maximum: usize) -> Result<(), ProxyConfigError> { + if value.is_empty() || value.len() > maximum || value.chars().any(char::is_control) { + return Err(ProxyConfigError::Authentication); + } + Ok(()) +} + +#[cfg(unix)] +fn read_proxy_secret(path: PathBuf, maximum: usize) -> Result, ProxyConfigError> { + let initial = fs::symlink_metadata(&path).map_err(|source| ProxyConfigError::SecretFile { + path: path.clone(), + source, + })?; + if !initial.file_type().is_file() || initial.permissions().mode() & 0o077 != 0 { + return Err(ProxyConfigError::SecretFileSecurity { path }); + } + let mut options = OpenOptions::new(); + options.read(true).custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC); + let mut file = options.open(&path).map_err(|source| ProxyConfigError::SecretFile { + path: path.clone(), + source, + })?; + let opened = file.metadata().map_err(|source| ProxyConfigError::SecretFile { + path: path.clone(), + source, + })?; + if !opened.is_file() + || opened.uid() != process_uid() + || opened.dev() != initial.dev() + || opened.ino() != initial.ino() + || opened.len() > maximum as u64 + 2 + { + return Err(ProxyConfigError::SecretFileSecurity { path }); + } + let mut bytes = Zeroizing::new(Vec::with_capacity(opened.len() as usize)); + file.read_to_end(&mut bytes).map_err(|source| ProxyConfigError::SecretFile { + path: path.clone(), + source, + })?; + while matches!(bytes.last(), Some(b'\n' | b'\r')) { + bytes.pop(); + } + let value = Zeroizing::new(String::from_utf8(std::mem::take(&mut *bytes)).map_err(|_| ProxyConfigError::Authentication)?); + validate_proxy_secret(&value, maximum)?; + Ok(value) +} + +#[cfg(unix)] +// SAFETY: geteuid has no pointer arguments or caller preconditions. +#[allow(unsafe_code)] +fn process_uid() -> u32 { + unsafe { libc::geteuid() } +} + +#[derive(Debug, thiserror::Error)] +pub enum ProxyConfigError { + #[error("Connect proxy configuration requires RUSTFS_CONNECT_PROXY_URL and both or neither authentication files")] + Partial, + #[error("Connect proxy configuration is not valid UTF-8")] + Encoding, + #[error("Connect proxy must be an HTTP base URL without credentials, path, query, or fragment")] + Url, + #[error("Connect proxy bypass rules are invalid")] + Bypass, + #[error("Connect proxy credentials are invalid")] + Authentication, + #[error("Connect proxy credential file must be an owner-only regular file: {path}")] + SecretFileSecurity { path: PathBuf }, + #[error("Connect proxy credential file could not be read: {path}")] + SecretFile { + path: PathBuf, + #[source] + source: std::io::Error, + }, + #[error("Connect proxy credential files require Unix filesystem security guarantees")] + PlatformSecurity, +} #[derive(Clone, Copy, Debug)] pub struct HeartbeatSchedule { @@ -54,6 +311,7 @@ pub struct HeartbeatConfig { pub credential_store: CredentialStore, pub state_path: PathBuf, pub schedule: HeartbeatSchedule, + pub proxy: Option, } impl HeartbeatConfig { @@ -72,6 +330,7 @@ impl HeartbeatConfig { credential_store, state_path, schedule: HeartbeatSchedule::default(), + proxy: None, } } @@ -84,6 +343,7 @@ impl HeartbeatConfig { credential_store: CredentialStore::new(state_root.join("credential")), state_path: state_root.join("heartbeat/state.json"), schedule: HeartbeatSchedule::default(), + proxy: None, } } @@ -100,6 +360,10 @@ impl HeartbeatConfig { env::var_os(ENV_CONNECT_ENDPOINT), env::var_os(ENV_CONNECT_ROOT_CA_FILE), env::var_os(ENV_CONNECT_STATE_DIR), + env::var_os(ENV_CONNECT_PROXY_URL), + env::var_os(ENV_CONNECT_PROXY_BYPASS), + env::var_os(ENV_CONNECT_PROXY_USERNAME_FILE), + env::var_os(ENV_CONNECT_PROXY_PASSWORD_FILE), ) } @@ -107,8 +371,20 @@ impl HeartbeatConfig { endpoint: Option, root_ca_file: Option, state_dir: Option, + proxy_url: Option, + proxy_bypass: Option, + proxy_username_file: Option, + proxy_password_file: Option, ) -> Result, HeartbeatConfigError> { - let configured = endpoint.is_some() || root_ca_file.is_some() || state_dir.is_some(); + let proxy_configured = + proxy_url.is_some() || proxy_bypass.is_some() || proxy_username_file.is_some() || proxy_password_file.is_some(); + let configured = endpoint.is_some() + || root_ca_file.is_some() + || state_dir.is_some() + || proxy_url.is_some() + || proxy_bypass.is_some() + || proxy_username_file.is_some() + || proxy_password_file.is_some(); if !configured { return Ok(None); } @@ -116,7 +392,10 @@ impl HeartbeatConfig { return Err(HeartbeatConfigError::Partial); }; let state_dir = PathBuf::from(state_dir); - if state_dir.as_os_str().is_empty() || endpoint.is_some() != root_ca_file.is_some() { + if state_dir.as_os_str().is_empty() + || endpoint.is_some() != root_ca_file.is_some() + || (proxy_configured && endpoint.is_none()) + { return Err(HeartbeatConfigError::Partial); } #[cfg(not(target_os = "linux"))] @@ -139,13 +418,19 @@ impl HeartbeatConfig { source, })?; #[cfg(target_os = "linux")] - Ok(Some(Self::new( - endpoint, - root_ca_pem, - IdentityStore::new(state_dir.join("identity")), - CredentialStore::new(state_dir.join("credential")), - state_dir.join("heartbeat/state.json"), - ))) + let proxy = ProxyConfig::from_env_values(proxy_url, proxy_bypass, proxy_username_file, proxy_password_file)?; + #[cfg(target_os = "linux")] + { + let mut config = Self::new( + endpoint, + root_ca_pem, + IdentityStore::new(state_dir.join("identity")), + CredentialStore::new(state_dir.join("credential")), + state_dir.join("heartbeat/state.json"), + ); + config.proxy = proxy; + Ok(Some(config)) + } } } @@ -165,17 +450,80 @@ pub enum HeartbeatConfigError { }, #[error("Connect inventory persistence requires Linux filesystem security guarantees")] PlatformSecurity, + #[error(transparent)] + Proxy(#[from] ProxyConfigError), } #[cfg(test)] mod tests { - use super::{HeartbeatConfig, HeartbeatConfigError}; + use super::{HeartbeatConfig, HeartbeatConfigError, ProxyConfig, ProxyConfigError}; use std::ffi::OsString; + #[test] + fn proxy_rejects_implicit_credentials_and_non_http_transport() { + assert!(matches!( + ProxyConfig::new("http://user:secret@proxy.example:8080", None), + Err(ProxyConfigError::Url) + )); + assert!(matches!(ProxyConfig::new("https://proxy.example:8443", None), Err(ProxyConfigError::Url))); + assert!(matches!( + ProxyConfig::new("http://proxy.example:8080/tunnel", None), + Err(ProxyConfigError::Url) + )); + } + + #[test] + fn proxy_debug_output_contains_no_endpoint_or_credentials() { + let proxy = ProxyConfig::new("http://sensitive-proxy.example:8080", Some("private.example")) + .expect("proxy") + .with_basic_auth("sensitive-user", "sensitive-password") + .expect("authentication"); + let debug = format!("{proxy:?}"); + + for secret in ["sensitive-proxy", "private.example", "sensitive-user", "sensitive-password"] { + assert!(!debug.contains(secret)); + } + assert!(debug.contains("authentication_configured: true")); + } + + #[test] + #[cfg(unix)] + fn proxy_authentication_requires_owner_only_regular_files() { + use std::os::unix::fs::PermissionsExt as _; + + let temp = tempfile::tempdir().expect("tempdir"); + let username = temp.path().join("username"); + let password = temp.path().join("password"); + std::fs::write(&username, b"proxy-user\n").expect("username"); + std::fs::write(&password, b"proxy-password\n").expect("password"); + std::fs::set_permissions(&username, std::fs::Permissions::from_mode(0o600)).expect("username mode"); + std::fs::set_permissions(&password, std::fs::Permissions::from_mode(0o644)).expect("password mode"); + + assert!(matches!( + ProxyConfig::from_env_values( + Some(OsString::from("http://proxy.example:8080")), + None, + Some(username.clone().into_os_string()), + Some(password.clone().into_os_string()), + ), + Err(ProxyConfigError::SecretFileSecurity { .. }) + )); + std::fs::set_permissions(&password, std::fs::Permissions::from_mode(0o600)).expect("private password mode"); + let proxy = ProxyConfig::from_env_values( + Some(OsString::from("http://proxy.example:8080")), + Some(OsString::from("localhost,127.0.0.1")), + Some(username.into_os_string()), + Some(password.into_os_string()), + ) + .expect("valid proxy") + .expect("configured proxy"); + assert!(!format!("{proxy:?}").contains("proxy-password")); + } + #[test] fn absent_environment_is_disabled_without_side_effects() { assert!( - HeartbeatConfig::from_env_values(None, None, None) + HeartbeatConfig::from_env_values(None, None, None, None, None, None, None) .expect("absent config") .is_none() ); @@ -184,7 +532,15 @@ mod tests { #[test] fn partial_environment_is_rejected() { assert!(matches!( - HeartbeatConfig::from_env_values(Some(OsString::from("https://connect.example/agent/")), None, None), + HeartbeatConfig::from_env_values( + Some(OsString::from("https://connect.example/agent/")), + None, + None, + None, + None, + None, + None, + ), Err(HeartbeatConfigError::Partial) )); assert!(matches!( @@ -192,11 +548,23 @@ mod tests { Some(OsString::from("https://connect.example/agent/")), Some(OsString::from("root.pem")), None, + None, + None, + None, + None, ), Err(HeartbeatConfigError::Partial) )); assert!(matches!( - HeartbeatConfig::from_env_values(None, Some(OsString::from("root.pem")), Some(OsString::from("state"))), + HeartbeatConfig::from_env_values( + None, + Some(OsString::from("root.pem")), + Some(OsString::from("state")), + None, + None, + None, + None, + ), Err(HeartbeatConfigError::Partial) )); } @@ -205,7 +573,7 @@ mod tests { #[cfg(target_os = "linux")] fn state_directory_alone_enables_local_inventory_without_transport() { let state = tempfile::tempdir().expect("tempdir").keep(); - let config = HeartbeatConfig::from_env_values(None, None, Some(state.clone().into_os_string())) + let config = HeartbeatConfig::from_env_values(None, None, Some(state.clone().into_os_string()), None, None, None, None) .expect("state-only config") .expect("enabled config"); @@ -224,6 +592,10 @@ mod tests { Some(OsString::from("https://connect.example/agent/")), Some(root.into_os_string()), Some(state.clone().into_os_string()), + None, + None, + None, + None, ) .expect("complete config") .expect("enabled config"); @@ -239,7 +611,7 @@ mod tests { #[cfg(not(target_os = "linux"))] fn configured_inventory_fails_without_linux_filesystem_guarantees() { assert!(matches!( - HeartbeatConfig::from_env_values(None, None, Some(OsString::from("state"))), + HeartbeatConfig::from_env_values(None, None, Some(OsString::from("state")), None, None, None, None), Err(HeartbeatConfigError::PlatformSecurity) )); assert!(matches!( @@ -247,6 +619,10 @@ mod tests { Some(OsString::from("https://connect.example/agent/")), Some(OsString::from("missing-root.pem")), Some(OsString::from("state")), + None, + None, + None, + None, ), Err(HeartbeatConfigError::PlatformSecurity) )); diff --git a/rustfs/src/connect/heartbeat.rs b/rustfs/src/connect/heartbeat.rs index 51e77ea64..f46b70713 100644 --- a/rustfs/src/connect/heartbeat.rs +++ b/rustfs/src/connect/heartbeat.rs @@ -395,6 +395,16 @@ pub enum HeartbeatError { Endpoint, #[error("Connect heartbeat root CA configuration is invalid")] RootCertificate, + #[error("Connect heartbeat proxy configuration is invalid")] + ProxyConfiguration, + #[error("Connect proxy authentication failed; verify the configured proxy credential files")] + ProxyAuthentication, + #[error( + "Connect proxy connection failed; verify proxy availability, credentials, the proxy allow-list, and the Connect endpoint" + )] + ProxyRejected, + #[error("Connect TLS peer certificate validation failed; verify the endpoint and configured root CA")] + TlsPeer, #[error("Connect heartbeat schedule is invalid")] Schedule, #[error("RustFS is not registered with Connect")] @@ -455,6 +465,10 @@ impl From for HeartbeatError { match error { TelemetryError::Endpoint => Self::Endpoint, TelemetryError::RootCertificate => Self::RootCertificate, + TelemetryError::ProxyConfiguration => Self::ProxyConfiguration, + TelemetryError::ProxyAuthentication => Self::ProxyAuthentication, + TelemetryError::ProxyRejected => Self::ProxyRejected, + TelemetryError::TlsPeer => Self::TlsPeer, TelemetryError::Schedule => Self::Schedule, TelemetryError::NotRegistered => Self::NotRegistered, TelemetryError::IdentityMissing => Self::IdentityMissing, diff --git a/rustfs/src/connect/mod.rs b/rustfs/src/connect/mod.rs index a4cf881b6..2380a093c 100644 --- a/rustfs/src/connect/mod.rs +++ b/rustfs/src/connect/mod.rs @@ -42,7 +42,10 @@ pub mod runtime; mod telemetry; pub use client::{ClientError, ConnectClient, ConnectConfig}; -pub use config::{HeartbeatConfig, HeartbeatConfigError, HeartbeatSchedule}; +pub use config::{ + ENV_CONNECT_PROXY_BYPASS, ENV_CONNECT_PROXY_PASSWORD_FILE, ENV_CONNECT_PROXY_URL, ENV_CONNECT_PROXY_USERNAME_FILE, + HeartbeatConfig, HeartbeatConfigError, HeartbeatSchedule, ProxyConfig, ProxyConfigError, +}; pub use credential_store::{CredentialStore, DeviceCredential}; pub use diagnostics::{ CLIENT_CAPABILITY, CLIENT_SCHEMA_VERSION, CPU_PROFILE_CAPABILITY, CaptureMode, ClientDiagnosticResult, ClientMeasurement, diff --git a/rustfs/src/connect/registration_bootstrap.rs b/rustfs/src/connect/registration_bootstrap.rs index 6231941ac..3c12f820c 100644 --- a/rustfs/src/connect/registration_bootstrap.rs +++ b/rustfs/src/connect/registration_bootstrap.rs @@ -25,7 +25,7 @@ use std::{ use super::TokenError; #[cfg(unix)] -use super::{ConnectClient, ConnectConfig, CredentialStore, IdentityStore, RegistrationToken}; +use super::{ConnectClient, ConnectConfig, CredentialStore, IdentityStore, ProxyConfig, RegistrationToken}; #[cfg(unix)] const REQUEST_TIMEOUT: Duration = Duration::from_secs(15); @@ -58,6 +58,16 @@ pub enum RegistrationBootstrapError { Input(#[source] io::Error), #[error("Connect registration configuration is invalid")] Configuration, + #[error("Connect registration proxy configuration is invalid")] + ProxyConfiguration, + #[error("Connect proxy authentication failed; verify the configured proxy credential files")] + ProxyAuthentication, + #[error( + "Connect proxy connection failed; verify proxy availability, credentials, the proxy allow-list, and the Connect endpoint" + )] + ProxyRejected, + #[error("Connect TLS peer certificate validation failed; verify the endpoint and configured root CA")] + TlsPeer, #[error("Connect registration exchange failed")] Exchange, #[error("Connect registration bootstrap requires Unix owner and permission guarantees")] @@ -85,10 +95,12 @@ pub async fn register_from_protected_input( token_file: Option<&Path>, ) -> Result { let root_ca_pem = read_regular_file(root_ca_file, false)?; + let proxy = ProxyConfig::from_env().map_err(|_| RegistrationBootstrapError::ProxyConfiguration)?; let client = ConnectClient::new(ConnectConfig { endpoint, root_ca_pem: &root_ca_pem, timeout: REQUEST_TIMEOUT, + proxy: proxy.as_ref(), }) .map_err(|_| RegistrationBootstrapError::Configuration)?; @@ -105,7 +117,12 @@ pub async fn register_from_protected_input( &token, ) .await - .map_err(|_| RegistrationBootstrapError::Exchange)?; + .map_err(|error| match error { + super::ClientError::ProxyAuthentication => RegistrationBootstrapError::ProxyAuthentication, + super::ClientError::ProxyRejected => RegistrationBootstrapError::ProxyRejected, + super::ClientError::TlsPeer => RegistrationBootstrapError::TlsPeer, + _ => RegistrationBootstrapError::Exchange, + })?; if credential.name != format!("{cluster_name}/clusterDevices/{}", credential.uid) { return Err(RegistrationBootstrapError::Exchange); } diff --git a/rustfs/src/connect/runtime.rs b/rustfs/src/connect/runtime.rs index e6d43da92..d6cbcd00f 100644 --- a/rustfs/src/connect/runtime.rs +++ b/rustfs/src/connect/runtime.rs @@ -127,6 +127,7 @@ where endpoint: &config.endpoint, root_ca_pem: &config.root_ca_pem, timeout: config.schedule.timeout, + proxy: config.proxy.as_ref(), }) .map_err(rotation_failure)?; let identity_store = config.identity_store.clone(); @@ -417,6 +418,10 @@ fn rotation_failure(error: ClientError) -> HeartbeatError { match error { ClientError::Endpoint => HeartbeatError::Endpoint, ClientError::RootCertificate => HeartbeatError::RootCertificate, + ClientError::ProxyConfiguration(_) => HeartbeatError::ProxyConfiguration, + ClientError::ProxyAuthentication => HeartbeatError::ProxyAuthentication, + ClientError::ProxyRejected => HeartbeatError::ProxyRejected, + ClientError::TlsPeer => HeartbeatError::TlsPeer, ClientError::NotRegistered => HeartbeatError::NotRegistered, ClientError::IdentityMissing => HeartbeatError::IdentityMissing, ClientError::CredentialExpired | ClientError::CredentialNotYetValid => HeartbeatError::CredentialExpired, @@ -441,6 +446,10 @@ pub(crate) fn heartbeat_failure_reason(error: &HeartbeatError) -> &'static str { match error { HeartbeatError::Endpoint => "connect_heartbeat_endpoint", HeartbeatError::RootCertificate => "connect_heartbeat_root_certificate", + HeartbeatError::ProxyConfiguration => "connect_heartbeat_proxy_configuration", + HeartbeatError::ProxyAuthentication => "connect_heartbeat_proxy_authentication", + HeartbeatError::ProxyRejected => "connect_heartbeat_proxy_rejected", + HeartbeatError::TlsPeer => "connect_heartbeat_tls_peer", HeartbeatError::Schedule => "connect_heartbeat_schedule", HeartbeatError::NotRegistered => "connect_heartbeat_not_registered", HeartbeatError::IdentityMissing => "connect_heartbeat_identity_missing", diff --git a/rustfs/src/connect/telemetry.rs b/rustfs/src/connect/telemetry.rs index 3baef8cc2..d3bc82a27 100644 --- a/rustfs/src/connect/telemetry.rs +++ b/rustfs/src/connect/telemetry.rs @@ -21,7 +21,7 @@ use rustls::pki_types::{CertificateDer, pem::PemObject as _}; use serde::{Deserialize, Serialize}; use zeroize::Zeroizing; -use super::client::{ClientError, ConnectClient}; +use super::client::{ClientError, ConnectClient, TransportFailure, build_client, classify_transport_failure}; use super::config::HeartbeatConfig; use super::credential_store::{CredentialStoreError, DeviceCredential}; use super::identity::IdentityError; @@ -104,6 +104,13 @@ impl TelemetryTransport { let response = match authenticated.client.post(url).json(value).send().await { Ok(response) => response, Err(error) if error.is_timeout() || error.is_connect() || error.is_request() => { + if let Some(failure) = classify_transport_failure(&error, self.config.proxy.is_some()) { + return Err(match failure { + TransportFailure::ProxyAuthentication => TelemetryError::ProxyAuthentication, + TransportFailure::ProxyRejected => TelemetryError::ProxyRejected, + TransportFailure::TlsPeer => TelemetryError::TlsPeer, + }); + } return Ok(TelemetryDelivery::Retry { retry_after: None }); } Err(error) => return Err(error.into()), @@ -177,19 +184,8 @@ impl TelemetryTransport { pem.push(b'\n'); pem.extend_from_slice(key.as_bytes()); let identity = reqwest::Identity::from_pem(&pem).map_err(|_| TelemetryError::IdentityCertificate)?; - let roots = self - .roots - .iter() - .map(|root| reqwest::Certificate::from_der(root.as_ref())) - .collect::, _>>()?; - Client::builder() - .https_only(true) - .redirect(reqwest::redirect::Policy::none()) - .timeout(self.config.schedule.timeout) - .tls_certs_only(roots) - .identity(identity) - .build() - .map_err(Into::into) + build_client(&self.roots, self.config.schedule.timeout, Some(identity), self.config.proxy.as_ref()) + .map_err(credential_recovery_error) } } @@ -197,6 +193,10 @@ fn credential_recovery_error(error: ClientError) -> TelemetryError { match error { ClientError::Endpoint => TelemetryError::Endpoint, ClientError::RootCertificate => TelemetryError::RootCertificate, + ClientError::ProxyConfiguration(_) => TelemetryError::ProxyConfiguration, + ClientError::ProxyAuthentication => TelemetryError::ProxyAuthentication, + ClientError::ProxyRejected => TelemetryError::ProxyRejected, + ClientError::TlsPeer => TelemetryError::TlsPeer, ClientError::PendingRegistration | ClientError::PendingRotation => TelemetryError::StateConflict, ClientError::NotRegistered => TelemetryError::NotRegistered, ClientError::IdentityMissing => TelemetryError::IdentityMissing, @@ -291,6 +291,16 @@ pub(crate) enum TelemetryError { Endpoint, #[error("Connect telemetry root CA configuration is invalid")] RootCertificate, + #[error("Connect telemetry proxy configuration is invalid")] + ProxyConfiguration, + #[error("Connect proxy authentication failed; verify the configured proxy credential files")] + ProxyAuthentication, + #[error( + "Connect proxy connection failed; verify proxy availability, credentials, the proxy allow-list, and the Connect endpoint" + )] + ProxyRejected, + #[error("Connect TLS peer certificate validation failed; verify the endpoint and configured root CA")] + TlsPeer, #[error("Connect telemetry retry schedule is invalid")] Schedule, #[error("RustFS is not registered with Connect")] diff --git a/rustfs/tests/connect_heartbeat.rs b/rustfs/tests/connect_heartbeat.rs index 4a72efbb2..7280f8779 100644 --- a/rustfs/tests/connect_heartbeat.rs +++ b/rustfs/tests/connect_heartbeat.rs @@ -282,6 +282,7 @@ fn config_with_stores( initial_backoff: Duration::from_millis(20), max_backoff: Duration::from_millis(80), }, + proxy: None, } } diff --git a/rustfs/tests/connect_inventory.rs b/rustfs/tests/connect_inventory.rs index 0261320fc..ffcb73240 100644 --- a/rustfs/tests/connect_inventory.rs +++ b/rustfs/tests/connect_inventory.rs @@ -264,6 +264,7 @@ fn config(temp: &tempfile::TempDir, pki: &TestPki, server: &TestServer) -> Heart initial_backoff: Duration::from_millis(20), max_backoff: Duration::from_millis(80), }, + proxy: None, } } diff --git a/rustfs/tests/connect_registration.rs b/rustfs/tests/connect_registration.rs index 14b7983f3..bf957e3b2 100644 --- a/rustfs/tests/connect_registration.rs +++ b/rustfs/tests/connect_registration.rs @@ -33,7 +33,8 @@ use rcgen::{ }; use rustfs::connect::{ ClientError, CoarseNodeSummary, ConnectClient, ConnectConfig, CredentialStore, DeviceIdentity, HeartbeatConfig, - HeartbeatError, HeartbeatSchedule, HeartbeatStatus, IdentityStore, RegistrationToken, TokenError, spawn_heartbeat_runtime, + HeartbeatError, HeartbeatSchedule, HeartbeatStatus, IdentityStore, ProxyConfig, RegistrationToken, TokenError, + spawn_heartbeat_runtime, }; #[cfg(target_os = "linux")] use rustfs::connect::{InventorySchedule, InventorySnapshot, InventoryStatus, spawn_inventory_runtime}; @@ -44,7 +45,8 @@ use serde_json::{Value, json}; use sha2::{Digest as _, Sha256}; use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; -use tokio::net::TcpListener; +use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; +use tokio::net::{TcpListener, TcpStream}; use tokio::sync::{Notify, watch}; use tokio_rustls::TlsAcceptor; use tokio_util::sync::CancellationToken; @@ -213,6 +215,117 @@ impl Drop for TestServer { } } +#[derive(Clone, Debug, PartialEq, Eq)] +struct ProxyObservation { + authority: String, + authenticated: bool, +} + +#[derive(Clone, Copy)] +enum ProxyBehavior { + Forward, + RejectAuthentication, + InterruptTunnel, +} + +struct TestProxy { + endpoint: String, + observations: Arc>>, + task: tokio::task::JoinHandle<()>, +} + +impl Drop for TestProxy { + fn drop(&mut self) { + self.task.abort(); + } +} + +async fn proxy(allowed_authority: &str, username: &str, password: &str, behavior: ProxyBehavior) -> TestProxy { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind proxy"); + let address = listener.local_addr().expect("proxy address"); + let allowed_authority = allowed_authority.to_owned(); + let expected_auth = format!("Basic {}", BASE64_STANDARD.encode_to_string(format!("{username}:{password}"))); + let observations = Arc::new(Mutex::new(Vec::new())); + let captured = observations.clone(); + let task = tokio::spawn(async move { + while let Ok((mut inbound, _)) = listener.accept().await { + let allowed_authority = allowed_authority.clone(); + let expected_auth = expected_auth.clone(); + let observations = captured.clone(); + tokio::spawn(async move { + let mut request = Vec::with_capacity(1024); + let mut byte = [0_u8; 1]; + while request.len() < 8192 && !request.ends_with(b"\r\n\r\n") { + if inbound.read_exact(&mut byte).await.is_err() { + return; + } + request.push(byte[0]); + } + let Ok(request) = std::str::from_utf8(&request) else { + return; + }; + let mut lines = request.split("\r\n"); + let Some(authority) = lines + .next() + .and_then(|line| line.strip_prefix("CONNECT ")) + .and_then(|line| line.strip_suffix(" HTTP/1.1")) + else { + let _ = inbound + .write_all(b"HTTP/1.1 405 Method Not Allowed\r\nContent-Length: 0\r\n\r\n") + .await; + return; + }; + let authenticated = lines.any(|line| { + line.strip_prefix("Proxy-Authorization: ") + .or_else(|| line.strip_prefix("proxy-authorization: ")) + == Some(expected_auth.as_str()) + }); + observations.lock().expect("proxy observations").push(ProxyObservation { + authority: authority.to_owned(), + authenticated, + }); + if matches!(behavior, ProxyBehavior::RejectAuthentication) || !authenticated { + let _ = inbound + .write_all(b"HTTP/1.1 407 Proxy Authentication Required\r\nContent-Length: 0\r\n\r\n") + .await; + return; + } + if authority != allowed_authority { + let _ = inbound + .write_all(b"HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\n\r\n") + .await; + return; + } + let Ok(mut outbound) = TcpStream::connect(&allowed_authority).await else { + let _ = inbound + .write_all(b"HTTP/1.1 502 Bad Gateway\r\nContent-Length: 0\r\n\r\n") + .await; + return; + }; + if inbound + .write_all(b"HTTP/1.1 200 Connection Established\r\n\r\n") + .await + .is_err() + || matches!(behavior, ProxyBehavior::InterruptTunnel) + { + return; + } + let _ = tokio::io::copy_bidirectional(&mut inbound, &mut outbound).await; + }); + } + }); + TestProxy { + endpoint: format!("http://{address}"), + observations, + task, + } +} + +fn endpoint_authority(endpoint: &str) -> String { + let url = reqwest::Url::parse(endpoint).expect("endpoint URL"); + format!("{}:{}", url.host_str().expect("endpoint host"), url.port().expect("endpoint port")) +} + async fn server(pki: &TestPki, replies: Vec) -> TestServer { server_with_client_auth(pki, replies, false).await } @@ -485,10 +598,205 @@ fn client(server: &TestServer, pki: &TestPki, timeout: Duration) -> ConnectClien endpoint: &server.endpoint, root_ca_pem: pki.root_pem.as_bytes(), timeout, + proxy: None, }) .expect("build Connect client") } +fn client_with_proxy(endpoint: &str, pki: &TestPki, proxy: &ProxyConfig, timeout: Duration) -> ConnectClient { + ConnectClient::new(ConnectConfig { + endpoint, + root_ca_pem: pki.root_pem.as_bytes(), + timeout, + proxy: Some(proxy), + }) + .expect("build proxied Connect client") +} + +#[tokio::test] +async fn explicit_proxy_carries_registration_rotation_and_heartbeat_with_mtls() { + let temp = tempfile::tempdir().expect("tempdir"); + let pki = TestPki::new(); + let next = stage_next_identity(&temp); + let (identity_store, credential_store) = stores(&temp); + let current = identity_store.load_or_create().expect("current identity"); + let registered = pki.credential(¤t, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 0x31); + let (rotated, _) = rotation_response(&pki, &next, 0x32); + let server = server( + &pki, + vec![ + Reply::Json(StatusCode::CREATED, registered.clone()), + Reply::VerifiedRotation { + response: rotated, + current_public_key: current.public_key_der(), + current_certificate_fingerprint: certificate_fingerprint( + registered["certificate"].as_str().expect("registered certificate"), + ), + device_name: registered["name"].as_str().expect("device name").to_owned(), + }, + Reply::Json(StatusCode::OK, heartbeat_response("2026-09-13T01:02:03Z")), + ], + ) + .await; + let authority = endpoint_authority(&server.endpoint); + let proxy_server = proxy(&authority, "proxy-user", "proxy-password", ProxyBehavior::Forward).await; + let proxy_config = ProxyConfig::new(&proxy_server.endpoint, None) + .expect("proxy URL") + .with_basic_auth("proxy-user", "proxy-password") + .expect("proxy authentication"); + let client = client_with_proxy(&server.endpoint, &pki, &proxy_config, Duration::from_secs(2)); + + let credential = client + .register(&identity_store, &credential_store, &token()) + .await + .expect("registration through proxy"); + let due = credential.not_after_unix - EXPECTED_ROTATION_THRESHOLD_SECONDS; + client + .rotate_if_due(&identity_store, &credential_store, due) + .await + .expect("rotation through proxy") + .expect("rotation due"); + + let mut config = HeartbeatConfig::new( + &server.endpoint, + pki.root_pem.as_bytes(), + identity_store, + credential_store, + temp.path().join("heartbeat/state.json"), + ); + config.proxy = Some(proxy_config); + config.schedule = HeartbeatSchedule { + cadence: Duration::from_secs(30), + jitter: Duration::ZERO, + timeout: Duration::from_secs(2), + initial_backoff: Duration::from_millis(20), + max_backoff: Duration::from_millis(80), + }; + let shutdown = CancellationToken::new(); + let runtime = spawn_heartbeat_runtime(Some(config), &shutdown, || CoarseNodeSummary::new(1, 1, 0).expect("node summary")) + .expect("heartbeat runtime") + .expect("configured heartbeat runtime"); + let mut status = runtime.status(); + assert!(matches!( + wait_for_heartbeat_status(&mut status, |status| matches!(status, HeartbeatStatus::Online { .. })).await, + HeartbeatStatus::Online { .. } + )); + runtime.shutdown().await; + + assert_eq!( + server.paths.lock().expect("paths").as_slice(), + [ + "/agent/registrationTokens:exchange", + "/agent/clusterDevices/0198f4b0-3c00-7e30-8f41-4a5b6c7d8e92:rotateCredential", + "/agent/clusters/0198f4b0-2b00-7d20-9e31-3f4a5b6c7d81/heartbeats", + ] + ); + let certificates = server.client_certificates.lock().expect("client certificates"); + assert_eq!(certificates.len(), 3); + assert!(certificates[0].is_none(), "registration is the unauthenticated bootstrap operation"); + assert!(certificates[1].is_some(), "rotation must retain mTLS through CONNECT"); + assert!(certificates[2].is_some(), "telemetry must retain mTLS through CONNECT"); + drop(certificates); + let observations = proxy_server.observations.lock().expect("proxy observations"); + assert_eq!(observations.len(), 3); + assert!( + observations + .iter() + .all(|observation| { observation.authority == authority && observation.authenticated }) + ); +} + +#[tokio::test] +async fn proxy_bypass_preserves_direct_connectivity() { + let temp = tempfile::tempdir().expect("tempdir"); + let pki = TestPki::new(); + let (identity_store, credential_store) = stores(&temp); + let identity = identity_store.load_or_create().expect("identity"); + let server = server( + &pki, + vec![Reply::Json( + StatusCode::CREATED, + pki.credential(&identity, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 0x33), + )], + ) + .await; + let proxy_server = proxy( + &endpoint_authority(&server.endpoint), + "proxy-user", + "proxy-password", + ProxyBehavior::RejectAuthentication, + ) + .await; + let proxy_config = ProxyConfig::new(&proxy_server.endpoint, Some("localhost")) + .expect("proxy URL") + .with_basic_auth("wrong-user", "wrong-password") + .expect("proxy authentication"); + + client_with_proxy(&server.endpoint, &pki, &proxy_config, Duration::from_secs(2)) + .register(&identity_store, &credential_store, &token()) + .await + .expect("bypassed direct registration"); + assert!(proxy_server.observations.lock().expect("proxy observations").is_empty()); +} + +#[tokio::test] +async fn proxy_failures_are_actionable_and_redacted() { + let pki = TestPki::new(); + let server = server(&pki, vec![]).await; + let authority = endpoint_authority(&server.endpoint); + let cases = [ + (ProxyBehavior::RejectAuthentication, "credentials"), + (ProxyBehavior::InterruptTunnel, "availability"), + ]; + for (behavior, expected) in cases { + let temp = tempfile::tempdir().expect("tempdir"); + let (identity_store, credential_store) = stores(&temp); + let proxy_server = proxy(&authority, "secret-user", "secret-password", behavior).await; + let proxy_config = ProxyConfig::new(&proxy_server.endpoint, None) + .expect("proxy URL") + .with_basic_auth("secret-user", "secret-password") + .expect("proxy authentication"); + let error = client_with_proxy(&server.endpoint, &pki, &proxy_config, Duration::from_millis(300)) + .register(&identity_store, &credential_store, &token()) + .await + .expect_err("proxy failure"); + let diagnostic = error.to_string().to_ascii_lowercase(); + assert!(diagnostic.contains(expected), "{diagnostic}"); + assert!(!diagnostic.contains("secret-user")); + assert!(!diagnostic.contains("secret-password")); + } +} + +#[tokio::test] +async fn proxy_preserves_endpoint_allow_list_and_tls_roots() { + let pki = TestPki::new(); + let server = server(&pki, vec![]).await; + let authority = endpoint_authority(&server.endpoint); + let proxy_server = proxy(&authority, "proxy-user", "proxy-password", ProxyBehavior::Forward).await; + let proxy_config = ProxyConfig::new(&proxy_server.endpoint, None) + .expect("proxy URL") + .with_basic_auth("proxy-user", "proxy-password") + .expect("proxy authentication"); + + let temp = tempfile::tempdir().expect("tempdir"); + let (identity_store, credential_store) = stores(&temp); + let wrong_target = server.endpoint.replace(&authority, "localhost:9"); + let error = client_with_proxy(&wrong_target, &pki, &proxy_config, Duration::from_millis(300)) + .register(&identity_store, &credential_store, &token()) + .await + .expect_err("unapproved CONNECT target"); + assert!(matches!(error, ClientError::ProxyRejected)); + + let temp = tempfile::tempdir().expect("tempdir"); + let (identity_store, credential_store) = stores(&temp); + let untrusted = TestPki::new(); + let error = client_with_proxy(&server.endpoint, &untrusted, &proxy_config, Duration::from_millis(300)) + .register(&identity_store, &credential_store, &token()) + .await + .expect_err("untrusted Connect certificate"); + assert!(matches!(error, ClientError::TlsPeer)); +} + fn due_runtime_config( temp: &tempfile::TempDir, pki: &TestPki,