gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-18 16:45:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(connect): support explicit enterprise proxies (#7745)

Browse Source 
feat(connect): add explicit proxy transport
This commit is contained in:
Chris
2026-09-13 19:40:40 +08:00
committed by GitHub
parent 4ac1a9fbb9
commit da56f5bf5b
10 changed files with 841 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/connect/client.rs
+66 -4
View File
@@ -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<CertificateDer<'static>>,
client: Client,
timeout: Duration,
proxy: Option<ProxyConfig>,
}
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<Utc>) -> Option<Durati
})
}
fn build_client(
pub(crate) fn build_client(
roots: &[CertificateDer<'static>],
timeout: Duration,
identity: Option<reqwest::Identity>,
proxy: Option<&ProxyConfig>,
) -> Result<Client, ClientError> {
let certificates = roots
.iter()
.map(|root| reqwest::Certificate::from_der(root.as_ref()))
.collect::<Result<Vec<_>, _>>()?;
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<TransportFailure> {
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<TransportFailure> 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<CredentialResponse, ClientError> {
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"
)]
rustfs/src/connect/config.rs
+393 -17
View File
@@ -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<String>,
username: Option<Zeroizing<String>>,
password: Option<Zeroizing<String>>,
}
impl ProxyConfig {
/// Creates an unauthenticated proxy configuration.
pub fn new(url: &str, bypass: Option<&str>) -> Result<Self, ProxyConfigError> {
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<Self, ProxyConfigError> {
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<Option<Self>, 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<ClientBuilder, ProxyConfigError> {
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<OsString>,
bypass: Option<OsString>,
username_file: Option<OsString>,
password_file: Option<OsString>,
) -> Result<Option<Self>, 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<OsString>,
bypass: Option<OsString>,
username_file: Option<OsString>,
password_file: Option<OsString>,
) -> Result<Option<Self>, 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<Url, ProxyConfigError> {
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<Option<String>, 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::<std::net::IpAddr>().is_ok() {
return true;
}
if let Some((network, prefix)) = value.split_once('/') {
let Ok(address) = network.parse::<std::net::IpAddr>() else {
return false;
};
let Ok(prefix) = prefix.parse::<u8>() 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<Zeroizing<String>, 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<ProxyConfig>,
}
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<OsString>,
root_ca_file: Option<OsString>,
state_dir: Option<OsString>,
proxy_url: Option<OsString>,
proxy_bypass: Option<OsString>,
proxy_username_file: Option<OsString>,
proxy_password_file: Option<OsString>,
) -> Result<Option<Self>, 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)
));
rustfs/src/connect/heartbeat.rs
+14
View File
@@ -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<TelemetryError> 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,
rustfs/src/connect/mod.rs
+4 -1
View File
@@ -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,
rustfs/src/connect/registration_bootstrap.rs
+19 -2
View File
@@ -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<RegistrationBootstrapResult, RegistrationBootstrapError> {
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);
}
rustfs/src/connect/runtime.rs
+9
View File
@@ -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",
rustfs/src/connect/telemetry.rs
+24 -14
View File
@@ -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::<Result<Vec<_>, _>>()?;
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")]
rustfs/tests/connect_heartbeat.rs
+1
View File
@@ -282,6 +282,7 @@ fn config_with_stores(
initial_backoff: Duration::from_millis(20),
max_backoff: Duration::from_millis(80),
},
proxy: None,
}
}
rustfs/tests/connect_inventory.rs
+1
View File
@@ -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,
}
}
rustfs/tests/connect_registration.rs
+310 -2
View File
@@ -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<Mutex<Vec<ProxyObservation>>>,
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<Reply>) -> 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(&current, &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,