From 5d22cc431ba8fad7f43dbc08983b2804d09c27b0 Mon Sep 17 00:00:00 2001 From: Maciej Modzelewski Date: Mon, 27 Jul 2026 13:11:10 +0200 Subject: [PATCH 1/4] feat(server-ng): add cluster.nodes.advertised_address for clients A node's `ip` is the address it binds for replica traffic, which in Docker, Kubernetes, or NAT deployments is private. Both client-facing surfaces reuse it: the cluster roster returned to clients and the follower's redirect to the metadata primary. Clients outside the replica network get an address they cannot reach. Nodes now take an optional `advertised_address` that overrides `ip` only where an address is handed to a client. Replica traffic still uses `ip`, so the two planes can live on separate networks. Left unset, behavior is unchanged. The field is typed as a string but validated as a literal IP, leaving room to accept hostnames later without a config-format break. Validation now checks advertised client endpoints for conflicts alongside the existing bind-endpoint check, since two nodes on distinct private IPs can otherwise publish the same client address and silently shadow each other. Node entries also reject unknown fields, so a misspelled advertised_address fails at startup instead of silently leaving the private ip in client responses. --- core/configs/src/server_ng_config/cluster.rs | 146 +++++++++++++++++- core/configs/src/server_ng_config/defaults.rs | 1 + .../src/server_ng_config/validators.rs | 1 + core/server-ng/config.toml | 7 + core/server-ng/src/cluster_meta.rs | 41 ++++- core/server-ng/src/http/error.rs | 56 ++++++- 6 files changed, 239 insertions(+), 13 deletions(-) diff --git a/core/configs/src/server_ng_config/cluster.rs b/core/configs/src/server_ng_config/cluster.rs index 44b565f581..0e61d64948 100644 --- a/core/configs/src/server_ng_config/cluster.rs +++ b/core/configs/src/server_ng_config/cluster.rs @@ -29,6 +29,7 @@ use configs::ConfigEnv; use iggy_common::{IggyDuration, Validatable}; use serde::{Deserialize, Serialize}; use serde_with::{DisplayFromStr, serde_as}; +use std::net::{IpAddr, SocketAddr}; use std::time::Duration; /// The primary heartbeats roughly every second (`PING_TICKS`); a window at @@ -144,9 +145,14 @@ pub struct ClusterTlsConfig { } #[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)] +#[serde(deny_unknown_fields)] pub struct ClusterNodeConfig { pub name: String, pub ip: String, + /// Optional client-facing address. Replica traffic continues to use + /// [`Self::ip`]. + #[serde(default)] + pub advertised_address: Option, /// Numeric replica ID for VSR consensus (0-based). /// /// Must be unique across [`ClusterConfig::nodes`] and strictly less than @@ -226,6 +232,8 @@ impl Validatable for ClusterConfig { let mut seen_ids = std::collections::HashSet::new(); let mut seen_names = std::collections::HashSet::new(); let mut used_endpoints = std::collections::HashSet::new(); + let mut used_advertised_endpoints = std::collections::HashSet::new(); + let mut used_raw_advertised_endpoints = std::collections::HashSet::new(); for node in &self.nodes { if node.name.trim().is_empty() { @@ -265,17 +273,17 @@ impl Validatable for ClusterConfig { return Err(ConfigurationError::InvalidConfigurationValue); } - let port_list = [ + let client_ports = [ ("TCP", node.ports.tcp), ("QUIC", node.ports.quic), ("HTTP", node.ports.http), ("WebSocket", node.ports.websocket), - ("TCP_REPLICA", node.ports.tcp_replica), ]; + let replica_port = ("TCP_REPLICA", node.ports.tcp_replica); - for (name, port_opt) in &port_list { + for (name, port_opt) in client_ports.into_iter().chain([replica_port]) { if let Some(port) = port_opt { - if *port == 0 { + if port == 0 { eprintln!( "Invalid cluster configuration: {} port cannot be 0 for node '{}'", name, node.name @@ -293,6 +301,47 @@ impl Validatable for ClusterConfig { } } } + + // Strict for now; the String type leaves room to accept hostnames + // later without breaking existing configs. + let client_ip = match node.advertised_address.as_deref() { + Some(advertised_address) => match advertised_address.parse::() { + Ok(ip) => Some(ip), + Err(_) => { + eprintln!( + "Invalid cluster configuration: advertised_address '{advertised_address}' is not a valid IP address for node '{}'", + node.name + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + }, + None => node.ip.parse::().ok(), + }; + + let client_address = node.advertised_address.as_deref().unwrap_or(&node.ip); + for (name, port) in &client_ports { + if let Some(port) = port { + let (endpoint, inserted) = client_ip.map_or_else( + || { + let endpoint = format!("{client_address}:{port}"); + let inserted = used_raw_advertised_endpoints.insert(endpoint.clone()); + (endpoint, inserted) + }, + |ip| { + let endpoint = SocketAddr::new(ip, *port); + let inserted = used_advertised_endpoints.insert(endpoint); + (endpoint.to_string(), inserted) + }, + ); + if !inserted { + eprintln!( + "Invalid cluster configuration: advertised client endpoint conflict - {endpoint} is already used (node '{}', transport {name})", + node.name + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + } + } } // Replica-auth PSK (only reached when the cluster is enabled; the early @@ -380,6 +429,27 @@ mod tests { "shared_secret field present in serialized config: {serialized}" ); } + + #[test] + fn cluster_node_rejects_unknown_fields() { + let error = serde_json::from_str::( + r#"{ + "name": "node-0", + "ip": "10.0.0.1", + "advertised_addres": "203.0.113.1", + "replica_id": 0, + "ports": {} + }"#, + ) + .expect_err("misspelled advertised_address must be rejected"); + + assert!( + error + .to_string() + .contains("unknown field `advertised_addres`"), + "unexpected deserialization error: {error}" + ); + } } #[cfg(test)] @@ -390,6 +460,7 @@ mod cluster_validate_tests { ClusterNodeConfig { name: name.to_string(), ip: "127.0.0.1".to_string(), + advertised_address: None, replica_id: id, ports: TransportPorts::default(), } @@ -513,6 +584,73 @@ mod cluster_validate_tests { assert!(c.validate().is_ok()); } + #[test] + fn validate_rejects_duplicate_advertised_client_endpoint() { + let mut n1 = node("n1", 0); + n1.ip = "10.0.0.1".to_owned(); + n1.advertised_address = Some("203.0.113.1".to_owned()); + n1.ports.tcp = Some(8090); + let mut n2 = node("n2", 1); + n2.ip = "10.0.0.2".to_owned(); + n2.advertised_address = n1.advertised_address.clone(); + n2.ports.tcp = Some(8090); + + assert!(cfg(vec![n1, n2]).validate().is_err()); + } + + #[test] + fn validate_rejects_equivalent_ipv6_advertised_client_endpoints() { + for equivalent_address in ["2001:DB8::1", "2001:db8:0:0:0:0:0:1"] { + let mut n1 = node("n1", 0); + n1.ip = "10.0.0.1".to_owned(); + n1.advertised_address = Some("2001:db8::1".to_owned()); + n1.ports.tcp = Some(8090); + let mut n2 = node("n2", 1); + n2.ip = "10.0.0.2".to_owned(); + n2.advertised_address = Some(equivalent_address.to_owned()); + n2.ports.tcp = Some(8090); + + assert!( + cfg(vec![n1, n2]).validate().is_err(), + "{equivalent_address} must conflict with 2001:db8::1" + ); + } + } + + #[test] + fn validate_rejects_equivalent_ipv6_client_endpoints_from_node_ip() { + let mut n1 = node("n1", 0); + n1.ip = "2001:db8::1".to_owned(); + n1.ports.tcp = Some(8090); + let mut n2 = node("n2", 1); + n2.ip = "2001:db8:0:0:0:0:0:1".to_owned(); + n2.ports.tcp = Some(8090); + + assert!(cfg(vec![n1, n2]).validate().is_err()); + } + + #[test] + fn validate_accepts_distinct_advertised_client_endpoints() { + let mut n1 = node("n1", 0); + n1.ip = "10.0.0.1".to_owned(); + n1.advertised_address = Some("203.0.113.1".to_owned()); + n1.ports.tcp = Some(8090); + let mut n2 = node("n2", 1); + n2.ip = "10.0.0.2".to_owned(); + n2.advertised_address = Some("203.0.113.2".to_owned()); + n2.ports.tcp = Some(8090); + + assert!(cfg(vec![n1, n2]).validate().is_ok()); + } + + #[test] + fn validate_rejects_non_ip_advertised_address() { + let mut n1 = node("n1", 0); + n1.advertised_address = Some("iggy-node-1.example.com".to_owned()); + + assert!(cfg(vec![n1, node("n2", 1)]).validate().is_err()); + } + #[test] fn validate_rejects_zero_tcp_replica_port() { let ports = TransportPorts { diff --git a/core/configs/src/server_ng_config/defaults.rs b/core/configs/src/server_ng_config/defaults.rs index 58887a70fe..d9c32dc8b0 100644 --- a/core/configs/src/server_ng_config/defaults.rs +++ b/core/configs/src/server_ng_config/defaults.rs @@ -84,6 +84,7 @@ impl Default for ClusterConfig { .map(|node| ClusterNodeConfig { name: node.name.parse().unwrap(), ip: node.ip.parse().unwrap(), + advertised_address: None, replica_id: u8::try_from(node.replica_id).expect( "static_toml replica_id must fit in u8 (0..=255); \ fix core/server-ng/config.toml", diff --git a/core/configs/src/server_ng_config/validators.rs b/core/configs/src/server_ng_config/validators.rs index 4736d90554..e824afe492 100644 --- a/core/configs/src/server_ng_config/validators.rs +++ b/core/configs/src/server_ng_config/validators.rs @@ -435,6 +435,7 @@ mod tests { ClusterNodeConfig { name: format!("node-{replica_id}"), ip: "127.0.0.1".to_string(), + advertised_address: None, replica_id, ports: TransportPorts { tcp: Some(8090 + u16::from(replica_id)), diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml index 7a185a80da..34baee0e6a 100644 --- a/core/server-ng/config.toml +++ b/core/server-ng/config.toml @@ -639,6 +639,12 @@ ca_file = "" # node's identity is resolved at launch from the '--replica-id ' CLI # flag, which selects the entry in this list that describes the current # node. All other entries are remote peers. +# +# Each node may also set 'advertised_address': the client-facing address +# handed out in cluster metadata and leader redirects. Set it when 'ip' is +# a private replica-network address unreachable by clients (Docker, +# Kubernetes, NAT). Must be a literal IP address for now. Replica-to-replica +# traffic always uses 'ip'. Unset = clients get 'ip'. [[cluster.nodes]] name = "iggy-node-1" ip = "127.0.0.1" @@ -655,6 +661,7 @@ ports = { tcp = 8091, quic = 8081, http = 3001, websocket = 8093, tcp_replica = # [[cluster.nodes]] # name = "iggy-node-3" # ip = "192.168.1.100" +# advertised_address = "203.0.113.10" # replica_id = 2 # ports = { tcp = 8092, http = 3002, tcp_replica = 9092 } diff --git a/core/server-ng/src/cluster_meta.rs b/core/server-ng/src/cluster_meta.rs index 9e969f3dfa..1337b879e2 100644 --- a/core/server-ng/src/cluster_meta.rs +++ b/core/server-ng/src/cluster_meta.rs @@ -106,7 +106,10 @@ impl ClusterRoster { .iter() .map(|node| ClusterNode { name: node.name.clone(), - ip: node.ip.clone(), + ip: node + .advertised_address + .clone() + .unwrap_or_else(|| node.ip.clone()), endpoints: ports_to_endpoints(&node.ports), role: role_for(primary_index, node.replica_id), status: ClusterNodeStatus::Healthy, @@ -150,3 +153,39 @@ fn ports_to_endpoints(ports: &TransportPorts) -> TransportEndpoints { ports.websocket.unwrap_or(0), ) } + +#[cfg(test)] +mod tests { + use super::*; + + fn roster(advertised_address: Option) -> ClusterRoster { + ClusterRoster { + enabled: true, + name: "test-cluster".to_owned(), + nodes: vec![ClusterNodeConfig { + name: "node-0".to_owned(), + ip: "10.0.0.1".to_owned(), + advertised_address, + replica_id: 0, + ports: TransportPorts::default(), + }], + self_ip: "127.0.0.1".to_owned(), + self_ports: TransportPorts::default(), + metadata_view: Arc::new(AtomicU64::new(METADATA_VIEW_UNKNOWN)), + } + } + + #[test] + fn cluster_metadata_uses_advertised_address_when_configured() { + let metadata = roster(Some("203.0.113.10".to_owned())).cluster_metadata(Some(0)); + + assert_eq!(metadata.nodes[0].ip, "203.0.113.10"); + } + + #[test] + fn cluster_metadata_falls_back_to_replica_ip() { + let metadata = roster(None).cluster_metadata(Some(0)); + + assert_eq!(metadata.nodes[0].ip, "10.0.0.1"); + } +} diff --git a/core/server-ng/src/http/error.rs b/core/server-ng/src/http/error.rs index 511a1dd22f..6799d4fa66 100644 --- a/core/server-ng/src/http/error.rs +++ b/core/server-ng/src/http/error.rs @@ -25,6 +25,7 @@ use axum::Json; use axum::http::header::{LOCATION, RETRY_AFTER}; use axum::http::{HeaderValue, StatusCode}; use axum::response::{IntoResponse, Response}; +use configs::ng_cluster::ClusterNodeConfig; use iggy_binary_protocol::Operation; use iggy_common::IggyError; use serde::{Deserialize, Serialize}; @@ -500,39 +501,54 @@ pub(in crate::http) fn primary_redirect_location( scheme: &str, path_and_query: &str, ) -> Option { - let socket = primary_http_socket(roster, primary_index)?; + let socket = primary_advertised_http_socket(roster, primary_index)?; Some(format!("{scheme}://{socket}{path_and_query}")) } /// Resolve the VSR primary's HTTP socket from the static roster: the node -/// whose `replica_id` equals `primary_index`, its `ports.http`, and its `ip` -/// parsed strictly. `None` on any miss so callers fail closed. Formatting the -/// returned [`SocketAddr`] brackets an IPv6 host (`[::1]:8080`) rather than -/// leaving it ambiguous. +/// whose `replica_id` equals `primary_index`, its `ports.http`, and its private +/// roster `ip` parsed strictly. Internal replica forwarding uses this address. pub(in crate::http) fn primary_http_socket( roster: &ClusterRoster, primary_index: u8, ) -> Option { + let (node, http_port) = primary_node(roster, primary_index)?; + let ip = node.ip.parse::().ok()?; + Some(SocketAddr::new(ip, http_port)) +} + +/// Resolve the client-facing HTTP socket for a redirect. The advertised +/// address is preferred, with the private roster IP retained as the +/// compatibility fallback. Constructing a socket address brackets IPv6 hosts, +/// which keeps the resulting redirect URL valid. +fn primary_advertised_http_socket(roster: &ClusterRoster, primary_index: u8) -> Option { + let (node, http_port) = primary_node(roster, primary_index)?; + let host = node.advertised_address.as_deref().unwrap_or(&node.ip); + let ip = host.parse::().ok()?; + Some(SocketAddr::new(ip, http_port)) +} + +fn primary_node(roster: &ClusterRoster, primary_index: u8) -> Option<(&ClusterNodeConfig, u16)> { let node = roster .nodes .iter() .find(|node| node.replica_id == primary_index)?; let http_port = node.ports.http?; - let ip = node.ip.parse::().ok()?; - Some(SocketAddr::new(ip, http_port)) + Some((node, http_port)) } #[cfg(test)] mod tests { use super::*; - use configs::ng_cluster::{ClusterNodeConfig, TransportPorts}; + use configs::ng_cluster::TransportPorts; const READ_PATH: &str = "/streams?consistency=linearizable"; fn node(replica_id: u8, ip: &str, http: Option) -> ClusterNodeConfig { ClusterNodeConfig { name: format!("node-{replica_id}"), ip: ip.to_owned(), + advertised_address: None, replica_id, ports: TransportPorts { tcp: None, @@ -614,6 +630,30 @@ mod tests { ); } + #[test] + fn primary_redirect_location_uses_advertised_address() { + let mut primary = node(0, "10.0.0.1", Some(8080)); + primary.advertised_address = Some("2001:db8::1".to_owned()); + let roster = roster(vec![primary]); + + assert_eq!( + primary_redirect_location(&roster, 0, "https", READ_PATH), + Some("https://[2001:db8::1]:8080/streams?consistency=linearizable".to_owned()) + ); + } + + #[test] + fn primary_http_socket_uses_private_roster_ip() { + let mut primary = node(0, "10.0.0.1", Some(8080)); + primary.advertised_address = Some("203.0.113.1".to_owned()); + let roster = roster(vec![primary]); + + assert_eq!( + primary_http_socket(&roster, 0), + Some("10.0.0.1:8080".parse().expect("valid socket address")) + ); + } + #[test] fn transient_not_committed_renders_503_with_retry_after() { let response = CustomError::from(IggyError::TransientNotCommitted).into_response(); From fa9e56d2406daaec5df673fa450674bbd6bed911 Mon Sep 17 00:00:00 2001 From: Maciej Modzelewski Date: Mon, 27 Jul 2026 16:20:37 +0200 Subject: [PATCH 2/4] allow hostname --- core/configs/src/server_ng_config/cluster.rs | 379 +++++++++++++++++-- core/server-ng/config.toml | 7 +- core/server-ng/src/cluster_meta.rs | 48 ++- core/server-ng/src/http/error.rs | 34 +- 4 files changed, 423 insertions(+), 45 deletions(-) diff --git a/core/configs/src/server_ng_config/cluster.rs b/core/configs/src/server_ng_config/cluster.rs index 0e61d64948..21f663b48f 100644 --- a/core/configs/src/server_ng_config/cluster.rs +++ b/core/configs/src/server_ng_config/cluster.rs @@ -29,7 +29,9 @@ use configs::ConfigEnv; use iggy_common::{IggyDuration, Validatable}; use serde::{Deserialize, Serialize}; use serde_with::{DisplayFromStr, serde_as}; -use std::net::{IpAddr, SocketAddr}; +use std::fmt; +use std::net::{IpAddr, Ipv6Addr, SocketAddr}; +use std::str::FromStr; use std::time::Duration; /// The primary heartbeats roughly every second (`PING_TICKS`); a window at @@ -41,6 +43,13 @@ pub const MIN_CLUSTER_HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(2); /// length is accepted. const MIN_SHARED_SECRET_LEN: usize = 32; +/// DNS caps a full name at 255 octets on the wire, which leaves 253 +/// characters of presentation text (RFC 1035). +const MAX_HOSTNAME_LEN: usize = 253; + +/// Per-label limit from RFC 1035. +const MAX_HOSTNAME_LABEL_LEN: usize = 63; + /// serde fallback for configs written before the field existed; the value /// itself lives in `core/server-ng/config.toml` like every other default. fn default_heartbeat_timeout() -> IggyDuration { @@ -149,8 +158,9 @@ pub struct ClusterTlsConfig { pub struct ClusterNodeConfig { pub name: String, pub ip: String, - /// Optional client-facing address. Replica traffic continues to use - /// [`Self::ip`]. + /// Optional client-facing address: a literal IP or a DNS hostname, + /// validated as [`AdvertisedAddress`] at boot. Replica traffic continues + /// to use [`Self::ip`]. #[serde(default)] pub advertised_address: Option, /// Numeric replica ID for VSR consensus (0-based). @@ -171,6 +181,166 @@ pub struct TransportPorts { pub tcp_replica: Option, } +/// A validated client-facing node address: a literal IP or a DNS hostname. +/// +/// Hostnames follow RFC 1123: ASCII letters, digits and hyphens in labels of +/// 1-63 characters that do not start or end with a hyphen, at most +/// [`MAX_HOSTNAME_LEN`] characters total, no port and no trailing dot. Names +/// consisting solely of digits and dots are rejected as malformed IPv4 rather +/// than accepted as hostnames, so `10.0.0.256` fails loudly instead of being +/// handed to DNS. Hostnames normalize to lowercase and IPs to their canonical +/// form ([`IpAddr`]), so textual variants of one address (`Broker.Example.COM`, +/// `2001:DB8::1`, `[2001:db8::1]`) compare equal. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub enum AdvertisedAddress { + Ip(IpAddr), + Hostname(String), +} + +impl AdvertisedAddress { + /// Render `host:port` for a URL or endpoint listing, bracketing IPv6 + /// hosts (`[::1]:8080`) so the port separator stays unambiguous. + pub fn authority(&self, port: u16) -> String { + match self { + Self::Ip(ip) => SocketAddr::new(*ip, port).to_string(), + Self::Hostname(hostname) => format!("{hostname}:{port}"), + } + } +} + +impl FromStr for AdvertisedAddress { + type Err = AdvertisedAddressError; + + fn from_str(address: &str) -> Result { + if address.is_empty() { + return Err(AdvertisedAddressError::Empty); + } + if let Ok(ip) = address.parse::() { + return Ok(Self::Ip(ip)); + } + // URL-style bracketed IPv6 (`[2001:db8::1]`) is unambiguous; accept + // it and store the inner address. + if let Some(inner) = address + .strip_prefix('[') + .and_then(|rest| rest.strip_suffix(']')) + && let Ok(ip) = inner.parse::() + { + return Ok(Self::Ip(IpAddr::V6(ip))); + } + if let Some((host, port)) = address.rsplit_once(':') { + // `host:port` and `[v6]:port` are the common misconfigurations; + // anything else with a colon can only be a broken IPv6 literal, + // since ':' never appears in a hostname. + let bracketed_host = host.starts_with('[') && host.ends_with(']'); + if !port.is_empty() + && port.bytes().all(|byte| byte.is_ascii_digit()) + && (bracketed_host || !host.contains(':')) + { + return Err(AdvertisedAddressError::PortNotAllowed); + } + return Err(AdvertisedAddressError::MalformedIpv6); + } + if address.len() > MAX_HOSTNAME_LEN { + return Err(AdvertisedAddressError::HostnameTooLong { + length: address.len(), + }); + } + let mut all_labels_numeric = true; + for label in address.split('.') { + if label.is_empty() { + return Err(AdvertisedAddressError::EmptyLabel); + } + if label.len() > MAX_HOSTNAME_LABEL_LEN { + return Err(AdvertisedAddressError::LabelTooLong { + label: label.to_owned(), + }); + } + if label.starts_with('-') || label.ends_with('-') { + return Err(AdvertisedAddressError::LabelHyphen { + label: label.to_owned(), + }); + } + if let Some(character) = label + .chars() + .find(|character| !character.is_ascii_alphanumeric() && *character != '-') + { + return Err(AdvertisedAddressError::InvalidCharacter { character }); + } + all_labels_numeric &= label.bytes().all(|byte| byte.is_ascii_digit()); + } + if all_labels_numeric { + return Err(AdvertisedAddressError::MalformedIpv4); + } + // DNS resolution is case-insensitive; normalizing here makes equality + // (and thus endpoint-conflict detection) case-insensitive too. + Ok(Self::Hostname(address.to_ascii_lowercase())) + } +} + +impl fmt::Display for AdvertisedAddress { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Ip(ip) => write!(formatter, "{ip}"), + Self::Hostname(hostname) => write!(formatter, "{hostname}"), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum AdvertisedAddressError { + Empty, + PortNotAllowed, + MalformedIpv4, + MalformedIpv6, + HostnameTooLong { length: usize }, + EmptyLabel, + LabelTooLong { label: String }, + LabelHyphen { label: String }, + InvalidCharacter { character: char }, +} + +impl fmt::Display for AdvertisedAddressError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Empty => write!(formatter, "address cannot be empty"), + Self::PortNotAllowed => write!( + formatter, + "address must not include a port; ports are configured in cluster.nodes.ports" + ), + Self::MalformedIpv4 => write!( + formatter, + "address consists only of digits and dots but is not a valid IPv4 address" + ), + Self::MalformedIpv6 => write!( + formatter, + "address contains ':' but is not a valid IPv6 address, and ':' cannot appear in a hostname" + ), + Self::HostnameTooLong { length } => write!( + formatter, + "hostname is {length} characters long; the limit is {MAX_HOSTNAME_LEN}" + ), + Self::EmptyLabel => write!( + formatter, + "hostname contains an empty label (leading, trailing, or doubled dot)" + ), + Self::LabelTooLong { label } => write!( + formatter, + "hostname label '{label}' exceeds {MAX_HOSTNAME_LABEL_LEN} characters" + ), + Self::LabelHyphen { label } => write!( + formatter, + "hostname label '{label}' cannot start or end with a hyphen" + ), + Self::InvalidCharacter { character } => write!( + formatter, + "character '{character}' is not allowed in a hostname (allowed: ASCII letters, digits, '-', '.')" + ), + } + } +} + +impl std::error::Error for AdvertisedAddressError {} + /// Whether cluster-wide JWT key material exists: a configured `http.jwt` /// secret, or the signing key derived from the cluster PSK. When it does, a /// bearer minted on any node verifies on every node - the invariant @@ -302,37 +472,41 @@ impl Validatable for ClusterConfig { } } - // Strict for now; the String type leaves room to accept hostnames - // later without breaking existing configs. - let client_ip = match node.advertised_address.as_deref() { - Some(advertised_address) => match advertised_address.parse::() { - Ok(ip) => Some(ip), - Err(_) => { + // An advertised address must parse strictly (IP or RFC 1123 + // hostname): the value is handed verbatim to every client via + // cluster metadata and redirect URLs, so a bad one poisons them + // all. The roster `ip` predates this check and is only validated + // as non-empty (Docker service names with underscores exist in + // the wild), so when it backs the client endpoints an unparseable + // value falls back to raw-string comparison instead of failing + // boot. + let client_address = match node.advertised_address.as_deref() { + Some(advertised_address) => match advertised_address.parse::() { + Ok(address) => Some(address), + Err(error) => { eprintln!( - "Invalid cluster configuration: advertised_address '{advertised_address}' is not a valid IP address for node '{}'", + "Invalid cluster configuration: advertised_address '{advertised_address}' for node '{}': {error}", node.name ); return Err(ConfigurationError::InvalidConfigurationValue); } }, - None => node.ip.parse::().ok(), + None => node.ip.parse::().ok(), }; - let client_address = node.advertised_address.as_deref().unwrap_or(&node.ip); for (name, port) in &client_ports { if let Some(port) = port { - let (endpoint, inserted) = client_ip.map_or_else( - || { - let endpoint = format!("{client_address}:{port}"); + let (endpoint, inserted) = match &client_address { + Some(address) => ( + address.authority(*port), + used_advertised_endpoints.insert((address.clone(), *port)), + ), + None => { + let endpoint = format!("{}:{port}", node.ip); let inserted = used_raw_advertised_endpoints.insert(endpoint.clone()); (endpoint, inserted) - }, - |ip| { - let endpoint = SocketAddr::new(ip, *port); - let inserted = used_advertised_endpoints.insert(endpoint); - (endpoint.to_string(), inserted) - }, - ); + } + }; if !inserted { eprintln!( "Invalid cluster configuration: advertised client endpoint conflict - {endpoint} is already used (node '{}', transport {name})", @@ -436,7 +610,7 @@ mod tests { r#"{ "name": "node-0", "ip": "10.0.0.1", - "advertised_addres": "203.0.113.1", + "advertise_address": "203.0.113.1", "replica_id": 0, "ports": {} }"#, @@ -446,12 +620,89 @@ mod tests { assert!( error .to_string() - .contains("unknown field `advertised_addres`"), + .contains("unknown field `advertise_address`"), "unexpected deserialization error: {error}" ); } } +#[cfg(test)] +mod advertised_address_tests { + use super::*; + + #[test] + fn parses_ip_literals_to_canonical_form() { + assert_eq!( + "203.0.113.1".parse::(), + Ok(AdvertisedAddress::Ip("203.0.113.1".parse().unwrap())) + ); + for equivalent_address in ["2001:DB8::1", "2001:db8:0:0:0:0:0:1", "[2001:db8::1]"] { + assert_eq!( + equivalent_address.parse::(), + Ok(AdvertisedAddress::Ip("2001:db8::1".parse().unwrap())), + "'{equivalent_address}' must parse to canonical 2001:db8::1" + ); + } + } + + #[test] + fn normalizes_hostname_to_lowercase() { + let address = "Broker-1.Example.COM".parse::(); + assert_eq!( + address, + Ok(AdvertisedAddress::Hostname( + "broker-1.example.com".to_owned() + )) + ); + } + + #[test] + fn authority_brackets_ipv6_hosts_only() { + let cases = [ + ("203.0.113.1", "203.0.113.1:8090"), + ("2001:db8::1", "[2001:db8::1]:8090"), + ("broker-1.example.com", "broker-1.example.com:8090"), + ]; + for (host, expected_authority) in cases { + let address = host.parse::().expect("valid address"); + assert_eq!(address.authority(8090), expected_authority); + } + } + + #[test] + fn rejects_port_suffixes() { + for address_with_port in ["example.com:8090", "10.0.0.1:8090", "[2001:db8::1]:8090"] { + assert_eq!( + address_with_port.parse::(), + Err(AdvertisedAddressError::PortNotAllowed), + "'{address_with_port}' must be rejected as host:port" + ); + } + } + + #[test] + fn rejects_dotted_numeric_strings_as_malformed_ipv4() { + for malformed_ip in ["10.0.0.256", "192.168.1", "12345"] { + assert_eq!( + malformed_ip.parse::(), + Err(AdvertisedAddressError::MalformedIpv4), + "'{malformed_ip}' must not pass as a hostname" + ); + } + } + + #[test] + fn rejects_broken_ipv6_literals() { + for broken_ipv6 in ["2001:db8:::1", "[2001:db8::zz]", "::1::2"] { + assert_eq!( + broken_ipv6.parse::(), + Err(AdvertisedAddressError::MalformedIpv6), + "'{broken_ipv6}' must be rejected as malformed IPv6" + ); + } + } +} + #[cfg(test)] mod cluster_validate_tests { use super::*; @@ -600,7 +851,7 @@ mod cluster_validate_tests { #[test] fn validate_rejects_equivalent_ipv6_advertised_client_endpoints() { - for equivalent_address in ["2001:DB8::1", "2001:db8:0:0:0:0:0:1"] { + for equivalent_address in ["2001:DB8::1", "2001:db8:0:0:0:0:0:1", "[2001:db8::1]"] { let mut n1 = node("n1", 0); n1.ip = "10.0.0.1".to_owned(); n1.advertised_address = Some("2001:db8::1".to_owned()); @@ -644,11 +895,85 @@ mod cluster_validate_tests { } #[test] - fn validate_rejects_non_ip_advertised_address() { + fn validate_accepts_hostname_advertised_address() { let mut n1 = node("n1", 0); n1.advertised_address = Some("iggy-node-1.example.com".to_owned()); + n1.ports.tcp = Some(8090); + + assert!(cfg(vec![n1, node("n2", 1)]).validate().is_ok()); + } + + #[test] + fn validate_rejects_malformed_advertised_addresses() { + let oversized_label = format!("{}.example.com", "a".repeat(64)); + let oversized_hostname = format!("{}example.com", "a.".repeat(130)); + for advertised_address in [ + "", + " 203.0.113.1", + "10.0.0.256", + "192.168.1", + "example.com:8090", + "[2001:db8::1]:8090", + "2001:db8:::1", + "iggy_node.example.com", + "-node.example.com", + "node-.example.com", + ".example.com", + "example..com", + "example.com.", + "ex\u{e4}mple.com", + oversized_label.as_str(), + oversized_hostname.as_str(), + ] { + let mut n1 = node("n1", 0); + n1.advertised_address = Some(advertised_address.to_owned()); + + assert!( + cfg(vec![n1, node("n2", 1)]).validate().is_err(), + "'{advertised_address}' must be rejected" + ); + } + } - assert!(cfg(vec![n1, node("n2", 1)]).validate().is_err()); + #[test] + fn validate_rejects_case_variant_hostname_advertised_endpoints() { + let mut n1 = node("n1", 0); + n1.ip = "10.0.0.1".to_owned(); + n1.advertised_address = Some("broker.example.com".to_owned()); + n1.ports.tcp = Some(8090); + let mut n2 = node("n2", 1); + n2.ip = "10.0.0.2".to_owned(); + n2.advertised_address = Some("Broker.Example.COM".to_owned()); + n2.ports.tcp = Some(8090); + + assert!(cfg(vec![n1, n2]).validate().is_err()); + } + + #[test] + fn validate_rejects_node_ip_hostname_clashing_with_advertised_hostname() { + let mut n1 = node("n1", 0); + n1.ip = "10.0.0.1".to_owned(); + n1.advertised_address = Some("broker.example.com".to_owned()); + n1.ports.tcp = Some(8090); + let mut n2 = node("n2", 1); + n2.ip = "broker.example.com".to_owned(); + n2.ports.tcp = Some(8090); + + assert!(cfg(vec![n1, n2]).validate().is_err()); + } + + #[test] + fn validate_accepts_distinct_hostname_advertised_endpoints() { + let mut n1 = node("n1", 0); + n1.ip = "10.0.0.1".to_owned(); + n1.advertised_address = Some("broker-1.example.com".to_owned()); + n1.ports.tcp = Some(8090); + let mut n2 = node("n2", 1); + n2.ip = "10.0.0.2".to_owned(); + n2.advertised_address = Some("broker-2.example.com".to_owned()); + n2.ports.tcp = Some(8090); + + assert!(cfg(vec![n1, n2]).validate().is_ok()); } #[test] diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml index 34baee0e6a..a4d709f87c 100644 --- a/core/server-ng/config.toml +++ b/core/server-ng/config.toml @@ -643,8 +643,9 @@ ca_file = "" # Each node may also set 'advertised_address': the client-facing address # handed out in cluster metadata and leader redirects. Set it when 'ip' is # a private replica-network address unreachable by clients (Docker, -# Kubernetes, NAT). Must be a literal IP address for now. Replica-to-replica -# traffic always uses 'ip'. Unset = clients get 'ip'. +# Kubernetes, NAT). Accepts a literal IPv4/IPv6 address or a DNS hostname +# (RFC 1123: ASCII letters, digits, '-' and '.'; no port, no trailing dot). +# Replica-to-replica traffic always uses 'ip'. Unset = clients get 'ip'. [[cluster.nodes]] name = "iggy-node-1" ip = "127.0.0.1" @@ -661,7 +662,7 @@ ports = { tcp = 8091, quic = 8081, http = 3001, websocket = 8093, tcp_replica = # [[cluster.nodes]] # name = "iggy-node-3" # ip = "192.168.1.100" -# advertised_address = "203.0.113.10" +# advertised_address = "iggy-node-3.example.com" # replica_id = 2 # ports = { tcp = 8092, http = 3002, tcp_replica = 9092 } diff --git a/core/server-ng/src/cluster_meta.rs b/core/server-ng/src/cluster_meta.rs index 1337b879e2..8ff23f6f2c 100644 --- a/core/server-ng/src/cluster_meta.rs +++ b/core/server-ng/src/cluster_meta.rs @@ -28,7 +28,7 @@ //! leader, but the full roster is still returned). The self-synthesized single //! node is the cluster-disabled fallback, shared by both callers. -use configs::ng_cluster::{ClusterNodeConfig, TransportPorts}; +use configs::ng_cluster::{AdvertisedAddress, ClusterNodeConfig, TransportPorts}; use iggy_common::{ ClusterMetadata, ClusterNode, ClusterNodeRole, ClusterNodeStatus, TransportEndpoints, }; @@ -106,10 +106,7 @@ impl ClusterRoster { .iter() .map(|node| ClusterNode { name: node.name.clone(), - ip: node - .advertised_address - .clone() - .unwrap_or_else(|| node.ip.clone()), + ip: client_host(node), endpoints: ports_to_endpoints(&node.ports), role: role_for(primary_index, node.replica_id), status: ClusterNodeStatus::Healthy, @@ -138,6 +135,18 @@ impl ClusterRoster { } } +/// Client-facing host in normalized form (lowercase hostname, canonical IP), +/// matching what boot validation compared and what redirect URLs render, so +/// textual config variants of one address publish identical metadata. A +/// roster `ip` that parses as neither (boot only requires it non-empty) +/// passes through verbatim; a configured `advertised_address` always parses, +/// validation rejects it otherwise. +fn client_host(node: &ClusterNodeConfig) -> String { + let host = node.advertised_address.as_deref().unwrap_or(&node.ip); + host.parse::() + .map_or_else(|_| host.to_owned(), |address| address.to_string()) +} + const fn role_for(primary_index: Option, replica_id: u8) -> ClusterNodeRole { match primary_index { Some(primary) if primary == replica_id => ClusterNodeRole::Leader, @@ -188,4 +197,33 @@ mod tests { assert_eq!(metadata.nodes[0].ip, "10.0.0.1"); } + + #[test] + fn cluster_metadata_normalizes_advertised_hostname_to_lowercase() { + let metadata = roster(Some("Broker.Example.COM".to_owned())).cluster_metadata(Some(0)); + + assert_eq!(metadata.nodes[0].ip, "broker.example.com"); + } + + #[test] + fn cluster_metadata_canonicalizes_advertised_ipv6_address() { + for equivalent_address in ["2001:DB8::1", "[2001:db8::1]"] { + let metadata = roster(Some(equivalent_address.to_owned())).cluster_metadata(Some(0)); + + assert_eq!( + metadata.nodes[0].ip, "2001:db8::1", + "'{equivalent_address}' must publish canonical form" + ); + } + } + + #[test] + fn cluster_metadata_passes_unparseable_replica_ip_verbatim() { + let mut cluster_roster = roster(None); + cluster_roster.nodes[0].ip = "iggy_node".to_owned(); + + let metadata = cluster_roster.cluster_metadata(Some(0)); + + assert_eq!(metadata.nodes[0].ip, "iggy_node"); + } } diff --git a/core/server-ng/src/http/error.rs b/core/server-ng/src/http/error.rs index 6799d4fa66..8ea07561e1 100644 --- a/core/server-ng/src/http/error.rs +++ b/core/server-ng/src/http/error.rs @@ -25,7 +25,7 @@ use axum::Json; use axum::http::header::{LOCATION, RETRY_AFTER}; use axum::http::{HeaderValue, StatusCode}; use axum::response::{IntoResponse, Response}; -use configs::ng_cluster::ClusterNodeConfig; +use configs::ng_cluster::{AdvertisedAddress, ClusterNodeConfig}; use iggy_binary_protocol::Operation; use iggy_common::IggyError; use serde::{Deserialize, Serialize}; @@ -501,8 +501,8 @@ pub(in crate::http) fn primary_redirect_location( scheme: &str, path_and_query: &str, ) -> Option { - let socket = primary_advertised_http_socket(roster, primary_index)?; - Some(format!("{scheme}://{socket}{path_and_query}")) + let authority = primary_advertised_http_authority(roster, primary_index)?; + Some(format!("{scheme}://{authority}{path_and_query}")) } /// Resolve the VSR primary's HTTP socket from the static roster: the node @@ -517,15 +517,17 @@ pub(in crate::http) fn primary_http_socket( Some(SocketAddr::new(ip, http_port)) } -/// Resolve the client-facing HTTP socket for a redirect. The advertised -/// address is preferred, with the private roster IP retained as the -/// compatibility fallback. Constructing a socket address brackets IPv6 hosts, -/// which keeps the resulting redirect URL valid. -fn primary_advertised_http_socket(roster: &ClusterRoster, primary_index: u8) -> Option { +/// Resolve the client-facing HTTP authority (`host:port`) for a redirect. The +/// advertised address is preferred, with the private roster IP retained as +/// the compatibility fallback. [`AdvertisedAddress::authority`] brackets IPv6 +/// hosts and passes hostnames through, so the redirect URL stays valid; a +/// host that is neither a valid IP nor a valid hostname yields `None` so +/// callers fail closed. +fn primary_advertised_http_authority(roster: &ClusterRoster, primary_index: u8) -> Option { let (node, http_port) = primary_node(roster, primary_index)?; let host = node.advertised_address.as_deref().unwrap_or(&node.ip); - let ip = host.parse::().ok()?; - Some(SocketAddr::new(ip, http_port)) + let address = host.parse::().ok()?; + Some(address.authority(http_port)) } fn primary_node(roster: &ClusterRoster, primary_index: u8) -> Option<(&ClusterNodeConfig, u16)> { @@ -642,6 +644,18 @@ mod tests { ); } + #[test] + fn primary_redirect_location_uses_advertised_hostname() { + let mut primary = node(0, "10.0.0.1", Some(8080)); + primary.advertised_address = Some("broker-1.example.com".to_owned()); + let roster = roster(vec![primary]); + + assert_eq!( + primary_redirect_location(&roster, 0, "https", READ_PATH), + Some("https://broker-1.example.com:8080/streams?consistency=linearizable".to_owned()) + ); + } + #[test] fn primary_http_socket_uses_private_roster_ip() { let mut primary = node(0, "10.0.0.1", Some(8080)); From 96a90b4f4c6bd3934ac9fd545171d12889b8f8c6 Mon Sep 17 00:00:00 2001 From: Maciej Modzelewski Date: Mon, 27 Jul 2026 17:01:50 +0200 Subject: [PATCH 3/4] fix typo --- core/configs/src/server_ng_config/cluster.rs | 2 +- core/server-ng/src/cluster_meta.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/core/configs/src/server_ng_config/cluster.rs b/core/configs/src/server_ng_config/cluster.rs index 21f663b48f..d65bfd27f6 100644 --- a/core/configs/src/server_ng_config/cluster.rs +++ b/core/configs/src/server_ng_config/cluster.rs @@ -477,7 +477,7 @@ impl Validatable for ClusterConfig { // cluster metadata and redirect URLs, so a bad one poisons them // all. The roster `ip` predates this check and is only validated // as non-empty (Docker service names with underscores exist in - // the wild), so when it backs the client endpoints an unparseable + // the wild), so when it backs the client endpoints an unparsable // value falls back to raw-string comparison instead of failing // boot. let client_address = match node.advertised_address.as_deref() { diff --git a/core/server-ng/src/cluster_meta.rs b/core/server-ng/src/cluster_meta.rs index 8ff23f6f2c..8a1bff5353 100644 --- a/core/server-ng/src/cluster_meta.rs +++ b/core/server-ng/src/cluster_meta.rs @@ -218,7 +218,7 @@ mod tests { } #[test] - fn cluster_metadata_passes_unparseable_replica_ip_verbatim() { + fn cluster_metadata_passes_unparsable_replica_ip_verbatim() { let mut cluster_roster = roster(None); cluster_roster.nodes[0].ip = "iggy_node".to_owned(); From 99b9b6d87b558b87a6d216336189ad0f7b6d346a Mon Sep 17 00:00:00 2001 From: Maciej Modzelewski Date: Tue, 28 Jul 2026 07:05:43 +0200 Subject: [PATCH 4/4] add missing property --- core/server-ng/src/bootstrap.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs index 1fcd78495f..9730f74fd0 100644 --- a/core/server-ng/src/bootstrap.rs +++ b/core/server-ng/src/bootstrap.rs @@ -3350,6 +3350,7 @@ mod tests { configs::ng_cluster::ClusterNodeConfig { name: "node".to_owned(), ip: ip.to_owned(), + advertised_address: None, replica_id: 0, ports: configs::ng_cluster::TransportPorts { tcp: Some(18070),