use std::{
collections::{BTreeMap, HashMap, HashSet},
env, fmt,
fs::{File, create_dir_all, metadata},
io::{ErrorKind, Read},
net::{SocketAddr, ToSocketAddrs},
ops::Range,
path::PathBuf,
};
use crate::{
ObjectKind,
certificate::split_certificate_chain,
listener_key::{InterfaceError, ListenerKey, validate_listener_interface},
logging::AccessLogFormat,
proto::command::{
ActivateListener, AddBackend, AddCertificate, CertificateAndKey, ClientAuthMode, Cluster,
CustomHttpAnswers, ForwardedHeaders, Header, HeaderMatchKind, HeaderPosition,
HeaderPredicate, HealthCheckConfig, HealthCheckHeader, HealthCheckHeaderValue,
HealthCheckMode, HstsConfig, HttpListenerConfig, HttpStatusRange, HttpsListenerConfig,
ListenerType, LoadBalancingAlgorithms, LoadBalancingParams, LoadMetric, MetricDetail,
MetricsConfiguration, PathRule, ProtobufAccessLogFormat, ProxyProtocolConfig,
RedirectPolicy, RedirectScheme, Request, RequestHttpFrontend, RequestTcpFrontend,
RequestUdpFrontend, RulePosition, ServerConfig, ServerMetricsConfig, ShardMode,
SocketAddress, TcpListenerConfig, TlsVersion, UdpAffinityKey, UdpClusterConfig,
UdpHealthConfig, UdpHealthMode, UdpListenerConfig, WorkerRequest, request::RequestType,
},
};
pub const DEFAULT_CIPHER_LIST: [&str; 9] = [
"TLS13_AES_256_GCM_SHA384",
"TLS13_AES_128_GCM_SHA256",
"TLS13_CHACHA20_POLY1305_SHA256",
"TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384",
"TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256",
"TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256",
"TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384",
"TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256",
"TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256",
];
pub const DEFAULT_SIGNATURE_ALGORITHMS: [&str; 9] = [
"ECDSA+SHA256",
"ECDSA+SHA384",
"ECDSA+SHA512",
"RSA+SHA256",
"RSA+SHA384",
"RSA+SHA512",
"RSA-PSS+SHA256",
"RSA-PSS+SHA384",
"RSA-PSS+SHA512",
];
pub const DEFAULT_GROUPS_LIST: [&str; 4] = ["X25519MLKEM768", "x25519", "P-256", "P-384"];
pub const DEFAULT_ALPN_PROTOCOLS: [&str; 2] = ["h2", "http/1.1"];
pub const DEFAULT_FRONT_TIMEOUT: u32 = 60;
pub const DEFAULT_COMMAND_SOCKET_MODE: u32 = 0o600;
pub const MAX_COMMAND_SOCKET_MODE: u32 = 0o777;
pub const DEFAULT_BACK_TIMEOUT: u32 = 30;
pub const DEFAULT_CONNECT_TIMEOUT: u32 = 3;
pub const DEFAULT_SNI_PREREAD_TIMEOUT: u32 = 5;
pub const DEFAULT_SNI_PREREAD_MAX_BYTES: u32 = 16384;
pub const MIN_SNI_PREREAD_MAX_BYTES: u32 = 5;
pub const DEFAULT_REQUEST_TIMEOUT: u32 = 10;
pub const DEFAULT_UDP_FRONT_TIMEOUT: u32 = 30;
pub const DEFAULT_UDP_BACK_TIMEOUT: u32 = 30;
pub const DEFAULT_UDP_MAX_RX_DATAGRAM_SIZE: u32 = 1500;
pub const DEFAULT_UDP_MAX_FLOWS: u32 = 0;
pub const DEFAULT_WORKER_TIMEOUT: u32 = 10;
pub const DEFAULT_STICKY_NAME: &str = "SOZUBALANCEID";
pub const DEFAULT_ZOMBIE_CHECK_INTERVAL: u32 = 1_800;
pub const DEFAULT_ACCEPT_QUEUE_TIMEOUT: u32 = 60;
pub const DEFAULT_HSTS_MAX_AGE: u32 = 31_536_000;
pub const DEFAULT_EVICT_ON_QUEUE_FULL: bool = false;
pub const DEFAULT_WORKER_COUNT: u16 = 2;
pub const DEFAULT_WORKER_AUTOMATIC_RESTART: bool = true;
pub const DEFAULT_AUTOMATIC_STATE_SAVE: bool = false;
pub const DEFAULT_MIN_BUFFERS: u64 = 1;
pub const DEFAULT_MAX_BUFFERS: u64 = 1_000;
pub const DEFAULT_BUFFER_SIZE: u64 = 16_393;
pub const H2_MIN_BUFFER_SIZE: u64 = 16_393;
pub const DEFAULT_MAX_CONNECTIONS: usize = 10_000;
pub const DEFAULT_COMMAND_BUFFER_SIZE: u64 = 1_000_000;
pub const DEFAULT_MAX_COMMAND_BUFFER_SIZE: u64 = 2_000_000;
pub const DEFAULT_DISABLE_CLUSTER_METRICS: bool = false;
pub const MAX_LOOP_ITERATIONS: usize = 100000;
pub const DEFAULT_SEND_TLS_13_TICKETS: u64 = 4;
pub const DEFAULT_LOG_TARGET: &str = "stdout";
pub const DEFAULT_MAX_CONNECTIONS_PER_IP: u64 = 0;
pub const DEFAULT_RETRY_AFTER: u32 = 60;
pub const DEFAULT_MAX_CONNECTIONS_PER_SUBNET: u64 = 0;
pub const DEFAULT_MAX_CONNECTION_ATTEMPTS: u32 = 5;
pub const MAX_CONNECTION_ATTEMPTS_UPPER_BOUND: u32 = u8::MAX as u32;
pub fn validate_max_connection_attempts(attempts: u32) -> Result<(), &'static str> {
if (1..=MAX_CONNECTION_ATTEMPTS_UPPER_BOUND).contains(&attempts) {
Ok(())
} else {
Err("max_connection_attempts must lie in 1..=255")
}
}
pub const HEALTH_CHECK_VALUE_FILE_UNSUPPORTED: &str = "health check header value_file references cannot be applied yet: the protected activation of their values is not available";
pub const HEALTH_CHECK_PORT_UNIX_REFUSED: &str = "health_check.port targets a backend IP address and cannot be combined with a unix_path backend; unset the probe port or the unix_path";
pub fn parse_backend_unix_path(
unix_path: &str,
) -> Result<std::os::unix::net::SocketAddr, &'static str> {
const TOO_LONG: &str = "unix_path does not fit a sockaddr_un (107 bytes at most)";
if unix_path.as_bytes().contains(&0) {
return Err("unix_path must not contain a NUL byte");
}
let address = if let Some(name) = unix_path.strip_prefix('@') {
if name.is_empty() {
return Err("an abstract unix_path needs a name after '@'");
}
parse_abstract_unix_name(name.as_bytes())
.map_err(|unsupported| unsupported.unwrap_or(TOO_LONG))?
} else if unix_path.starts_with('/') {
std::os::unix::net::SocketAddr::from_pathname(unix_path).map_err(|_| TOO_LONG)?
} else {
return Err("unix_path must be an absolute path or an @abstract name");
};
debug_assert!(
!address.is_unnamed(),
"a parsed unix_path never yields an unnamed address"
);
debug_assert_eq!(
address.as_pathname().is_some(),
!unix_path.starts_with('@'),
"only a path without '@' parses to a pathname address"
);
Ok(address)
}
#[cfg(any(target_os = "linux", target_os = "android"))]
fn parse_abstract_unix_name(
name: &[u8],
) -> Result<std::os::unix::net::SocketAddr, Option<&'static str>> {
#[cfg(target_os = "android")]
use std::os::android::net::SocketAddrExt;
#[cfg(target_os = "linux")]
use std::os::linux::net::SocketAddrExt;
std::os::unix::net::SocketAddr::from_abstract_name(name).map_err(|_| None)
}
#[cfg(not(any(target_os = "linux", target_os = "android")))]
fn parse_abstract_unix_name(
_name: &[u8],
) -> Result<std::os::unix::net::SocketAddr, Option<&'static str>> {
Err(Some("an @abstract unix_path needs Linux"))
}
pub const DEFAULT_SUBNET_IPV4_PREFIX: u32 = 32;
pub const DEFAULT_SUBNET_IPV6_PREFIX: u32 = 128;
pub const MAX_SUBNET_IPV4_PREFIX: u32 = 32;
pub const MAX_SUBNET_IPV6_PREFIX: u32 = 128;
#[derive(Debug)]
pub enum IncompatibilityKind {
PublicAddress,
ProxyProtocol,
}
#[derive(Debug)]
pub enum MissingKind {
Field(String),
Protocol,
SavedState,
}
#[derive(thiserror::Error, Debug)]
pub enum ConfigError {
#[error("env path not found: {0}")]
Env(String),
#[error("Could not open file {path_to_open}: {io_error}")]
FileOpen {
path_to_open: String,
io_error: std::io::Error,
},
#[error("Could not read file {path_to_read}: {io_error}")]
FileRead {
path_to_read: String,
io_error: std::io::Error,
},
#[error(
"the field {kind:?} of {object:?} with id or address {id} is incompatible with the rest of the options"
)]
Incompatible {
kind: IncompatibilityKind,
object: ObjectKind,
id: String,
},
#[error("Invalid '{0}' field for a TCP frontend")]
InvalidFrontendConfig(String),
#[error("invalid path {0:?}")]
InvalidPath(PathBuf),
#[error("listening address {0:?} is already used in the configuration")]
ListenerAddressAlreadyInUse(SocketAddr),
#[error("listener {0} is already declared in the configuration")]
ListenerAlreadyDeclared(ListenerKey),
#[error(
"listeners sharing the address {0} on different interfaces must all use the same protocol"
)]
ListenerProtocolMismatchOnAddress(SocketAddr),
#[error("listener {address}: {error}")]
ListenerInterface {
address: SocketAddr,
error: InterfaceError,
},
#[error("missing {0:?}")]
Missing(MissingKind),
#[error("could not get parent directory for file {0}")]
NoFileParent(String),
#[error("Could not get the path of the saved state")]
SaveStatePath(String),
#[error("Can not determine path to sozu socket: {0}")]
SocketPathError(String),
#[error("toml decoding error: {0}")]
DeserializeToml(String),
#[error("Can not set this frontend on a {0:?} listener")]
WrongFrontendProtocol(ListenerProtocol),
#[error("Can not build a {expected:?} listener from a {found:?} config")]
WrongListenerProtocol {
expected: ListenerProtocol,
found: Option<ListenerProtocol>,
},
#[error("Invalid ALPN protocol '{0}'. Valid values: \"h2\", \"http/1.1\"")]
InvalidAlpnProtocol(String),
#[error("cluster {cluster_id}: {reason}")]
InvalidAffinityKey {
cluster_id: String,
reason: &'static str,
},
#[error("cluster {cluster_id}: {reason}")]
InvalidShuffleSharding {
cluster_id: String,
reason: &'static str,
},
#[error("max_connection_attempts = {value}{}: {reason}", cluster_id.as_deref().map(|id| format!(" in cluster {id}")).unwrap_or_default())]
InvalidMaxConnectionAttempts {
cluster_id: Option<String>,
value: u32,
reason: &'static str,
},
#[error("cluster {cluster_id}, backend {address}: {reason}")]
InvalidBackendUnixPath {
cluster_id: String,
address: SocketAddr,
reason: &'static str,
},
#[error("cluster {cluster_id}: {reason}")]
InvalidHealthCheck {
cluster_id: String,
reason: &'static str,
},
#[error(
"{key} = {value} is out of range: a {family} prefix length must be between 0 and \
{maximum} bits inclusive."
)]
InvalidSubnetPrefix {
key: &'static str,
family: &'static str,
value: u32,
maximum: u32,
},
#[error("command_socket_mode = {value:?} is invalid: {reason}")]
InvalidCommandSocketMode { value: String, reason: &'static str },
#[error(
"disable_http11 = true is incompatible with alpn_protocols containing \"http/1.1\" \
on listener {address}. The proxy would advertise http/1.1 then refuse every \
connection that negotiates it. Drop \"http/1.1\" from alpn_protocols or unset \
disable_http11."
)]
DisableHttp11WithHttp11Alpn { address: String },
#[error(
"buffer_size = {buffer_size} is below the H2 minimum of {minimum} but \
{listeners} listener(s) serve HTTP/2 (an HTTPS listener advertising H2 \
ALPN, or an HTTP listener with h2c_prior_knowledge). The H2 mux deadlocks \
on full-size frames with smaller buffers. Raise buffer_size to >= {minimum}, \
or remove \"h2\" from those HTTPS listeners' alpn_protocols and turn \
h2c_prior_knowledge off on those HTTP listeners."
)]
BufferSizeTooSmallForH2 {
buffer_size: u64,
minimum: u64,
listeners: usize,
},
#[error(
"command_buffer_size = {command_buffer_size} exceeds max_command_buffer_size = \
{max_command_buffer_size}: lower command_buffer_size to <= {max_command_buffer_size} \
or raise max_command_buffer_size to >= {command_buffer_size}"
)]
CommandBufferSizeExceedsMax {
command_buffer_size: u64,
max_command_buffer_size: u64,
},
#[error(
"listener {address}: {key} = {value} is below the minimum of {minimum} the runtime can \
honour. Below the minimum the value does not relax the limit, it breaks it: the H2 flood \
detector compares count > threshold, so 0 is already exceeded the first time the counter \
it guards leaves zero. How far that reaches depends on the key — \
h2_max_header_list_size and h2_max_header_fields are also the HPACK decode budget, so 0 \
refuses every request on this listener with ENHANCE_YOUR_CALM (as a stream reset, or as \
a connection GOAWAY once a header block spans CONTINUATION frames), while a per-window \
or lifetime frame counter only trips once a client sends the frame it counts, and \
h2_max_concurrent_streams / h2_stream_shrink_ratio are instead promoted to their floor \
behind a warning. Raise {key} to >= {minimum}, or remove it to keep the default."
)]
H2ThresholdBelowMinimum {
address: String,
key: &'static str,
value: u64,
minimum: u64,
},
#[error(
"invalid redirect policy '{0}'. Valid values: \"forward\", \"permanent\", \"unauthorized\", \"silent_close\""
)]
InvalidRedirectPolicy(String),
#[error(
"invalid redirect scheme '{0}'. Valid values: \"use-same\", \"use-http\", \"use-https\""
)]
InvalidRedirectScheme(String),
#[error(
"invalid header position '{position}' at headers[{index}]. Valid values: \"request\", \"response\", \"both\""
)]
InvalidHeaderPosition { index: usize, position: String },
#[error(
"invalid header bytes in {field} at headers[{index}]: control characters \
(NUL / CR / LF / other C0) are forbidden in header keys and values"
)]
InvalidHeaderBytes { index: usize, field: &'static str },
#[error("invalid header predicate at header_predicates[{index}]: {reason}")]
InvalidHeaderPredicate { index: usize, reason: &'static str },
#[error("invalid HSTS config at {0}: `enabled` is required when an [hsts] block is present")]
HstsEnabledRequired(String),
#[error(
"invalid HSTS config at {0}: HSTS is only valid on HTTPS listeners and frontends \
(RFC 6797 §7.2 forbids the header over plaintext HTTP)"
)]
HstsOnPlainHttp(String),
#[error(
"invalid client auth config at {0}: client_auth / client_ca_certificates / \
client_ca_crls are only valid on HTTPS listeners"
)]
ClientAuthOnNonHttps(String),
#[error(
"invalid client auth config at {0}: client_ca_certificates / client_ca_crls are set \
but client_auth is not; set client_auth to \"optional\" or \"required\", or to \
\"none\" to keep them unused"
)]
ClientAuthModeMissing(String),
#[error(
"invalid SNI pattern '{sni}' for a TCP frontend: expected an exact hostname or a \
single leading \"*.\" wildcard label (e.g. \"example.com\" or \"*.example.com\"); \
'/' and non-leading '*' are rejected"
)]
InvalidSniPattern { sni: String },
#[error(
"non-ASCII SNI pattern '{sni}' for a TCP frontend: on-wire SNI is always an ASCII \
A-label (RFC 6066), so this pattern would never match a ClientHello. Write the \
punycode A-label form instead (e.g. \"xn--mnchen-3ya.example\" for \
\"münchen.example\")"
)]
NonAsciiSniPattern { sni: String },
#[error(
"TCP frontend {address} sets alpn but no hostname (sni): alpn only matches within an \
SNI-scoped preread, so a frontend without hostname would silently ignore its alpn list. \
Set hostname or drop alpn."
)]
AlpnWithoutSni { address: SocketAddr },
#[error(
"TCP frontends on {address} with sni {sni:?} both match ALPN protocol '{protocol}': \
ALPN matchers for the same (address, sni) must not overlap"
)]
TcpFrontendAlpnOverlap {
address: SocketAddr,
sni: Option<String>,
protocol: String,
},
#[error(
"more than one TCP frontend on {address} with sni {sni:?} leaves alpn empty (the \
catch-all match): at most one frontend per (address, sni) may omit alpn"
)]
TcpFrontendMultipleAlpnCatchAll {
address: SocketAddr,
sni: Option<String>,
},
#[error(
"TCP listener {address} is targeted by both a no-SNI frontend and at least one \
SNI-scoped frontend: an SNI-enabled listener must not also have a raw-TCP fallback \
frontend on the same address"
)]
TcpListenerMixesSniAndNoSni { address: SocketAddr },
#[error(
"sni_preread_timeout = {sni_preread_timeout}s on TCP listener {address} exceeds its \
front_timeout = {front_timeout}s: the preread phase cannot outlive the frontend \
inactivity timeout that would already have closed the connection"
)]
SniPrereadTimeoutExceedsFrontTimeout {
address: SocketAddr,
sni_preread_timeout: u32,
front_timeout: u32,
},
#[error(
"sni_preread_max_bytes = {sni_preread_max_bytes} on TCP listener {address} exceeds \
buffer_size = {buffer_size}: raise buffer_size to >= {sni_preread_max_bytes} or lower \
sni_preread_max_bytes"
)]
SniPrereadMaxBytesExceedsBufferSize {
address: SocketAddr,
sni_preread_max_bytes: u32,
buffer_size: u64,
},
#[error(
"sni_preread_max_bytes = {sni_preread_max_bytes} on TCP listener {address} is below the \
minimum of {minimum} bytes (a full TLS record header): the preread shell could never \
read enough bytes to make progress. Raise sni_preread_max_bytes to >= {minimum}"
)]
SniPrereadMaxBytesTooSmall {
address: SocketAddr,
sni_preread_max_bytes: u32,
minimum: u32,
},
#[error(
"cluster {cluster_id}: cannot resolve backend address {address:?}: {reason}. \
Expected an IP:port socket address or a resolvable host:port"
)]
BackendAddressResolution {
cluster_id: String,
address: String,
reason: String,
},
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "lowercase")]
pub enum ClientAuthConfig {
#[default]
None,
Optional,
Required,
}
impl From<ClientAuthConfig> for ClientAuthMode {
fn from(value: ClientAuthConfig) -> Self {
match value {
ClientAuthConfig::None => ClientAuthMode::ClientAuthNone,
ClientAuthConfig::Optional => ClientAuthMode::ClientAuthOptional,
ClientAuthConfig::Required => ClientAuthMode::ClientAuthRequired,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ListenerBuilder {
pub address: SocketAddr,
pub interface: Option<String>,
pub protocol: Option<ListenerProtocol>,
pub public_address: Option<SocketAddr>,
pub answer_301: Option<String>,
pub answer_400: Option<String>,
pub answer_401: Option<String>,
pub answer_404: Option<String>,
pub answer_408: Option<String>,
pub answer_413: Option<String>,
pub answer_421: Option<String>,
pub answer_502: Option<String>,
pub answer_503: Option<String>,
pub answer_504: Option<String>,
pub answer_507: Option<String>,
pub answer_429: Option<String>,
pub tls_versions: Option<Vec<TlsVersion>>,
pub cipher_list: Option<Vec<String>>,
pub cipher_suites: Option<Vec<String>>,
pub groups_list: Option<Vec<String>>,
pub expect_proxy: Option<bool>,
#[serde(default = "default_sticky_name")]
pub sticky_name: String,
pub certificate: Option<String>,
pub certificate_chain: Option<String>,
pub key: Option<String>,
pub front_timeout: Option<u32>,
pub back_timeout: Option<u32>,
pub connect_timeout: Option<u32>,
pub request_timeout: Option<u32>,
pub config: Option<Config>,
pub send_tls13_tickets: Option<u64>,
pub alpn_protocols: Option<Vec<String>>,
pub h2_max_rst_stream_per_window: Option<u32>,
pub h2_max_ping_per_window: Option<u32>,
pub h2_max_settings_per_window: Option<u32>,
pub h2_max_empty_data_per_window: Option<u32>,
pub h2_max_window_update_stream0_per_window: Option<u32>,
pub sozu_id_header: Option<String>,
pub h2_max_continuation_frames: Option<u32>,
pub h2_max_glitch_count: Option<u32>,
pub h2_initial_connection_window: Option<u32>,
pub h2_max_concurrent_streams: Option<u32>,
pub h2_stream_shrink_ratio: Option<u32>,
pub h2_max_rst_stream_lifetime: Option<u64>,
pub h2_max_rst_stream_abusive_lifetime: Option<u64>,
pub h2_max_rst_stream_emitted_lifetime: Option<u64>,
pub h2_max_header_list_size: Option<u32>,
pub h2_max_header_table_size: Option<u32>,
pub h2_max_header_fields: Option<u32>,
pub h2_stream_refusal_percent: Option<u32>,
pub h2_stream_idle_timeout_seconds: Option<u32>,
pub h2_graceful_shutdown_deadline_seconds: Option<u32>,
pub strict_sni_binding: Option<bool>,
pub disable_http11: Option<bool>,
pub elide_x_real_ip: Option<bool>,
pub send_x_real_ip: Option<bool>,
pub h2c_prior_knowledge: Option<bool>,
pub forwarded_headers: Option<ForwardedHeadersMode>,
pub answers: Option<BTreeMap<String, String>>,
pub hsts: Option<FileHstsConfig>,
pub max_rx_datagram_size: Option<u32>,
pub max_flows: Option<u32>,
pub sni_preread_timeout: Option<u32>,
pub sni_preread_max_bytes: Option<u32>,
pub client_auth: Option<ClientAuthConfig>,
pub client_ca_certificates: Option<Vec<String>>,
pub client_ca_crls: Option<Vec<String>>,
}
pub fn default_sticky_name() -> String {
DEFAULT_STICKY_NAME.to_string()
}
impl ListenerBuilder {
pub fn new_http(address: SocketAddress) -> ListenerBuilder {
Self::new(address, ListenerProtocol::Http)
}
pub fn new_tcp(address: SocketAddress) -> ListenerBuilder {
Self::new(address, ListenerProtocol::Tcp)
}
pub fn new_https(address: SocketAddress) -> ListenerBuilder {
Self::new(address, ListenerProtocol::Https)
}
pub fn new_udp(address: SocketAddress) -> ListenerBuilder {
Self::new(address, ListenerProtocol::Udp)
}
fn new(address: SocketAddress, protocol: ListenerProtocol) -> ListenerBuilder {
ListenerBuilder {
address: address.into(),
interface: None,
answer_301: None,
answer_401: None,
answer_400: None,
answer_404: None,
answer_408: None,
answer_413: None,
answer_421: None,
answer_502: None,
answer_503: None,
answer_504: None,
answer_507: None,
answer_429: None,
back_timeout: None,
certificate_chain: None,
certificate: None,
cipher_list: None,
cipher_suites: None,
groups_list: None,
config: None,
connect_timeout: None,
expect_proxy: None,
front_timeout: None,
key: None,
protocol: Some(protocol),
public_address: None,
request_timeout: None,
send_tls13_tickets: None,
sticky_name: DEFAULT_STICKY_NAME.to_string(),
tls_versions: None,
alpn_protocols: None,
h2_max_rst_stream_per_window: None,
h2_max_ping_per_window: None,
h2_max_settings_per_window: None,
h2_max_empty_data_per_window: None,
h2_max_window_update_stream0_per_window: None,
sozu_id_header: None,
h2_max_continuation_frames: None,
h2_max_glitch_count: None,
h2_initial_connection_window: None,
h2_max_concurrent_streams: None,
h2_stream_shrink_ratio: None,
h2_max_rst_stream_lifetime: None,
h2_max_rst_stream_abusive_lifetime: None,
h2_max_rst_stream_emitted_lifetime: None,
h2_max_header_list_size: None,
h2_max_header_table_size: None,
h2_max_header_fields: None,
h2_stream_refusal_percent: None,
h2_stream_idle_timeout_seconds: None,
h2_graceful_shutdown_deadline_seconds: None,
strict_sni_binding: None,
disable_http11: None,
elide_x_real_ip: None,
send_x_real_ip: None,
h2c_prior_knowledge: None,
forwarded_headers: None,
answers: None,
hsts: None,
max_rx_datagram_size: None,
max_flows: None,
sni_preread_timeout: None,
sni_preread_max_bytes: None,
client_auth: None,
client_ca_certificates: None,
client_ca_crls: None,
}
}
pub fn with_interface<S>(&mut self, interface: Option<S>) -> &mut Self
where
S: ToString,
{
self.interface = interface.map(|interface| interface.to_string());
self
}
fn validated_interface(&self) -> Result<Option<String>, ConfigError> {
if let Some(interface) = &self.interface {
validate_listener_interface(interface).map_err(|error| {
ConfigError::ListenerInterface {
address: self.address,
error,
}
})?;
}
Ok(self.interface.clone())
}
pub fn with_public_address(&mut self, public_address: Option<SocketAddr>) -> &mut Self {
if let Some(address) = public_address {
self.public_address = Some(address);
}
self
}
pub fn with_answer_404_path<S>(&mut self, answer_404_path: Option<S>) -> &mut Self
where
S: ToString,
{
if let Some(path) = answer_404_path {
self.answer_404 = Some(path.to_string());
}
self
}
pub fn with_answer_503_path<S>(&mut self, answer_503_path: Option<S>) -> &mut Self
where
S: ToString,
{
if let Some(path) = answer_503_path {
self.answer_503 = Some(path.to_string());
}
self
}
pub fn with_tls_versions(&mut self, tls_versions: Vec<TlsVersion>) -> &mut Self {
self.tls_versions = Some(tls_versions);
self
}
pub fn with_cipher_list(&mut self, cipher_list: Option<Vec<String>>) -> &mut Self {
self.cipher_list = cipher_list;
self
}
pub fn with_cipher_suites(&mut self, cipher_suites: Option<Vec<String>>) -> &mut Self {
self.cipher_suites = cipher_suites;
self
}
pub fn with_alpn_protocols(&mut self, alpn_protocols: Option<Vec<String>>) -> &mut Self {
self.alpn_protocols = alpn_protocols;
self
}
pub fn with_elide_x_real_ip(&mut self, elide_x_real_ip: bool) -> &mut Self {
self.elide_x_real_ip = Some(elide_x_real_ip);
self
}
pub fn with_send_x_real_ip(&mut self, send_x_real_ip: bool) -> &mut Self {
self.send_x_real_ip = Some(send_x_real_ip);
self
}
pub fn with_h2c_prior_knowledge(&mut self, h2c_prior_knowledge: bool) -> &mut Self {
self.h2c_prior_knowledge = Some(h2c_prior_knowledge);
self
}
pub fn with_forwarded_headers(&mut self, forwarded_headers: ForwardedHeadersMode) -> &mut Self {
self.forwarded_headers = Some(forwarded_headers);
self
}
pub fn with_expect_proxy(&mut self, expect_proxy: bool) -> &mut Self {
self.expect_proxy = Some(expect_proxy);
self
}
pub fn with_sticky_name<S>(&mut self, sticky_name: Option<S>) -> &mut Self
where
S: ToString,
{
if let Some(name) = sticky_name {
self.sticky_name = name.to_string();
}
self
}
pub fn with_certificate<S>(&mut self, certificate: S) -> &mut Self
where
S: ToString,
{
self.certificate = Some(certificate.to_string());
self
}
pub fn with_certificate_chain(&mut self, certificate_chain: String) -> &mut Self {
self.certificate = Some(certificate_chain);
self
}
pub fn with_key<S>(&mut self, key: String) -> &mut Self
where
S: ToString,
{
self.key = Some(key);
self
}
pub fn with_front_timeout(&mut self, front_timeout: Option<u32>) -> &mut Self {
self.front_timeout = front_timeout;
self
}
pub fn with_back_timeout(&mut self, back_timeout: Option<u32>) -> &mut Self {
self.back_timeout = back_timeout;
self
}
pub fn with_connect_timeout(&mut self, connect_timeout: Option<u32>) -> &mut Self {
self.connect_timeout = connect_timeout;
self
}
pub fn with_request_timeout(&mut self, request_timeout: Option<u32>) -> &mut Self {
self.request_timeout = request_timeout;
self
}
pub fn with_answer<S, P>(&mut self, code: S, path: P) -> &mut Self
where
S: ToString,
P: ToString,
{
self.answers
.get_or_insert_with(BTreeMap::new)
.insert(code.to_string(), path.to_string());
self
}
pub fn with_answers(&mut self, answers: BTreeMap<String, String>) -> &mut Self {
self.answers = Some(answers);
self
}
fn get_http_answers(&self) -> Result<Option<CustomHttpAnswers>, ConfigError> {
let http_answers = CustomHttpAnswers {
answer_301: read_http_answer_file(&self.answer_301)?,
answer_400: read_http_answer_file(&self.answer_400)?,
answer_401: read_http_answer_file(&self.answer_401)?,
answer_404: read_http_answer_file(&self.answer_404)?,
answer_408: read_http_answer_file(&self.answer_408)?,
answer_413: read_http_answer_file(&self.answer_413)?,
answer_421: read_http_answer_file(&self.answer_421)?,
answer_502: read_http_answer_file(&self.answer_502)?,
answer_503: read_http_answer_file(&self.answer_503)?,
answer_504: read_http_answer_file(&self.answer_504)?,
answer_507: read_http_answer_file(&self.answer_507)?,
answer_429: read_http_answer_file(&self.answer_429)?,
};
Ok(Some(http_answers))
}
fn get_listener_answers(&self) -> Result<BTreeMap<String, String>, ConfigError> {
let mut out = BTreeMap::new();
macro_rules! merge_legacy {
($code:literal, $field:ident) => {
if let Some(body) = read_http_answer_file(&self.$field)? {
out.insert($code.to_owned(), body);
}
};
}
merge_legacy!("301", answer_301);
merge_legacy!("400", answer_400);
merge_legacy!("401", answer_401);
merge_legacy!("404", answer_404);
merge_legacy!("408", answer_408);
merge_legacy!("413", answer_413);
merge_legacy!("421", answer_421);
merge_legacy!("502", answer_502);
merge_legacy!("503", answer_503);
merge_legacy!("504", answer_504);
merge_legacy!("507", answer_507);
merge_legacy!("429", answer_429);
if let Some(map) = &self.answers {
let loaded = load_answers(map)?;
out.extend(loaded);
}
Ok(out)
}
fn assign_config_timeouts(&mut self, config: &Config) {
self.front_timeout = Some(self.front_timeout.unwrap_or(config.front_timeout));
self.back_timeout = Some(self.back_timeout.unwrap_or(config.back_timeout));
self.connect_timeout = Some(self.connect_timeout.unwrap_or(config.connect_timeout));
self.request_timeout = Some(self.request_timeout.unwrap_or(config.request_timeout));
}
fn validate_h2_thresholds(&self) -> Result<(), ConfigError> {
macro_rules! require_at_least {
($builder:ident, $field:ident, $minimum:literal) => {
if let Some(value) = $builder.$field
&& u64::from(value) < $minimum
{
return Err(ConfigError::H2ThresholdBelowMinimum {
address: $builder.address.to_string(),
key: stringify!($field),
value: u64::from(value),
minimum: $minimum,
});
}
};
}
crate::for_each_h2_knob_floor!(require_at_least, self);
Ok(())
}
fn reject_client_auth_fields(&self, listener_kind: &str) -> Result<(), ConfigError> {
let has_mtls = self
.client_auth
.is_some_and(|m| m != ClientAuthConfig::None)
|| self
.client_ca_certificates
.as_ref()
.is_some_and(|v| !v.is_empty())
|| self.client_ca_crls.as_ref().is_some_and(|v| !v.is_empty());
if has_mtls {
return Err(ConfigError::ClientAuthOnNonHttps(format!(
"{} listener {}",
listener_kind, self.address
)));
}
Ok(())
}
pub fn to_http(&mut self, config: Option<&Config>) -> Result<HttpListenerConfig, ConfigError> {
if self.protocol != Some(ListenerProtocol::Http) {
return Err(ConfigError::WrongListenerProtocol {
expected: ListenerProtocol::Http,
found: self.protocol.to_owned(),
});
}
self.reject_client_auth_fields("HTTP")?;
self.validate_h2_thresholds()?;
if self.hsts.is_some() {
return Err(ConfigError::HstsOnPlainHttp(format!(
"HTTP listener {}",
self.address
)));
}
if let Some(config) = config {
self.assign_config_timeouts(config);
}
let http_answers = self.get_http_answers()?;
let answers = self.get_listener_answers()?;
let interface = self.validated_interface()?;
let configuration = HttpListenerConfig {
address: self.address.into(),
interface,
public_address: self.public_address.map(|a| a.into()),
expect_proxy: self.expect_proxy.unwrap_or(false),
sticky_name: self.sticky_name.clone(),
front_timeout: self.front_timeout.unwrap_or(DEFAULT_FRONT_TIMEOUT),
back_timeout: self.back_timeout.unwrap_or(DEFAULT_BACK_TIMEOUT),
connect_timeout: self.connect_timeout.unwrap_or(DEFAULT_CONNECT_TIMEOUT),
request_timeout: self.request_timeout.unwrap_or(DEFAULT_REQUEST_TIMEOUT),
http_answers,
answers,
h2_max_rst_stream_per_window: self.h2_max_rst_stream_per_window,
h2_max_ping_per_window: self.h2_max_ping_per_window,
h2_max_settings_per_window: self.h2_max_settings_per_window,
h2_max_empty_data_per_window: self.h2_max_empty_data_per_window,
h2_max_window_update_stream0_per_window: self.h2_max_window_update_stream0_per_window,
h2_max_continuation_frames: self.h2_max_continuation_frames,
h2_max_glitch_count: self.h2_max_glitch_count,
h2_initial_connection_window: self.h2_initial_connection_window,
h2_max_concurrent_streams: self.h2_max_concurrent_streams,
h2_stream_shrink_ratio: self.h2_stream_shrink_ratio,
h2_max_rst_stream_lifetime: self.h2_max_rst_stream_lifetime,
h2_max_rst_stream_abusive_lifetime: self.h2_max_rst_stream_abusive_lifetime,
h2_max_rst_stream_emitted_lifetime: self.h2_max_rst_stream_emitted_lifetime,
h2_max_header_list_size: self.h2_max_header_list_size,
h2_max_header_table_size: self.h2_max_header_table_size,
h2_max_header_fields: self.h2_max_header_fields,
h2_stream_refusal_percent: self.h2_stream_refusal_percent,
h2_stream_idle_timeout_seconds: self.h2_stream_idle_timeout_seconds,
h2_graceful_shutdown_deadline_seconds: self.h2_graceful_shutdown_deadline_seconds,
sozu_id_header: self.sozu_id_header.clone(),
elide_x_real_ip: Some(self.elide_x_real_ip.unwrap_or(false)),
send_x_real_ip: Some(self.send_x_real_ip.unwrap_or(false)),
forwarded_headers: Some(ForwardedHeaders::from(
self.forwarded_headers.unwrap_or_default(),
) as i32),
h2c_prior_knowledge: self.h2c_prior_knowledge,
..Default::default()
};
debug_assert_eq!(
configuration.address,
self.address.into(),
"HTTP listener must bind the requested address"
);
Ok(configuration)
}
pub fn to_tls(&mut self, config: Option<&Config>) -> Result<HttpsListenerConfig, ConfigError> {
if self.protocol != Some(ListenerProtocol::Https) {
return Err(ConfigError::WrongListenerProtocol {
expected: ListenerProtocol::Https,
found: self.protocol.to_owned(),
});
}
self.validate_h2_thresholds()?;
let default_cipher_list = DEFAULT_CIPHER_LIST.into_iter().map(String::from).collect();
let cipher_list = self.cipher_list.clone().unwrap_or(default_cipher_list);
let cipher_suites = self
.cipher_suites
.clone()
.unwrap_or_else(|| DEFAULT_CIPHER_LIST.into_iter().map(String::from).collect());
let signature_algorithms: Vec<String> = DEFAULT_SIGNATURE_ALGORITHMS
.into_iter()
.map(String::from)
.collect();
let groups_list = self
.groups_list
.clone()
.unwrap_or_else(|| DEFAULT_GROUPS_LIST.into_iter().map(String::from).collect());
let alpn_protocols: Vec<String> = match &self.alpn_protocols {
Some(protos) if !protos.is_empty() => {
for proto in protos {
match proto.as_str() {
"h2" | "http/1.1" => {}
other => return Err(ConfigError::InvalidAlpnProtocol(other.to_owned())),
}
}
if self.disable_http11.unwrap_or(false) && protos.iter().any(|p| p == "http/1.1") {
return Err(ConfigError::DisableHttp11WithHttp11Alpn {
address: self.address.to_string(),
});
}
if !protos.iter().any(|p| p == "http/1.1") {
warn!(
"ALPN protocols do not include 'http/1.1'. Clients without H2 support will fail TLS negotiation."
);
}
let mut seen = std::collections::HashSet::new();
protos
.iter()
.filter(|p| seen.insert(p.as_str()))
.cloned()
.collect()
}
_ => {
if self.disable_http11.unwrap_or(false)
&& DEFAULT_ALPN_PROTOCOLS.contains(&"http/1.1")
{
return Err(ConfigError::DisableHttp11WithHttp11Alpn {
address: self.address.to_string(),
});
}
DEFAULT_ALPN_PROTOCOLS
.iter()
.map(|s| s.to_string())
.collect()
}
};
let versions = match self.tls_versions {
None => vec![TlsVersion::TlsV12 as i32, TlsVersion::TlsV13 as i32],
Some(ref v) => v.iter().map(|v| *v as i32).collect(),
};
let key = self.key.as_ref().and_then(|path| {
Config::load_file(path)
.map_err(|e| {
error!("cannot load key at path '{}': {:?}", path, e);
e
})
.ok()
});
let certificate = self.certificate.as_ref().and_then(|path| {
Config::load_file(path)
.map_err(|e| {
error!("cannot load certificate at path '{}': {:?}", path, e);
e
})
.ok()
});
let certificate_chain = self
.certificate_chain
.as_ref()
.and_then(|path| {
Config::load_file(path)
.map_err(|e| {
error!("cannot load certificate chain at path '{}': {:?}", path, e);
e
})
.ok()
})
.map(split_certificate_chain)
.unwrap_or_default();
let has_client_auth_files = self
.client_ca_certificates
.as_ref()
.is_some_and(|paths| !paths.is_empty())
|| self
.client_ca_crls
.as_ref()
.is_some_and(|paths| !paths.is_empty());
match self.client_auth {
None if has_client_auth_files => {
return Err(ConfigError::ClientAuthModeMissing(format!(
"HTTPS listener {}",
self.address
)));
}
Some(ClientAuthConfig::None) if has_client_auth_files => {
warn!(
"HTTPS listener {}: client_auth is \"none\", so its client_ca_certificates \
and client_ca_crls are ignored and clients are not authenticated",
self.address
);
}
_ => {}
}
let mode = self.client_auth.unwrap_or_default();
let client_auth = self
.client_auth
.map(|mode| ClientAuthMode::from(mode) as i32);
let load_pem_paths = |paths: &Option<Vec<String>>| -> Result<Vec<String>, ConfigError> {
if mode == ClientAuthConfig::None {
return Ok(Vec::new());
}
paths
.as_ref()
.map(|list| list.iter().map(|path| Config::load_file(path)).collect())
.unwrap_or_else(|| Ok(Vec::new()))
};
let client_ca_certificates = load_pem_paths(&self.client_ca_certificates)?;
let client_ca_crls = load_pem_paths(&self.client_ca_crls)?;
let http_answers = self.get_http_answers()?;
let answers = self.get_listener_answers()?;
if let Some(config) = config {
self.assign_config_timeouts(config);
}
let https_listener_config = HttpsListenerConfig {
interface: self.validated_interface()?,
address: self.address.into(),
sticky_name: self.sticky_name.clone(),
public_address: self.public_address.map(|a| a.into()),
cipher_list,
versions,
expect_proxy: self.expect_proxy.unwrap_or(false),
key,
certificate,
certificate_chain,
front_timeout: self.front_timeout.unwrap_or(DEFAULT_FRONT_TIMEOUT),
back_timeout: self.back_timeout.unwrap_or(DEFAULT_BACK_TIMEOUT),
connect_timeout: self.connect_timeout.unwrap_or(DEFAULT_CONNECT_TIMEOUT),
request_timeout: self.request_timeout.unwrap_or(DEFAULT_REQUEST_TIMEOUT),
cipher_suites,
signature_algorithms,
groups_list,
active: false,
send_tls13_tickets: self
.send_tls13_tickets
.unwrap_or(DEFAULT_SEND_TLS_13_TICKETS),
http_answers,
answers,
alpn_protocols,
h2_max_rst_stream_per_window: self.h2_max_rst_stream_per_window,
h2_max_ping_per_window: self.h2_max_ping_per_window,
h2_max_settings_per_window: self.h2_max_settings_per_window,
h2_max_empty_data_per_window: self.h2_max_empty_data_per_window,
h2_max_window_update_stream0_per_window: self.h2_max_window_update_stream0_per_window,
h2_max_continuation_frames: self.h2_max_continuation_frames,
h2_max_glitch_count: self.h2_max_glitch_count,
h2_initial_connection_window: self.h2_initial_connection_window,
h2_max_concurrent_streams: self.h2_max_concurrent_streams,
h2_stream_shrink_ratio: self.h2_stream_shrink_ratio,
h2_max_rst_stream_lifetime: self.h2_max_rst_stream_lifetime,
h2_max_rst_stream_abusive_lifetime: self.h2_max_rst_stream_abusive_lifetime,
h2_max_rst_stream_emitted_lifetime: self.h2_max_rst_stream_emitted_lifetime,
h2_max_header_list_size: self.h2_max_header_list_size,
h2_max_header_table_size: self.h2_max_header_table_size,
h2_max_header_fields: self.h2_max_header_fields,
h2_stream_refusal_percent: self.h2_stream_refusal_percent,
strict_sni_binding: self.strict_sni_binding,
disable_http11: self.disable_http11,
h2_stream_idle_timeout_seconds: self.h2_stream_idle_timeout_seconds,
h2_graceful_shutdown_deadline_seconds: self.h2_graceful_shutdown_deadline_seconds,
sozu_id_header: self.sozu_id_header.clone(),
elide_x_real_ip: Some(self.elide_x_real_ip.unwrap_or(false)),
send_x_real_ip: Some(self.send_x_real_ip.unwrap_or(false)),
forwarded_headers: Some(ForwardedHeaders::from(
self.forwarded_headers.unwrap_or_default(),
) as i32),
hsts: match self.hsts.as_ref() {
Some(h) => Some(h.to_proto("listener")?),
None => None,
},
client_auth,
client_ca_certificates,
client_ca_crls,
};
debug_assert_eq!(
https_listener_config.address,
self.address.into(),
"HTTPS listener must bind the requested address"
);
debug_assert!(
!https_listener_config.active,
"a freshly built HTTPS listener must start inactive"
);
debug_assert!(
!https_listener_config.alpn_protocols.is_empty(),
"resolved ALPN list must not be empty"
);
debug_assert!(
https_listener_config
.alpn_protocols
.iter()
.all(|p| p == "h2" || p == "http/1.1"),
"resolved ALPN list must contain only h2 and http/1.1"
);
debug_assert!(
{
let mut seen = std::collections::HashSet::new();
https_listener_config
.alpn_protocols
.iter()
.all(|p| seen.insert(p))
},
"resolved ALPN list must be duplicate-free"
);
debug_assert!(
!(self.disable_http11.unwrap_or(false)
&& https_listener_config
.alpn_protocols
.iter()
.any(|p| p == "http/1.1")),
"disable_http11 with http/1.1 in ALPN must have been rejected"
);
Ok(https_listener_config)
}
pub fn to_tcp(&mut self, config: Option<&Config>) -> Result<TcpListenerConfig, ConfigError> {
if self.protocol != Some(ListenerProtocol::Tcp) {
return Err(ConfigError::WrongListenerProtocol {
expected: ListenerProtocol::Tcp,
found: self.protocol.to_owned(),
});
}
self.reject_client_auth_fields("TCP")?;
if let Some(config) = config {
self.assign_config_timeouts(config);
}
let tcp_listener_config = TcpListenerConfig {
interface: self.validated_interface()?,
address: self.address.into(),
public_address: self.public_address.map(|a| a.into()),
expect_proxy: self.expect_proxy.unwrap_or(false),
front_timeout: self.front_timeout.unwrap_or(DEFAULT_FRONT_TIMEOUT),
back_timeout: self.back_timeout.unwrap_or(DEFAULT_BACK_TIMEOUT),
connect_timeout: self.connect_timeout.unwrap_or(DEFAULT_CONNECT_TIMEOUT),
active: false,
sni_preread_timeout: Some(
self.sni_preread_timeout
.unwrap_or(DEFAULT_SNI_PREREAD_TIMEOUT),
),
sni_preread_max_bytes: Some(
self.sni_preread_max_bytes
.unwrap_or(DEFAULT_SNI_PREREAD_MAX_BYTES),
),
};
debug_assert_eq!(
tcp_listener_config.address,
self.address.into(),
"TCP listener must bind the requested address"
);
debug_assert!(
!tcp_listener_config.active,
"a freshly built TCP listener must start inactive"
);
Ok(tcp_listener_config)
}
pub fn to_udp(&mut self, config: Option<&Config>) -> Result<UdpListenerConfig, ConfigError> {
if self.protocol != Some(ListenerProtocol::Udp) {
return Err(ConfigError::WrongListenerProtocol {
expected: ListenerProtocol::Udp,
found: self.protocol.to_owned(),
});
}
self.reject_client_auth_fields("UDP")?;
let mut max_rx_datagram_size = self
.max_rx_datagram_size
.unwrap_or(DEFAULT_UDP_MAX_RX_DATAGRAM_SIZE);
let buffer_size = config.map(|c| c.buffer_size).unwrap_or(DEFAULT_BUFFER_SIZE);
if u64::from(max_rx_datagram_size) > buffer_size {
warn!(
"UDP listener {}: max_rx_datagram_size = {} exceeds buffer_size = {}, clamping to buffer_size",
self.address, max_rx_datagram_size, buffer_size
);
max_rx_datagram_size = buffer_size as u32;
}
let max_flows = self.max_flows.unwrap_or(DEFAULT_UDP_MAX_FLOWS);
if max_flows > 0
&& let Some(soft_limit) = soft_rlimit_nofile()
{
let advisory = soft_limit.saturating_mul(7) / 10;
if u64::from(max_flows) > advisory {
warn!(
"UDP listener {}: max_flows = {} exceeds ~70% of the soft RLIMIT_NOFILE ({}); \
per-flow connected sockets may hit EMFILE",
self.address, max_flows, advisory
);
}
}
Ok(UdpListenerConfig {
interface: self.validated_interface()?,
address: self.address.into(),
public_address: self.public_address.map(|a| a.into()),
front_timeout: self.front_timeout.unwrap_or(DEFAULT_UDP_FRONT_TIMEOUT),
back_timeout: self.back_timeout.unwrap_or(DEFAULT_UDP_BACK_TIMEOUT),
max_rx_datagram_size,
max_flows,
active: false,
})
}
}
fn soft_rlimit_nofile() -> Option<u64> {
let mut limit = libc::rlimit {
rlim_cur: 0,
rlim_max: 0,
};
let rc = unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &mut limit) };
if rc == 0 {
Some(limit.rlim_cur)
} else {
None
}
}
fn read_http_answer_file(path: &Option<String>) -> Result<Option<String>, ConfigError> {
match path {
Some(path) => {
let mut content = String::new();
let mut file = File::open(path).map_err(|io_error| ConfigError::FileOpen {
path_to_open: path.to_owned(),
io_error,
})?;
file.read_to_string(&mut content)
.map_err(|io_error| ConfigError::FileRead {
path_to_read: path.to_owned(),
io_error,
})?;
Ok(Some(content))
}
None => Ok(None),
}
}
pub fn resolve_answer_source(value: &str) -> Result<String, ConfigError> {
if let Some(path) = value.strip_prefix("file://") {
let mut content = String::new();
let mut file = File::open(path).map_err(|io_error| ConfigError::FileOpen {
path_to_open: path.to_owned(),
io_error,
})?;
file.read_to_string(&mut content)
.map_err(|io_error| ConfigError::FileRead {
path_to_read: path.to_owned(),
io_error,
})?;
return Ok(content);
}
Ok(value.to_owned())
}
pub fn load_answers(
answers: &BTreeMap<String, String>,
) -> Result<BTreeMap<String, String>, ConfigError> {
let mut out = BTreeMap::new();
for (code, value) in answers {
if value.is_empty() {
continue;
}
out.insert(code.to_owned(), resolve_answer_source(value)?);
}
debug_assert!(
out.len() <= answers.len(),
"load_answers must not synthesize entries"
);
debug_assert!(
out.keys().all(|k| answers.contains_key(k)),
"every loaded status code must come from the input map"
);
Ok(out)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
#[derive(Default)]
pub enum MetricDetailLevel {
Process,
Frontend,
#[default]
Cluster,
Backend,
}
impl From<MetricDetailLevel> for MetricDetail {
fn from(level: MetricDetailLevel) -> Self {
match level {
MetricDetailLevel::Process => MetricDetail::DetailProcess,
MetricDetailLevel::Frontend => MetricDetail::DetailFrontend,
MetricDetailLevel::Cluster => MetricDetail::DetailCluster,
MetricDetailLevel::Backend => MetricDetail::DetailBackend,
}
}
}
impl From<MetricDetail> for MetricDetailLevel {
fn from(detail: MetricDetail) -> Self {
match detail {
MetricDetail::DetailProcess => MetricDetailLevel::Process,
MetricDetail::DetailFrontend => MetricDetailLevel::Frontend,
MetricDetail::DetailCluster => MetricDetailLevel::Cluster,
MetricDetail::DetailBackend => MetricDetailLevel::Backend,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ForwardedHeadersMode {
#[default]
Both,
XForwarded,
Rfc7239,
None,
}
impl From<ForwardedHeadersMode> for ForwardedHeaders {
fn from(mode: ForwardedHeadersMode) -> Self {
match mode {
ForwardedHeadersMode::Both => ForwardedHeaders::Both,
ForwardedHeadersMode::XForwarded => ForwardedHeaders::XForwarded,
ForwardedHeadersMode::Rfc7239 => ForwardedHeaders::Rfc7239,
ForwardedHeadersMode::None => ForwardedHeaders::None,
}
}
}
impl From<ForwardedHeaders> for ForwardedHeadersMode {
fn from(headers: ForwardedHeaders) -> Self {
match headers {
ForwardedHeaders::Both => ForwardedHeadersMode::Both,
ForwardedHeaders::XForwarded => ForwardedHeadersMode::XForwarded,
ForwardedHeaders::Rfc7239 => ForwardedHeadersMode::Rfc7239,
ForwardedHeaders::None => ForwardedHeadersMode::None,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MetricsConfig {
pub address: SocketAddr,
#[serde(default)]
pub tagged_metrics: bool,
#[serde(default)]
pub prefix: Option<String>,
#[serde(default)]
pub detail: MetricDetailLevel,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
#[serde(deny_unknown_fields)]
pub enum PathRuleType {
Prefix,
Regex,
Equals,
RegexPath,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct FileClusterFrontendConfig {
pub address: SocketAddr,
pub hostname: Option<String>,
#[serde(default)]
pub alpn: Vec<String>,
pub path: Option<String>,
pub path_type: Option<PathRuleType>,
pub method: Option<String>,
pub certificate: Option<String>,
pub key: Option<String>,
pub certificate_chain: Option<String>,
#[serde(default)]
pub tls_versions: Vec<TlsVersion>,
#[serde(default)]
pub position: RulePosition,
pub tags: Option<BTreeMap<String, String>>,
pub redirect: Option<String>,
pub redirect_scheme: Option<String>,
pub redirect_template: Option<String>,
pub rewrite_host: Option<String>,
pub rewrite_path: Option<String>,
pub rewrite_port: Option<u32>,
pub required_auth: Option<bool>,
pub headers: Option<Vec<HeaderEditConfig>>,
pub hsts: Option<FileHstsConfig>,
pub header_predicates: Option<Vec<HeaderPredicateConfig>>,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct HeaderPredicateConfig {
pub name: String,
#[serde(rename = "match")]
pub kind: String,
pub value: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct HeaderEditConfig {
pub position: String,
pub key: String,
pub value: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct FileHstsConfig {
pub enabled: Option<bool>,
pub max_age: Option<u32>,
pub include_subdomains: Option<bool>,
pub preload: Option<bool>,
pub force_replace_backend: Option<bool>,
}
impl FileHstsConfig {
pub fn to_proto(&self, scope: &str) -> Result<HstsConfig, ConfigError> {
let enabled = match self.enabled {
Some(v) => v,
None => return Err(ConfigError::HstsEnabledRequired(scope.to_owned())),
};
let max_age = match (enabled, self.max_age) {
(true, None) => Some(DEFAULT_HSTS_MAX_AGE),
(_, m) => m,
};
if let Some(value) = max_age
&& value > 0
&& value < 86_400
{
warn!(
"HSTS max_age = {}s on {} is below 1 day — this is almost certainly a \
misconfiguration. RFC 6797 §11.4 reserves max_age = 0 as the explicit kill \
switch.",
value, scope
);
}
let include_subdomains = self.include_subdomains;
let preload = self.preload;
if matches!(preload, Some(true)) {
let max_age_value = max_age.unwrap_or(0);
if max_age_value < DEFAULT_HSTS_MAX_AGE {
warn!(
"HSTS preload = true on {} with max_age = {}s; the Chrome HSTS preload \
list requires max_age >= {} (https://hstspreload.org/).",
scope, max_age_value, DEFAULT_HSTS_MAX_AGE
);
}
if include_subdomains != Some(true) {
warn!(
"HSTS preload = true on {} without include_subdomains = true; the Chrome \
HSTS preload list requires includeSubDomains \
(https://hstspreload.org/).",
scope
);
}
}
let config = HstsConfig {
enabled: Some(enabled),
max_age,
include_subdomains,
preload,
force_replace_backend: self.force_replace_backend,
};
debug_assert_eq!(
config.enabled,
Some(enabled),
"built HSTS config must record the resolved enabled flag"
);
debug_assert!(
!enabled || config.max_age.is_some(),
"an enabled HSTS policy must carry a max_age"
);
Ok(config)
}
}
impl FileClusterFrontendConfig {
pub fn to_tcp_front(&self) -> Result<TcpFrontendConfig, ConfigError> {
if self.path.is_some() {
return Err(ConfigError::InvalidFrontendConfig(
"path_prefix".to_string(),
));
}
if self.certificate.is_some() {
return Err(ConfigError::InvalidFrontendConfig(
"certificate".to_string(),
));
}
if self.certificate_chain.is_some() {
return Err(ConfigError::InvalidFrontendConfig(
"certificate_chain".to_string(),
));
}
if self.header_predicates.is_some() {
return Err(ConfigError::InvalidFrontendConfig(
"header_predicates".to_string(),
));
}
let sni = match &self.hostname {
Some(hostname) => Some(validate_sni_pattern(hostname)?),
None => None,
};
if sni.is_none() && !self.alpn.is_empty() {
return Err(ConfigError::AlpnWithoutSni {
address: self.address,
});
}
let tcp_front = TcpFrontendConfig {
address: self.address,
tags: self.tags.clone(),
sni,
alpn: self.alpn.clone(),
udp: false,
};
debug_assert_eq!(
tcp_front.address, self.address,
"TCP frontend must bind the requested address"
);
debug_assert!(
self.path.is_none() && self.certificate.is_none() && self.certificate_chain.is_none(),
"a built TCP frontend must carry no HTTP-only attributes"
);
debug_assert!(
tcp_front.sni.is_some() || tcp_front.alpn.is_empty(),
"a built TCP frontend without sni must never carry a non-empty alpn"
);
Ok(tcp_front)
}
pub fn to_http_front(&self, _cluster_id: &str) -> Result<HttpFrontendConfig, ConfigError> {
if !self.alpn.is_empty() {
return Err(ConfigError::InvalidFrontendConfig("alpn".to_string()));
}
let hostname = match &self.hostname {
Some(hostname) => hostname.to_owned(),
None => {
return Err(ConfigError::Missing(MissingKind::Field(
"hostname".to_string(),
)));
}
};
let key_opt = match self.key.as_ref() {
None => None,
Some(path) => {
let key = Config::load_file(path)?;
Some(key)
}
};
let certificate_opt = match self.certificate.as_ref() {
None => None,
Some(path) => {
let certificate = Config::load_file(path)?;
Some(certificate)
}
};
let certificate_chain = match self.certificate_chain.as_ref() {
None => None,
Some(path) => {
let certificate_chain = Config::load_file(path)?;
Some(split_certificate_chain(certificate_chain))
}
};
let path = match (self.path.as_ref(), self.path_type.as_ref()) {
(None, _) => PathRule::prefix("".to_string()),
(Some(s), Some(PathRuleType::Prefix)) => PathRule::prefix(s.to_string()),
(Some(s), Some(PathRuleType::Regex)) => PathRule::regex(s.to_string()),
(Some(s), Some(PathRuleType::Equals)) => PathRule::equals(s.to_string()),
(Some(s), Some(PathRuleType::RegexPath)) => PathRule::regex_path(s.to_string()),
(Some(s), None) => PathRule::prefix(s.clone()),
};
let redirect = match self.redirect.as_deref() {
Some(v) => Some(parse_redirect_policy(v)?),
None => None,
};
let redirect_scheme = match self.redirect_scheme.as_deref() {
Some(v) => Some(parse_redirect_scheme(v)?),
None => None,
};
let headers = match self.headers.as_ref() {
Some(entries) => {
let mut out = Vec::with_capacity(entries.len());
for (index, entry) in entries.iter().enumerate() {
out.push(parse_header_edit(index, entry)?);
}
out
}
None => Vec::new(),
};
let header_predicates = match self.header_predicates.as_ref() {
Some(entries) => entries
.iter()
.enumerate()
.map(|(index, entry)| parse_header_predicate(index, entry))
.collect::<Result<Vec<_>, _>>()?,
None => Vec::new(),
};
let frontend_serves_https = key_opt.is_some() && certificate_opt.is_some();
let hsts = match self.hsts.as_ref() {
Some(h) => {
if !frontend_serves_https {
return Err(ConfigError::HstsOnPlainHttp(format!(
"frontend {_cluster_id}/{hostname}"
)));
}
Some(h.to_proto(&format!("frontend {_cluster_id}/{hostname}"))?)
}
None => None,
};
Ok(HttpFrontendConfig {
address: self.address,
hostname,
certificate: certificate_opt,
key: key_opt,
certificate_chain,
tls_versions: self.tls_versions.clone(),
position: self.position,
path,
method: self.method.clone(),
tags: self.tags.clone(),
redirect,
redirect_scheme,
redirect_template: self.redirect_template.clone(),
rewrite_host: self.rewrite_host.clone(),
rewrite_path: self.rewrite_path.clone(),
rewrite_port: self.rewrite_port,
required_auth: self.required_auth,
headers,
hsts,
header_predicates,
})
}
}
pub fn validate_sni_pattern(sni: &str) -> Result<String, ConfigError> {
let invalid = || ConfigError::InvalidSniPattern {
sni: sni.to_string(),
};
if sni.is_empty() {
return Err(invalid());
}
if !sni.is_ascii() {
return Err(ConfigError::NonAsciiSniPattern {
sni: sni.to_string(),
});
}
let remainder = sni.strip_prefix("*.").unwrap_or(sni);
if remainder.is_empty() || remainder.contains('*') || remainder.contains('/') {
return Err(invalid());
}
if remainder.split('.').any(|label| label.is_empty()) {
return Err(invalid());
}
let normalized = sni.to_ascii_lowercase();
debug_assert!(
normalized.matches('*').count() <= 1,
"a validated SNI pattern must carry at most one wildcard marker"
);
debug_assert!(
normalized.is_ascii(),
"a validated SNI pattern must be pure ASCII"
);
Ok(normalized)
}
pub(crate) fn parse_redirect_policy(value: &str) -> Result<RedirectPolicy, ConfigError> {
match value.to_ascii_lowercase().as_str() {
"forward" => Ok(RedirectPolicy::Forward),
"permanent" => Ok(RedirectPolicy::Permanent),
"unauthorized" => Ok(RedirectPolicy::Unauthorized),
"silent_close" => Ok(RedirectPolicy::SilentClose),
_ => Err(ConfigError::InvalidRedirectPolicy(value.to_owned())),
}
}
pub(crate) fn parse_redirect_scheme(value: &str) -> Result<RedirectScheme, ConfigError> {
match value.to_ascii_lowercase().as_str() {
"use-same" | "use_same" => Ok(RedirectScheme::UseSame),
"use-http" | "use_http" => Ok(RedirectScheme::UseHttp),
"use-https" | "use_https" => Ok(RedirectScheme::UseHttps),
_ => Err(ConfigError::InvalidRedirectScheme(value.to_owned())),
}
}
pub(crate) fn parse_header_edit(
index: usize,
entry: &HeaderEditConfig,
) -> Result<Header, ConfigError> {
let position = match entry.position.to_ascii_lowercase().as_str() {
"request" => HeaderPosition::Request,
"response" => HeaderPosition::Response,
"both" => HeaderPosition::Both,
_ => {
return Err(ConfigError::InvalidHeaderPosition {
index,
position: entry.position.clone(),
});
}
};
if !header_name_is_valid_token(entry.key.as_bytes()) {
return Err(ConfigError::InvalidHeaderBytes {
index,
field: "key",
});
}
if header_value_contains_forbidden_controls(entry.value.as_bytes()) {
return Err(ConfigError::InvalidHeaderBytes {
index,
field: "value",
});
}
let header = Header {
position: position as i32,
key: entry.key.clone(),
val: entry.value.clone(),
};
debug_assert!(
header_name_is_valid_token(header.key.as_bytes()),
"an emitted header key must be a valid token"
);
debug_assert!(
!header_value_contains_forbidden_controls(header.val.as_bytes()),
"an emitted header value must be free of forbidden control bytes"
);
Ok(header)
}
pub(crate) fn parse_header_predicate(
index: usize,
entry: &HeaderPredicateConfig,
) -> Result<HeaderPredicate, ConfigError> {
let kind = match entry.kind.to_ascii_lowercase().as_str() {
"exact" => HeaderMatchKind::Exact,
"regex" => HeaderMatchKind::Regex,
"present" => HeaderMatchKind::Present,
_ => {
return Err(ConfigError::InvalidHeaderPredicate {
index,
reason: "match must be \"exact\", \"regex\" or \"present\"",
});
}
};
let predicate = HeaderPredicate {
name: entry.name.to_owned(),
kind: kind as i32,
value: entry.value.to_owned(),
};
validate_header_predicate(&predicate)
.map_err(|reason| ConfigError::InvalidHeaderPredicate { index, reason })?;
Ok(predicate)
}
pub fn validate_header_predicate(predicate: &HeaderPredicate) -> Result<(), &'static str> {
let name = predicate.name.as_str();
if !header_name_is_valid_token(name.as_bytes()) {
return Err("name must be a non-empty RFC 9110 token (a header field name)");
}
if name.eq_ignore_ascii_case("host") {
return Err(
"name cannot be Host: the request authority is not a header Sōzu keeps \
(H1 folds Host into it, H2 sends :authority); route on the hostname instead",
);
}
if name.eq_ignore_ascii_case("cookie") {
return Err("name cannot be Cookie: cookies are parsed into a jar, not kept as a header");
}
match (
HeaderMatchKind::try_from(predicate.kind),
predicate.value.as_deref(),
) {
(Err(_), _) => Err("unknown match kind"),
(Ok(HeaderMatchKind::Present), Some(_)) => Err("a present predicate takes no value"),
(Ok(HeaderMatchKind::Present), None) => Ok(()),
(Ok(HeaderMatchKind::Exact | HeaderMatchKind::Regex), None) => {
Err("an exact or regex predicate requires a value")
}
(Ok(HeaderMatchKind::Exact), Some(value))
if header_value_contains_forbidden_controls(value.as_bytes()) =>
{
Err("an exact value cannot carry a control character no header value can hold")
}
(Ok(HeaderMatchKind::Exact | HeaderMatchKind::Regex), Some(_)) => Ok(()),
}
}
pub(crate) fn header_name_is_valid_token(bytes: &[u8]) -> bool {
if bytes.is_empty() {
return false;
}
bytes.iter().all(|&b| is_tchar(b))
}
fn is_tchar(b: u8) -> bool {
b.is_ascii_alphanumeric()
|| matches!(
b,
b'!' | b'#'
| b'$'
| b'%'
| b'&'
| b'\''
| b'*'
| b'+'
| b'-'
| b'.'
| b'^'
| b'_'
| b'`'
| b'|'
| b'~'
)
}
pub(crate) fn header_value_contains_forbidden_controls(bytes: &[u8]) -> bool {
bytes
.iter()
.any(|&b| matches!(b, 0x00..=0x08 | 0x0A..=0x1F | 0x7F))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(deny_unknown_fields, rename_all = "lowercase")]
pub enum ListenerProtocol {
Http,
Https,
Tcp,
Udp,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(deny_unknown_fields, rename_all = "lowercase")]
pub enum FileClusterProtocolConfig {
Http,
Tcp,
}
fn default_health_check_interval() -> u32 {
10
}
fn default_health_check_timeout() -> u32 {
5
}
fn default_health_check_threshold() -> u32 {
3
}
fn default_health_check_uri() -> String {
"/".to_owned()
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct FileHealthCheckConfig {
#[serde(default = "default_health_check_uri")]
pub uri: String,
#[serde(default = "default_health_check_interval")]
pub interval: u32,
#[serde(default = "default_health_check_timeout")]
pub timeout: u32,
#[serde(default = "default_health_check_threshold")]
pub healthy_threshold: u32,
#[serde(default = "default_health_check_threshold")]
pub unhealthy_threshold: u32,
#[serde(default)]
pub expected_status: u32,
#[serde(default)]
pub mode: Option<HealthCheckMode>,
#[serde(default)]
pub accepted_statuses: Vec<String>,
#[serde(default)]
pub port: Option<u32>,
#[serde(default)]
pub headers: Vec<FileHealthCheckHeader>,
}
#[derive(Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct FileHealthCheckHeader {
pub name: String,
pub value_file: String,
}
impl std::fmt::Debug for FileHealthCheckHeader {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("FileHealthCheckHeader")
.field("name_len", &self.name.len())
.field("value_file", &"<redacted>")
.finish_non_exhaustive()
}
}
impl FileHealthCheckConfig {
pub fn to_proto(&self) -> Result<HealthCheckConfig, &'static str> {
if !self.headers.is_empty() {
return Err(
"health check headers from the configuration file are not available yet: their values need the protected channel",
);
}
self.to_parts().map(|(config, _)| config)
}
pub fn to_parts(
&self,
) -> Result<(HealthCheckConfig, Vec<FileHealthCheckHeader>), &'static str> {
let accepted_statuses = self
.accepted_statuses
.iter()
.map(|spec| parse_http_status_range(spec))
.collect::<Result<Vec<_>, _>>()?;
let proto = HealthCheckConfig {
uri: self.uri.to_owned(),
interval: self.interval,
timeout: self.timeout,
healthy_threshold: self.healthy_threshold,
unhealthy_threshold: self.unhealthy_threshold,
expected_status: self.expected_status,
mode: self.mode.unwrap_or(HealthCheckMode::Http) as i32,
accepted_statuses,
port: self.port,
headers: self
.headers
.iter()
.map(|header| HealthCheckHeader {
name: header.name.to_owned(),
fingerprint: String::new(),
})
.collect(),
};
debug_assert_eq!(proto.uri, self.uri, "proto URI must mirror the file config");
debug_assert!(
proto.interval == self.interval
&& proto.timeout == self.timeout
&& proto.healthy_threshold == self.healthy_threshold
&& proto.unhealthy_threshold == self.unhealthy_threshold,
"proto timing knobs must mirror the file config"
);
debug_assert_eq!(
proto.port, self.port,
"proto probe port must mirror the file config"
);
debug_assert_eq!(
proto.accepted_statuses.len(),
self.accepted_statuses.len(),
"every accepted status spec must map to exactly one range"
);
debug_assert_eq!(
proto.headers.len(),
self.headers.len(),
"every configured header must map to exactly one proto header"
);
validate_health_check_config(&proto)?;
Ok((proto, self.headers.clone()))
}
}
pub const HTTP_STATUS_MIN: u32 = 100;
pub const HTTP_STATUS_MAX: u32 = 599;
pub fn parse_http_status_range(spec: &str) -> Result<HttpStatusRange, &'static str> {
const INVALID: &str = "health check accepted status must be a code (\"404\"), a range \
(\"200-399\"), a class (\"2xx\") or \"any\"";
let spec = spec.trim();
let range = if spec.eq_ignore_ascii_case("any") {
HttpStatusRange {
start: HTTP_STATUS_MIN,
end: HTTP_STATUS_MAX,
}
} else if let [class @ b'1'..=b'5', x1, x2] = spec.as_bytes()
&& x1.eq_ignore_ascii_case(&b'x')
&& x2.eq_ignore_ascii_case(&b'x')
{
let start = u32::from(class - b'0') * 100;
HttpStatusRange {
start,
end: start + 99,
}
} else if let Some((start, end)) = spec.split_once('-') {
HttpStatusRange {
start: parse_status_code(start).ok_or(INVALID)?,
end: parse_status_code(end).ok_or(INVALID)?,
}
} else {
let code = parse_status_code(spec).ok_or(INVALID)?;
HttpStatusRange {
start: code,
end: code,
}
};
validate_http_status_range(&range)?;
Ok(range)
}
fn parse_status_code(code: &str) -> Option<u32> {
let code = code.trim();
if code.is_empty() || !code.bytes().all(|b| b.is_ascii_digit()) {
return None;
}
code.parse().ok()
}
fn validate_http_status_range(range: &HttpStatusRange) -> Result<(), &'static str> {
if range.start < HTTP_STATUS_MIN || range.end > HTTP_STATUS_MAX {
return Err("health check accepted statuses must lie within 100-599");
}
if range.start > range.end {
return Err("health check accepted status range must not end before it starts");
}
debug_assert!(
(HTTP_STATUS_MIN..=HTTP_STATUS_MAX).contains(&range.start)
&& (HTTP_STATUS_MIN..=HTTP_STATUS_MAX).contains(&range.end),
"a validated range must lie within the valid status codes"
);
Ok(())
}
pub fn validate_affinity_key(
affinity_header: Option<&str>,
affinity_cookie: Option<&str>,
) -> Result<(), &'static str> {
match (affinity_header, affinity_cookie) {
(Some(_), Some(_)) => {
Err("affinity_header and affinity_cookie are mutually exclusive: set at most one")
}
(Some(name), None) if !header_name_is_valid_token(name.as_bytes()) => {
Err("affinity_header must be a non-empty RFC 9110 token (a header field name)")
}
(Some(name), None) if name.eq_ignore_ascii_case("host") => Err(
"affinity_header cannot be Host: the request authority is not a header Sōzu keeps \
(H1 folds Host into it, H2 sends :authority)",
),
(Some(name), None) if name.eq_ignore_ascii_case("cookie") => Err(
"affinity_header cannot be Cookie: cookies are parsed into a jar; key on one with \
affinity_cookie instead",
),
(None, Some(name)) if !header_name_is_valid_token(name.as_bytes()) => {
Err("affinity_cookie must be a non-empty RFC 6265 cookie name (an RFC 9110 token)")
}
_ => Ok(()),
}
}
pub const DEFAULT_SHARD_MIN_BACKENDS: u32 = 8;
pub fn validate_shuffle_sharding(
shard_percent: Option<u32>,
shard_min_backends: Option<u32>,
shard_mode: Option<i32>,
) -> Result<(), &'static str> {
let Some(percent) = shard_percent else {
if shard_min_backends.is_some() || shard_mode.is_some() {
return Err("shard_min_backends and shard_mode require shard_percent");
}
return Ok(());
};
if !(1..=100).contains(&percent) {
return Err("shard_percent must lie in 1..=100");
}
if shard_min_backends.is_some_and(|minimum| minimum < 2) {
return Err("shard_min_backends must be at least 2, the smallest shard");
}
if shard_mode.is_some_and(|mode| ShardMode::try_from(mode).is_err()) {
return Err("shard_mode must be FALLBACK or STRICT");
}
Ok(())
}
pub const HEALTH_CHECK_MAX_HEADERS: usize = 32;
pub const HEALTH_CHECK_MAX_HEADER_BYTES: usize = 8192;
pub const HEALTH_CHECK_MAX_HPACK_BLOCK: usize = 16384;
pub const HEALTH_CHECK_MAX_AUTHORITY_LEN: usize = 1 + 39 + 11 + 1 + 6;
const HEALTH_CHECK_FORBIDDEN_HEADERS: [&str; 10] = [
"host",
"connection",
"keep-alive",
"proxy-connection",
"transfer-encoding",
"content-length",
"te",
"trailer",
"upgrade",
"expect",
];
fn hpack_string_length_bytes(length: usize) -> usize {
if length < 127 {
return 1;
}
let mut rest = length - 127;
let mut bytes = 1;
loop {
bytes += 1;
rest >>= 7;
if rest == 0 {
return bytes;
}
}
}
pub fn health_check_hpack_upper_bound(cfg: &HealthCheckConfig, value_lens: &[usize]) -> usize {
let mut bound = 2;
bound += 1 + hpack_string_length_bytes(cfg.uri.len()) + cfg.uri.len();
bound += 1
+ hpack_string_length_bytes(HEALTH_CHECK_MAX_AUTHORITY_LEN)
+ HEALTH_CHECK_MAX_AUTHORITY_LEN;
for (header, value_len) in cfg.headers.iter().zip(value_lens) {
bound += 1
+ hpack_string_length_bytes(header.name.len())
+ header.name.len()
+ hpack_string_length_bytes(*value_len)
+ value_len;
}
bound
}
pub(crate) fn check_health_check_header_value(value: &str) -> Result<(), &'static str> {
if header_value_contains_forbidden_controls(value.as_bytes()) {
return Err("health check header value must not contain control bytes");
}
if value.starts_with([' ', '\t']) || value.ends_with([' ', '\t']) {
return Err("health check header value must not begin or end with a space or a tab");
}
Ok(())
}
fn validate_health_check_headers(
cfg: &HealthCheckConfig,
mode: HealthCheckMode,
) -> Result<(), &'static str> {
if cfg.headers.is_empty() {
return Ok(());
}
if mode == HealthCheckMode::Tcp {
return Err("health check headers apply to HTTP mode only: leave them unset with mode TCP");
}
if cfg.headers.len() > HEALTH_CHECK_MAX_HEADERS {
return Err("health check carries more than 32 headers");
}
for (index, header) in cfg.headers.iter().enumerate() {
if header.name.starts_with(':')
|| HEALTH_CHECK_FORBIDDEN_HEADERS
.iter()
.any(|forbidden| header.name.eq_ignore_ascii_case(forbidden))
{
return Err("health check header name is reserved for the probe or its framing");
}
if !header_name_is_valid_token(header.name.as_bytes()) {
return Err("health check header name must be a non-empty RFC 9110 token");
}
if cfg.headers[..index]
.iter()
.any(|earlier| earlier.name.eq_ignore_ascii_case(&header.name))
{
return Err("health check header names must be unique, compared case-insensitively");
}
if !header.fingerprint.is_empty()
&& header
.fingerprint
.parse::<crate::secret::HeaderFingerprint>()
.is_err()
{
return Err("health check header fingerprint is malformed");
}
}
Ok(())
}
pub fn validate_health_check_header_values(
cfg: &HealthCheckConfig,
values: &[HealthCheckHeaderValue],
) -> Result<(), &'static str> {
if values.len() != cfg.headers.len() {
return Err("health check carries a header value count different from its header count");
}
let mut wire_bytes = 0usize;
for (header, carrier) in cfg.headers.iter().zip(values) {
if header.name != carrier.name {
return Err(
"health check header values must follow the order and names of the headers",
);
}
let text = std::str::from_utf8(&carrier.value)
.map_err(|_| "health check header value must be valid UTF-8")?;
check_health_check_header_value(text)?;
wire_bytes = wire_bytes
.saturating_add(header.name.len())
.saturating_add(carrier.value.len())
.saturating_add(4);
}
if wire_bytes > HEALTH_CHECK_MAX_HEADER_BYTES {
return Err("health check headers exceed 8192 bytes (name + value + 4 per header)");
}
let value_lens: Vec<usize> = values.iter().map(|carrier| carrier.value.len()).collect();
if health_check_hpack_upper_bound(cfg, &value_lens) > HEALTH_CHECK_MAX_HPACK_BLOCK {
return Err("health check URI and headers may exceed the 16384-byte HTTP/2 probe block");
}
Ok(())
}
pub fn validate_health_check_config(cfg: &HealthCheckConfig) -> Result<(), &'static str> {
let Ok(mode) = HealthCheckMode::try_from(cfg.mode) else {
return Err("health check mode must be HTTP or TCP");
};
if cfg.interval == 0 {
return Err("health check interval must be > 0");
}
if cfg.timeout == 0 {
return Err("health check timeout must be > 0");
}
if cfg.healthy_threshold == 0 {
return Err("health check healthy_threshold must be > 0");
}
if cfg.unhealthy_threshold == 0 {
return Err("health check unhealthy_threshold must be > 0");
}
if mode == HealthCheckMode::Http && !cfg.uri.starts_with('/') {
return Err("health check URI must start with '/'");
}
if cfg
.uri
.bytes()
.any(|b| b == b'\r' || b == b'\n' || b == 0 || (b < 0x20 && b != b'\t'))
{
return Err("health check URI must not contain CR, LF, NUL, or other C0 control bytes");
}
for range in &cfg.accepted_statuses {
validate_http_status_range(range)?;
}
if cfg.expected_status != 0 && !cfg.accepted_statuses.is_empty() {
return Err(
"health check expected_status and accepted_statuses are mutually exclusive: \
set at most one",
);
}
if mode == HealthCheckMode::Tcp
&& (cfg.expected_status != 0 || !cfg.accepted_statuses.is_empty())
{
return Err(
"health check expected_status and accepted_statuses apply to HTTP mode only: \
leave them unset with mode TCP",
);
}
if cfg
.port
.is_some_and(|port| !(1..=u32::from(u16::MAX)).contains(&port))
{
return Err("health check port must lie in 1..=65535");
}
validate_health_check_headers(cfg, mode)?;
debug_assert!(
cfg.interval > 0
&& cfg.timeout > 0
&& cfg.healthy_threshold > 0
&& cfg.unhealthy_threshold > 0,
"validated health-check thresholds must all be strictly positive"
);
debug_assert!(
mode == HealthCheckMode::Tcp || cfg.uri.starts_with('/'),
"a validated HTTP health-check URI must be an absolute path"
);
debug_assert!(
cfg.expected_status == 0 || cfg.accepted_statuses.is_empty(),
"a validated health check sets at most one of expected_status and accepted_statuses"
);
debug_assert!(
mode == HealthCheckMode::Http
|| (cfg.expected_status == 0 && cfg.accepted_statuses.is_empty()),
"a validated TCP health check sets no status field"
);
debug_assert!(
cfg.port
.is_none_or(|port| port != 0 && u16::try_from(port).is_ok()),
"a validated health-check probe port is a non-zero 16-bit port"
);
Ok(())
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct FileClusterConfig {
pub frontends: Vec<FileClusterFrontendConfig>,
pub backends: Vec<FileBackendConfig>,
pub protocol: FileClusterProtocolConfig,
pub sticky_session: Option<bool>,
pub https_redirect: Option<bool>,
#[serde(default)]
pub send_proxy: Option<bool>,
#[serde(default)]
pub load_balancing: LoadBalancingAlgorithms,
pub answer_503: Option<String>,
#[serde(default)]
pub load_metric: Option<LoadMetric>,
pub http2: Option<bool>,
pub answers: Option<BTreeMap<String, String>>,
pub https_redirect_port: Option<u32>,
pub authorized_hashes: Option<Vec<String>>,
pub www_authenticate: Option<String>,
pub max_connections_per_ip: Option<u64>,
pub max_connections_per_subnet: Option<u64>,
pub retry_after: Option<u32>,
#[serde(default)]
pub health_check: Option<FileHealthCheckConfig>,
#[serde(default)]
pub udp: Option<FileUdpClusterConfig>,
#[serde(default)]
pub affinity_header: Option<String>,
#[serde(default)]
pub affinity_cookie: Option<String>,
#[serde(default)]
pub shard_percent: Option<u32>,
#[serde(default)]
pub shard_min_backends: Option<u32>,
#[serde(default)]
pub shard_mode: Option<ShardMode>,
#[serde(default)]
pub max_connection_attempts: Option<u32>,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct FileUdpHealthConfig {
pub mode: Option<UdpHealthMode>,
pub tcp_port: Option<u32>,
pub rise: Option<u32>,
pub fall: Option<u32>,
pub fail_open: Option<bool>,
pub udp_probe_payload: Option<String>,
pub probe_interval_seconds: Option<u32>,
pub probe_timeout_seconds: Option<u32>,
}
impl FileUdpHealthConfig {
pub fn to_proto(&self) -> UdpHealthConfig {
UdpHealthConfig {
mode: self.mode.map(|m| m as i32),
tcp_port: self.tcp_port,
rise: self.rise,
fall: self.fall,
fail_open: self.fail_open,
udp_probe_payload: self
.udp_probe_payload
.as_ref()
.map(|p| p.as_bytes().to_owned()),
probe_interval_seconds: self.probe_interval_seconds,
probe_timeout_seconds: self.probe_timeout_seconds,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct FileUdpClusterConfig {
pub affinity_key: Option<UdpAffinityKey>,
pub responses: Option<u32>,
pub requests: Option<u32>,
pub send_proxy_protocol: Option<bool>,
pub proxy_protocol_every_datagram: Option<bool>,
pub health: Option<FileUdpHealthConfig>,
}
impl FileUdpClusterConfig {
pub fn to_proto(&self) -> UdpClusterConfig {
UdpClusterConfig {
affinity_key: self.affinity_key.map(|k| k as i32),
responses: self.responses,
requests: self.requests,
send_proxy_protocol: self.send_proxy_protocol,
proxy_protocol_every_datagram: self.proxy_protocol_every_datagram,
health: self.health.as_ref().map(|h| h.to_proto()),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct BackendConfig {
pub address: SocketAddr,
pub weight: Option<u8>,
pub sticky_id: Option<String>,
pub backup: Option<bool>,
pub backend_id: Option<String>,
pub unix_path: Option<String>,
}
fn validate_backend_unix_paths(
cluster_id: &str,
backends: &[BackendConfig],
) -> Result<(), ConfigError> {
for (index, backend) in backends.iter().enumerate() {
let Some(unix_path) = backend.unix_path.as_deref() else {
continue;
};
let invalid = |reason| ConfigError::InvalidBackendUnixPath {
cluster_id: cluster_id.to_owned(),
address: backend.address,
reason,
};
parse_backend_unix_path(unix_path).map_err(invalid)?;
if backends
.iter()
.enumerate()
.any(|(other, b)| other != index && b.address == backend.address)
{
return Err(invalid(
"a unix backend's address is its identity and must be unique in the cluster",
));
}
}
Ok(())
}
pub type BackendResolver<'a> = dyn Fn(&str) -> std::io::Result<Vec<SocketAddr>> + 'a;
pub fn system_resolver(address: &str) -> std::io::Result<Vec<SocketAddr>> {
address.to_socket_addrs().map(Iterator::collect)
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct FileBackendConfig {
pub address: String,
pub weight: Option<u8>,
pub sticky_id: Option<String>,
pub backup: Option<bool>,
pub backend_id: Option<String>,
pub unix_path: Option<String>,
}
fn is_legacy_numeric_host(host: &str) -> bool {
host.split('.').all(|component| {
!component.is_empty()
&& match component
.strip_prefix("0x")
.or_else(|| component.strip_prefix("0X"))
{
Some(hex) => hex.bytes().all(|byte| byte.is_ascii_hexdigit()),
None => component.bytes().all(|byte| byte.is_ascii_digit()),
}
})
}
impl FileBackendConfig {
pub fn resolve(
&self,
cluster_id: &str,
index: usize,
resolver: &BackendResolver<'_>,
) -> Result<Vec<BackendConfig>, ConfigError> {
if let Ok(address) = self.address.parse::<SocketAddr>() {
return Ok(vec![BackendConfig {
address,
weight: self.weight,
sticky_id: self.sticky_id.clone(),
backup: self.backup,
backend_id: Some(
self.backend_id
.to_owned()
.unwrap_or_else(|| format!("{cluster_id}-{index}-{address}")),
),
unix_path: self.unix_path.clone(),
}]);
}
let error = |reason: String| ConfigError::BackendAddressResolution {
cluster_id: cluster_id.to_owned(),
address: self.address.clone(),
reason,
};
if let Some((host, _port)) = self.address.rsplit_once(':') {
if host.contains(':') {
return Err(error(format!(
"{host:?} is an unbracketed IPv6 literal, IPv6 literals must use [address]:port"
)));
}
if !host.is_empty() && is_legacy_numeric_host(host) {
return Err(error(format!(
"{host:?} is not a valid IP address literal and numeric shorthand is not accepted"
)));
}
}
let addresses = resolver(&self.address).map_err(|io_error| error(io_error.to_string()))?;
self.expand(cluster_id, addresses)
}
fn expand(
&self,
cluster_id: &str,
mut addresses: Vec<SocketAddr>,
) -> Result<Vec<BackendConfig>, ConfigError> {
addresses.sort_unstable();
addresses.dedup();
if addresses.is_empty() {
return Err(ConfigError::BackendAddressResolution {
cluster_id: cluster_id.to_owned(),
address: self.address.clone(),
reason: "the name resolved to no address".to_owned(),
});
}
let backends: Vec<BackendConfig> = addresses
.iter()
.map(|address| BackendConfig {
address: *address,
weight: self.weight,
sticky_id: self
.sticky_id
.as_ref()
.map(|sticky_id| format!("{sticky_id}-{address}")),
backup: self.backup,
backend_id: Some(match &self.backend_id {
Some(backend_id) => format!("{backend_id}-{address}"),
None => format!("{cluster_id}-{}-{address}", self.address),
}),
unix_path: self.unix_path.clone(),
})
.collect();
debug_assert_eq!(backends.len(), addresses.len());
debug_assert!(
backends
.iter()
.map(|backend| backend.backend_id.as_deref())
.collect::<HashSet<_>>()
.len()
== backends.len(),
"resolved backends must carry distinct ids"
);
Ok(backends)
}
}
impl FileClusterConfig {
pub fn to_cluster_config(
self,
cluster_id: &str,
expect_proxy: &HashSet<SocketAddr>,
) -> Result<ClusterConfig, ConfigError> {
self.to_cluster_config_with(cluster_id, expect_proxy, Some(&system_resolver))
}
pub fn to_cluster_config_with(
self,
cluster_id: &str,
expect_proxy: &HashSet<SocketAddr>,
resolver: Option<&BackendResolver<'_>>,
) -> Result<ClusterConfig, ConfigError> {
let requested_frontend_count = self.frontends.len();
validate_affinity_key(
self.affinity_header.as_deref(),
self.affinity_cookie.as_deref(),
)
.map_err(|reason| ConfigError::InvalidAffinityKey {
cluster_id: cluster_id.to_owned(),
reason,
})?;
validate_shuffle_sharding(
self.shard_percent,
self.shard_min_backends,
self.shard_mode.map(|mode| mode as i32),
)
.map_err(|reason| ConfigError::InvalidShuffleSharding {
cluster_id: cluster_id.to_owned(),
reason,
})?;
if let Some(attempts) = self.max_connection_attempts {
validate_max_connection_attempts(attempts).map_err(|reason| {
ConfigError::InvalidMaxConnectionAttempts {
cluster_id: Some(cluster_id.to_owned()),
value: attempts,
reason,
}
})?;
}
let mut backends = Vec::with_capacity(self.backends.len());
for (index, backend) in self.backends.iter().enumerate() {
match resolver {
Some(resolver) => backends.extend(backend.resolve(cluster_id, index, resolver)?),
None if backend.address.parse::<SocketAddr>().is_ok() => {
backends.extend(backend.resolve(cluster_id, index, &system_resolver)?)
}
None => {}
}
}
debug_assert!(resolver.is_none() || backends.len() >= self.backends.len());
let (health_check, health_check_header_files) = match self
.health_check
.as_ref()
.map(FileHealthCheckConfig::to_parts)
.transpose()
.map_err(|reason| ConfigError::InvalidHealthCheck {
cluster_id: cluster_id.to_owned(),
reason,
})? {
Some((config, files)) => (Some(config), files),
None => (None, Vec::new()),
};
validate_backend_unix_paths(cluster_id, &backends)?;
if health_check.as_ref().is_some_and(|hc| hc.port.is_some())
&& self.backends.iter().any(|b| b.unix_path.is_some())
{
return Err(ConfigError::InvalidHealthCheck {
cluster_id: cluster_id.to_owned(),
reason: HEALTH_CHECK_PORT_UNIX_REFUSED,
});
}
match self.protocol {
FileClusterProtocolConfig::Tcp => {
if self.affinity_header.is_some() || self.affinity_cookie.is_some() {
return Err(ConfigError::InvalidAffinityKey {
cluster_id: cluster_id.to_owned(),
reason: "a TCP cluster always keys on the client source IP: \
affinity_header and affinity_cookie apply to HTTP clusters only",
});
}
let mut has_expect_proxy = None;
let mut frontends = Vec::new();
for f in self.frontends {
if expect_proxy.contains(&f.address) {
match has_expect_proxy {
Some(true) => {}
Some(false) => {
return Err(ConfigError::Incompatible {
object: ObjectKind::Cluster,
id: cluster_id.to_owned(),
kind: IncompatibilityKind::ProxyProtocol,
});
}
None => has_expect_proxy = Some(true),
}
} else {
match has_expect_proxy {
Some(false) => {}
Some(true) => {
return Err(ConfigError::Incompatible {
object: ObjectKind::Cluster,
id: cluster_id.to_owned(),
kind: IncompatibilityKind::ProxyProtocol,
});
}
None => has_expect_proxy = Some(false),
}
}
let tcp_frontend = f.to_tcp_front()?;
frontends.push(tcp_frontend);
}
let send_proxy = self.send_proxy.unwrap_or(false);
let expect_proxy = has_expect_proxy.unwrap_or(false);
let proxy_protocol = match (send_proxy, expect_proxy) {
(true, true) => Some(ProxyProtocolConfig::RelayHeader),
(true, false) => Some(ProxyProtocolConfig::SendHeader),
(false, true) => Some(ProxyProtocolConfig::ExpectHeader),
_ => None,
};
let answers = match self.answers.as_ref() {
Some(map) => load_answers(map)?,
None => BTreeMap::new(),
};
let udp = self.udp.as_ref().map(|u| u.to_proto());
debug_assert_eq!(
frontends.len(),
requested_frontend_count,
"every TCP frontend must survive conversion"
);
debug_assert_eq!(
proxy_protocol,
match (send_proxy, expect_proxy) {
(true, true) => Some(ProxyProtocolConfig::RelayHeader),
(true, false) => Some(ProxyProtocolConfig::SendHeader),
(false, true) => Some(ProxyProtocolConfig::ExpectHeader),
(false, false) => None,
},
"proxy_protocol must be the (send, expect) function"
);
Ok(ClusterConfig::Tcp(TcpClusterConfig {
cluster_id: cluster_id.to_string(),
frontends,
backends,
proxy_protocol,
load_balancing: self.load_balancing,
load_metric: self.load_metric,
answers,
https_redirect_port: self.https_redirect_port,
authorized_hashes: self.authorized_hashes.unwrap_or_default(),
www_authenticate: self.www_authenticate,
max_connections_per_ip: self.max_connections_per_ip,
max_connections_per_subnet: self.max_connections_per_subnet,
retry_after: self.retry_after,
health_check,
health_check_header_files,
udp,
shard_percent: self.shard_percent,
shard_min_backends: self.shard_min_backends,
shard_mode: self.shard_mode,
max_connection_attempts: self.max_connection_attempts,
}))
}
FileClusterProtocolConfig::Http => {
let mut frontends = Vec::new();
for frontend in self.frontends {
let http_frontend = frontend.to_http_front(cluster_id)?;
frontends.push(http_frontend);
}
let answer_503 = self.answer_503.as_ref().and_then(|path| {
Config::load_file(path)
.map_err(|e| {
error!("cannot load 503 error page at path '{}': {:?}", path, e);
e
})
.ok()
});
let answers = match self.answers.as_ref() {
Some(map) => load_answers(map)?,
None => BTreeMap::new(),
};
let udp = self.udp.as_ref().map(|u| u.to_proto());
debug_assert_eq!(
frontends.len(),
requested_frontend_count,
"every HTTP frontend must survive conversion"
);
Ok(ClusterConfig::Http(HttpClusterConfig {
cluster_id: cluster_id.to_string(),
frontends,
backends,
sticky_session: self.sticky_session.unwrap_or(false),
https_redirect: self.https_redirect.unwrap_or(false),
load_balancing: self.load_balancing,
load_metric: self.load_metric,
answer_503,
http2: self.http2,
answers,
https_redirect_port: self.https_redirect_port,
authorized_hashes: self.authorized_hashes.unwrap_or_default(),
www_authenticate: self.www_authenticate,
max_connections_per_ip: self.max_connections_per_ip,
max_connections_per_subnet: self.max_connections_per_subnet,
retry_after: self.retry_after,
health_check,
health_check_header_files,
udp,
affinity_header: self.affinity_header,
affinity_cookie: self.affinity_cookie,
shard_percent: self.shard_percent,
shard_min_backends: self.shard_min_backends,
shard_mode: self.shard_mode,
max_connection_attempts: self.max_connection_attempts,
}))
}
}
}
}
#[derive(Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct HttpFrontendConfig {
pub address: SocketAddr,
pub hostname: String,
pub path: PathRule,
pub method: Option<String>,
pub certificate: Option<String>,
pub key: Option<String>,
pub certificate_chain: Option<Vec<String>>,
#[serde(default)]
pub tls_versions: Vec<TlsVersion>,
#[serde(default)]
pub position: RulePosition,
pub tags: Option<BTreeMap<String, String>>,
#[serde(default)]
pub redirect: Option<RedirectPolicy>,
#[serde(default)]
pub redirect_scheme: Option<RedirectScheme>,
#[serde(default)]
pub redirect_template: Option<String>,
#[serde(default)]
pub rewrite_host: Option<String>,
#[serde(default)]
pub rewrite_path: Option<String>,
#[serde(default)]
pub rewrite_port: Option<u32>,
#[serde(default)]
pub required_auth: Option<bool>,
#[serde(default)]
pub headers: Vec<Header>,
#[serde(default)]
pub hsts: Option<HstsConfig>,
#[serde(default)]
pub header_predicates: Vec<HeaderPredicate>,
}
impl fmt::Debug for HttpFrontendConfig {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let certificate = self.certificate.as_ref().map(|_| "[redacted]");
let certificate_len = self.certificate.as_ref().map(String::len);
let key = self.key.as_ref().map(|_| "[redacted]");
let key_len = self.key.as_ref().map(String::len);
let certificate_chain = self.certificate_chain.as_ref().map(|_| "[redacted]");
let certificate_chain_count = self.certificate_chain.as_ref().map(Vec::len);
let certificate_chain_len = self.certificate_chain.as_ref().map(|chain| {
chain
.iter()
.map(String::len)
.fold(0usize, usize::saturating_add)
});
let method_len = self.method.as_ref().map(String::len);
let redirect_template_len = self.redirect_template.as_ref().map(String::len);
let rewrite_host_len = self.rewrite_host.as_ref().map(String::len);
let rewrite_path_len = self.rewrite_path.as_ref().map(String::len);
f.debug_struct("HttpFrontendConfig")
.field("address", &self.address)
.field("hostname_len", &self.hostname.len())
.field("path_kind", &self.path.kind)
.field("path_len", &self.path.value.len())
.field("method_len", &method_len)
.field("certificate", &certificate)
.field("certificate_len", &certificate_len)
.field("key", &key)
.field("key_len", &key_len)
.field("certificate_chain", &certificate_chain)
.field("certificate_chain_count", &certificate_chain_count)
.field("certificate_chain_len", &certificate_chain_len)
.field("tls_versions_count", &self.tls_versions.len())
.field("position", &self.position)
.field(
"tags_count",
&self.tags.as_ref().map(BTreeMap::len).unwrap_or_default(),
)
.field("redirect", &self.redirect)
.field("redirect_scheme", &self.redirect_scheme)
.field("redirect_template_len", &redirect_template_len)
.field("rewrite_host_len", &rewrite_host_len)
.field("rewrite_path_len", &rewrite_path_len)
.field("rewrite_port", &self.rewrite_port)
.field("required_auth", &self.required_auth)
.field("headers_count", &self.headers.len())
.field("hsts", &self.hsts)
.field("header_predicates_count", &self.header_predicates.len())
.finish()
}
}
impl HttpFrontendConfig {
pub fn generate_requests(&self, cluster_id: &str) -> Vec<Request> {
let mut v = Vec::new();
let tags = self.tags.clone().unwrap_or_default();
if self.key.is_some() && self.certificate.is_some() {
v.push(
RequestType::AddCertificate(AddCertificate {
address: self.address.into(),
certificate: CertificateAndKey {
key: self.key.clone().unwrap(),
certificate: self.certificate.clone().unwrap(),
certificate_chain: self.certificate_chain.clone().unwrap_or_default(),
versions: self.tls_versions.iter().map(|v| *v as i32).collect(),
names: vec![],
},
expired_at: None,
})
.into(),
);
v.push(
RequestType::add_https_frontend(RequestHttpFrontend {
cluster_id: Some(cluster_id.to_string()),
address: self.address.into(),
hostname: self.hostname.clone(),
path: self.path.clone(),
method: self.method.clone(),
position: self.position.into(),
tags,
redirect: self.redirect.map(|r| r as i32),
required_auth: self.required_auth,
redirect_scheme: self.redirect_scheme.map(|s| s as i32),
redirect_template: self.redirect_template.clone(),
rewrite_host: self.rewrite_host.clone(),
rewrite_path: self.rewrite_path.clone(),
rewrite_port: self.rewrite_port,
headers: self.headers.clone(),
hsts: self.hsts,
header_predicates: self.header_predicates.clone(),
})
.into(),
);
} else {
v.push(
RequestType::add_http_frontend(RequestHttpFrontend {
cluster_id: Some(cluster_id.to_string()),
address: self.address.into(),
hostname: self.hostname.clone(),
path: self.path.clone(),
method: self.method.clone(),
position: self.position.into(),
tags,
redirect: self.redirect.map(|r| r as i32),
required_auth: self.required_auth,
redirect_scheme: self.redirect_scheme.map(|s| s as i32),
redirect_template: self.redirect_template.clone(),
rewrite_host: self.rewrite_host.clone(),
rewrite_path: self.rewrite_path.clone(),
rewrite_port: self.rewrite_port,
headers: self.headers.clone(),
hsts: self.hsts,
header_predicates: self.header_predicates.clone(),
})
.into(),
);
}
v
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct HttpClusterConfig {
pub cluster_id: String,
pub frontends: Vec<HttpFrontendConfig>,
pub backends: Vec<BackendConfig>,
pub sticky_session: bool,
pub https_redirect: bool,
pub load_balancing: LoadBalancingAlgorithms,
pub load_metric: Option<LoadMetric>,
pub answer_503: Option<String>,
pub http2: Option<bool>,
#[serde(default)]
pub answers: BTreeMap<String, String>,
#[serde(default)]
pub https_redirect_port: Option<u32>,
#[serde(default)]
pub authorized_hashes: Vec<String>,
#[serde(default)]
pub www_authenticate: Option<String>,
#[serde(default)]
pub max_connections_per_ip: Option<u64>,
#[serde(default)]
pub max_connections_per_subnet: Option<u64>,
#[serde(default)]
pub retry_after: Option<u32>,
#[serde(default)]
pub health_check: Option<HealthCheckConfig>,
#[serde(default)]
pub health_check_header_files: Vec<FileHealthCheckHeader>,
#[serde(default)]
pub udp: Option<UdpClusterConfig>,
#[serde(default)]
pub affinity_header: Option<String>,
#[serde(default)]
pub affinity_cookie: Option<String>,
#[serde(default)]
pub shard_percent: Option<u32>,
#[serde(default)]
pub shard_min_backends: Option<u32>,
#[serde(default)]
pub shard_mode: Option<ShardMode>,
#[serde(default)]
pub max_connection_attempts: Option<u32>,
}
impl HttpClusterConfig {
pub fn generate_requests(&self) -> Result<Vec<Request>, ConfigError> {
if !self.health_check_header_files.is_empty() {
return Err(ConfigError::InvalidHealthCheck {
cluster_id: self.cluster_id.clone(),
reason: HEALTH_CHECK_VALUE_FILE_UNSUPPORTED,
});
}
let mut v: Vec<Request> = vec![
RequestType::AddCluster(Cluster {
cluster_id: self.cluster_id.clone(),
sticky_session: self.sticky_session,
https_redirect: self.https_redirect,
proxy_protocol: None,
load_balancing: self.load_balancing as i32,
answer_503: self.answer_503.clone(),
load_metric: self.load_metric.map(|s| s as i32),
http2: self.http2,
answers: self.answers.clone(),
https_redirect_port: self.https_redirect_port,
authorized_hashes: self.authorized_hashes.clone(),
www_authenticate: self.www_authenticate.clone(),
max_connections_per_ip: self.max_connections_per_ip,
max_connections_per_subnet: self.max_connections_per_subnet,
retry_after: self.retry_after,
health_check: self.health_check.clone(),
udp: self.udp.clone(),
affinity_header: self.affinity_header.clone(),
affinity_cookie: self.affinity_cookie.clone(),
shard_percent: self.shard_percent,
shard_min_backends: self.shard_min_backends,
shard_mode: self.shard_mode.map(|mode| mode as i32),
max_connection_attempts: self.max_connection_attempts,
})
.into(),
];
for frontend in &self.frontends {
let mut orders = frontend.generate_requests(&self.cluster_id);
v.append(&mut orders);
}
for (backend_count, backend) in self.backends.iter().enumerate() {
let load_balancing_parameters = Some(LoadBalancingParams {
weight: backend.weight.unwrap_or(100) as i32,
});
v.push(
RequestType::add_backend(AddBackend {
cluster_id: self.cluster_id.clone(),
backend_id: backend.backend_id.clone().unwrap_or_else(|| {
format!("{}-{}-{}", self.cluster_id, backend_count, backend.address)
}),
address: backend.address.into(),
load_balancing_parameters,
sticky_id: backend.sticky_id.clone(),
backup: backend.backup,
unix_path: backend.unix_path.clone(),
})
.into(),
);
}
debug_assert!(
matches!(
v.first().and_then(|r| r.request_type.as_ref()),
Some(RequestType::AddCluster(_))
),
"HTTP cluster orders must lead with an AddCluster"
);
debug_assert_eq!(
v.iter()
.filter(|r| matches!(
r.request_type,
Some(RequestType::AddBackend(_)) | Some(RequestType::AddUnixBackend(_))
))
.count(),
self.backends.len(),
"one AddBackend order per configured backend"
);
Ok(v)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct TcpFrontendConfig {
pub address: SocketAddr,
pub tags: Option<BTreeMap<String, String>>,
#[serde(default)]
pub udp: bool,
#[serde(default)]
pub sni: Option<String>,
#[serde(default)]
pub alpn: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct TcpClusterConfig {
pub cluster_id: String,
pub frontends: Vec<TcpFrontendConfig>,
pub backends: Vec<BackendConfig>,
#[serde(default)]
pub proxy_protocol: Option<ProxyProtocolConfig>,
pub load_balancing: LoadBalancingAlgorithms,
pub load_metric: Option<LoadMetric>,
#[serde(default)]
pub answers: BTreeMap<String, String>,
#[serde(default)]
pub https_redirect_port: Option<u32>,
#[serde(default)]
pub authorized_hashes: Vec<String>,
#[serde(default)]
pub www_authenticate: Option<String>,
#[serde(default)]
pub max_connections_per_ip: Option<u64>,
#[serde(default)]
pub max_connections_per_subnet: Option<u64>,
#[serde(default)]
pub retry_after: Option<u32>,
#[serde(default)]
pub health_check: Option<HealthCheckConfig>,
#[serde(default)]
pub health_check_header_files: Vec<FileHealthCheckHeader>,
#[serde(default)]
pub udp: Option<UdpClusterConfig>,
#[serde(default)]
pub shard_percent: Option<u32>,
#[serde(default)]
pub shard_min_backends: Option<u32>,
#[serde(default)]
pub shard_mode: Option<ShardMode>,
#[serde(default)]
pub max_connection_attempts: Option<u32>,
}
impl TcpClusterConfig {
pub fn generate_requests(&self) -> Result<Vec<Request>, ConfigError> {
if !self.health_check_header_files.is_empty() {
return Err(ConfigError::InvalidHealthCheck {
cluster_id: self.cluster_id.clone(),
reason: HEALTH_CHECK_VALUE_FILE_UNSUPPORTED,
});
}
let mut v: Vec<Request> = vec![
RequestType::AddCluster(Cluster {
cluster_id: self.cluster_id.clone(),
sticky_session: false,
https_redirect: false,
proxy_protocol: self.proxy_protocol.map(|s| s as i32),
load_balancing: self.load_balancing as i32,
load_metric: self.load_metric.map(|s| s as i32),
answer_503: None,
http2: None,
answers: self.answers.clone(),
https_redirect_port: self.https_redirect_port,
authorized_hashes: self.authorized_hashes.clone(),
www_authenticate: self.www_authenticate.clone(),
max_connections_per_ip: self.max_connections_per_ip,
max_connections_per_subnet: self.max_connections_per_subnet,
retry_after: self.retry_after,
health_check: self.health_check.clone(),
udp: self.udp.clone(),
affinity_header: None,
affinity_cookie: None,
shard_percent: self.shard_percent,
shard_min_backends: self.shard_min_backends,
shard_mode: self.shard_mode.map(|mode| mode as i32),
max_connection_attempts: self.max_connection_attempts,
})
.into(),
];
for frontend in &self.frontends {
if frontend.udp {
v.push(
RequestType::AddUdpFrontend(RequestUdpFrontend {
cluster_id: self.cluster_id.clone(),
address: frontend.address.into(),
tags: frontend.tags.clone().unwrap_or(BTreeMap::new()),
})
.into(),
);
} else {
v.push(
RequestType::AddTcpFrontend(RequestTcpFrontend {
cluster_id: self.cluster_id.clone(),
address: frontend.address.into(),
tags: frontend.tags.clone().unwrap_or(BTreeMap::new()),
sni: frontend.sni.clone(),
alpn: frontend.alpn.clone(),
})
.into(),
);
}
}
for (backend_count, backend) in self.backends.iter().enumerate() {
let load_balancing_parameters = Some(LoadBalancingParams {
weight: backend.weight.unwrap_or(100) as i32,
});
v.push(
RequestType::add_backend(AddBackend {
cluster_id: self.cluster_id.clone(),
backend_id: backend.backend_id.clone().unwrap_or_else(|| {
format!("{}-{}-{}", self.cluster_id, backend_count, backend.address)
}),
address: backend.address.into(),
load_balancing_parameters,
sticky_id: backend.sticky_id.clone(),
backup: backend.backup,
unix_path: backend.unix_path.clone(),
})
.into(),
);
}
debug_assert!(
matches!(
v.first().and_then(|r| r.request_type.as_ref()),
Some(RequestType::AddCluster(_))
),
"TCP cluster orders must lead with an AddCluster"
);
debug_assert_eq!(
v.iter()
.filter(|r| matches!(
r.request_type,
Some(RequestType::AddTcpFrontend(_)) | Some(RequestType::AddUdpFrontend(_))
))
.count(),
self.frontends.len(),
"one AddTcpFrontend or AddUdpFrontend order per configured frontend"
);
debug_assert_eq!(
v.iter()
.filter(|r| matches!(
r.request_type,
Some(RequestType::AddBackend(_)) | Some(RequestType::AddUnixBackend(_))
))
.count(),
self.backends.len(),
"one AddBackend order per configured backend"
);
Ok(v)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum ClusterConfig {
Http(HttpClusterConfig),
Tcp(TcpClusterConfig),
}
impl ClusterConfig {
pub fn generate_requests(&self) -> Result<Vec<Request>, ConfigError> {
match *self {
ClusterConfig::Http(ref http) => http.generate_requests(),
ClusterConfig::Tcp(ref tcp) => tcp.generate_requests(),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Default, Deserialize)]
pub struct FileConfig {
pub command_socket: Option<String>,
#[serde(default)]
pub command_socket_mode: Option<String>,
pub command_buffer_size: Option<u64>,
pub max_command_buffer_size: Option<u64>,
pub max_connections: Option<usize>,
pub min_buffers: Option<u64>,
pub max_buffers: Option<u64>,
pub buffer_size: Option<u64>,
#[serde(default)]
pub slab_entries_per_connection: Option<u64>,
#[serde(default)]
pub basic_auth_max_credential_bytes: Option<u64>,
#[serde(default)]
pub max_connections_per_ip: Option<u64>,
#[serde(default)]
pub max_connections_per_subnet: Option<u64>,
#[serde(default)]
pub subnet_ipv4_prefix: Option<u32>,
#[serde(default)]
pub subnet_ipv6_prefix: Option<u32>,
#[serde(default)]
pub retry_after: Option<u32>,
#[serde(default)]
pub max_connection_attempts: Option<u32>,
#[serde(default)]
pub splice_pipe_capacity_bytes: Option<u64>,
#[serde(default)]
pub command_allowed_uids: Option<Vec<u32>>,
pub saved_state: Option<String>,
#[serde(default)]
pub automatic_state_save: Option<bool>,
pub log_level: Option<String>,
pub log_target: Option<String>,
#[serde(default)]
pub log_colored: bool,
#[serde(default)]
pub audit_logs_target: Option<String>,
#[serde(default)]
pub audit_logs_json_target: Option<String>,
#[serde(default)]
pub access_logs_target: Option<String>,
#[serde(default)]
pub access_logs_format: Option<AccessLogFormat>,
#[serde(default)]
pub access_logs_colored: Option<bool>,
pub worker_count: Option<u16>,
pub worker_automatic_restart: Option<bool>,
pub metrics: Option<MetricsConfig>,
pub disable_cluster_metrics: Option<bool>,
pub listeners: Option<Vec<ListenerBuilder>>,
pub clusters: Option<HashMap<String, FileClusterConfig>>,
pub handle_process_affinity: Option<bool>,
pub ctl_command_timeout: Option<u64>,
pub pid_file_path: Option<String>,
pub activate_listeners: Option<bool>,
#[serde(default)]
pub front_timeout: Option<u32>,
#[serde(default)]
pub back_timeout: Option<u32>,
#[serde(default)]
pub connect_timeout: Option<u32>,
#[serde(default)]
pub zombie_check_interval: Option<u32>,
#[serde(default)]
pub accept_queue_timeout: Option<u32>,
#[serde(default)]
pub evict_on_queue_full: Option<bool>,
#[serde(default)]
pub request_timeout: Option<u32>,
#[serde(default)]
pub worker_timeout: Option<u32>,
}
impl FileConfig {
pub fn load_from_path(path: &str) -> Result<FileConfig, ConfigError> {
let data = Config::load_file(path)?;
let config: FileConfig = match toml::from_str(&data) {
Ok(config) => config,
Err(e) => {
return Err(ConfigError::DeserializeToml(toml_error_diagnostic(
path, &e,
)));
}
};
let mut reserved_address: HashSet<ListenerKey> = HashSet::new();
if let Some(listeners) = config.listeners.as_ref() {
for listener in listeners.iter() {
let key = ListenerKey::new(listener.address, listener.interface.as_deref());
if reserved_address.contains(&key) {
return Err(ConfigError::ListenerAlreadyDeclared(key));
}
reserved_address.insert(key);
}
}
Ok(config)
}
}
pub struct ConfigBuilder {
file: FileConfig,
known_addresses: HashMap<SocketAddr, ListenerProtocol>,
expect_proxy_addresses: HashSet<SocketAddr>,
resolve_backends: bool,
built: Config,
}
impl ConfigBuilder {
pub fn new<S>(file_config: FileConfig, config_path: S) -> Self
where
S: ToString,
{
let built = Config {
accept_queue_timeout: file_config
.accept_queue_timeout
.unwrap_or(DEFAULT_ACCEPT_QUEUE_TIMEOUT),
evict_on_queue_full: file_config
.evict_on_queue_full
.unwrap_or(DEFAULT_EVICT_ON_QUEUE_FULL),
activate_listeners: file_config.activate_listeners.unwrap_or(true),
automatic_state_save: file_config
.automatic_state_save
.unwrap_or(DEFAULT_AUTOMATIC_STATE_SAVE),
back_timeout: file_config.back_timeout.unwrap_or(DEFAULT_BACK_TIMEOUT),
buffer_size: file_config.buffer_size.unwrap_or(DEFAULT_BUFFER_SIZE),
command_buffer_size: file_config
.command_buffer_size
.unwrap_or(DEFAULT_COMMAND_BUFFER_SIZE),
config_path: config_path.to_string(),
connect_timeout: file_config
.connect_timeout
.unwrap_or(DEFAULT_CONNECT_TIMEOUT),
ctl_command_timeout: file_config.ctl_command_timeout.unwrap_or(1_000),
front_timeout: file_config.front_timeout.unwrap_or(DEFAULT_FRONT_TIMEOUT),
handle_process_affinity: file_config.handle_process_affinity.unwrap_or(false),
access_logs_target: file_config.access_logs_target.clone(),
audit_logs_target: file_config.audit_logs_target.clone(),
audit_logs_json_target: file_config.audit_logs_json_target.clone(),
access_logs_format: file_config.access_logs_format.clone(),
access_logs_colored: file_config.access_logs_colored,
log_level: file_config
.log_level
.clone()
.unwrap_or_else(|| String::from("info")),
log_target: file_config
.log_target
.clone()
.unwrap_or_else(|| String::from("stdout")),
log_colored: file_config.log_colored,
max_buffers: file_config.max_buffers.unwrap_or(DEFAULT_MAX_BUFFERS),
max_command_buffer_size: file_config
.max_command_buffer_size
.unwrap_or(DEFAULT_MAX_COMMAND_BUFFER_SIZE),
max_connections: file_config
.max_connections
.unwrap_or(DEFAULT_MAX_CONNECTIONS),
metrics: file_config.metrics.clone(),
disable_cluster_metrics: file_config
.disable_cluster_metrics
.unwrap_or(DEFAULT_DISABLE_CLUSTER_METRICS),
min_buffers: std::cmp::min(
file_config.min_buffers.unwrap_or(DEFAULT_MIN_BUFFERS),
file_config.max_buffers.unwrap_or(DEFAULT_MAX_BUFFERS),
),
pid_file_path: file_config.pid_file_path.clone(),
request_timeout: file_config
.request_timeout
.unwrap_or(DEFAULT_REQUEST_TIMEOUT),
saved_state: file_config.saved_state.clone(),
worker_automatic_restart: file_config
.worker_automatic_restart
.unwrap_or(DEFAULT_WORKER_AUTOMATIC_RESTART),
worker_count: file_config.worker_count.unwrap_or(DEFAULT_WORKER_COUNT),
zombie_check_interval: file_config
.zombie_check_interval
.unwrap_or(DEFAULT_ZOMBIE_CHECK_INTERVAL),
worker_timeout: file_config.worker_timeout.unwrap_or(DEFAULT_WORKER_TIMEOUT),
slab_entries_per_connection: file_config.slab_entries_per_connection.map(|n| {
n.clamp(
ServerConfig::MIN_SLAB_ENTRIES_PER_CONNECTION,
ServerConfig::MAX_SLAB_ENTRIES_PER_CONNECTION,
)
}),
command_allowed_uids: file_config.command_allowed_uids.clone(),
basic_auth_max_credential_bytes: file_config.basic_auth_max_credential_bytes,
max_connections_per_ip: file_config
.max_connections_per_ip
.unwrap_or(DEFAULT_MAX_CONNECTIONS_PER_IP),
max_connections_per_subnet: file_config
.max_connections_per_subnet
.unwrap_or(DEFAULT_MAX_CONNECTIONS_PER_SUBNET),
subnet_ipv4_prefix: file_config
.subnet_ipv4_prefix
.unwrap_or(DEFAULT_SUBNET_IPV4_PREFIX),
subnet_ipv6_prefix: file_config
.subnet_ipv6_prefix
.unwrap_or(DEFAULT_SUBNET_IPV6_PREFIX),
retry_after: file_config.retry_after.unwrap_or(DEFAULT_RETRY_AFTER),
max_connection_attempts: file_config
.max_connection_attempts
.unwrap_or(DEFAULT_MAX_CONNECTION_ATTEMPTS),
splice_pipe_capacity_bytes: file_config.splice_pipe_capacity_bytes,
..Default::default()
};
debug_assert!(
built.min_buffers <= built.max_buffers,
"min_buffers must be clamped to <= max_buffers in the builder"
);
debug_assert!(
built.slab_entries_per_connection.is_none_or(|n| {
(ServerConfig::MIN_SLAB_ENTRIES_PER_CONNECTION
..=ServerConfig::MAX_SLAB_ENTRIES_PER_CONNECTION)
.contains(&n)
}),
"a set slab_entries_per_connection must be clamped into [MIN, MAX]"
);
Self {
file: file_config,
known_addresses: HashMap::new(),
expect_proxy_addresses: HashSet::new(),
resolve_backends: true,
built,
}
}
fn push_tls_listener(&mut self, mut listener: ListenerBuilder) -> Result<(), ConfigError> {
let listener = listener.to_tls(Some(&self.built))?;
self.built.https_listeners.push(listener);
Ok(())
}
fn push_http_listener(&mut self, mut listener: ListenerBuilder) -> Result<(), ConfigError> {
let listener = listener.to_http(Some(&self.built))?;
self.built.http_listeners.push(listener);
Ok(())
}
fn push_tcp_listener(&mut self, mut listener: ListenerBuilder) -> Result<(), ConfigError> {
let listener = listener.to_tcp(Some(&self.built))?;
self.built.tcp_listeners.push(listener);
Ok(())
}
fn push_udp_listener(&mut self, mut listener: ListenerBuilder) -> Result<(), ConfigError> {
let listener = listener.to_udp(Some(&self.built))?;
self.built.udp_listeners.push(listener);
Ok(())
}
fn populate_listeners(&mut self, listeners: Vec<ListenerBuilder>) -> Result<(), ConfigError> {
let mut declared: HashSet<ListenerKey> = HashSet::new();
for listener in listeners.iter() {
let key = ListenerKey::new(listener.address, listener.interface.as_deref());
if !declared.insert(key.clone()) {
if key.interface.is_none() {
return Err(ConfigError::ListenerAddressAlreadyInUse(listener.address));
}
return Err(ConfigError::ListenerAlreadyDeclared(key));
}
let protocol = listener
.protocol
.ok_or(ConfigError::Missing(MissingKind::Protocol))?;
if let Some(known) = self.known_addresses.get(&listener.address)
&& *known != protocol
{
return Err(ConfigError::ListenerProtocolMismatchOnAddress(
listener.address,
));
}
self.known_addresses.insert(listener.address, protocol);
if listener.expect_proxy == Some(true) {
self.expect_proxy_addresses.insert(listener.address);
}
if listener.public_address.is_some() && listener.expect_proxy == Some(true) {
return Err(ConfigError::Incompatible {
object: ObjectKind::Listener,
id: listener.address.to_string(),
kind: IncompatibilityKind::PublicAddress,
});
}
match protocol {
ListenerProtocol::Https => self.push_tls_listener(listener.clone())?,
ListenerProtocol::Http => self.push_http_listener(listener.clone())?,
ListenerProtocol::Tcp => self.push_tcp_listener(listener.clone())?,
ListenerProtocol::Udp => self.push_udp_listener(listener.clone())?,
}
}
Ok(())
}
fn populate_clusters(
&mut self,
mut file_cluster_configs: HashMap<String, FileClusterConfig>,
) -> Result<(), ConfigError> {
for (id, file_cluster_config) in file_cluster_configs.drain() {
let resolver: Option<&BackendResolver<'_>> =
self.resolve_backends.then_some(&system_resolver);
let mut cluster_config = file_cluster_config.to_cluster_config_with(
id.as_str(),
&self.expect_proxy_addresses,
resolver,
)?;
match cluster_config {
ClusterConfig::Http(ref mut http) => {
for frontend in http.frontends.iter_mut() {
match self.known_addresses.get(&frontend.address) {
Some(ListenerProtocol::Tcp) => {
return Err(ConfigError::WrongFrontendProtocol(
ListenerProtocol::Tcp,
));
}
Some(ListenerProtocol::Udp) => {
return Err(ConfigError::WrongFrontendProtocol(
ListenerProtocol::Udp,
));
}
Some(ListenerProtocol::Http) => {
if frontend.certificate.is_some() {
return Err(ConfigError::WrongFrontendProtocol(
ListenerProtocol::Http,
));
}
}
Some(ListenerProtocol::Https) => {
if frontend.certificate.is_none() {
if let Some(https_listener) =
self.built.https_listeners.iter().find(|listener| {
listener.address == frontend.address.into()
&& listener.certificate.is_some()
})
{
frontend
.certificate
.clone_from(&https_listener.certificate);
frontend.certificate_chain =
Some(https_listener.certificate_chain.clone());
frontend.key.clone_from(&https_listener.key);
}
if frontend.certificate.is_none() {
debug!("known addresses: {:?}", self.known_addresses);
debug!("frontend: {:?}", frontend);
return Err(ConfigError::WrongFrontendProtocol(
ListenerProtocol::Https,
));
}
}
}
None => {
let file_listener_protocol = if frontend.certificate.is_some() {
self.push_tls_listener(ListenerBuilder::new(
frontend.address.into(),
ListenerProtocol::Https,
))?;
ListenerProtocol::Https
} else {
self.push_http_listener(ListenerBuilder::new(
frontend.address.into(),
ListenerProtocol::Http,
))?;
ListenerProtocol::Http
};
self.known_addresses
.insert(frontend.address, file_listener_protocol);
}
}
}
}
ClusterConfig::Tcp(ref mut tcp) => {
for frontend in tcp.frontends.iter_mut() {
match self.known_addresses.get(&frontend.address) {
Some(ListenerProtocol::Http) | Some(ListenerProtocol::Https) => {
return Err(ConfigError::WrongFrontendProtocol(
ListenerProtocol::Http,
));
}
Some(ListenerProtocol::Udp) => {
frontend.udp = true;
if let Some(backend) =
tcp.backends.iter().find(|b| b.unix_path.is_some())
{
return Err(ConfigError::InvalidBackendUnixPath {
cluster_id: tcp.cluster_id.to_owned(),
address: backend.address,
reason: "a UDP cluster cannot use unix_path backends",
});
}
}
Some(ListenerProtocol::Tcp) => {}
None => {
self.push_tcp_listener(ListenerBuilder::new(
frontend.address.into(),
ListenerProtocol::Tcp,
))?;
self.known_addresses
.insert(frontend.address, ListenerProtocol::Tcp);
}
}
}
}
}
self.built.clusters.insert(id, cluster_config);
}
Ok(())
}
pub fn without_backend_resolution(mut self) -> Self {
self.resolve_backends = false;
self
}
pub fn into_config(&mut self) -> Result<Config, ConfigError> {
if self.built.subnet_ipv4_prefix > MAX_SUBNET_IPV4_PREFIX {
return Err(ConfigError::InvalidSubnetPrefix {
key: "subnet_ipv4_prefix",
family: "IPv4",
value: self.built.subnet_ipv4_prefix,
maximum: MAX_SUBNET_IPV4_PREFIX,
});
}
if self.built.subnet_ipv6_prefix > MAX_SUBNET_IPV6_PREFIX {
return Err(ConfigError::InvalidSubnetPrefix {
key: "subnet_ipv6_prefix",
family: "IPv6",
value: self.built.subnet_ipv6_prefix,
maximum: MAX_SUBNET_IPV6_PREFIX,
});
}
validate_max_connection_attempts(self.built.max_connection_attempts).map_err(|reason| {
ConfigError::InvalidMaxConnectionAttempts {
cluster_id: None,
value: self.built.max_connection_attempts,
reason,
}
})?;
if let Some(listeners) = &self.file.listeners {
self.populate_listeners(listeners.clone())?;
}
if let Some(file_cluster_configs) = &self.file.clusters {
self.populate_clusters(file_cluster_configs.clone())?;
}
type SniAlpnByAddress = HashMap<SocketAddr, Vec<(Option<String>, Vec<String>)>>;
let mut frontends_by_address: SniAlpnByAddress = HashMap::new();
let mut addresses_with_no_sni_frontend: HashSet<SocketAddr> = HashSet::new();
for cluster in self.built.clusters.values() {
if let ClusterConfig::Tcp(tcp) = cluster {
for frontend in &tcp.frontends {
if frontend.udp {
continue;
}
if frontend.sni.is_none() {
addresses_with_no_sni_frontend.insert(frontend.address);
}
frontends_by_address
.entry(frontend.address)
.or_default()
.push((frontend.sni.clone(), frontend.alpn.clone()));
}
}
}
let mut addresses_with_sni_frontend: HashSet<SocketAddr> = HashSet::new();
for (address, frontends) in &frontends_by_address {
if !frontends.iter().any(|(sni, _)| sni.is_some()) {
continue;
}
addresses_with_sni_frontend.insert(*address);
if addresses_with_no_sni_frontend.contains(address) {
return Err(ConfigError::TcpListenerMixesSniAndNoSni { address: *address });
}
let mut alpn_lists_by_sni: HashMap<Option<String>, Vec<&Vec<String>>> = HashMap::new();
for (sni, alpn) in frontends {
alpn_lists_by_sni.entry(sni.clone()).or_default().push(alpn);
}
for (sni, alpn_lists) in alpn_lists_by_sni {
let mut seen_protocols: HashSet<&str> = HashSet::new();
let mut catch_all_count = 0usize;
for alpn in alpn_lists {
if alpn.is_empty() {
catch_all_count += 1;
if catch_all_count > 1 {
return Err(ConfigError::TcpFrontendMultipleAlpnCatchAll {
address: *address,
sni: sni.clone(),
});
}
continue;
}
for protocol in alpn {
if !seen_protocols.insert(protocol.as_str()) {
return Err(ConfigError::TcpFrontendAlpnOverlap {
address: *address,
sni: sni.clone(),
protocol: protocol.clone(),
});
}
}
}
}
}
for listener in &self.built.tcp_listeners {
let address: SocketAddr = listener.address.into();
if !addresses_with_sni_frontend.contains(&address) {
continue;
}
let sni_preread_timeout = listener
.sni_preread_timeout
.unwrap_or(DEFAULT_SNI_PREREAD_TIMEOUT);
if sni_preread_timeout > listener.front_timeout {
return Err(ConfigError::SniPrereadTimeoutExceedsFrontTimeout {
address,
sni_preread_timeout,
front_timeout: listener.front_timeout,
});
}
let sni_preread_max_bytes = listener
.sni_preread_max_bytes
.unwrap_or(DEFAULT_SNI_PREREAD_MAX_BYTES);
if sni_preread_max_bytes < MIN_SNI_PREREAD_MAX_BYTES {
return Err(ConfigError::SniPrereadMaxBytesTooSmall {
address,
sni_preread_max_bytes,
minimum: MIN_SNI_PREREAD_MAX_BYTES,
});
}
if u64::from(sni_preread_max_bytes) > self.built.buffer_size {
return Err(ConfigError::SniPrereadMaxBytesExceedsBufferSize {
address,
sni_preread_max_bytes,
buffer_size: self.built.buffer_size,
});
}
}
let h2_listeners = self
.built
.https_listeners
.iter()
.filter(|l| l.alpn_protocols.iter().any(|p| p == "h2"))
.count()
+ self
.built
.http_listeners
.iter()
.filter(|l| l.h2c_prior_knowledge == Some(true))
.count();
if h2_listeners > 0 && self.built.buffer_size < H2_MIN_BUFFER_SIZE {
return Err(ConfigError::BufferSizeTooSmallForH2 {
buffer_size: self.built.buffer_size,
minimum: H2_MIN_BUFFER_SIZE,
listeners: h2_listeners,
});
}
if self.built.command_buffer_size > self.built.max_command_buffer_size {
return Err(ConfigError::CommandBufferSizeExceedsMax {
command_buffer_size: self.built.command_buffer_size,
max_command_buffer_size: self.built.max_command_buffer_size,
});
}
if let Some(cap) = self.built.basic_auth_max_credential_bytes {
let third = self.built.buffer_size / 3;
if cap >= third {
warn!(
"basic_auth_max_credential_bytes = {} is >= buffer_size / 3 ({}); \
a hostile peer can pin ~33% of the per-frontend buffer per failed auth \
attempt. Consider lowering basic_auth_max_credential_bytes (typical \
credentials are <100 bytes) or raising buffer_size.",
cap, third
);
}
}
if self.built.evict_on_queue_full && self.built.max_connections < 100 {
let pct = 100usize.div_ceil(self.built.max_connections);
warn!(
"evict_on_queue_full enabled with max_connections = {}; the eviction batch \
clamps to 1, equivalent to ~{}% of capacity per cap event (the knob is \
documented as 1%). Confirm this is intended.",
self.built.max_connections, pct
);
}
let command_socket_path = self.file.command_socket.clone().unwrap_or({
let mut path = env::current_dir().map_err(|e| ConfigError::Env(e.to_string()))?;
path.push("sozu.sock");
let verified_path = path
.to_str()
.ok_or(ConfigError::InvalidPath(path.clone()))?;
verified_path.to_owned()
});
if let (None, Some(true)) = (&self.file.saved_state, &self.file.automatic_state_save) {
return Err(ConfigError::Missing(MissingKind::SavedState));
}
let command_socket_mode = match self.file.command_socket_mode.as_deref() {
Some(mode) => parse_command_socket_mode(mode)?,
None => DEFAULT_COMMAND_SOCKET_MODE,
};
let config = Config {
command_socket: command_socket_path,
command_socket_mode,
..self.built.clone()
};
debug_assert!(
config.min_buffers <= config.max_buffers,
"min_buffers must not exceed max_buffers"
);
debug_assert!(
!config
.https_listeners
.iter()
.any(|l| l.alpn_protocols.iter().any(|p| p == "h2"))
&& !config
.http_listeners
.iter()
.any(|l| l.h2c_prior_knowledge == Some(true))
|| config.buffer_size >= H2_MIN_BUFFER_SIZE,
"an h2-advertising config must satisfy the H2 minimum buffer size"
);
Ok(config)
}
}
#[derive(Clone, PartialEq, Eq, Serialize, Default, Deserialize)]
pub struct Config {
pub config_path: String,
pub command_socket: String,
#[serde(default = "default_command_socket_mode")]
pub command_socket_mode: u32,
pub command_buffer_size: u64,
pub max_command_buffer_size: u64,
pub max_connections: usize,
pub min_buffers: u64,
pub max_buffers: u64,
pub buffer_size: u64,
pub saved_state: Option<String>,
#[serde(default)]
pub automatic_state_save: bool,
pub log_level: String,
pub log_target: String,
pub log_colored: bool,
#[serde(default)]
pub audit_logs_target: Option<String>,
#[serde(default)]
pub audit_logs_json_target: Option<String>,
#[serde(default)]
pub access_logs_target: Option<String>,
pub access_logs_format: Option<AccessLogFormat>,
pub access_logs_colored: Option<bool>,
pub worker_count: u16,
pub worker_automatic_restart: bool,
pub metrics: Option<MetricsConfig>,
#[serde(default = "default_disable_cluster_metrics")]
pub disable_cluster_metrics: bool,
pub http_listeners: Vec<HttpListenerConfig>,
pub https_listeners: Vec<HttpsListenerConfig>,
pub tcp_listeners: Vec<TcpListenerConfig>,
#[serde(default)]
pub udp_listeners: Vec<UdpListenerConfig>,
pub clusters: HashMap<String, ClusterConfig>,
pub handle_process_affinity: bool,
pub ctl_command_timeout: u64,
pub pid_file_path: Option<String>,
pub activate_listeners: bool,
#[serde(default = "default_front_timeout")]
pub front_timeout: u32,
#[serde(default = "default_back_timeout")]
pub back_timeout: u32,
#[serde(default = "default_connect_timeout")]
pub connect_timeout: u32,
#[serde(default = "default_zombie_check_interval")]
pub zombie_check_interval: u32,
#[serde(default = "default_accept_queue_timeout")]
pub accept_queue_timeout: u32,
#[serde(default = "default_evict_on_queue_full")]
pub evict_on_queue_full: bool,
#[serde(default = "default_request_timeout")]
pub request_timeout: u32,
#[serde(default = "default_worker_timeout")]
pub worker_timeout: u32,
#[serde(default)]
pub slab_entries_per_connection: Option<u64>,
#[serde(default)]
pub command_allowed_uids: Option<Vec<u32>>,
#[serde(default)]
pub basic_auth_max_credential_bytes: Option<u64>,
#[serde(default = "default_max_connections_per_ip")]
pub max_connections_per_ip: u64,
#[serde(default = "default_max_connections_per_subnet")]
pub max_connections_per_subnet: u64,
#[serde(default = "default_subnet_ipv4_prefix")]
pub subnet_ipv4_prefix: u32,
#[serde(default = "default_subnet_ipv6_prefix")]
pub subnet_ipv6_prefix: u32,
#[serde(default = "default_retry_after")]
pub retry_after: u32,
#[serde(default = "default_max_connection_attempts")]
pub max_connection_attempts: u32,
#[serde(default)]
pub splice_pipe_capacity_bytes: Option<u64>,
}
fn default_command_socket_mode() -> u32 {
DEFAULT_COMMAND_SOCKET_MODE
}
pub fn parse_command_socket_mode(value: &str) -> Result<u32, ConfigError> {
let invalid = |reason| ConfigError::InvalidCommandSocketMode {
value: value.to_owned(),
reason,
};
if value.is_empty() || value.len() > 4 || !value.bytes().all(|b| (b'0'..=b'7').contains(&b)) {
return Err(invalid(
"expected one to four octal digits, such as \"0660\" or \"660\"",
));
}
let mode =
u32::from_str_radix(value, 8).map_err(|_| invalid("expected one to four octal digits"))?;
if mode > MAX_COMMAND_SOCKET_MODE {
return Err(invalid(
"only permission bits (at most 0777) are allowed; setuid, setgid and sticky are refused",
));
}
if mode & 0o600 != 0o600 {
return Err(invalid("the owner must keep read and write (0600)"));
}
Ok(mode)
}
fn default_front_timeout() -> u32 {
DEFAULT_FRONT_TIMEOUT
}
fn default_back_timeout() -> u32 {
DEFAULT_BACK_TIMEOUT
}
fn default_connect_timeout() -> u32 {
DEFAULT_CONNECT_TIMEOUT
}
fn default_request_timeout() -> u32 {
DEFAULT_REQUEST_TIMEOUT
}
fn default_zombie_check_interval() -> u32 {
DEFAULT_ZOMBIE_CHECK_INTERVAL
}
fn default_accept_queue_timeout() -> u32 {
DEFAULT_ACCEPT_QUEUE_TIMEOUT
}
fn default_evict_on_queue_full() -> bool {
DEFAULT_EVICT_ON_QUEUE_FULL
}
fn default_disable_cluster_metrics() -> bool {
DEFAULT_DISABLE_CLUSTER_METRICS
}
fn default_worker_timeout() -> u32 {
DEFAULT_WORKER_TIMEOUT
}
fn default_max_connections_per_ip() -> u64 {
DEFAULT_MAX_CONNECTIONS_PER_IP
}
fn default_max_connections_per_subnet() -> u64 {
DEFAULT_MAX_CONNECTIONS_PER_SUBNET
}
fn default_subnet_ipv4_prefix() -> u32 {
DEFAULT_SUBNET_IPV4_PREFIX
}
fn default_subnet_ipv6_prefix() -> u32 {
DEFAULT_SUBNET_IPV6_PREFIX
}
fn default_retry_after() -> u32 {
DEFAULT_RETRY_AFTER
}
fn default_max_connection_attempts() -> u32 {
DEFAULT_MAX_CONNECTION_ATTEMPTS
}
impl Config {
pub fn load_from_path(path: &str) -> Result<Config, ConfigError> {
let file_config = FileConfig::load_from_path(path)?;
let mut config = ConfigBuilder::new(file_config, path).into_config()?;
config.saved_state = config.saved_state_path()?;
Ok(config)
}
pub fn load_from_path_for_control(path: &str) -> Result<Config, ConfigError> {
let file_config = FileConfig::load_from_path(path)?;
let mut config = ConfigBuilder::new(file_config, path)
.without_backend_resolution()
.into_config()?;
config.saved_state = config.saved_state_path()?;
Ok(config)
}
pub fn generate_config_messages(&self) -> Result<Vec<WorkerRequest>, ConfigError> {
let mut v = Vec::new();
let mut count = 0u8;
for listener in &self.http_listeners {
v.push(WorkerRequest {
id: format!("CONFIG-{count}"),
content: RequestType::AddHttpListener(listener.clone()).into(),
});
count += 1;
}
for listener in &self.https_listeners {
v.push(WorkerRequest {
id: format!("CONFIG-{count}"),
content: RequestType::add_https_listener(listener.clone()).into(),
});
count += 1;
}
for listener in &self.tcp_listeners {
v.push(WorkerRequest {
id: format!("CONFIG-{count}"),
content: RequestType::AddTcpListener(listener.clone()).into(),
});
count += 1;
}
for listener in &self.udp_listeners {
v.push(WorkerRequest {
id: format!("CONFIG-{count}"),
content: RequestType::AddUdpListener(listener.clone()).into(),
});
count += 1;
}
for cluster in self.clusters.values() {
let mut orders = cluster.generate_requests()?;
for content in orders.drain(..) {
v.push(WorkerRequest {
id: format!("CONFIG-{count}"),
content,
});
count += 1;
}
}
if self.activate_listeners {
for listener in &self.http_listeners {
v.push(WorkerRequest {
id: format!("CONFIG-{count}"),
content: RequestType::ActivateListener(ActivateListener {
interface: listener.interface.clone(),
address: listener.address,
proxy: ListenerType::Http.into(),
from_scm: false,
})
.into(),
});
count += 1;
}
for listener in &self.https_listeners {
v.push(WorkerRequest {
id: format!("CONFIG-{count}"),
content: RequestType::ActivateListener(ActivateListener {
interface: listener.interface.clone(),
address: listener.address,
proxy: ListenerType::Https.into(),
from_scm: false,
})
.into(),
});
count += 1;
}
for listener in &self.tcp_listeners {
v.push(WorkerRequest {
id: format!("CONFIG-{count}"),
content: RequestType::ActivateListener(ActivateListener {
interface: listener.interface.clone(),
address: listener.address,
proxy: ListenerType::Tcp.into(),
from_scm: false,
})
.into(),
});
count += 1;
}
for listener in &self.udp_listeners {
v.push(WorkerRequest {
id: format!("CONFIG-{count}"),
content: RequestType::ActivateListener(ActivateListener {
interface: listener.interface.clone(),
address: listener.address,
proxy: ListenerType::Udp.into(),
from_scm: false,
})
.into(),
});
count += 1;
}
}
if self.disable_cluster_metrics {
v.push(WorkerRequest {
id: format!("CONFIG-{count}"),
content: RequestType::ConfigureMetrics(MetricsConfiguration::Disabled.into())
.into(),
});
}
Ok(v)
}
pub fn command_socket_path(&self) -> Result<String, ConfigError> {
let config_path_buf = PathBuf::from(self.config_path.clone());
let mut config_dir = config_path_buf
.parent()
.ok_or(ConfigError::NoFileParent(
config_path_buf.to_string_lossy().to_string(),
))?
.to_path_buf();
let socket_path = PathBuf::from(self.command_socket.clone());
let mut socket_parent_dir = match socket_path.parent() {
None => config_dir,
Some(path) => {
config_dir.push(path);
config_dir.canonicalize().map_err(|io_error| {
ConfigError::SocketPathError(format!(
"Could not canonicalize path {config_dir:?}: {io_error}"
))
})?
}
};
let socket_name = socket_path
.file_name()
.ok_or(ConfigError::SocketPathError(format!(
"could not get command socket file name from {socket_path:?}"
)))?;
socket_parent_dir.push(socket_name);
let command_socket_path = socket_parent_dir
.to_str()
.ok_or(ConfigError::SocketPathError(format!(
"Invalid socket path {socket_parent_dir:?}"
)))?
.to_string();
Ok(command_socket_path)
}
fn saved_state_path(&self) -> Result<Option<String>, ConfigError> {
let path = match self.saved_state.as_ref() {
Some(path) => path,
None => return Ok(None),
};
debug!("saved_stated path in the config: {}", path);
let config_path = PathBuf::from(self.config_path.clone());
debug!("Config path buffer: {:?}", config_path);
let config_dir = config_path
.parent()
.ok_or(ConfigError::SaveStatePath(format!(
"Could get parent directory of config file {config_path:?}"
)))?;
debug!("Config folder: {:?}", config_dir);
if !config_dir.exists() {
create_dir_all(config_dir).map_err(|io_error| {
ConfigError::SaveStatePath(format!(
"failed to create state parent directory '{config_dir:?}': {io_error}"
))
})?;
}
let mut saved_state_path_raw = config_dir.to_path_buf();
saved_state_path_raw.push(path);
debug!(
"Looking for saved state on the path {:?}",
saved_state_path_raw
);
match metadata(path) {
Err(err) if matches!(err.kind(), ErrorKind::NotFound) => {
info!("Create an empty state file at '{}'", path);
File::create(path).map_err(|io_error| {
ConfigError::SaveStatePath(format!(
"failed to create state file '{path:?}': {io_error}"
))
})?;
}
_ => {}
}
saved_state_path_raw.canonicalize().map_err(|io_error| {
ConfigError::SaveStatePath(format!(
"could not get saved state path from config file input {path:?}: {io_error}"
))
})?;
let stringified_path = saved_state_path_raw
.to_str()
.ok_or(ConfigError::SaveStatePath(format!(
"Invalid path {saved_state_path_raw:?}"
)))?
.to_string();
Ok(Some(stringified_path))
}
pub fn load_file(path: &str) -> Result<String, ConfigError> {
std::fs::read_to_string(path).map_err(|io_error| ConfigError::FileRead {
path_to_read: path.to_owned(),
io_error,
})
}
pub fn load_file_bytes(path: &str) -> Result<Vec<u8>, ConfigError> {
std::fs::read(path).map_err(|io_error| ConfigError::FileRead {
path_to_read: path.to_owned(),
io_error,
})
}
}
impl fmt::Debug for Config {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Config")
.field("config_path", &self.config_path)
.field("command_socket", &self.command_socket)
.field(
"command_socket_mode",
&format_args!("{:o}", self.command_socket_mode),
)
.field("command_buffer_size", &self.command_buffer_size)
.field("max_command_buffer_size", &self.max_command_buffer_size)
.field("max_connections", &self.max_connections)
.field("min_buffers", &self.min_buffers)
.field("max_buffers", &self.max_buffers)
.field("buffer_size", &self.buffer_size)
.field("saved_state", &self.saved_state)
.field("automatic_state_save", &self.automatic_state_save)
.field("log_level", &self.log_level)
.field("log_target", &self.log_target)
.field("access_logs_target", &self.access_logs_target)
.field("audit_logs_target", &self.audit_logs_target)
.field("audit_logs_json_target", &self.audit_logs_json_target)
.field("access_logs_format", &self.access_logs_format)
.field("worker_count", &self.worker_count)
.field("worker_automatic_restart", &self.worker_automatic_restart)
.field("metrics", &self.metrics)
.field("disable_cluster_metrics", &self.disable_cluster_metrics)
.field("handle_process_affinity", &self.handle_process_affinity)
.field("ctl_command_timeout", &self.ctl_command_timeout)
.field("pid_file_path", &self.pid_file_path)
.field("activate_listeners", &self.activate_listeners)
.field("front_timeout", &self.front_timeout)
.field("back_timeout", &self.back_timeout)
.field("connect_timeout", &self.connect_timeout)
.field("zombie_check_interval", &self.zombie_check_interval)
.field("accept_queue_timeout", &self.accept_queue_timeout)
.field("evict_on_queue_full", &self.evict_on_queue_full)
.field("request_timeout", &self.request_timeout)
.field("worker_timeout", &self.worker_timeout)
.finish()
}
}
fn toml_error_diagnostic(path: &str, error: &toml::de::Error) -> String {
match error.span() {
Some(Range { start, end }) => format!(
"error parsing the configuration file '{path}' at bytes {start}..{end}: invalid TOML syntax or unexpected value (the offending text is not shown)"
),
None => format!(
"error parsing the configuration file '{path}': invalid TOML syntax or unexpected value (the offending text is not shown)"
),
}
}
impl ServerConfig {
pub const DEFAULT_SLAB_ENTRIES_PER_CONNECTION: u64 = 4;
pub const MIN_SLAB_ENTRIES_PER_CONNECTION: u64 = 2;
pub const MAX_SLAB_ENTRIES_PER_CONNECTION: u64 = 32;
pub fn effective_slab_entries_per_connection(&self) -> u64 {
let effective = match self.slab_entries_per_connection {
Some(0) | None => Self::DEFAULT_SLAB_ENTRIES_PER_CONNECTION,
Some(n) => n.clamp(
Self::MIN_SLAB_ENTRIES_PER_CONNECTION,
Self::MAX_SLAB_ENTRIES_PER_CONNECTION,
),
};
debug_assert!(
(Self::MIN_SLAB_ENTRIES_PER_CONNECTION..=Self::MAX_SLAB_ENTRIES_PER_CONNECTION)
.contains(&effective),
"effective slab entries per connection must stay within [MIN, MAX]"
);
effective
}
pub fn slab_capacity(&self) -> u64 {
let per_conn = self.effective_slab_entries_per_connection();
let capacity = 10 + per_conn * self.max_connections;
debug_assert!(
capacity >= 10,
"slab capacity must reserve the base entries"
);
debug_assert!(
self.max_connections == 0 || capacity > 10,
"a non-zero connection cap must reserve per-connection slab entries"
);
capacity
}
}
impl From<&Config> for ServerConfig {
fn from(config: &Config) -> Self {
let metrics = config.metrics.clone().map(|m| ServerMetricsConfig {
address: m.address.to_string(),
tagged_metrics: m.tagged_metrics,
prefix: m.prefix,
detail: Some(MetricDetail::from(m.detail) as i32),
});
let server_config = Self {
max_connections: config.max_connections as u64,
front_timeout: config.front_timeout,
back_timeout: config.back_timeout,
connect_timeout: config.connect_timeout,
zombie_check_interval: config.zombie_check_interval,
accept_queue_timeout: config.accept_queue_timeout,
min_buffers: config.min_buffers,
max_buffers: config.max_buffers,
buffer_size: config.buffer_size,
log_level: config.log_level.clone(),
log_target: config.log_target.clone(),
access_logs_target: config.access_logs_target.clone(),
audit_logs_target: config.audit_logs_target.clone(),
audit_logs_json_target: config.audit_logs_json_target.clone(),
command_buffer_size: config.command_buffer_size,
max_command_buffer_size: config.max_command_buffer_size,
metrics,
access_log_format: ProtobufAccessLogFormat::from(&config.access_logs_format) as i32,
log_colored: config.log_colored,
slab_entries_per_connection: config.slab_entries_per_connection,
basic_auth_max_credential_bytes: config.basic_auth_max_credential_bytes,
evict_on_queue_full: Some(config.evict_on_queue_full),
max_connections_per_ip: Some(config.max_connections_per_ip),
max_connections_per_subnet: Some(config.max_connections_per_subnet),
subnet_ipv4_prefix: Some(config.subnet_ipv4_prefix),
subnet_ipv6_prefix: Some(config.subnet_ipv6_prefix),
retry_after: Some(config.retry_after),
splice_pipe_capacity_bytes: config.splice_pipe_capacity_bytes,
max_connection_attempts: Some(config.max_connection_attempts),
};
debug_assert!(
server_config.min_buffers <= server_config.max_buffers,
"ServerConfig must preserve min_buffers <= max_buffers"
);
debug_assert_eq!(
server_config.buffer_size, config.buffer_size,
"ServerConfig buffer_size must mirror the source config"
);
debug_assert_eq!(
server_config.max_connections, config.max_connections as u64,
"ServerConfig max_connections must mirror the source config"
);
server_config
}
}
#[cfg(test)]
mod tests {
use toml::to_string;
use super::*;
#[test]
fn silent_close_redirect_policy_parses_case_insensitively() {
for spelling in ["silent_close", "SILENT_CLOSE", "Silent_Close"] {
assert_eq!(
parse_redirect_policy(spelling).ok(),
Some(RedirectPolicy::SilentClose),
"{spelling} must parse to SilentClose"
);
}
assert!(matches!(
parse_redirect_policy("silent-close"),
Err(ConfigError::InvalidRedirectPolicy(value)) if value == "silent-close"
));
}
#[test]
fn http_frontend_debug_redacts_pem_material() {
const CERTIFICATE_SECRET: &str = "HTTP_FRONTEND_CERTIFICATE_PEM_SECRET_SENTINEL";
const CHAIN_SECRET: &str = "HTTP_FRONTEND_CHAIN_PEM_SECRET_SENTINEL";
const KEY_SECRET: &str = "HTTP_FRONTEND_KEY_PEM_SECRET_SENTINEL";
const HOSTNAME_SECRET: &str = "HTTP_FRONTEND_HOSTNAME_SECRET_SENTINEL";
const PATH_SECRET: &str = "HTTP_FRONTEND_PATH_SECRET_SENTINEL";
const METHOD_SECRET: &str = "HTTP_FRONTEND_METHOD_SECRET_SENTINEL";
const TAG_KEY_SECRET: &str = "HTTP_FRONTEND_TAG_KEY_SECRET_SENTINEL";
const TAG_SECRET: &str = "HTTP_FRONTEND_TAG_VALUE_SECRET_SENTINEL";
const HEADER_KEY_SECRET: &str = "HTTP_FRONTEND_HEADER_KEY_SECRET_SENTINEL";
const HEADER_SECRET: &str = "HTTP_FRONTEND_HEADER_VALUE_SECRET_SENTINEL";
const REDIRECT_TEMPLATE_SECRET: &str = "HTTP_FRONTEND_REDIRECT_TEMPLATE_SECRET_SENTINEL";
const REWRITE_HOST_SECRET: &str = "HTTP_FRONTEND_REWRITE_HOST_SECRET_SENTINEL";
const REWRITE_PATH_SECRET: &str = "HTTP_FRONTEND_REWRITE_PATH_SECRET_SENTINEL";
let long_value = |marker: &str| format!("{marker}{}", "x".repeat(4096));
let certificate = long_value(CERTIFICATE_SECRET);
let certificate_chain = long_value(CHAIN_SECRET);
let key = long_value(KEY_SECRET);
let hostname = long_value(HOSTNAME_SECRET);
let path = long_value(PATH_SECRET);
let method = long_value(METHOD_SECRET);
let redirect_template = long_value(REDIRECT_TEMPLATE_SECRET);
let rewrite_host = long_value(REWRITE_HOST_SECRET);
let rewrite_path = long_value(REWRITE_PATH_SECRET);
let frontend = HttpFrontendConfig {
address: "127.0.0.1:8443".parse().unwrap(),
hostname,
path: PathRule::prefix(path),
method: Some(method),
certificate: Some(certificate),
key: Some(key),
certificate_chain: Some(vec![certificate_chain]),
tls_versions: vec![TlsVersion::TlsV13],
position: RulePosition::Tree,
tags: Some(BTreeMap::from([(
long_value(TAG_KEY_SECRET),
long_value(TAG_SECRET),
)])),
redirect: None,
redirect_scheme: None,
redirect_template: Some(redirect_template),
rewrite_host: Some(rewrite_host),
rewrite_path: Some(rewrite_path),
rewrite_port: None,
required_auth: None,
headers: vec![Header {
position: HeaderPosition::Request as i32,
key: long_value(HEADER_KEY_SECRET),
val: long_value(HEADER_SECRET),
}],
hsts: None,
header_predicates: Vec::new(),
};
let output = format!("{frontend:?}");
let secrets = [
CERTIFICATE_SECRET,
CHAIN_SECRET,
KEY_SECRET,
HOSTNAME_SECRET,
PATH_SECRET,
METHOD_SECRET,
TAG_KEY_SECRET,
TAG_SECRET,
HEADER_KEY_SECRET,
HEADER_SECRET,
REDIRECT_TEMPLATE_SECRET,
REWRITE_HOST_SECRET,
REWRITE_PATH_SECRET,
];
for secret in secrets {
assert!(
!output.contains(secret),
"HttpFrontendConfig Debug leaked secret marker {secret}: {output}"
);
}
let expected_metadata = [
"address: 127.0.0.1:8443".to_owned(),
format!("hostname_len: {}", long_value(HOSTNAME_SECRET).len()),
format!("path_kind: {}", PathRule::prefix(String::new()).kind),
format!("path_len: {}", long_value(PATH_SECRET).len()),
format!("method_len: Some({})", long_value(METHOD_SECRET).len()),
"certificate: Some(\"[redacted]\")".to_owned(),
format!(
"certificate_len: Some({})",
long_value(CERTIFICATE_SECRET).len()
),
"key: Some(\"[redacted]\")".to_owned(),
format!("key_len: Some({})", long_value(KEY_SECRET).len()),
"certificate_chain: Some(\"[redacted]\")".to_owned(),
"certificate_chain_count: Some(1)".to_owned(),
format!(
"certificate_chain_len: Some({})",
long_value(CHAIN_SECRET).len()
),
"tls_versions_count: 1".to_owned(),
"tags_count: 1".to_owned(),
format!(
"redirect_template_len: Some({})",
long_value(REDIRECT_TEMPLATE_SECRET).len()
),
format!(
"rewrite_host_len: Some({})",
long_value(REWRITE_HOST_SECRET).len()
),
format!(
"rewrite_path_len: Some({})",
long_value(REWRITE_PATH_SECRET).len()
),
"headers_count: 1".to_owned(),
];
for safe_metadata in expected_metadata {
assert!(
output.contains(&safe_metadata),
"HttpFrontendConfig Debug omitted safe metadata {safe_metadata}: {output}"
);
}
assert!(
output.len() <= 1024,
"HttpFrontendConfig Debug output is not bounded: {} bytes",
output.len()
);
for (index, request) in frontend
.generate_requests("safe-cluster-id")
.into_iter()
.enumerate()
{
let generated_output = format!("{request:?}");
for secret in secrets {
assert!(
!generated_output.contains(secret),
"generated frontend request {index} Debug leaked secret marker {secret}: {generated_output}"
);
}
assert!(
generated_output.len() <= 2048,
"generated frontend request {index} Debug output is not bounded: {} bytes",
generated_output.len()
);
}
}
#[test]
fn hsts_to_proto_enabled_substitutes_default_max_age() {
let cfg = FileHstsConfig {
enabled: Some(true),
max_age: None,
include_subdomains: None,
preload: None,
force_replace_backend: None,
};
let proto = cfg.to_proto("test").expect("should validate");
assert_eq!(proto.enabled, Some(true));
assert_eq!(proto.max_age, Some(DEFAULT_HSTS_MAX_AGE));
}
#[test]
fn hsts_to_proto_explicit_max_age_kept() {
let cfg = FileHstsConfig {
enabled: Some(true),
max_age: Some(63_072_000),
include_subdomains: Some(true),
preload: Some(true),
force_replace_backend: None,
};
let proto = cfg.to_proto("test").expect("should validate");
assert_eq!(proto.max_age, Some(63_072_000));
assert_eq!(proto.include_subdomains, Some(true));
assert_eq!(proto.preload, Some(true));
}
#[test]
fn hsts_to_proto_disabled_keeps_zero_intent() {
let cfg = FileHstsConfig {
enabled: Some(false),
max_age: None,
include_subdomains: None,
preload: None,
force_replace_backend: None,
};
let proto = cfg.to_proto("test").expect("should validate");
assert_eq!(proto.enabled, Some(false));
}
#[test]
fn hsts_to_proto_kill_switch_max_age_zero_allowed() {
let cfg = FileHstsConfig {
enabled: Some(true),
max_age: Some(0),
include_subdomains: None,
preload: None,
force_replace_backend: None,
};
let proto = cfg.to_proto("test").expect("kill-switch must validate");
assert_eq!(proto.max_age, Some(0));
}
#[test]
fn hsts_to_proto_missing_enabled_errors() {
let cfg = FileHstsConfig {
enabled: None,
max_age: Some(31_536_000),
include_subdomains: None,
preload: None,
force_replace_backend: None,
};
match cfg.to_proto("test").unwrap_err() {
ConfigError::HstsEnabledRequired(scope) => assert_eq!(scope, "test"),
other => panic!("expected HstsEnabledRequired, got {other:?}"),
}
}
#[test]
fn hsts_rejected_on_http_listener() {
let mut listener = ListenerBuilder::new(
SocketAddress::new_v4(127, 0, 0, 1, 8080),
ListenerProtocol::Http,
);
listener.hsts = Some(FileHstsConfig {
enabled: Some(true),
max_age: Some(31_536_000),
include_subdomains: None,
preload: None,
force_replace_backend: None,
});
match listener.to_http(None).unwrap_err() {
ConfigError::HstsOnPlainHttp(scope) => assert!(
scope.contains("HTTP listener"),
"expected scope to mention 'HTTP listener', got {scope:?}"
),
other => panic!("expected HstsOnPlainHttp, got {other:?}"),
}
}
#[test]
fn h2_flood_threshold_zero_rejected_on_http_listener() {
let mut listener = ListenerBuilder::new(
SocketAddress::new_v4(127, 0, 0, 1, 8080),
ListenerProtocol::Http,
);
listener.h2_max_rst_stream_per_window = Some(0);
match listener.to_http(None).unwrap_err() {
ConfigError::H2ThresholdBelowMinimum {
key,
value,
minimum,
..
} => {
assert_eq!(key, "h2_max_rst_stream_per_window");
assert_eq!(value, 0);
assert_eq!(minimum, 1);
}
other => panic!("expected H2ThresholdBelowMinimum, got {other:?}"),
}
}
#[test]
fn h2_flood_threshold_zero_rejected_on_https_listener() {
let mut listener = ListenerBuilder::new_https(SocketAddress::new_v4(127, 0, 0, 1, 8443));
listener.h2_max_rst_stream_abusive_lifetime = Some(0);
match listener.to_tls(None).unwrap_err() {
ConfigError::H2ThresholdBelowMinimum {
key,
value,
minimum,
..
} => {
assert_eq!(key, "h2_max_rst_stream_abusive_lifetime");
assert_eq!(value, 0);
assert_eq!(minimum, 1);
}
other => panic!("expected H2ThresholdBelowMinimum, got {other:?}"),
}
}
#[test]
fn h2_stream_shrink_ratio_below_two_rejected_on_http_listener() {
let mut listener = ListenerBuilder::new(
SocketAddress::new_v4(127, 0, 0, 1, 8080),
ListenerProtocol::Http,
);
listener.h2_stream_shrink_ratio = Some(1);
match listener.to_http(None).unwrap_err() {
ConfigError::H2ThresholdBelowMinimum {
key,
value,
minimum,
..
} => {
assert_eq!(key, "h2_stream_shrink_ratio");
assert_eq!(value, 1);
assert_eq!(minimum, 2);
}
other => panic!("expected H2ThresholdBelowMinimum, got {other:?}"),
}
}
#[test]
fn h2_thresholds_at_their_minimum_accepted() {
let mut listener = ListenerBuilder::new(
SocketAddress::new_v4(127, 0, 0, 1, 8080),
ListenerProtocol::Http,
);
listener.h2_max_rst_stream_per_window = Some(1);
listener.h2_max_rst_stream_abusive_lifetime = Some(1);
listener.h2_stream_shrink_ratio = Some(2);
let config = listener
.to_http(None)
.expect("thresholds at their minimum must be accepted");
assert_eq!(config.h2_max_rst_stream_per_window, Some(1));
assert_eq!(config.h2_max_rst_stream_abusive_lifetime, Some(1));
assert_eq!(config.h2_stream_shrink_ratio, Some(2));
}
#[test]
fn hsts_rejected_on_http_frontend() {
let frontend = FileClusterFrontendConfig {
address: "127.0.0.1:8080".parse().unwrap(),
hostname: Some("example.com".to_owned()),
alpn: vec![],
path: None,
path_type: None,
method: None,
certificate: None,
key: None,
certificate_chain: None,
tls_versions: vec![],
position: RulePosition::Tree,
tags: None,
redirect: None,
redirect_scheme: None,
redirect_template: None,
rewrite_host: None,
rewrite_path: None,
rewrite_port: None,
required_auth: None,
headers: None,
hsts: Some(FileHstsConfig {
enabled: Some(true),
max_age: Some(31_536_000),
include_subdomains: None,
preload: None,
force_replace_backend: None,
}),
header_predicates: None,
};
match frontend.to_http_front("api").unwrap_err() {
ConfigError::HstsOnPlainHttp(scope) => {
assert!(
scope.contains("api") && scope.contains("example.com"),
"expected scope to mention 'api' and 'example.com', got {scope:?}"
);
}
other => panic!("expected HstsOnPlainHttp, got {other:?}"),
}
}
#[test]
fn path_type_regex_path_parses_to_the_path_only_regex_kind() {
let parse = |path_type: &str| {
toml::from_str::<FileClusterFrontendConfig>(&format!(
"address = \"127.0.0.1:8080\"\n\
hostname = \"example.com\"\n\
path = \"^/api/v2$\"\n\
path_type = \"{path_type}\"\n"
))
};
let frontend = parse("REGEX_PATH").expect("REGEX_PATH must parse");
assert_eq!(frontend.path_type, Some(PathRuleType::RegexPath));
let http = frontend
.to_http_front("api")
.expect("a REGEX_PATH frontend must convert");
assert_eq!(http.path, PathRule::regex_path("^/api/v2$"));
assert_eq!(
crate::proto::command::PathRuleKind::try_from(http.path.kind),
Ok(crate::proto::command::PathRuleKind::RegexPath)
);
assert_ne!(
http.path,
PathRule::regex("^/api/v2$"),
"REGEX_PATH must not collapse into REGEX"
);
let regex = parse("REGEX").expect("REGEX must still parse");
assert_eq!(regex.path_type, Some(PathRuleType::Regex));
assert!(
parse("PATH_REGEX").is_err(),
"an unknown path_type is refused"
);
}
#[test]
fn serialize() {
let http = ListenerBuilder::new(
SocketAddress::new_v4(127, 0, 0, 1, 8080),
ListenerProtocol::Http,
)
.with_answer_404_path(Some("404.html"))
.to_owned();
println!("http: {:?}", to_string(&http));
let https = ListenerBuilder::new(
SocketAddress::new_v4(127, 0, 0, 1, 8443),
ListenerProtocol::Https,
)
.with_answer_404_path(Some("404.html"))
.to_owned();
println!("https: {:?}", to_string(&https));
let listeners = vec![http, https];
let config = FileConfig {
command_socket: Some(String::from("./command_folder/sock")),
worker_count: Some(2),
worker_automatic_restart: Some(true),
max_connections: Some(500),
min_buffers: Some(1),
max_buffers: Some(500),
buffer_size: Some(16393),
metrics: Some(MetricsConfig {
address: "127.0.0.1:8125".parse().unwrap(),
tagged_metrics: false,
prefix: Some(String::from("sozu-metrics")),
detail: MetricDetailLevel::default(),
}),
listeners: Some(listeners),
..Default::default()
};
println!("config: {:?}", to_string(&config));
let encoded = to_string(&config).unwrap();
println!("conf:\n{encoded}");
}
#[test]
fn parse() {
let path = "assets/config.toml";
let config = Config::load_from_path(path).unwrap_or_else(|load_error| {
panic!("Cannot load config from path {path}: {load_error:?}")
});
println!("config: {config:#?}");
}
#[test]
fn multiple_listeners_preserve_per_address_expect_proxy() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
[[listeners]]
protocol = "http"
address = "172.16.20.1:80"
expect_proxy = true
[[listeners]]
protocol = "http"
address = "10.22.0.1:80"
expect_proxy = false
[[listeners]]
protocol = "https"
address = "192.168.1.1:443"
expect_proxy = true
[[listeners]]
protocol = "https"
address = "192.168.2.1:443"
expect_proxy = false
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let listeners = file_config.listeners.as_ref().expect("No listeners found");
assert_eq!(listeners.len(), 4);
let config = ConfigBuilder::new(file_config, "/tmp/test_config.toml")
.into_config()
.expect("Could not build config");
assert_eq!(config.http_listeners.len(), 2);
assert_eq!(config.https_listeners.len(), 2);
let http_proxy = config
.http_listeners
.iter()
.find(|l| SocketAddr::from(l.address) == "172.16.20.1:80".parse().unwrap())
.expect("Listener on 172.16.20.1:80 not found");
let http_direct = config
.http_listeners
.iter()
.find(|l| SocketAddr::from(l.address) == "10.22.0.1:80".parse().unwrap())
.expect("Listener on 10.22.0.1:80 not found");
assert!(http_proxy.expect_proxy);
assert!(!http_direct.expect_proxy);
let https_proxy = config
.https_listeners
.iter()
.find(|l| SocketAddr::from(l.address) == "192.168.1.1:443".parse().unwrap())
.expect("Listener on 192.168.1.1:443 not found");
let https_direct = config
.https_listeners
.iter()
.find(|l| SocketAddr::from(l.address) == "192.168.2.1:443".parse().unwrap())
.expect("Listener on 192.168.2.1:443 not found");
assert!(https_proxy.expect_proxy);
assert!(!https_direct.expect_proxy);
}
#[test]
fn h2c_prior_knowledge_is_parsed_from_toml_and_absent_when_unset() {
let toml_content = r#"
[[listeners]]
protocol = "http"
address = "127.0.0.1:8080"
h2c_prior_knowledge = true
[[listeners]]
protocol = "http"
address = "127.0.0.1:8081"
h2c_prior_knowledge = false
[[listeners]]
protocol = "http"
address = "127.0.0.1:8082"
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let config = ConfigBuilder::new(file_config, "/tmp/test_config.toml")
.into_config()
.expect("Could not build config");
let knob = |address: &str| {
config
.http_listeners
.iter()
.find(|l| SocketAddr::from(l.address) == address.parse().unwrap())
.unwrap_or_else(|| panic!("listener on {address} not found"))
.h2c_prior_knowledge
};
assert_eq!(knob("127.0.0.1:8080"), Some(true));
assert_eq!(knob("127.0.0.1:8081"), Some(false));
assert_eq!(knob("127.0.0.1:8082"), None);
}
fn http_cluster_from_backends(backends: &str) -> Result<HttpClusterConfig, ConfigError> {
let file: FileClusterConfig = toml::from_str(&format!(
"protocol = \"http\"\nfrontends = []\nbackends = [{backends}]\n"
))
.expect("the cluster must parse");
match file.to_cluster_config("cluster", &HashSet::new())? {
ClusterConfig::Http(http) => Ok(http),
ClusterConfig::Tcp(_) => panic!("an http cluster must build as one"),
}
}
#[test]
fn a_backend_hostname_resolves_when_the_file_is_loaded() {
let backends = r#"{ address = "localhost:8080", weight = 50 }"#;
let http = http_cluster_from_backends(backends).expect("localhost must resolve");
assert!(
!http.backends.is_empty(),
"localhost resolves to an address"
);
let mut ids = HashSet::new();
for backend in &http.backends {
assert!(backend.address.ip().is_loopback(), "{backend:?}");
assert_eq!(backend.address.port(), 8080);
assert_eq!(backend.weight, Some(50));
assert_eq!(
backend.backend_id.as_deref(),
Some(format!("cluster-localhost:8080-{}", backend.address).as_str())
);
assert!(ids.insert(backend.backend_id.clone()), "ids are distinct");
}
assert_eq!(
http_cluster_from_backends(backends)
.expect("localhost must resolve")
.backends,
http.backends,
"the same file yields the same backends"
);
let orders = http.generate_requests().expect("the orders must build");
let added: Vec<&AddBackend> = orders
.iter()
.filter_map(|order| match order.request_type.as_ref() {
Some(RequestType::AddBackend(add)) => Some(add),
_ => None,
})
.collect();
assert_eq!(added.len(), http.backends.len());
for (add, backend) in added.iter().zip(&http.backends) {
assert_eq!(Some(&add.backend_id), backend.backend_id.as_ref());
assert_eq!(SocketAddr::from(add.address), backend.address);
}
}
#[test]
fn a_literal_backend_address_keeps_its_positional_id() {
let http = http_cluster_from_backends(
r#"{ address = "127.0.0.1:8080" }, { address = "[::1]:8081", backend_id = "six" }"#,
)
.expect("literals must build");
assert_eq!(
http.backends,
vec![
BackendConfig {
address: "127.0.0.1:8080".parse().unwrap(),
weight: None,
sticky_id: None,
backup: None,
backend_id: Some("cluster-0-127.0.0.1:8080".to_owned()),
unix_path: None,
},
BackendConfig {
address: "[::1]:8081".parse().unwrap(),
weight: None,
sticky_id: None,
backup: None,
backend_id: Some("six".to_owned()),
unix_path: None,
},
]
);
}
#[test]
fn a_literal_after_a_name_keeps_its_file_position_id() {
let one: SocketAddr = "10.0.0.2:8080".parse().unwrap();
let two: SocketAddr = "10.0.0.3:8080".parse().unwrap();
let answer = |addresses: Vec<SocketAddr>| {
move |_: &str| -> std::io::Result<Vec<SocketAddr>> { Ok(addresses.clone()) }
};
for (protocol, addresses) in [
("http", vec![one]),
("http", vec![one, two]),
("tcp", vec![one]),
("tcp", vec![one, two]),
] {
let resolver = answer(addresses.clone());
let file: FileClusterConfig = toml::from_str(&format!(
"protocol = \"{protocol}\"\nfrontends = []\n\
backends = [{{ address = \"name:8080\" }}, {{ address = \"127.0.0.1:9000\" }}]\n"
))
.expect("the cluster must parse");
let backends = match file
.to_cluster_config_with("cluster", &HashSet::new(), Some(&resolver))
.expect("the cluster must build")
{
ClusterConfig::Http(http) => http.backends,
ClusterConfig::Tcp(tcp) => tcp.backends,
};
assert_eq!(backends.len(), addresses.len() + 1, "{protocol}");
assert_eq!(
backends.last().and_then(|b| b.backend_id.as_deref()),
Some("cluster-1-127.0.0.1:9000"),
"{protocol} with {} resolved address(es)",
addresses.len()
);
}
let error = toml::from_str::<FileClusterConfig>(
"protocol = \"tcp\"\nfrontends = []\nbackends = [{ address = \"localhost\" }]\n",
)
.expect("the cluster must parse")
.to_cluster_config("cluster", &HashSet::new())
.expect_err("a missing port refuses the tcp cluster");
assert!(matches!(
error,
ConfigError::BackendAddressResolution { .. }
));
}
#[test]
fn numeric_shorthand_backend_hosts_never_reach_the_resolver() {
let shorthand = |_: &str| -> std::io::Result<Vec<SocketAddr>> {
Ok(vec!["8.0.0.1:8080".parse().unwrap()])
};
for address in [
"010.0.0.1:8080",
"1.2.3:8080",
"2130706433:8080",
"0x7f000001:8080",
"0x7f.1:8080",
"0X7F.0.0.1:8080",
"0177.1:8080",
"0x7f.0177.1:8080",
] {
let entry = FileBackendConfig {
address: address.to_owned(),
weight: None,
sticky_id: None,
backup: None,
backend_id: None,
unix_path: None,
};
let error = entry
.resolve("cluster", 0, &shorthand)
.expect_err("numeric shorthand must be refused");
let ConfigError::BackendAddressResolution {
cluster_id,
address: entry_address,
..
} = &error
else {
panic!("{address}: unexpected error {error:?}");
};
assert_eq!(cluster_id, "cluster");
assert_eq!(entry_address, address);
}
}
#[test]
fn unbracketed_ipv6_backend_hosts_never_reach_the_resolver() {
let unreachable = |address: &str| -> std::io::Result<Vec<SocketAddr>> {
panic!("{address}: must not reach the resolver")
};
for address in ["::1:8080", "2001:db8::1:8080", "fe80::1%eth0:80"] {
let entry = FileBackendConfig {
address: address.to_owned(),
weight: None,
sticky_id: None,
backup: None,
backend_id: None,
unix_path: None,
};
let error = entry
.resolve("cluster", 0, &unreachable)
.expect_err("an unbracketed IPv6 literal must be refused");
let ConfigError::BackendAddressResolution {
cluster_id,
address: entry_address,
reason,
} = &error
else {
panic!("{address}: unexpected error {error:?}");
};
assert_eq!(cluster_id, "cluster");
assert_eq!(entry_address, address);
assert!(reason.contains("[address]:port"), "{address}: {reason}");
}
let entry = FileBackendConfig {
address: "[2001:db8::1]:8080".to_owned(),
weight: Some(3),
sticky_id: Some("sticky".to_owned()),
backup: Some(true),
backend_id: None,
unix_path: None,
};
let backends = entry
.resolve("cluster", 2, &unreachable)
.expect("a bracketed IPv6 literal parses directly");
let address: SocketAddr = "[2001:db8::1]:8080".parse().unwrap();
assert_eq!(backends.len(), 1);
assert_eq!(backends[0].address, address);
assert_eq!(backends[0].weight, Some(3));
assert_eq!(backends[0].sticky_id.as_deref(), Some("sticky"));
assert_eq!(backends[0].backup, Some(true));
assert_eq!(
backends[0].backend_id.as_deref(),
Some("cluster-2-[2001:db8::1]:8080")
);
}
#[test]
fn benign_backend_hostnames_are_not_numeric_shorthand() {
let resolved = "10.0.0.2:8080".parse::<SocketAddr>().unwrap();
let resolver = |_: &str| -> std::io::Result<Vec<SocketAddr>> { Ok(vec![resolved]) };
for address in [
"123.example.com:8080",
"dead.beef:8080",
"0x.example.com:8080",
"1.2.3.example:8080",
"localhost:8080",
] {
let entry = FileBackendConfig {
address: address.to_owned(),
weight: None,
sticky_id: None,
backup: None,
backend_id: None,
unix_path: None,
};
let backends = entry
.resolve("cluster", 0, &resolver)
.unwrap_or_else(|error| panic!("{address}: {error:?}"));
assert_eq!(backends.len(), 1, "{address}");
}
assert!(!is_legacy_numeric_host("dead.beef"));
assert!(!is_legacy_numeric_host("1..2"));
assert!(is_legacy_numeric_host("0x7f.1"));
}
#[test]
fn a_control_load_does_not_resolve_backend_names() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("config.toml");
std::fs::write(
&path,
"command_socket = \"/tmp/sozu_test.sock\"\n\
ctl_command_timeout = 4242\n\
[clusters.app]\n\
protocol = \"http\"\n\
frontends = [{ address = \"127.0.0.1:8080\", hostname = \"example.org\" }]\n\
backends = [{ address = \"1.2.3:8080\" }, { address = \"127.0.0.1:9000\" }]\n",
)
.expect("write config");
let path = path.to_str().expect("utf-8 path");
assert!(matches!(
Config::load_from_path(path),
Err(ConfigError::BackendAddressResolution { .. })
));
let config = Config::load_from_path_for_control(path).expect("control load");
assert_eq!(config.ctl_command_timeout, 4242);
let Some(ClusterConfig::Http(app)) = config.clusters.get("app") else {
panic!("the cluster must still be validated and built");
};
assert_eq!(app.backends.len(), 1, "only the literal is kept");
}
#[test]
fn an_unresolvable_backend_address_is_a_config_error() {
for address in ["localhost", "localhost:notaport"] {
let error = http_cluster_from_backends(&format!(r#"{{ address = "{address}" }}"#))
.expect_err("the address must not resolve");
let ConfigError::BackendAddressResolution {
cluster_id,
address: entry,
..
} = &error
else {
panic!("{address}: unexpected error {error:?}");
};
assert_eq!(cluster_id, "cluster");
assert_eq!(entry, address);
assert!(error.to_string().contains(address), "{error}");
}
}
#[test]
fn every_expanded_backend_keeps_its_unix_path() {
let entry = |address: &str| FileBackendConfig {
address: address.to_owned(),
weight: None,
sticky_id: None,
backup: None,
backend_id: None,
unix_path: Some("/run/app.sock".to_owned()),
};
let v4: SocketAddr = "10.0.0.2:8080".parse().unwrap();
let v6: SocketAddr = "[fd00::1]:8080".parse().unwrap();
let expanded = entry("ingress:8080")
.expand("cluster", vec![v4, v6])
.expect("two addresses expand");
assert_eq!(expanded.len(), 2);
assert!(
expanded
.iter()
.all(|backend| backend.unix_path.as_deref() == Some("/run/app.sock"))
);
let literal = entry("127.0.0.1:8080")
.resolve("cluster", 0, &system_resolver)
.expect("a literal resolves");
assert_eq!(literal[0].unix_path.as_deref(), Some("/run/app.sock"));
}
#[test]
fn a_name_with_several_addresses_expands_into_distinct_backends() {
let entry = FileBackendConfig {
address: "ingress:8080".to_owned(),
weight: None,
sticky_id: Some("pinned".to_owned()),
backup: Some(true),
backend_id: Some("ingress".to_owned()),
unix_path: None,
};
let v4: SocketAddr = "10.0.0.2:8080".parse().unwrap();
let v6: SocketAddr = "[fd00::1]:8080".parse().unwrap();
let backends = entry
.expand("cluster", vec![v6, v4, v6])
.expect("two addresses expand");
assert_eq!(
backends,
vec![
BackendConfig {
address: v4,
weight: None,
sticky_id: Some(format!("pinned-{v4}")),
backup: Some(true),
backend_id: Some(format!("ingress-{v4}")),
unix_path: None,
},
BackendConfig {
address: v6,
weight: None,
sticky_id: Some(format!("pinned-{v6}")),
backup: Some(true),
backend_id: Some(format!("ingress-{v6}")),
unix_path: None,
},
]
);
let single = entry.expand("cluster", vec![v4]).expect("one address");
assert_eq!(single[0].sticky_id, Some(format!("pinned-{v4}")));
assert!(matches!(
entry.expand("cluster", Vec::new()),
Err(ConfigError::BackendAddressResolution { .. })
));
}
#[test]
fn an_affinity_header_reaches_the_add_cluster_order() {
let file: FileClusterConfig = toml::from_str(
r#"
protocol = "http"
load_balancing = "HRW"
affinity_header = "X-Tenant"
frontends = []
backends = []
"#,
)
.expect("the cluster must parse");
let ClusterConfig::Http(http) = file
.to_cluster_config("keyed", &HashSet::new())
.expect("the cluster must build")
else {
panic!("an http cluster must build as one");
};
let orders = http.generate_requests().expect("the orders must build");
let Some(RequestType::AddCluster(cluster)) =
orders.first().and_then(|order| order.request_type.as_ref())
else {
panic!("the orders must lead with an AddCluster");
};
assert_eq!(cluster.affinity_header.as_deref(), Some("X-Tenant"));
assert_eq!(cluster.affinity_cookie, None);
}
#[test]
fn a_unix_path_backend_reaches_the_add_unix_backend_order() {
let file: FileClusterConfig = toml::from_str(
r#"
protocol = "http"
frontends = []
backends = [
{ address = "127.0.0.1:1", unix_path = "/run/app.sock" },
{ address = "127.0.0.1:2" },
]
"#,
)
.expect("the cluster must parse");
let ClusterConfig::Http(http) = file
.to_cluster_config("app", &HashSet::new())
.expect("the cluster must build")
else {
panic!("an http cluster must build as one");
};
let unix_paths: Vec<_> = http
.generate_requests()
.expect("the orders must build")
.into_iter()
.filter_map(|order| match order.request_type {
Some(RequestType::AddUnixBackend(backend)) => {
Some(("AddUnixBackend", backend.unix_path))
}
Some(RequestType::AddBackend(backend)) => Some(("AddBackend", backend.unix_path)),
_ => None,
})
.collect();
assert_eq!(
unix_paths,
vec![
("AddUnixBackend", Some("/run/app.sock".to_owned())),
("AddBackend", None)
]
);
}
#[test]
fn a_unix_path_backend_needs_a_unique_address() {
for protocol in ["http", "tcp"] {
let file: FileClusterConfig = toml::from_str(&format!(
"protocol = \"{protocol}\"\nfrontends = []\nbackends = [\n\
{{ address = \"127.0.0.1:1\", unix_path = \"/run/a.sock\" }},\n\
{{ address = \"127.0.0.1:1\", unix_path = \"/run/b.sock\" }},\n]\n"
))
.expect("the cluster must parse");
assert!(
matches!(
file.to_cluster_config("app", &HashSet::new()),
Err(ConfigError::InvalidBackendUnixPath { .. })
),
"a {protocol} cluster must refuse a shared unix backend address"
);
}
}
#[test]
fn a_health_check_port_and_a_unix_path_backend_never_share_a_file_cluster() {
let resolver = |_: &str| -> std::io::Result<Vec<SocketAddr>> {
Ok(vec!["10.0.0.2:8080".parse().unwrap()])
};
for protocol in ["http", "tcp"] {
for address in ["127.0.0.1:1", "ingress:8080"] {
let build = |health_check: &str, resolver: Option<&BackendResolver<'_>>| {
toml::from_str::<FileClusterConfig>(&format!(
"protocol = \"{protocol}\"\nfrontends = []\nbackends = [\n\
{{ address = \"{address}\", unix_path = \"/run/app.sock\" }},\n]\n\
[health_check]\n{health_check}"
))
.expect("the cluster must parse")
.to_cluster_config_with("app", &HashSet::new(), resolver)
};
for resolver in [Some(&resolver as &BackendResolver<'_>), None] {
let error = build("mode = \"TCP\"\nport = 8081\n", resolver)
.expect_err("a probe port next to a unix backend must be refused");
assert!(
matches!(
&error,
ConfigError::InvalidHealthCheck { cluster_id, reason }
if cluster_id == "app" && *reason == HEALTH_CHECK_PORT_UNIX_REFUSED
),
"{protocol} {address}: unexpected {error:?}"
);
build("mode = \"TCP\"\n", resolver)
.expect("a unix backend without a probe port stays valid");
}
}
}
}
#[test]
fn parse_backend_unix_path_accepts_absolute_and_abstract_names_only() {
let path = parse_backend_unix_path("/run/app.sock").expect("an absolute path parses");
assert_eq!(
path.as_pathname(),
Some(std::path::Path::new("/run/app.sock"))
);
#[cfg(target_os = "linux")]
{
use std::os::linux::net::SocketAddrExt;
let named = parse_backend_unix_path("@app").expect("an abstract name parses");
assert_eq!(named.as_abstract_name(), Some(&b"app"[..]));
assert_eq!(named.as_pathname(), None);
}
let too_long = format!("/{}", "a".repeat(200));
for refused in ["", "run/app.sock", "@", "/run/a\0b", too_long.as_str()] {
assert!(
parse_backend_unix_path(refused).is_err(),
"{refused:?} must be refused"
);
}
}
#[test]
fn header_predicates_reach_the_add_http_frontend_order() {
let file: FileClusterConfig = toml::from_str(
r#"
protocol = "http"
frontends = [
{ address = "0.0.0.0:80", hostname = "api.example.com", header_predicates = [
{ name = "X-Canary", match = "EXACT", value = "always" },
{ name = "X-Region", match = "regex", value = "eu-.*" },
{ name = "X-Beta", match = "present" },
] },
]
backends = []
"#,
)
.expect("the cluster must parse");
let ClusterConfig::Http(http) = file
.to_cluster_config("canary", &HashSet::new())
.expect("the cluster must build")
else {
panic!("an http cluster must build as one");
};
let orders = http.generate_requests().expect("the orders must build");
let front = orders
.iter()
.find_map(|order| match order.request_type.as_ref() {
Some(RequestType::AddHttpFrontendWithHeaderPredicates(front)) => Some(front),
_ => None,
})
.expect("the orders must add the frontend");
let predicate = |name: &str, kind: HeaderMatchKind, value: Option<&str>| HeaderPredicate {
name: name.to_owned(),
kind: kind as i32,
value: value.map(str::to_owned),
};
assert_eq!(
front.header_predicates,
vec![
predicate("X-Canary", HeaderMatchKind::Exact, Some("always")),
predicate("X-Region", HeaderMatchKind::Regex, Some("eu-.*")),
predicate("X-Beta", HeaderMatchKind::Present, None),
]
);
}
#[test]
fn unusable_header_predicates_are_refused_at_config_load() {
for entry in [
r#"{ name = "X-Canary", match = "equals", value = "1" }"#,
r#"{ name = "Host", match = "exact", value = "example.com" }"#,
r#"{ name = "Cookie", match = "present" }"#,
r#"{ name = "X-Canary", match = "present", value = "1" }"#,
r#"{ name = "X-Canary", match = "exact" }"#,
r#"{ name = "X Canary", match = "present" }"#,
] {
let file: FileClusterConfig = toml::from_str(&format!(
r#"
protocol = "http"
frontends = [
{{ address = "0.0.0.0:80", hostname = "api.example.com", header_predicates = [
{{ name = "X-Valid", match = "present" }},
{entry},
] }},
]
backends = []
"#
))
.expect("the cluster must parse");
match file.to_cluster_config("canary", &HashSet::new()) {
Err(ConfigError::InvalidHeaderPredicate { index: 1, .. }) => {}
other => panic!("{entry}: expected InvalidHeaderPredicate at 1, got {other:?}"),
}
}
}
#[test]
fn a_tcp_cluster_refuses_an_affinity_header_or_cookie() {
for key in [
"affinity_header = \"X-Tenant\"",
"affinity_cookie = \"tenant\"",
] {
let file: FileClusterConfig = toml::from_str(&format!(
"protocol = \"tcp\"\nload_balancing = \"MAGLEV\"\n{key}\nfrontends = []\nbackends = []\n"
))
.expect("the cluster must parse");
assert!(
matches!(
file.to_cluster_config("keyed", &HashSet::new()),
Err(ConfigError::InvalidAffinityKey { .. })
),
"a TCP cluster must refuse `{key}`"
);
}
}
#[test]
fn validate_affinity_key_accepts_one_token_and_refuses_the_rest() {
assert!(validate_affinity_key(None, None).is_ok());
assert!(validate_affinity_key(Some("X-Tenant"), None).is_ok());
assert!(validate_affinity_key(None, Some("tenant_id")).is_ok());
assert!(validate_affinity_key(Some("X-Tenant"), Some("tenant")).is_err());
for unreachable in ["Host", "host", "HOST", "Cookie", "cookie"] {
assert!(
validate_affinity_key(Some(unreachable), None).is_err(),
"header {unreachable:?} can never be found and must be refused"
);
}
assert!(validate_affinity_key(None, Some("host")).is_ok());
for bad in ["", "X Tenant", "X-Tenant:", "tenant\r\n", "t=1"] {
assert!(
validate_affinity_key(Some(bad), None).is_err(),
"header {bad:?} must be refused"
);
assert!(
validate_affinity_key(None, Some(bad)).is_err(),
"cookie {bad:?} must be refused"
);
}
}
#[test]
fn shuffle_sharding_reaches_the_add_cluster_order_and_is_validated() {
let parse = |extra: &str| -> FileClusterConfig {
toml::from_str(&format!(
"protocol = \"http\"\nfrontends = []\nbackends = []\n{extra}"
))
.expect("the cluster must parse")
};
let ClusterConfig::Http(http) =
parse("shard_percent = 25\nshard_min_backends = 4\nshard_mode = \"STRICT\"\n")
.to_cluster_config("sharded", &HashSet::new())
.expect("a valid sharded cluster builds")
else {
panic!("an http cluster must build as one");
};
let orders = http.generate_requests().expect("the orders must build");
let Some(RequestType::AddCluster(cluster)) =
orders.first().and_then(|order| order.request_type.as_ref())
else {
panic!("the orders must lead with an AddCluster");
};
assert_eq!(
(
cluster.shard_percent,
cluster.shard_min_backends,
cluster.shard_mode
),
(Some(25), Some(4), Some(ShardMode::Strict as i32))
);
for bad in [
"shard_percent = 0\n",
"shard_percent = 101\n",
"shard_percent = 25\nshard_min_backends = 1\n",
"shard_min_backends = 8\n",
"shard_mode = \"STRICT\"\n",
] {
assert!(
matches!(
parse(bad).to_cluster_config("sharded", &HashSet::new()),
Err(ConfigError::InvalidShuffleSharding { .. })
),
"{bad:?} must be refused"
);
}
}
#[test]
fn http_status_range_parses_codes_ranges_classes_and_any() {
let range = |start, end| HttpStatusRange { start, end };
for (spec, expected) in [
("404", range(404, 404)),
("200-399", range(200, 399)),
(" 200 - 399 ", range(200, 399)),
("2xx", range(200, 299)),
("5XX", range(500, 599)),
("any", range(100, 599)),
("ANY", range(100, 599)),
("100-599", range(100, 599)),
] {
assert_eq!(parse_http_status_range(spec), Ok(expected), "{spec:?}");
}
for spec in [
"", "abc", "99", "600", "0", "6xx", "0xx", "2x", "2xxx", "399-200", "200-", "-200",
"+200", "200-600", "1-2-3", "2xx-3xx",
] {
assert!(
parse_http_status_range(spec).is_err(),
"{spec:?} must be refused"
);
}
}
#[test]
fn http_status_range_display_round_trips_through_the_parser() {
for spec in ["404", "200-399", "2xx", "any", "101-102"] {
let range = parse_http_status_range(spec).unwrap();
assert_eq!(range.to_string(), spec);
assert_eq!(parse_http_status_range(&range.to_string()), Ok(range));
}
}
#[test]
fn health_check_validation_covers_mode_uri_and_accepted_statuses() {
let http = HealthCheckConfig {
uri: "/health".to_owned(),
interval: 10,
timeout: 5,
healthy_threshold: 3,
unhealthy_threshold: 3,
expected_status: 0,
mode: HealthCheckMode::Http as i32,
accepted_statuses: Vec::new(),
port: None,
headers: Vec::new(),
};
assert_eq!(validate_health_check_config(&http), Ok(()));
let tcp_no_uri = HealthCheckConfig {
uri: String::new(),
mode: HealthCheckMode::Tcp as i32,
..http.clone()
};
assert_eq!(validate_health_check_config(&tcp_no_uri), Ok(()));
let http_no_uri = HealthCheckConfig {
uri: String::new(),
..http.clone()
};
assert!(validate_health_check_config(&http_no_uri).is_err());
let tcp_crlf = HealthCheckConfig {
uri: "/a\r\nGET /b".to_owned(),
..tcp_no_uri.clone()
};
assert!(validate_health_check_config(&tcp_crlf).is_err());
let tcp_expected = HealthCheckConfig {
expected_status: 200,
..tcp_no_uri.clone()
};
assert!(validate_health_check_config(&tcp_expected).is_err());
let tcp_accepted = HealthCheckConfig {
accepted_statuses: vec![HttpStatusRange {
start: 100,
end: 599,
}],
..tcp_no_uri.clone()
};
assert!(validate_health_check_config(&tcp_accepted).is_err());
let unknown_mode = HealthCheckConfig {
mode: 7,
..http.clone()
};
assert!(validate_health_check_config(&unknown_mode).is_err());
let accepted = HealthCheckConfig {
accepted_statuses: vec![HttpStatusRange {
start: 200,
end: 499,
}],
..http.clone()
};
assert_eq!(validate_health_check_config(&accepted), Ok(()));
let both = HealthCheckConfig {
expected_status: 200,
..accepted.clone()
};
assert!(
validate_health_check_config(&both).is_err(),
"expected_status and accepted_statuses must not both be set"
);
for (start, end) in [(99, 200), (200, 600), (300, 200)] {
let bad_range = HealthCheckConfig {
accepted_statuses: vec![HttpStatusRange { start, end }],
..http.clone()
};
assert!(
validate_health_check_config(&bad_range).is_err(),
"{start}-{end} must be refused"
);
}
}
#[test]
fn health_check_toml_mode_and_accepted_statuses_reach_the_add_cluster_order() {
let parse = |block: &str| -> FileClusterConfig {
toml::from_str(&format!(
"protocol = \"http\"\nfrontends = []\nbackends = []\n[health_check]\n{block}"
))
.expect("the cluster must parse")
};
let health_check = |block: &str| -> HealthCheckConfig {
let ClusterConfig::Http(http) = parse(block)
.to_cluster_config("checked", &HashSet::new())
.expect("a valid health check builds")
else {
panic!("an http cluster must build as one");
};
let orders = http.generate_requests().expect("the orders must build");
let Some(RequestType::AddCluster(cluster)) =
orders.first().and_then(|order| order.request_type.as_ref())
else {
panic!("the orders must lead with an AddCluster");
};
cluster
.health_check
.clone()
.expect("the health check must travel")
};
let legacy = health_check("uri = \"/health\"\nexpected_status = 204\n");
assert_eq!(legacy.mode, HealthCheckMode::Http as i32);
assert_eq!(
(legacy.uri.as_str(), legacy.expected_status),
("/health", 204)
);
assert!(legacy.accepted_statuses.is_empty());
let defaulted = health_check("");
assert_eq!(defaulted.mode, HealthCheckMode::Http as i32);
assert_eq!(defaulted.uri, "/");
let tcp = health_check("mode = \"TCP\"\n");
assert_eq!(tcp.mode, HealthCheckMode::Tcp as i32);
let accepted = health_check("uri = \"/livez\"\naccepted_statuses = [\"2xx\", \"404\"]\n");
assert_eq!(
accepted.accepted_statuses,
vec![
HttpStatusRange {
start: 200,
end: 299
},
HttpStatusRange {
start: 404,
end: 404
},
]
);
for bad in [
"accepted_statuses = [\"6xx\"]\n",
"expected_status = 200\naccepted_statuses = [\"any\"]\n",
"uri = \"health\"\n",
"mode = \"TCP\"\nexpected_status = 200\n",
"mode = \"TCP\"\naccepted_statuses = [\"any\"]\n",
] {
assert!(
matches!(
parse(bad).to_cluster_config("checked", &HashSet::new()),
Err(ConfigError::InvalidHealthCheck { .. })
),
"{bad:?} must be refused"
);
}
}
#[test]
fn health_check_validation_bounds_the_probe_port() {
for mode in [HealthCheckMode::Http, HealthCheckMode::Tcp] {
let config = |port: Option<u32>| HealthCheckConfig {
uri: "/health".to_owned(),
interval: 10,
timeout: 5,
healthy_threshold: 3,
unhealthy_threshold: 3,
mode: mode as i32,
port,
..Default::default()
};
for port in [None, Some(1), Some(8080), Some(65535)] {
assert_eq!(
validate_health_check_config(&config(port)),
Ok(()),
"{mode:?} probe port {port:?} must be accepted"
);
}
for port in [0, 65536, u32::MAX] {
assert_eq!(
validate_health_check_config(&config(Some(port))),
Err("health check port must lie in 1..=65535"),
"{mode:?} probe port {port} must be refused"
);
}
}
}
#[test]
fn health_check_toml_port_reaches_the_add_cluster_order() {
let build = |block: &str| {
toml::from_str::<FileClusterConfig>(&format!(
"protocol = \"tcp\"\nfrontends = []\nbackends = []\n[health_check]\n{block}"
))
.expect("the cluster must parse")
.to_cluster_config("checked", &HashSet::new())
};
let port = |block: &str| -> Option<u32> {
let Ok(ClusterConfig::Tcp(tcp)) = build(block) else {
panic!("a tcp cluster with a valid health check must build as one");
};
tcp.health_check.expect("the health check must travel").port
};
assert_eq!(port("mode = \"TCP\"\n"), None);
assert_eq!(port("mode = \"TCP\"\nport = 8081\n"), Some(8081));
assert_eq!(port("uri = \"/health\"\nport = 65535\n"), Some(65535));
for bad in ["port = 0\n", "port = 65536\n"] {
assert!(
matches!(build(bad), Err(ConfigError::InvalidHealthCheck { .. })),
"{bad:?} must be refused"
);
}
}
fn health_check_with_headers(headers: &[(&str, &str)]) -> HealthCheckConfig {
HealthCheckConfig {
uri: "/".to_owned(),
interval: 10,
timeout: 5,
healthy_threshold: 3,
unhealthy_threshold: 3,
expected_status: 0,
mode: HealthCheckMode::Http as i32,
accepted_statuses: Vec::new(),
port: None,
headers: headers
.iter()
.map(|(name, _)| HealthCheckHeader {
name: (*name).to_owned(),
fingerprint: String::new(),
})
.collect(),
}
}
fn check_headers(headers: &[(&str, &str)]) -> Result<(), &'static str> {
let config = health_check_with_headers(headers);
validate_health_check_config(&config)?;
let values: Vec<HealthCheckHeaderValue> = headers
.iter()
.map(|(name, value)| HealthCheckHeaderValue {
name: (*name).to_owned(),
value: value.as_bytes().to_vec(),
})
.collect();
validate_health_check_header_values(&config, &values)
}
#[test]
fn health_check_header_value_refuses_leading_and_trailing_ows() {
for value in [" v", "v ", "\tv", "v\t", " ", "\t", " v ", "a b \t"] {
assert_eq!(
check_headers(&[("x-a", value)]),
Err("health check header value must not begin or end with a space or a tab"),
"{value:?} must be refused"
);
}
for value in ["", "v", "a b", "a\tb", "é ü", "\u{a0}v"] {
assert_eq!(
check_headers(&[("x-a", value)]),
Ok(()),
"{value:?} must be accepted"
);
}
}
#[test]
fn health_check_header_validation_covers_every_class() {
let refused = |headers: &[(&str, &str)], reason: &str| {
assert_eq!(
check_headers(headers),
Err(reason),
"{headers:?} must be refused"
);
};
assert_eq!(
check_headers(&[
("Authorization", "Bearer t"),
("X-Empty", ""),
("x-tab", "a\tb"),
("x-utf8", "caf\u{e9} \u{1F600}"),
]),
Ok(())
);
let mut tcp = health_check_with_headers(&[("x-a", "1")]);
tcp.mode = HealthCheckMode::Tcp as i32;
assert_eq!(
validate_health_check_config(&tcp),
Err("health check headers apply to HTTP mode only: leave them unset with mode TCP")
);
let names: Vec<String> = (0..33).map(|i| format!("x-{i}")).collect();
let many = |count: usize| -> Vec<(&str, &str)> {
names[..count].iter().map(|n| (n.as_str(), "v")).collect()
};
assert_eq!(check_headers(&many(32)), Ok(()));
refused(&many(33), "health check carries more than 32 headers");
for name in [
":authority",
":x",
"Host",
"HOST",
"connection",
"Keep-Alive",
"proxy-connection",
"Transfer-Encoding",
"content-length",
"TE",
"trailer",
"upgrade",
"Expect",
] {
refused(
&[(name, "v")],
"health check header name is reserved for the probe or its framing",
);
}
for name in [
"",
" x",
"x ",
"a b",
"a:b",
"caf\u{e9}",
"a\r",
"a\"b",
"a(b",
] {
refused(
&[(name, "v")],
"health check header name must be a non-empty RFC 9110 token",
);
}
for value in [
"a\rb", "a\nb", "a\0b", "a\x01b", "a\x08b", "a\x0bb", "a\x1fb", "a\x7fb",
] {
refused(
&[("x-a", value)],
"health check header value must not contain control bytes",
);
}
refused(
&[("X-Token", "1"), ("x-token", "2")],
"health check header names must be unique, compared case-insensitively",
);
let value_fits = "v".repeat(8192 - 1 - 4);
assert_eq!(check_headers(&[("x", &value_fits)]), Ok(()));
let value_over = "v".repeat(8192 - 1 - 4 + 1);
refused(
&[("x", &value_over)],
"health check headers exceed 8192 bytes (name + value + 4 per header)",
);
let mut long_uri = health_check_with_headers(&[]);
long_uri.uri = format!("/{}", "a".repeat(9000));
assert_eq!(validate_health_check_config(&long_uri), Ok(()));
long_uri.headers = vec![HealthCheckHeader {
name: "x".to_owned(),
fingerprint: String::new(),
}];
assert_eq!(validate_health_check_config(&long_uri), Ok(()));
let carrier = HealthCheckHeaderValue {
name: "x".to_owned(),
value: "v".repeat(8000).into_bytes(),
};
assert_eq!(
validate_health_check_header_values(&long_uri, &[carrier]),
Err("health check URI and headers may exceed the 16384-byte HTTP/2 probe block")
);
let one = health_check_with_headers(&[("x-a", ""), ("x-b", "")]);
let carrier = |name: &str, value: &[u8]| HealthCheckHeaderValue {
name: name.to_owned(),
value: value.to_vec(),
};
assert_eq!(
validate_health_check_header_values(&one, &[carrier("x-a", b"1")]),
Err("health check carries a header value count different from its header count")
);
assert_eq!(
validate_health_check_header_values(
&one,
&[carrier("x-b", b"1"), carrier("x-a", b"2")]
),
Err("health check header values must follow the order and names of the headers")
);
assert_eq!(
validate_health_check_header_values(
&one,
&[carrier("x-a", b"1"), carrier("x-b", &[0xff])]
),
Err("health check header value must be valid UTF-8")
);
let mut spoofed = health_check_with_headers(&[("x-a", "")]);
spoofed.headers[0].fingerprint = "not a fingerprint".to_owned();
assert_eq!(
validate_health_check_config(&spoofed),
Err("health check header fingerprint is malformed")
);
}
#[test]
fn health_check_headers_default_empty_and_round_trip_in_order() {
use prost::Message;
let config = health_check_with_headers(&[
("Z-Last-Alphabetically", "1"),
("authorization", "Bearer \u{e9}\tx"),
("A-First-Alphabetically", ""),
]);
let mut value = serde_json::to_value(&config).expect("serialize");
value
.as_object_mut()
.expect("an object")
.remove("headers")
.expect("the key is serialized");
let legacy: HealthCheckConfig = serde_json::from_value(value).expect("legacy record loads");
assert!(legacy.headers.is_empty());
let json = serde_json::to_string(&config).expect("serialize");
let back: HealthCheckConfig = serde_json::from_str(&json).expect("deserialize");
assert_eq!(back, config);
assert_eq!(
back.headers
.iter()
.map(|h| h.name.as_str())
.collect::<Vec<_>>(),
[
"Z-Last-Alphabetically",
"authorization",
"A-First-Alphabetically"
]
);
let headerless = HealthCheckConfig {
headers: Vec::new(),
..config.clone()
};
let decoded =
HealthCheckConfig::decode(headerless.encode_to_vec().as_slice()).expect("decodes");
assert!(decoded.headers.is_empty());
let decoded =
HealthCheckConfig::decode(config.encode_to_vec().as_slice()).expect("decodes");
assert_eq!(decoded, config);
}
#[test]
fn health_check_toml_headers_split_into_metadata_and_ordered_references() {
let parse = |block: &str| {
toml::from_str::<FileHealthCheckConfig>(&format!("uri = \"/health\"\n{block}"))
.expect("the table parses")
};
let file = parse(
"[[headers]]\nname = \"Z-Second\"\nvalue_file = \"/missing/z\"\n\
[[headers]]\nname = \"Authorization\"\nvalue_file = \"rel/a\"\n\
[[headers]]\nname = \"X-Empty\"\nvalue_file = \"/missing/e\"\n",
);
let (config, files) = file.to_parts().expect("a valid list builds");
assert_eq!(
config
.headers
.iter()
.map(|h| (h.name.as_str(), h.fingerprint.as_str()))
.collect::<Vec<_>>(),
[("Z-Second", ""), ("Authorization", ""), ("X-Empty", "")],
"metadata names the headers and carries no value or fingerprint"
);
assert_eq!(
files
.iter()
.map(|f| (f.name.as_str(), f.value_file.as_str()))
.collect::<Vec<_>>(),
[
("Z-Second", "/missing/z"),
("Authorization", "rel/a"),
("X-Empty", "/missing/e")
]
);
assert_eq!(
file.to_proto(),
Err(
"health check headers from the configuration file are not available yet: their values need the protected channel"
)
);
let bare = parse("");
assert!(bare.to_proto().expect("builds").headers.is_empty());
assert!(bare.to_parts().expect("builds").1.is_empty());
for bad in [
"mode = \"TCP\"\n[[headers]]\nname = \"x\"\nvalue_file = \"/v\"\n",
"[[headers]]\nname = \"Host\"\nvalue_file = \"/v\"\n",
"[[headers]]\nname = \"x\"\nvalue_file = \"/v\"\n\
[[headers]]\nname = \"X\"\nvalue_file = \"/w\"\n",
] {
assert!(parse(bad).to_parts().is_err(), "{bad:?} must be refused");
}
assert!(
toml::from_str::<FileHealthCheckConfig>(
"uri = \"/\"\n[[headers]]\nname = \"x\"\nvalue_file = \"/v\"\nextra = 1\n"
)
.is_err()
);
}
const HEADER_SENTINEL: &str = "SENTINEL-HEADER-VALUE-7f3a91";
const REF_SENTINEL: &str = "sentinel-ref-dir-9c1e42";
fn value_file_cluster(protocol: &str, headers: &[(&str, &str)]) -> FileClusterConfig {
let tables: String = headers
.iter()
.map(|(name, path)| {
format!("[[health_check.headers]]\nname = \"{name}\"\nvalue_file = \"{path}\"\n")
})
.collect();
toml::from_str(&format!(
"protocol = \"{protocol}\"\nfrontends = []\nbackends = []\n\
[health_check]\nuri = \"/health\"\n{tables}"
))
.expect("a value_file reference parses")
}
fn cluster_json(cluster: &ClusterConfig) -> serde_json::Value {
match cluster {
ClusterConfig::Http(http) => serde_json::to_value(http),
ClusterConfig::Tcp(tcp) => serde_json::to_value(tcp),
}
.expect("a cluster serializes")
}
#[test]
fn health_check_value_file_is_kept_unopened_in_order_for_http_and_tcp() {
let missing_a = format!("/{REF_SENTINEL}/missing/a.secret");
let missing_b = format!("relative/../{REF_SENTINEL}/b.secret");
let headers = [
("Z-Second", missing_a.as_str()),
("Authorization", &missing_b),
];
for protocol in ["http", "tcp"] {
let file = value_file_cluster(protocol, &headers);
let cluster = file
.to_cluster_config("checked", &HashSet::new())
.expect("a reference to a missing file builds: nothing opens it");
let json = cluster_json(&cluster);
let metadata: Vec<(&str, &str)> = json["health_check"]["headers"]
.as_array()
.expect("the metadata lists the headers")
.iter()
.map(|h| {
(
h["name"].as_str().expect("a name"),
h["fingerprint"].as_str().expect("a fingerprint"),
)
})
.collect();
assert_eq!(
metadata,
[("Z-Second", ""), ("Authorization", "")],
"{protocol}: ordered names, empty fingerprints"
);
assert_eq!(
json["health_check_header_files"],
serde_json::json!([
{ "name": "Z-Second", "value_file": missing_a },
{ "name": "Authorization", "value_file": missing_b },
]),
"{protocol}: references are stored unchanged and in order"
);
let mut config = Config::default();
config
.clusters
.insert("checked".to_owned(), cluster.clone());
let text = serde_json::to_string(&config).expect("a Config serializes");
let back: Config = serde_json::from_str(&text).expect("a Config deserializes");
assert_eq!(
back.clusters.get("checked"),
Some(&cluster),
"{protocol}: references survive Config serde"
);
}
let bare: FileClusterConfig = toml::from_str(
"protocol = \"http\"\nfrontends = []\nbackends = []\n[health_check]\nuri = \"/\"\n",
)
.expect("parses");
let built = bare
.to_cluster_config("bare", &HashSet::new())
.expect("builds");
assert_eq!(
cluster_json(&built)["health_check_header_files"],
serde_json::json!([])
);
}
#[test]
fn health_check_header_refuses_a_literal_or_env_value_and_never_echoes() {
let parse = |table: &str| {
toml::from_str::<FileHealthCheckConfig>(&format!("uri = \"/\"\n[[headers]]\n{table}\n"))
};
for bad in [
format!("name = \"x\"\nvalue = \"{HEADER_SENTINEL}\""),
format!("name = \"x\"\nvalue_env = \"{HEADER_SENTINEL}\""),
"name = \"x\"".to_owned(),
"value_file = \"/p\"".to_owned(),
format!("name = \"x\"\nvalue_file = \"/p\"\nvalue = \"{HEADER_SENTINEL}\""),
format!("name = \"x\"\nvalue_file = \"/p\"\nvalue_env = \"{HEADER_SENTINEL}\""),
format!("name = \"x\"\nvalue_file = \"/p\"\nextra = \"{HEADER_SENTINEL}\""),
format!("name = \"x\"\nvalue_file = 12\nextra = \"{HEADER_SENTINEL}\""),
] {
let error = parse(&bad).expect_err(&format!("{bad:?} must be refused"));
let diagnostic = toml_error_diagnostic("/etc/sozu.toml", &error);
assert!(
!diagnostic.contains(HEADER_SENTINEL),
"the diagnostic must not echo a value: {diagnostic}"
);
}
let accepted = parse(&format!("name = \"x\"\nvalue_file = \"/{REF_SENTINEL}/v\""))
.expect("a name and a value_file parse");
assert_eq!(accepted.headers.len(), 1);
let debug = format!("{accepted:?}");
assert!(
!debug.contains(REF_SENTINEL),
"Debug must redact the reference path: {debug}"
);
}
#[test]
fn health_check_value_file_cluster_generates_no_worker_request() {
for protocol in ["http", "tcp"] {
let path = format!("/{REF_SENTINEL}/v");
let file = value_file_cluster(protocol, &[("x-token", &path)]);
let cluster = file
.to_cluster_config("guarded", &HashSet::new())
.expect("builds");
let direct = cluster
.generate_requests()
.expect_err("no order may carry a name without its value");
assert!(
matches!(direct, ConfigError::InvalidHealthCheck { .. }),
"{protocol}: {direct:?}"
);
let mut config = Config::default();
config.clusters.insert("guarded".to_owned(), cluster);
let error = config
.generate_config_messages()
.expect_err("the whole generation is refused before any order is returned");
let shown = format!("{error} {error:?}");
assert!(
matches!(error, ConfigError::InvalidHealthCheck { .. })
&& shown.contains("guarded")
&& !shown.contains(REF_SENTINEL),
"{protocol}: {shown}"
);
}
let headerless: FileClusterConfig = toml::from_str(
"protocol = \"http\"\nfrontends = []\nbackends = []\n[health_check]\nuri = \"/\"\n",
)
.expect("parses");
let cluster = headerless
.to_cluster_config("plain", &HashSet::new())
.expect("builds");
assert!(cluster.generate_requests().is_ok());
}
#[test]
fn multiple_listeners_generate_correct_worker_requests() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
activate_listeners = true
[[listeners]]
protocol = "http"
address = "172.16.20.1:80"
expect_proxy = true
[[listeners]]
protocol = "http"
address = "10.22.0.1:80"
expect_proxy = false
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let config = ConfigBuilder::new(file_config, "/tmp/test_config.toml")
.into_config()
.expect("Could not build config");
let messages = config
.generate_config_messages()
.expect("Could not generate config messages");
let add_listener_count = messages
.iter()
.filter(|m| {
matches!(
m.content.request_type,
Some(RequestType::AddHttpListener(_))
)
})
.count();
let activate_listener_count = messages
.iter()
.filter(|m| {
matches!(
m.content.request_type,
Some(RequestType::ActivateListener(ActivateListener {
proxy,
..
})) if proxy == ListenerType::Http as i32
)
})
.count();
assert_eq!(add_listener_count, 2);
assert_eq!(activate_listener_count, 2);
}
#[test]
fn documented_udp_dns_example_loads_and_emits_udp_requests() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
activate_listeners = true
[[listeners]]
protocol = "udp"
address = "0.0.0.0:53"
[clusters.dns]
protocol = "tcp"
load_balancing = "HRW"
frontends = [
{ address = "0.0.0.0:53" }
]
backends = [
{ address = "10.0.0.10:53" },
{ address = "10.0.0.11:53" }
]
[clusters.dns.udp]
affinity_key = "SOURCE_IP"
responses = 1
requests = 0
send_proxy_protocol = true
[clusters.dns.udp.health]
mode = "TCP_PROBE"
tcp_port = 53
rise = 2
fall = 3
fail_open = true
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse documented DNS TOML");
let config = ConfigBuilder::new(file_config, "/tmp/test_config.toml")
.into_config()
.expect("documented UDP DNS example must load without WrongFrontendProtocol");
assert_eq!(
config.udp_listeners.len(),
1,
"the protocol=\"udp\" listener must be built"
);
let messages = config
.generate_config_messages()
.expect("Could not generate config messages");
let add_udp_listener_count = messages
.iter()
.filter(|m| matches!(m.content.request_type, Some(RequestType::AddUdpListener(_))))
.count();
assert_eq!(
add_udp_listener_count, 1,
"must emit exactly one AddUdpListener"
);
let add_udp_frontend_count = messages
.iter()
.filter(|m| matches!(m.content.request_type, Some(RequestType::AddUdpFrontend(_))))
.count();
assert_eq!(
add_udp_frontend_count, 1,
"the cluster frontend on the UDP listener must emit AddUdpFrontend"
);
let add_tcp_frontend_count = messages
.iter()
.filter(|m| matches!(m.content.request_type, Some(RequestType::AddTcpFrontend(_))))
.count();
assert_eq!(
add_tcp_frontend_count, 0,
"a UDP-listener-addressed frontend must not be emitted as AddTcpFrontend"
);
let udp_frontend = messages
.iter()
.find_map(|m| match &m.content.request_type {
Some(RequestType::AddUdpFrontend(f)) => Some(f),
_ => None,
})
.expect("AddUdpFrontend must be present");
assert_eq!(udp_frontend.cluster_id, "dns");
assert_eq!(
SocketAddr::from(udp_frontend.address),
"0.0.0.0:53".parse().unwrap()
);
let cluster = messages
.iter()
.find_map(|m| match &m.content.request_type {
Some(RequestType::AddCluster(c)) if c.cluster_id == "dns" => Some(c),
_ => None,
})
.expect("AddCluster for 'dns' must be present");
let udp = cluster
.udp
.as_ref()
.expect("[clusters.dns.udp] block must carry onto the cluster");
assert_eq!(udp.responses, Some(1));
}
#[test]
fn listener_interface_round_trips_from_the_config_file() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
activate_listeners = true
[[listeners]]
protocol = "http"
address = "0.0.0.0:8080"
interface = "wg0"
[[listeners]]
protocol = "http"
address = "0.0.0.0:8080"
interface = "lo"
[[listeners]]
protocol = "http"
address = "0.0.0.0:8081"
[[listeners]]
protocol = "https"
address = "0.0.0.0:8443"
interface = "wg0"
[[listeners]]
protocol = "tcp"
address = "0.0.0.0:9000"
interface = "wg0"
[[listeners]]
protocol = "udp"
address = "0.0.0.0:5353"
interface = "wg0"
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("the interface key must parse");
let config = ConfigBuilder::new(file_config, "/tmp/test_config.toml")
.into_config()
.expect("listeners on one address with different interfaces must build");
let http: Vec<(String, Option<&str>)> = config
.http_listeners
.iter()
.map(|l| {
(
SocketAddr::from(l.address).to_string(),
l.interface.as_deref(),
)
})
.collect();
assert_eq!(
http,
vec![
("0.0.0.0:8080".to_owned(), Some("wg0")),
("0.0.0.0:8080".to_owned(), Some("lo")),
("0.0.0.0:8081".to_owned(), None),
]
);
assert_eq!(config.https_listeners[0].interface.as_deref(), Some("wg0"));
assert_eq!(config.tcp_listeners[0].interface.as_deref(), Some("wg0"));
assert_eq!(config.udp_listeners[0].interface.as_deref(), Some("wg0"));
let activations: Vec<Option<String>> = config
.generate_config_messages()
.expect("generate the config messages")
.into_iter()
.filter_map(|request| match request.content.request_type {
Some(RequestType::ActivateListener(activate)) => Some(activate.interface),
_ => None,
})
.collect();
assert_eq!(
activations,
vec![
Some("wg0".to_owned()),
Some("lo".to_owned()),
None,
Some("wg0".to_owned()),
Some("wg0".to_owned()),
Some("wg0".to_owned()),
]
);
let duplicate = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
[[listeners]]
protocol = "http"
address = "0.0.0.0:8080"
interface = "wg0"
[[listeners]]
protocol = "http"
address = "0.0.0.0:8080"
interface = "wg0"
"#;
let file_config: FileConfig = toml::from_str(duplicate).expect("parse");
assert!(
ConfigBuilder::new(file_config, "/tmp/test_config.toml")
.into_config()
.is_err(),
"the same (address, interface) twice must be refused"
);
}
#[test]
fn invalid_listener_interface_rejected_at_config_load() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
[[listeners]]
protocol = "http"
address = "0.0.0.0:8080"
interface = "wg/0"
"#;
let file_config: FileConfig = toml::from_str(toml_content).expect("parse");
let error = ConfigBuilder::new(file_config, "/tmp/test_config.toml")
.into_config()
.expect_err("an invalid interface name must be refused");
let message = error.to_string();
assert!(message.contains("0.0.0.0:8080"), "{message}");
assert!(message.contains("wg/0"), "{message}");
}
#[test]
fn duplicate_listener_address_rejected() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
[[listeners]]
protocol = "http"
address = "0.0.0.0:80"
[[listeners]]
protocol = "http"
address = "0.0.0.0:80"
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let result = ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config();
assert!(
result.is_err(),
"Should reject duplicate listener addresses"
);
}
#[test]
fn buffer_size_below_h2_minimum_rejected() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
buffer_size = 8192
[[listeners]]
protocol = "https"
address = "127.0.0.1:8443"
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let result = ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config();
match result {
Err(ConfigError::BufferSizeTooSmallForH2 {
buffer_size: 8192,
minimum: 16_393,
listeners: 1,
}) => {}
other => panic!("expected BufferSizeTooSmallForH2, got {other:?}"),
}
}
#[test]
fn buffer_size_below_h2_minimum_accepted_when_no_h2_listener() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
buffer_size = 8192
[[listeners]]
protocol = "https"
address = "127.0.0.1:8443"
alpn_protocols = ["http/1.1"]
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let result = ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config();
assert!(
result.is_ok(),
"non-H2 HTTPS listener with sub-16393 buffer should be accepted: {result:?}"
);
}
#[test]
fn buffer_size_below_h2_minimum_rejected_for_an_h2c_http_listener() {
let parse = |h2c: bool| {
let toml_content = format!(
r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
buffer_size = 8192
[[listeners]]
protocol = "http"
address = "127.0.0.1:8080"
h2c_prior_knowledge = {h2c}
"#
);
let file_config: FileConfig =
toml::from_str(&toml_content).expect("Could not parse TOML config");
ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config()
};
match parse(true) {
Err(ConfigError::BufferSizeTooSmallForH2 {
buffer_size: 8192,
minimum: 16_393,
listeners: 1,
}) => {}
other => panic!("expected BufferSizeTooSmallForH2, got {other:?}"),
}
let accepted = parse(false);
assert!(
accepted.is_ok(),
"an HTTP/1-only HTTP listener with a sub-16393 buffer must be accepted: {accepted:?}"
);
}
fn command_socket_mode_of(mode_line: &str) -> Result<Config, ConfigError> {
let toml_content = format!(
r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
{mode_line}
"#
);
let file_config: FileConfig =
toml::from_str(&toml_content).expect("Could not parse TOML config");
ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config()
}
#[test]
fn command_socket_mode_defaults_to_0600() {
let config = command_socket_mode_of("").expect("config without the key must load");
assert_eq!(config.command_socket_mode, 0o600);
assert_eq!(DEFAULT_COMMAND_SOCKET_MODE, 0o600);
}
#[test]
fn command_socket_mode_accepts_octal_strings() {
for (value, expected) in [
("0660", 0o660),
("660", 0o660),
("0600", 0o600),
("0666", 0o666),
("777", 0o777),
("0777", 0o777),
("600", 0o600),
("0640", 0o640),
] {
let config = command_socket_mode_of(&format!("command_socket_mode = \"{value}\""))
.unwrap_or_else(|e| panic!("{value:?} must load: {e}"));
assert_eq!(config.command_socket_mode, expected, "{value:?}");
}
}
#[test]
fn command_socket_mode_rejects_invalid_values() {
for value in [
"", "8", "0668", "abc", "0o660", "0x1b6", "-660", "+660", " 660", "660 ", "1660",
"2770", "4755", "7777", "01000", "00660", "0", "0000", "0066", "0200", "0400", "0466",
] {
match command_socket_mode_of(&format!("command_socket_mode = \"{value}\"")) {
Err(ConfigError::InvalidCommandSocketMode {
value: rejected, ..
}) => {
assert_eq!(rejected, value)
}
other => panic!("{value:?} must be rejected, got {other:?}"),
}
}
}
#[test]
fn command_socket_mode_rejects_a_toml_integer() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("config.toml");
std::fs::write(
&path,
"command_socket = \"/tmp/sozu_test.sock\"\ncommand_socket_mode = 660\n",
)
.expect("write config");
match FileConfig::load_from_path(path.to_str().expect("utf-8 path")) {
Err(ConfigError::DeserializeToml(message)) => assert!(
message.contains("config.toml") && message.contains(" at bytes "),
"the diagnostic keeps the path and the byte range: {message}"
),
other => panic!("an integer mode must fail as DeserializeToml, got {other:?}"),
}
}
#[test]
fn config_without_command_socket_mode_deserializes_to_0600() {
let mut encoded = serde_json::to_value(Config::default()).expect("serialize Config");
encoded
.as_object_mut()
.expect("Config serializes to an object")
.remove("command_socket_mode")
.expect("Config serializes command_socket_mode");
let decoded: Config = serde_json::from_value(encoded).expect("deserialize Config");
assert_eq!(decoded.command_socket_mode, 0o600);
}
#[test]
fn command_buffer_size_exceeds_max_rejected() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
command_buffer_size = 2000000
max_command_buffer_size = 1000000
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let result = ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config();
match result {
Err(ConfigError::CommandBufferSizeExceedsMax {
command_buffer_size: 2_000_000,
max_command_buffer_size: 1_000_000,
}) => {}
other => panic!("expected CommandBufferSizeExceedsMax, got {other:?}"),
}
}
#[test]
fn command_buffer_size_equal_to_max_accepted() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
command_buffer_size = 1000000
max_command_buffer_size = 1000000
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let result = ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config();
assert!(
result.is_ok(),
"command_buffer_size == max_command_buffer_size should be accepted: {result:?}"
);
}
#[test]
fn buffer_size_at_h2_minimum_accepted() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
buffer_size = 16393
[[listeners]]
protocol = "https"
address = "127.0.0.1:8443"
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let result = ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config();
assert!(
result.is_ok(),
"buffer_size at the H2 minimum should be accepted: {result:?}"
);
}
#[test]
fn alpn_protocols_default() {
let mut builder = ListenerBuilder::new_https(SocketAddress::new_v4(127, 0, 0, 1, 8443));
let config = builder.to_tls(None).expect("to_tls should succeed");
assert_eq!(config.alpn_protocols, vec!["h2", "http/1.1"]);
}
#[test]
fn alpn_protocols_custom() {
let mut builder = ListenerBuilder::new_https(SocketAddress::new_v4(127, 0, 0, 1, 8443));
builder.with_alpn_protocols(Some(vec!["http/1.1".to_owned()]));
let config = builder.to_tls(None).expect("to_tls should succeed");
assert_eq!(config.alpn_protocols, vec!["http/1.1"]);
}
#[test]
fn alpn_protocols_invalid_rejected() {
let mut builder = ListenerBuilder::new_https(SocketAddress::new_v4(127, 0, 0, 1, 8443));
builder.with_alpn_protocols(Some(vec!["h3".to_owned()]));
let result = builder.to_tls(None);
assert!(result.is_err());
let err = result.unwrap_err();
assert!(
err.to_string().contains("h3"),
"error should mention the invalid protocol: {err}"
);
}
#[test]
fn alpn_protocols_empty_uses_default() {
let mut builder = ListenerBuilder::new_https(SocketAddress::new_v4(127, 0, 0, 1, 8443));
builder.with_alpn_protocols(Some(vec![]));
let config = builder.to_tls(None).expect("to_tls should succeed");
assert_eq!(config.alpn_protocols, vec!["h2", "http/1.1"]);
}
#[test]
fn alpn_protocols_deduplicated() {
let mut builder = ListenerBuilder::new_https(SocketAddress::new_v4(127, 0, 0, 1, 8443));
builder.with_alpn_protocols(Some(vec![
"h2".to_owned(),
"h2".to_owned(),
"http/1.1".to_owned(),
]));
let config = builder.to_tls(None).expect("to_tls should succeed");
assert_eq!(config.alpn_protocols, vec!["h2", "http/1.1"]);
}
#[test]
fn alpn_protocols_order_preserved() {
let mut builder = ListenerBuilder::new_https(SocketAddress::new_v4(127, 0, 0, 1, 8443));
builder.with_alpn_protocols(Some(vec!["http/1.1".to_owned(), "h2".to_owned()]));
let config = builder.to_tls(None).expect("to_tls should succeed");
assert_eq!(config.alpn_protocols, vec!["http/1.1", "h2"]);
}
#[test]
fn parse_header_edit_rejects_crlf_in_value() {
let entry = HeaderEditConfig {
position: "request".to_owned(),
key: "X-Test".to_owned(),
value: "value\r\nEvil-Header: stolen".to_owned(),
};
let err = parse_header_edit(0, &entry).expect_err("CRLF in value must be rejected");
match err {
ConfigError::InvalidHeaderBytes { index, field } => {
assert_eq!(index, 0);
assert_eq!(field, "value");
}
other => panic!("expected InvalidHeaderBytes, got {other:?}"),
}
}
#[test]
fn parse_header_edit_rejects_lf_in_key() {
let entry = HeaderEditConfig {
position: "response".to_owned(),
key: "X-\nTest".to_owned(),
value: "ok".to_owned(),
};
let err = parse_header_edit(2, &entry).expect_err("LF in key must be rejected");
match err {
ConfigError::InvalidHeaderBytes { index, field } => {
assert_eq!(index, 2);
assert_eq!(field, "key");
}
other => panic!("expected InvalidHeaderBytes, got {other:?}"),
}
}
#[test]
fn parse_header_edit_rejects_nul() {
let entry = HeaderEditConfig {
position: "both".to_owned(),
key: "X-Test".to_owned(),
value: "with\0nul".to_owned(),
};
assert!(matches!(
parse_header_edit(0, &entry),
Err(ConfigError::InvalidHeaderBytes { .. })
));
}
#[test]
fn parse_header_edit_accepts_tab_in_value() {
let entry = HeaderEditConfig {
position: "request".to_owned(),
key: "X-Test".to_owned(),
value: "with\ttab".to_owned(),
};
let header = parse_header_edit(0, &entry).expect("tab in value must be accepted");
assert_eq!(header.val, "with\ttab");
}
#[test]
fn parse_header_edit_rejects_tab_in_key() {
let entry = HeaderEditConfig {
position: "request".to_owned(),
key: "Host\t".to_owned(),
value: "ok".to_owned(),
};
let err = parse_header_edit(0, &entry).expect_err("HTAB in key must be rejected");
match err {
ConfigError::InvalidHeaderBytes { field, .. } => assert_eq!(field, "key"),
other => panic!("expected InvalidHeaderBytes{{field=\"key\"}}, got {other:?}"),
}
}
#[test]
fn parse_header_edit_rejects_space_in_key() {
let entry = HeaderEditConfig {
position: "request".to_owned(),
key: "X Test".to_owned(),
value: "ok".to_owned(),
};
let err = parse_header_edit(0, &entry).expect_err("SP in key must be rejected");
assert!(matches!(err, ConfigError::InvalidHeaderBytes { .. }));
}
#[test]
fn parse_header_edit_rejects_empty_key() {
let entry = HeaderEditConfig {
position: "request".to_owned(),
key: String::new(),
value: "ok".to_owned(),
};
let err = parse_header_edit(0, &entry).expect_err("empty key must be rejected");
assert!(matches!(
err,
ConfigError::InvalidHeaderBytes { field: "key", .. }
));
}
#[test]
fn parse_header_edit_accepts_clean_value() {
let entry = HeaderEditConfig {
position: "request".to_owned(),
key: "X-Tenant".to_owned(),
value: "alpha".to_owned(),
};
let header = parse_header_edit(0, &entry).expect("clean value must be accepted");
assert_eq!(header.key, "X-Tenant");
assert_eq!(header.val, "alpha");
}
#[test]
fn resolve_answer_source_bare_string_is_literal() {
let body = resolve_answer_source("HTTP/1.1 503 Service Unavailable\r\n\r\nbusy")
.expect("bare-string source must resolve");
assert_eq!(body, "HTTP/1.1 503 Service Unavailable\r\n\r\nbusy");
}
#[test]
fn resolve_answer_source_empty_string_is_legitimate() {
let body = resolve_answer_source("").expect("empty source must resolve");
assert_eq!(body, "");
}
#[test]
fn resolve_answer_source_file_scheme_missing_file_errors() {
let err = resolve_answer_source("file:///nonexistent/sozu-test/never.http")
.expect_err("missing path must error");
assert!(matches!(err, ConfigError::FileOpen { .. }));
}
#[test]
fn resolve_answer_source_file_scheme_empty_path_errors() {
let err = resolve_answer_source("file://").expect_err("empty path must error");
assert!(matches!(err, ConfigError::FileOpen { .. }));
}
#[test]
fn legacy_tcp_frontend_toml_without_sni_alpn_parses_unchanged() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
[[listeners]]
protocol = "tcp"
address = "127.0.0.1:9000"
[clusters.legacy]
protocol = "tcp"
load_balancing = "ROUND_ROBIN"
frontends = [
{ address = "127.0.0.1:9000" }
]
backends = [
{ address = "10.0.0.1:9000" }
]
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse legacy TCP TOML");
let config = ConfigBuilder::new(file_config, "/tmp/test_config.toml")
.into_config()
.expect("legacy TCP config without sni/alpn must load unchanged");
assert_eq!(config.tcp_listeners.len(), 1);
let listener = &config.tcp_listeners[0];
assert_eq!(
listener.sni_preread_timeout,
Some(DEFAULT_SNI_PREREAD_TIMEOUT),
"proto default sni_preread_timeout must be populated even though unused"
);
assert_eq!(
listener.sni_preread_max_bytes,
Some(DEFAULT_SNI_PREREAD_MAX_BYTES),
"proto default sni_preread_max_bytes must be populated even though unused"
);
let messages = config
.generate_config_messages()
.expect("Could not generate config messages");
let tcp_frontend = messages
.iter()
.find_map(|m| match &m.content.request_type {
Some(RequestType::AddTcpFrontend(f)) => Some(f),
_ => None,
})
.expect("AddTcpFrontend must be present");
assert_eq!(tcp_frontend.sni, None, "legacy frontend must carry no sni");
assert!(
tcp_frontend.alpn.is_empty(),
"legacy frontend must carry no alpn"
);
}
#[test]
fn a_catch_all_post_frontend_loads_from_toml_and_rejects_a_mixed_case_position() {
let catch_all = |position: &str| {
format!(
r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
[[listeners]]
protocol = "http"
address = "127.0.0.1:8080"
[clusters.fallback]
protocol = "http"
frontends = [
{{ address = "127.0.0.1:8080", hostname = "*", position = "{position}" }}
]
backends = [
{{ address = "10.0.0.1:8000" }}
]
"#
)
};
let file_config: FileConfig = toml::from_str(&catch_all("POST"))
.expect("an upper-case `position = \"POST\"` must parse");
let config = ConfigBuilder::new(file_config, "/tmp/test_config.toml")
.into_config()
.expect("a catch-all POST frontend must load");
let messages = config
.generate_config_messages()
.expect("Could not generate config messages");
let frontend = messages
.iter()
.find_map(|m| match &m.content.request_type {
Some(RequestType::AddHttpFrontend(f)) => Some(f),
_ => None,
})
.expect("AddHttpFrontend must be present");
assert_eq!(frontend.hostname, "*");
assert_eq!(frontend.position, RulePosition::Post as i32);
let error = toml::from_str::<FileConfig>(&catch_all("Post"))
.expect_err("a mixed-case `position = \"Post\"` must be refused")
.to_string();
assert!(
error.contains("unknown variant `Post`, expected one of `PRE`, `POST`, `TREE`"),
"unexpected error: {error}",
);
}
#[test]
fn forwarded_headers_parses_every_mode_defaults_to_both_and_rejects_others() {
let listeners = |value: Option<&str>| {
let line = value
.map(|value| format!("forwarded_headers = \"{value}\""))
.unwrap_or_default();
format!(
r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
[[listeners]]
protocol = "http"
address = "127.0.0.1:8080"
{line}
[[listeners]]
protocol = "https"
address = "127.0.0.1:8443"
{line}
"#
)
};
let load = |value: Option<&str>| {
let file_config: FileConfig =
toml::from_str(&listeners(value)).expect("the listeners must parse");
let config = ConfigBuilder::new(file_config, "/tmp/test_config.toml")
.into_config()
.expect("the listeners must load");
(
config.http_listeners[0].forwarded_headers,
config.https_listeners[0].forwarded_headers,
)
};
for (spelling, expected) in [
(None, ForwardedHeaders::Both),
(Some("both"), ForwardedHeaders::Both),
(Some("x_forwarded"), ForwardedHeaders::XForwarded),
(Some("rfc7239"), ForwardedHeaders::Rfc7239),
(Some("none"), ForwardedHeaders::None),
] {
let expected = Some(expected as i32);
assert_eq!(load(spelling), (expected, expected), "{spelling:?}");
}
for invalid in ["RFC7239", "x-forwarded", "forwarded", ""] {
let error = toml::from_str::<FileConfig>(&listeners(Some(invalid)))
.expect_err("an unknown forwarded_headers value must be refused")
.to_string();
assert!(
error.contains(&format!(
"unknown variant `{invalid}`, expected one of `both`, `x_forwarded`, `rfc7239`, `none`"
)),
"{invalid:?}: unexpected error: {error}",
);
}
}
#[test]
fn tcp_frontend_hostname_maps_to_sni_exact_and_wildcard() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
[[listeners]]
protocol = "tcp"
address = "127.0.0.1:9010"
[clusters.exact]
protocol = "tcp"
load_balancing = "ROUND_ROBIN"
frontends = [
{ address = "127.0.0.1:9010", hostname = "example.com", alpn = ["h2"] }
]
backends = [ { address = "10.0.0.1:9010" } ]
[clusters.wildcard]
protocol = "tcp"
load_balancing = "ROUND_ROBIN"
frontends = [
{ address = "127.0.0.1:9010", hostname = "*.example.com" }
]
backends = [ { address = "10.0.0.2:9010" } ]
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let config = ConfigBuilder::new(file_config, "/tmp/test_config.toml")
.into_config()
.expect("exact + wildcard SNI frontends on distinct sni must load");
let messages = config
.generate_config_messages()
.expect("Could not generate config messages");
let mut frontends: Vec<_> = messages
.iter()
.filter_map(|m| match &m.content.request_type {
Some(RequestType::AddTcpFrontend(f)) => Some(f.clone()),
_ => None,
})
.collect();
frontends.sort_by(|a, b| a.cluster_id.cmp(&b.cluster_id));
assert_eq!(frontends.len(), 2);
assert_eq!(frontends[0].sni, Some("example.com".to_string()));
assert_eq!(frontends[0].alpn, vec!["h2".to_string()]);
assert_eq!(frontends[1].sni, Some("*.example.com".to_string()));
assert!(frontends[1].alpn.is_empty());
}
#[test]
fn tcp_frontend_invalid_sni_pattern_rejected() {
for invalid in [
"*.*.example.com",
"foo.*.com",
"*",
"example..com",
"",
"/[a-z]+/.example.com",
"foo/bar.example.com",
] {
let frontend = FileClusterFrontendConfig {
address: "127.0.0.1:8080".parse().unwrap(),
hostname: Some(invalid.to_string()),
alpn: vec![],
path: None,
path_type: None,
method: None,
certificate: None,
key: None,
certificate_chain: None,
tls_versions: vec![],
position: RulePosition::Tree,
tags: None,
redirect: None,
redirect_scheme: None,
redirect_template: None,
rewrite_host: None,
rewrite_path: None,
rewrite_port: None,
required_auth: None,
headers: None,
hsts: None,
header_predicates: None,
};
match frontend.to_tcp_front() {
Err(ConfigError::InvalidSniPattern { sni }) => assert_eq!(sni, invalid),
other => panic!("expected InvalidSniPattern for {invalid:?}, got {other:?}"),
}
}
}
#[test]
fn tcp_frontend_non_ascii_sni_pattern_rejected() {
for non_ascii in ["münchen.example", "*.bücher.example", "日本.example"] {
let frontend = FileClusterFrontendConfig {
address: "127.0.0.1:8080".parse().unwrap(),
hostname: Some(non_ascii.to_string()),
alpn: vec![],
path: None,
path_type: None,
method: None,
certificate: None,
key: None,
certificate_chain: None,
tls_versions: vec![],
position: RulePosition::Tree,
tags: None,
redirect: None,
redirect_scheme: None,
redirect_template: None,
rewrite_host: None,
rewrite_path: None,
rewrite_port: None,
required_auth: None,
headers: None,
hsts: None,
header_predicates: None,
};
match frontend.to_tcp_front() {
Err(ConfigError::NonAsciiSniPattern { sni }) => assert_eq!(sni, non_ascii),
other => panic!("expected NonAsciiSniPattern for {non_ascii:?}, got {other:?}"),
}
}
}
#[test]
fn tcp_frontend_punycode_sni_pattern_accepted() {
assert_eq!(
validate_sni_pattern("xn--mnchen-3ya.example").expect("A-label must be accepted"),
"xn--mnchen-3ya.example"
);
assert_eq!(
validate_sni_pattern("*.xn--bcher-kva.example")
.expect("wildcarded A-label must be accepted"),
"*.xn--bcher-kva.example"
);
assert_eq!(
validate_sni_pattern("XN--MNCHEN-3YA.Example")
.expect("mixed-case A-label must be accepted"),
"xn--mnchen-3ya.example",
"A-label patterns are ASCII-lowercased like any other pattern"
);
}
#[test]
fn alpn_rejected_on_http_frontend() {
let frontend = FileClusterFrontendConfig {
address: "127.0.0.1:8080".parse().unwrap(),
hostname: Some("example.com".to_owned()),
alpn: vec!["h2".to_string()],
path: None,
path_type: None,
method: None,
certificate: None,
key: None,
certificate_chain: None,
tls_versions: vec![],
position: RulePosition::Tree,
tags: None,
redirect: None,
redirect_scheme: None,
redirect_template: None,
rewrite_host: None,
rewrite_path: None,
rewrite_port: None,
required_auth: None,
headers: None,
hsts: None,
header_predicates: None,
};
match frontend.to_http_front("api") {
Err(ConfigError::InvalidFrontendConfig(field)) => assert_eq!(field, "alpn"),
other => panic!("expected InvalidFrontendConfig(\"alpn\"), got {other:?}"),
}
}
#[test]
fn header_predicates_rejected_on_tcp_frontend() {
let frontend = FileClusterFrontendConfig {
address: "127.0.0.1:8080".parse().unwrap(),
hostname: None,
alpn: vec![],
path: None,
path_type: None,
method: None,
certificate: None,
key: None,
certificate_chain: None,
tls_versions: vec![],
position: RulePosition::Tree,
tags: None,
redirect: None,
redirect_scheme: None,
redirect_template: None,
rewrite_host: None,
rewrite_path: None,
rewrite_port: None,
required_auth: None,
headers: None,
hsts: None,
header_predicates: Some(vec![]),
};
match frontend.to_tcp_front() {
Err(ConfigError::InvalidFrontendConfig(field)) => {
assert_eq!(field, "header_predicates")
}
other => panic!("expected InvalidFrontendConfig(\"header_predicates\"), got {other:?}"),
}
}
#[test]
fn tcp_frontend_alpn_without_sni_rejected() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
[[listeners]]
protocol = "tcp"
address = "127.0.0.1:9019"
[clusters.a]
protocol = "tcp"
load_balancing = "ROUND_ROBIN"
frontends = [
{ address = "127.0.0.1:9019", alpn = ["h2"] }
]
backends = [ { address = "10.0.0.1:9019" } ]
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let result = ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config();
match result {
Err(ConfigError::AlpnWithoutSni { address }) => {
assert_eq!(address.to_string(), "127.0.0.1:9019");
}
other => panic!("expected AlpnWithoutSni, got {other:?}"),
}
}
#[test]
fn tcp_frontend_alpn_overlap_rejected() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
[[listeners]]
protocol = "tcp"
address = "127.0.0.1:9020"
[clusters.a]
protocol = "tcp"
load_balancing = "ROUND_ROBIN"
frontends = [
{ address = "127.0.0.1:9020", hostname = "example.com", alpn = ["h2"] }
]
backends = [ { address = "10.0.0.1:9020" } ]
[clusters.b]
protocol = "tcp"
load_balancing = "ROUND_ROBIN"
frontends = [
{ address = "127.0.0.1:9020", hostname = "example.com", alpn = ["h2", "http/1.1"] }
]
backends = [ { address = "10.0.0.2:9020" } ]
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let result = ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config();
match result {
Err(ConfigError::TcpFrontendAlpnOverlap { protocol, .. }) => {
assert_eq!(protocol, "h2");
}
other => panic!("expected TcpFrontendAlpnOverlap, got {other:?}"),
}
}
#[test]
fn tcp_frontend_multiple_alpn_catch_all_rejected() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
[[listeners]]
protocol = "tcp"
address = "127.0.0.1:9021"
[clusters.a]
protocol = "tcp"
load_balancing = "ROUND_ROBIN"
frontends = [
{ address = "127.0.0.1:9021", hostname = "example.com" }
]
backends = [ { address = "10.0.0.1:9021" } ]
[clusters.b]
protocol = "tcp"
load_balancing = "ROUND_ROBIN"
frontends = [
{ address = "127.0.0.1:9021", hostname = "example.com" }
]
backends = [ { address = "10.0.0.2:9021" } ]
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let result = ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config();
assert!(
matches!(
result,
Err(ConfigError::TcpFrontendMultipleAlpnCatchAll { .. })
),
"expected TcpFrontendMultipleAlpnCatchAll, got {result:?}"
);
}
#[test]
fn tcp_listener_mixes_sni_and_no_sni_rejected() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
[[listeners]]
protocol = "tcp"
address = "127.0.0.1:9022"
[clusters.a]
protocol = "tcp"
load_balancing = "ROUND_ROBIN"
frontends = [
{ address = "127.0.0.1:9022", hostname = "example.com" }
]
backends = [ { address = "10.0.0.1:9022" } ]
[clusters.b]
protocol = "tcp"
load_balancing = "ROUND_ROBIN"
frontends = [
{ address = "127.0.0.1:9022" }
]
backends = [ { address = "10.0.0.2:9022" } ]
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let result = ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config();
assert!(
matches!(result, Err(ConfigError::TcpListenerMixesSniAndNoSni { .. })),
"expected TcpListenerMixesSniAndNoSni, got {result:?}"
);
}
#[test]
fn sni_preread_timeout_exceeding_front_timeout_rejected() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
[[listeners]]
protocol = "tcp"
address = "127.0.0.1:9030"
front_timeout = 2
[clusters.a]
protocol = "tcp"
load_balancing = "ROUND_ROBIN"
frontends = [
{ address = "127.0.0.1:9030", hostname = "example.com" }
]
backends = [ { address = "10.0.0.1:9030" } ]
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let result = ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config();
match result {
Err(ConfigError::SniPrereadTimeoutExceedsFrontTimeout {
sni_preread_timeout: 5,
front_timeout: 2,
..
}) => {}
other => panic!("expected SniPrereadTimeoutExceedsFrontTimeout, got {other:?}"),
}
}
#[test]
fn sni_preread_max_bytes_exceeding_buffer_size_rejected() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
buffer_size = 8192
[[listeners]]
protocol = "tcp"
address = "127.0.0.1:9031"
[clusters.a]
protocol = "tcp"
load_balancing = "ROUND_ROBIN"
frontends = [
{ address = "127.0.0.1:9031", hostname = "example.com" }
]
backends = [ { address = "10.0.0.1:9031" } ]
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let result = ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config();
match result {
Err(ConfigError::SniPrereadMaxBytesExceedsBufferSize {
sni_preread_max_bytes: 16384,
buffer_size: 8192,
..
}) => {}
other => panic!("expected SniPrereadMaxBytesExceedsBufferSize, got {other:?}"),
}
}
#[test]
fn sni_preread_max_bytes_zero_rejected() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
[[listeners]]
protocol = "tcp"
address = "127.0.0.1:9033"
sni_preread_max_bytes = 0
[clusters.a]
protocol = "tcp"
load_balancing = "ROUND_ROBIN"
frontends = [
{ address = "127.0.0.1:9033", hostname = "example.com" }
]
backends = [ { address = "10.0.0.1:9033" } ]
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let result = ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config();
match result {
Err(ConfigError::SniPrereadMaxBytesTooSmall {
sni_preread_max_bytes: 0,
minimum: 5,
..
}) => {}
other => panic!("expected SniPrereadMaxBytesTooSmall, got {other:?}"),
}
}
#[test]
fn sni_preread_max_bytes_at_the_floor_loads() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
[[listeners]]
protocol = "tcp"
address = "127.0.0.1:9034"
sni_preread_max_bytes = 5
[clusters.a]
protocol = "tcp"
load_balancing = "ROUND_ROBIN"
frontends = [
{ address = "127.0.0.1:9034", hostname = "example.com" }
]
backends = [ { address = "10.0.0.1:9034" } ]
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let result = ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config();
assert!(
result.is_ok(),
"sni_preread_max_bytes at the exact floor must load: {result:?}"
);
}
#[test]
fn sni_preread_validation_ignored_without_sni_frontend() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
buffer_size = 8192
[[listeners]]
protocol = "tcp"
address = "127.0.0.1:9032"
front_timeout = 2
[clusters.a]
protocol = "tcp"
load_balancing = "ROUND_ROBIN"
frontends = [
{ address = "127.0.0.1:9032" }
]
backends = [ { address = "10.0.0.1:9032" } ]
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let result = ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config();
assert!(
result.is_ok(),
"a no-SNI TCP listener must ignore sni_preread validation entirely: {result:?}"
);
}
#[test]
fn tcp_frontend_disjoint_alpn_same_sni_both_load() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
[[listeners]]
protocol = "tcp"
address = "127.0.0.1:9040"
[clusters.h2_cluster]
protocol = "tcp"
load_balancing = "ROUND_ROBIN"
frontends = [
{ address = "127.0.0.1:9040", hostname = "example.com", alpn = ["h2"] }
]
backends = [ { address = "10.0.0.1:9040" } ]
[clusters.http11_cluster]
protocol = "tcp"
load_balancing = "ROUND_ROBIN"
frontends = [
{ address = "127.0.0.1:9040", hostname = "example.com", alpn = ["http/1.1"] }
]
backends = [ { address = "10.0.0.2:9040" } ]
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let config = ConfigBuilder::new(file_config, "/tmp/test_config.toml")
.into_config()
.expect("disjoint non-empty ALPN lists on the same (address, sni) must load");
let messages = config
.generate_config_messages()
.expect("Could not generate config messages");
let mut frontends: Vec<_> = messages
.iter()
.filter_map(|m| match &m.content.request_type {
Some(RequestType::AddTcpFrontend(f)) => Some(f.clone()),
_ => None,
})
.collect();
frontends.sort_by(|a, b| a.cluster_id.cmp(&b.cluster_id));
assert_eq!(frontends.len(), 2, "both frontends must be emitted");
assert_eq!(frontends[0].cluster_id, "h2_cluster");
assert_eq!(frontends[0].sni, Some("example.com".to_string()));
assert_eq!(frontends[0].alpn, vec!["h2".to_string()]);
assert_eq!(frontends[1].cluster_id, "http11_cluster");
assert_eq!(frontends[1].sni, Some("example.com".to_string()));
assert_eq!(frontends[1].alpn, vec!["http/1.1".to_string()]);
}
#[test]
fn udp_routed_tcp_frontends_are_excluded_from_sni_invariants() {
let toml_content = r#"
command_socket = "/tmp/sozu_test.sock"
worker_count = 1
[[listeners]]
protocol = "udp"
address = "127.0.0.1:9050"
[clusters.a]
protocol = "tcp"
load_balancing = "ROUND_ROBIN"
frontends = [
{ address = "127.0.0.1:9050", hostname = "example.com" }
]
backends = [ { address = "10.0.0.1:9050" } ]
[clusters.b]
protocol = "tcp"
load_balancing = "ROUND_ROBIN"
frontends = [
{ address = "127.0.0.1:9050" }
]
backends = [ { address = "10.0.0.2:9050" } ]
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let result = ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config();
assert!(
result.is_ok(),
"UDP-routed tcp-cluster frontends must not trip the SNI mixing-ban: {result:?}"
);
let config = result.expect("checked is_ok above");
let messages = config
.generate_config_messages()
.expect("Could not generate config messages");
let add_udp_frontend_count = messages
.iter()
.filter(|m| matches!(m.content.request_type, Some(RequestType::AddUdpFrontend(_))))
.count();
assert_eq!(
add_udp_frontend_count, 2,
"both cluster frontends on the udp listener must emit AddUdpFrontend"
);
}
#[test]
fn subnet_connection_limit_keys_round_trip_from_toml() {
let toml_content = r#"
command_socket = "/tmp/sozu.sock"
saved_state = "./state.json"
worker_count = 1
max_connections_per_ip = 10
max_connections_per_subnet = 100
subnet_ipv4_prefix = 24
subnet_ipv6_prefix = 56
[[listeners]]
protocol = "http"
address = "127.0.0.1:8080"
[clusters.api]
protocol = "http"
load_balancing = "ROUND_ROBIN"
max_connections_per_subnet = 7
frontends = [
{ address = "127.0.0.1:8080", hostname = "example.com" }
]
backends = [ { address = "10.0.0.1:8080" } ]
"#;
let file_config: FileConfig =
toml::from_str(toml_content).expect("Could not parse TOML config");
let config = ConfigBuilder::new(file_config, "/tmp/test_config.toml")
.into_config()
.expect("a config with valid subnet prefixes must load");
assert_eq!(
config.max_connections_per_subnet, 100,
"the global subnet cap must survive the TOML round trip"
);
assert_eq!(
config.subnet_ipv4_prefix, 24,
"subnet_ipv4_prefix must survive the TOML round trip"
);
assert_eq!(
config.subnet_ipv6_prefix, 56,
"subnet_ipv6_prefix must survive the TOML round trip — a swap with the IPv6 \
key would leave a cap that silently never fires"
);
assert_eq!(
config.max_connections_per_ip, 10,
"the pre-existing per-IP cap must be unaffected: the two are independent"
);
let cluster = config
.clusters
.get("api")
.expect("the staged cluster must be present");
let override_value = match cluster {
ClusterConfig::Http(http) => http.max_connections_per_subnet,
other => panic!("expected an HTTP cluster, got {other:?}"),
};
assert_eq!(
override_value,
Some(7),
"the per-cluster subnet override must reach the cluster config"
);
}
#[test]
fn max_connection_attempts_defaults_to_five_and_round_trips_from_toml() {
let config_with = |global: &str, cluster: &str| -> Config {
let toml_content = format!(
r#"
command_socket = "/tmp/sozu.sock"
saved_state = "./state.json"
worker_count = 1
{global}
[[listeners]]
protocol = "http"
address = "127.0.0.1:8080"
[clusters.api]
protocol = "http"
load_balancing = "ROUND_ROBIN"
{cluster}
frontends = [
{{ address = "127.0.0.1:8080", hostname = "example.com" }}
]
backends = [ {{ address = "10.0.0.1:8080" }} ]
"#
);
let file_config: FileConfig =
toml::from_str(&toml_content).expect("Could not parse TOML config");
ConfigBuilder::new(file_config, "/tmp/test_config.toml")
.into_config()
.expect("a valid attempt budget must load")
};
let cluster_attempts = |config: &Config| match config.clusters.get("api") {
Some(ClusterConfig::Http(http)) => http.max_connection_attempts,
other => panic!("expected the HTTP cluster, got {other:?}"),
};
let add_cluster_attempts = |config: &Config| {
config
.generate_config_messages()
.expect("Could not generate config messages")
.into_iter()
.find_map(|message| match message.content.request_type {
Some(RequestType::AddCluster(cluster)) => Some(cluster.max_connection_attempts),
_ => None,
})
.expect("the cluster is added")
};
let defaults = config_with("", "");
assert_eq!(DEFAULT_MAX_CONNECTION_ATTEMPTS, 5);
assert_eq!(defaults.max_connection_attempts, 5);
assert_eq!(
ServerConfig::from(&defaults).max_connection_attempts,
Some(5),
"the worker receives the default"
);
assert_eq!(
cluster_attempts(&defaults),
None,
"a cluster inherits by default"
);
assert_eq!(add_cluster_attempts(&defaults), None);
let set = config_with("max_connection_attempts = 2", "max_connection_attempts = 7");
assert_eq!(set.max_connection_attempts, 2);
assert_eq!(ServerConfig::from(&set).max_connection_attempts, Some(2));
assert_eq!(cluster_attempts(&set), Some(7));
assert_eq!(add_cluster_attempts(&set), Some(7));
}
#[test]
fn an_out_of_range_max_connection_attempts_is_rejected() {
let load = |global: &str, cluster: &str| {
let toml_content = format!(
r#"
command_socket = "/tmp/sozu.sock"
saved_state = "./state.json"
worker_count = 1
{global}
[[listeners]]
protocol = "http"
address = "127.0.0.1:8080"
[clusters.api]
protocol = "http"
load_balancing = "ROUND_ROBIN"
{cluster}
frontends = [
{{ address = "127.0.0.1:8080", hostname = "example.com" }}
]
backends = [ {{ address = "10.0.0.1:8080" }} ]
"#
);
let file_config: FileConfig =
toml::from_str(&toml_content).expect("Could not parse TOML config");
ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config()
};
for value in ["0", "256"] {
let global = format!("max_connection_attempts = {value}");
assert!(
matches!(
load(&global, ""),
Err(ConfigError::InvalidMaxConnectionAttempts {
cluster_id: None,
..
})
),
"a global budget of {value} must be rejected"
);
assert!(
matches!(
load("", &global),
Err(ConfigError::InvalidMaxConnectionAttempts {
cluster_id: Some(_),
..
})
),
"a cluster budget of {value} must be rejected"
);
}
for value in ["1", "255"] {
let budget = format!("max_connection_attempts = {value}");
assert!(load(&budget, &budget).is_ok(), "{value} is in range");
}
}
#[test]
fn an_out_of_range_subnet_prefix_is_rejected_not_clamped() {
let with_prefixes = |v4: &str, v6: &str| -> String {
format!(
r#"
command_socket = "/tmp/sozu.sock"
saved_state = "./state.json"
worker_count = 1
max_connections_per_subnet = 100
subnet_ipv4_prefix = {v4}
subnet_ipv6_prefix = {v6}
[[listeners]]
protocol = "http"
address = "127.0.0.1:8080"
"#
)
};
for (v4, v6, key) in [
("33", "56", "subnet_ipv4_prefix"),
("24", "129", "subnet_ipv6_prefix"),
] {
let file_config: FileConfig =
toml::from_str(&with_prefixes(v4, v6)).expect("Could not parse TOML config");
let result = ConfigBuilder::new(file_config, "/tmp/test_config.toml").into_config();
match result {
Err(ConfigError::InvalidSubnetPrefix { key: reported, .. }) => {
assert_eq!(
reported, key,
"the rejection must name the key that is out of range"
);
}
other => panic!("{key} out of range must be rejected, got {other:?}"),
}
}
let file_config: FileConfig =
toml::from_str(&with_prefixes("32", "128")).expect("Could not parse TOML config");
assert!(
ConfigBuilder::new(file_config, "/tmp/test_config.toml")
.into_config()
.is_ok(),
"the full-width prefixes are legal and are the defaults — they must load"
);
}
#[test]
fn old_state_file_without_client_auth_fields_deserializes() {
let address = SocketAddress::new_v4(127, 0, 0, 1, 8443);
let config = ListenerBuilder::new_https(address)
.to_tls(None)
.expect("default HTTPS listener config");
let mut json: serde_json::Value =
serde_json::to_value(&config).expect("serialize HttpsListenerConfig");
let obj = json.as_object_mut().expect("object");
obj.remove("client_ca_certificates");
obj.remove("client_ca_crls");
obj.remove("client_auth");
let restored: HttpsListenerConfig =
serde_json::from_value(json).expect("old state file must deserialize");
assert!(restored.client_ca_certificates.is_empty());
assert!(restored.client_ca_crls.is_empty());
assert_eq!(restored.client_auth, None);
}
#[test]
fn client_auth_fields_rejected_on_non_https_listener() {
let address = SocketAddress::new_v4(127, 0, 0, 1, 9000);
let mut tcp = ListenerBuilder::new_tcp(address);
tcp.client_auth = Some(ClientAuthConfig::Required);
assert!(matches!(
tcp.to_tcp(None),
Err(ConfigError::ClientAuthOnNonHttps(_))
));
let mut http = ListenerBuilder::new_http(address);
http.client_ca_certificates = Some(vec!["ca.pem".to_owned()]);
assert!(matches!(
http.to_http(None),
Err(ConfigError::ClientAuthOnNonHttps(_))
));
}
#[test]
fn none_mode_ignores_stale_ca_paths() {
let address = SocketAddress::new_v4(127, 0, 0, 1, 9443);
let mut https = ListenerBuilder::new_https(address);
https.client_auth = Some(ClientAuthConfig::None);
https.client_ca_certificates = Some(vec!["/nonexistent/ca.pem".to_owned()]);
https.client_ca_crls = Some(vec!["/nonexistent/crl.pem".to_owned()]);
let config = https.to_tls(None).expect("NONE mode must ignore CA paths");
assert!(config.client_ca_certificates.is_empty());
assert!(config.client_ca_crls.is_empty());
}
#[test]
fn client_auth_listener_inlines_its_ca_and_crl_files_into_the_listener_request() {
let asset =
|name: &str| format!("{}/../lib/assets/mtls/{name}", env!("CARGO_MANIFEST_DIR"));
let (ca_path, crl_path) = (asset("ca-cert.pem"), asset("crl-revoked.pem"));
let toml_content = format!(
r#"
command_socket = "/tmp/sozu.sock"
saved_state = "./state.json"
worker_count = 1
[[listeners]]
protocol = "https"
address = "127.0.0.1:8443"
client_auth = "required"
client_ca_certificates = ["{ca_path}"]
client_ca_crls = ["{crl_path}"]
"#
);
let file_config: FileConfig =
toml::from_str(&toml_content).expect("Could not parse TOML config");
let config = ConfigBuilder::new(file_config, "/tmp/test_config.toml")
.into_config()
.expect("a client auth listener with readable CA and CRL files must load");
let listeners: Vec<HttpsListenerConfig> = config
.generate_config_messages()
.expect("the config must produce its worker requests")
.into_iter()
.filter_map(|request| match request.content.request_type {
Some(RequestType::AddHttpsListenerWithClientAuth(listener)) => Some(listener),
Some(RequestType::AddHttpsListener(_)) => {
panic!("a client auth listener must not go out as AddHttpsListener")
}
_ => None,
})
.collect();
assert_eq!(listeners.len(), 1, "one HTTPS listener is configured");
let listener = &listeners[0];
assert_eq!(
listener.client_auth,
Some(ClientAuthMode::ClientAuthRequired as i32)
);
assert_eq!(
listener.client_ca_certificates,
vec![std::fs::read_to_string(&ca_path).expect("read the CA fixture")],
"the CA file content, not its path, must reach the worker"
);
assert_eq!(
listener.client_ca_crls,
vec![std::fs::read_to_string(&crl_path).expect("read the CRL fixture")],
"the CRL file content, not its path, must reach the worker"
);
assert!(
listener.client_ca_crls[0].contains("BEGIN X509 CRL"),
"the inlined CRL is the PEM body"
);
}
#[test]
fn client_auth_listener_with_an_unreadable_crl_file_does_not_load() {
let address = SocketAddress::new_v4(127, 0, 0, 1, 9443);
let mut https = ListenerBuilder::new_https(address);
https.client_auth = Some(ClientAuthConfig::Required);
https.client_ca_certificates = Some(vec![format!(
"{}/../lib/assets/mtls/ca-cert.pem",
env!("CARGO_MANIFEST_DIR")
)]);
https.client_ca_crls = Some(vec!["/nonexistent/crl.pem".to_owned()]);
assert!(
https.to_tls(None).is_err(),
"an unreadable CRL file must fail the listener, not drop the revocation check"
);
}
#[test]
fn client_auth_files_without_a_mode_are_rejected() {
let address = SocketAddress::new_v4(127, 0, 0, 1, 9443);
for (cas, crls) in [
(Some(vec!["/etc/sozu/client-ca.pem".to_owned()]), None),
(None, Some(vec!["/etc/sozu/client-ca.crl.pem".to_owned()])),
] {
let mut https = ListenerBuilder::new_https(address);
https.client_ca_certificates = cas;
https.client_ca_crls = crls;
assert!(matches!(
https.to_tls(None),
Err(ConfigError::ClientAuthModeMissing(_))
));
}
let mut explicit_none = ListenerBuilder::new_https(address);
explicit_none.client_auth = Some(ClientAuthConfig::None);
explicit_none.client_ca_certificates = Some(vec!["/nonexistent/ca.pem".to_owned()]);
assert!(explicit_none.to_tls(None).is_ok());
assert!(ListenerBuilder::new_https(address).to_tls(None).is_ok());
}
#[test]
fn client_auth_config_parses_documented_lowercase_names() {
assert_eq!(
serde_json::from_str::<ClientAuthConfig>("\"required\"").unwrap(),
ClientAuthConfig::Required
);
assert_eq!(
serde_json::from_str::<ClientAuthConfig>("\"optional\"").unwrap(),
ClientAuthConfig::Optional
);
assert_eq!(
serde_json::from_str::<ClientAuthConfig>("\"none\"").unwrap(),
ClientAuthConfig::None
);
assert_eq!(
ClientAuthMode::from(ClientAuthConfig::Required),
ClientAuthMode::ClientAuthRequired
);
}
}