fix(server)!: reject a wildcard bind with no advertised address

.devcontainer/devcontainer.json

@@ -33,7 +33,8 @@
     "IGGY_HTTP_ADDRESS": "0.0.0.0:3000",
     "IGGY_QUIC_ADDRESS": "0.0.0.0:8080",
     "IGGY_TCP_ADDRESS": "0.0.0.0:8090",
-    "IGGY_WEBSOCKET_ADDRESS": "0.0.0.0:8092"
+    "IGGY_WEBSOCKET_ADDRESS": "0.0.0.0:8092",
+    "IGGY_NODE_ADVERTISED_ADDRESS": "localhost"
   },
   "forwardPorts": [3000, 3050, 8080, 8090, 8092],
   "portsAttributes": {

README.md

@@ -288,7 +288,7 @@ For configuration options and detailed help:
 You can also use environment variables to override any configuration setting:
 
 - Override TCP address
-   `IGGY_TCP_ADDRESS=0.0.0.0:8090 cargo run --bin iggy-server`
+   `IGGY_TCP_ADDRESS=127.0.0.1:8090 cargo run --bin iggy-server`
 
 - Set custom data path
    `IGGY_SYSTEM_PATH=/data/iggy cargo run --bin iggy-server`

bdd/docker-compose.server.yml

@@ -65,6 +65,7 @@ services:
       - IGGY_ROOT_PASSWORD=iggy
       - IGGY_SYSTEM_PATH=local_data
       - IGGY_TCP_ADDRESS=0.0.0.0:8090
+      - IGGY_NODE_ADVERTISED_ADDRESS=iggy-server
       - IGGY_HTTP_ADDRESS=0.0.0.0:3000
       - IGGY_QUIC_ADDRESS=0.0.0.0:8080
       - IGGY_WEBSOCKET_ADDRESS=0.0.0.0:8070

core/configs/src/server_config/defaults.rs

@@ -29,6 +29,7 @@ use super::cluster::{
 };
 use super::message_bus::MessageBusConfig;
 use super::metadata::MetadataConfig;
+use super::node::NodeConfig;
 use super::partition::PartitionConfig;
 use super::quic::{QuicCertificateConfig, QuicConfig};
 use super::server::ServerConfig;
@@ -52,6 +53,7 @@ impl Default for ServerConfig {
             consumer_group: ConsumerGroupConfig::default(),
             data_maintenance: DataMaintenanceConfig::default(),
             heartbeat: HeartbeatConfig::default(),
+            node: NodeConfig::default(),
             personal_access_token: PersonalAccessTokenConfig::default(),
             system: Arc::new(ServerSystemConfig::default()),
             quic: QuicConfig::default(),

core/configs/src/server_config/mod.rs

@@ -26,6 +26,7 @@ pub mod defaults;
 pub mod displays;
 pub mod message_bus;
 pub mod metadata;
+pub mod node;
 pub mod partition;
 pub mod quic;
 pub mod server;

core/configs/src/server_config/node.rs

@@ -0,0 +1,89 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// This node's own client-facing identity, for the cluster-disabled server.
+
+use super::COMPONENT;
+use super::cluster::AdvertisedAddress;
+use crate::ConfigurationError;
+use configs::ConfigEnv;
+use iggy_common::Validatable;
+use serde::{Deserialize, Serialize};
+
+/// Named to match its roster counterpart: `advertised_address` here and
+/// `cluster.nodes[*].advertised_address` there are the same setting for the
+/// same question, and an operator moving between the two modes should not have
+/// to learn a second spelling.
+#[derive(Debug, Default, Deserialize, Serialize, Clone, ConfigEnv)]
+#[serde(deny_unknown_fields)]
+pub struct NodeConfig {
+    /// Client-facing address: a literal IP or a DNS hostname. `None` leaves
+    /// the server deriving one from its bind address.
+    #[serde(default)]
+    pub advertised_address: Option<String>,
+}
+
+impl Validatable<ConfigurationError> for NodeConfig {
+    fn validate(&self) -> Result<(), ConfigurationError> {
+        let Some(address) = self.advertised_address.as_deref() else {
+            return Ok(());
+        };
+
+        address.parse::<AdvertisedAddress>().map_err(|error| {
+            eprintln!("{COMPONENT} - node.advertised_address '{address}': {error}");
+            ConfigurationError::InvalidConfigurationValue
+        })?;
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn advertised(address: &str) -> NodeConfig {
+        NodeConfig {
+            advertised_address: Some(address.to_owned()),
+        }
+    }
+
+    #[test]
+    fn validate_accepts_an_unset_address() {
+        assert!(NodeConfig::default().validate().is_ok());
+    }
+
+    #[test]
+    fn validate_accepts_a_routable_address() {
+        assert!(advertised("203.0.113.10").validate().is_ok());
+        assert!(advertised("broker-1.example.com").validate().is_ok());
+        assert!(advertised("2001:db8::1").validate().is_ok());
+    }
+
+    #[test]
+    fn validate_rejects_an_unspecified_address() {
+        assert!(advertised("0.0.0.0").validate().is_err());
+        assert!(advertised("::").validate().is_err());
+    }
+
+    #[test]
+    fn validate_rejects_an_unparsable_address() {
+        assert!(advertised("broker-1.example.com:8090").validate().is_err());
+        assert!(advertised("10.0.0.256").validate().is_err());
+        assert!(advertised("").validate().is_err());
+    }
+}

core/configs/src/server_config/server.rs

@@ -19,6 +19,7 @@ use super::COMPONENT;
 use super::cluster::ClusterConfig;
 use super::message_bus::MessageBusConfig;
 use super::metadata::MetadataConfig;
+use super::node::NodeConfig;
 use super::partition::PartitionConfig;
 use super::quic::QuicConfig;
 use super::tcp::TcpConfig;
@@ -111,6 +112,8 @@ pub struct ServerConfig {
     pub consumer_group: ConsumerGroupConfig,
     pub data_maintenance: DataMaintenanceConfig,
     #[serde(default)]
+    pub node: NodeConfig,
+    #[serde(default)]
     pub personal_access_token: PersonalAccessTokenConfig,
     pub heartbeat: HeartbeatConfig,
     pub system: Arc<ServerSystemConfig>,
@@ -125,7 +128,58 @@ pub struct ServerConfig {
     pub message_bus: MessageBusConfig,
 }
 
+/// One client-facing listener, as the client-facing address derivation and
+/// boot validation see it: the config key naming its bind address, that
+/// address as written, and whether the listener is switched on.
+pub struct ClientListener<'a> {
+    pub key: &'static str,
+    pub address: &'a str,
+    pub enabled: bool,
+}
+
 impl ServerConfig {
+    /// The client-facing listeners, in the order the derived client-facing
+    /// address prefers them. TCP leads: it is the binary protocol every SDK
+    /// speaks, so it is the listener an address derived for clients should
+    /// describe whenever it is running.
+    #[must_use]
+    pub fn client_listeners(&self) -> [ClientListener<'_>; 4] {
+        [
+            ClientListener {
+                key: "tcp.address",
+                address: &self.tcp.address,
+                enabled: self.tcp.enabled,
+            },
+            ClientListener {
+                key: "websocket.address",
+                address: &self.websocket.address,
+                enabled: self.websocket.enabled,
+            },
+            ClientListener {
+                key: "quic.address",
+                address: &self.quic.address,
+                enabled: self.quic.enabled,
+            },
+            ClientListener {
+                key: "http.address",
+                address: &self.http.address,
+                enabled: self.http.enabled,
+            },
+        ]
+    }
+
+    /// The listener whose bind address cluster metadata derives this node's
+    /// client-facing address from when `node.advertised_address` is unset:
+    /// the first enabled one. `None` when every client-facing listener is
+    /// off, which leaves no address for a client to dial and nothing to
+    /// publish.
+    #[must_use]
+    pub fn derived_address_listener(&self) -> Option<ClientListener<'_>> {
+        self.client_listeners()
+            .into_iter()
+            .find(|listener| listener.enabled)
+    }
+
     /// Load server configuration from file and environment variables.
     ///
     /// The path comes from `IGGY_CONFIG_PATH` or defaults to

core/configs/src/server_config/validators.rs

@@ -31,6 +31,7 @@ use crate::common::http::HMAC_JWT_ALGORITHMS;
 use crate::common::validators::SEGMENT_MAX_SIZE_BYTES;
 use err_trail::ErrContext;
 use iggy_common::{IggyExpiry, MAX_MESSAGE_SIZE_UPPER_BYTES, Validatable};
+use std::net::SocketAddr;
 
 /// compio-ws (tungstenite 0.29) `write_buffer_size` default. Used to
 /// evaluate the `max_write_buffer_size > write_buffer_size` invariant
@@ -76,6 +77,11 @@ impl Validatable<ConfigurationError> for ServerConfig {
         self.cluster.validate().error(|e: &ConfigurationError| {
             format!("{COMPONENT} (error: {e}) - failed to validate cluster config")
         })?;
+        self.node.validate().error(|e: &ConfigurationError| {
+            format!("{COMPONENT} (error: {e}) - failed to validate node config")
+        })?;
+        self.validate_tcp_bind_address()?;
+        self.validate_client_facing_address()?;
         self.metadata.validate().error(|e: &ConfigurationError| {
             format!("{COMPONENT} (error: {e}) - failed to validate metadata config")
         })?;
@@ -424,6 +430,54 @@ fn reject_unsupported(config: &ServerConfig) -> Result<(), ConfigurationError> {
     Ok(())
 }
 
+impl ServerConfig {
+    fn validate_tcp_bind_address(&self) -> Result<(), ConfigurationError> {
+        parse_bind_address("tcp.address", &self.tcp.address)?;
+        Ok(())
+    }
+
+    /// The listener the client-facing address is derived from must not bind a
+    /// wildcard unless that address is declared outright.
+    fn validate_client_facing_address(&self) -> Result<(), ConfigurationError> {
+        if self.cluster.enabled || self.node.advertised_address.is_some() {
+            return Ok(());
+        }
+        // No client-facing listener runs, so no client dials this node and
+        // there is no address to demand.
+        let Some(listener) = self.derived_address_listener() else {
+            return Ok(());
+        };
+        let bind = parse_bind_address(listener.key, listener.address)?;
+        if !bind.ip().to_canonical().is_unspecified() {
+            return Ok(());
+        }
+
+        eprintln!(
+            "{COMPONENT} - {} binds the wildcard {bind}, which says which interfaces this node \
+             accepts on rather than where a client reaches it, so cluster metadata would carry no \
+             address for this node. Set node.advertised_address to the address clients dial, or \
+             bind a concrete address.",
+            listener.key
+        );
+        Err(ConfigurationError::InvalidConfigurationValue)
+    }
+}
+
+/// A listener's bind address, which is a literal IP and a port and nothing
+/// else. `context` names the config key so the operator reads back the one
+/// they wrote.
+fn parse_bind_address(context: &str, address: &str) -> Result<SocketAddr, ConfigurationError> {
+    address.parse::<SocketAddr>().map_err(|error| {
+        eprintln!(
+            "{COMPONENT} - {context} '{address}' is not an address and port: {error}. The host \
+             is required and must be a literal IP, so ':PORT' and 'hostname:PORT' are both \
+             rejected; use 127.0.0.1:PORT for loopback or 0.0.0.0:PORT to accept on every \
+             interface."
+        );
+        ConfigurationError::InvalidConfigurationValue
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::super::cluster::{ClusterNodeConfig, TransportPorts};
@@ -470,6 +524,85 @@ mod tests {
         );
     }
 
+    #[test]
+    fn given_wildcard_bind_without_advertised_address_when_validating_should_reject() {
+        for wildcard in ["0.0.0.0:8090", "[::]:8090", "[::ffff:0.0.0.0]:8090"] {
+            let config = config_with_override(&format!(
+                "[tcp]\naddress = \"{wildcard}\"\n[cluster]\nenabled = false\n"
+            ));
+            assert!(
+                config.validate().is_err(),
+                "{wildcard} names no address a client can dial"
+            );
+        }
+    }
+
+    #[test]
+    fn given_a_hostless_or_named_bind_address_when_validating_should_reject() {
+        for address in [":8090", "localhost:8090", "0.0.0.0", "not-an-address"] {
+            let config = config_with_override(&format!("[tcp]\naddress = \"{address}\"\n"));
+            assert!(
+                config.validate().is_err(),
+                "{address} does not name a bind address"
+            );
+        }
+    }
+
+    #[test]
+    fn given_wildcard_bind_with_advertised_address_when_validating_should_pass() {
+        let config = config_with_override(
+            "[tcp]\naddress = \"0.0.0.0:8090\"\n[cluster]\nenabled = false\n\
+             [node]\nadvertised_address = \"broker-1.example.com\"\n",
+        );
+        assert!(config.validate().is_ok());
+    }
+
+    #[test]
+    fn given_concrete_bind_without_advertised_address_when_validating_should_pass() {
+        let config = config_with_override(
+            "[tcp]\naddress = \"192.0.2.10:8090\"\n[cluster]\nenabled = false\n",
+        );
+        assert!(config.validate().is_ok());
+    }
+
+    #[test]
+    fn given_wildcard_bind_on_a_disabled_listener_when_validating_should_pass() {
+        let config = config_with_override(
+            "[tcp]\nenabled = false\naddress = \"0.0.0.0:8090\"\n[cluster]\nenabled = false\n",
+        );
+        assert!(config.validate().is_ok());
+    }
+
+    #[test]
+    fn given_wildcard_bind_on_the_first_enabled_listener_when_validating_should_reject() {
+        let config = config_with_override(
+            "[tcp]\nenabled = false\n[websocket]\nenabled = false\n[quic]\nenabled = false\n\
+             [http]\naddress = \"0.0.0.0:3000\"\n[cluster]\nenabled = false\n",
+        );
+        assert!(
+            config.validate().is_err(),
+            "an http-only server derives its address from http.address"
+        );
+    }
+
+    #[test]
+    fn given_every_client_listener_disabled_when_validating_should_pass() {
+        let config = config_with_override(
+            "[tcp]\nenabled = false\naddress = \"0.0.0.0:8090\"\n[websocket]\nenabled = false\n\
+             [quic]\nenabled = false\n[http]\nenabled = false\n[cluster]\nenabled = false\n",
+        );
+        assert!(config.validate().is_ok());
+    }
+
+    #[test]
+    fn given_clustered_wildcard_bind_without_advertised_address_when_validating_should_pass() {
+        // The roster answers the client-facing address per node, so the bind
+        // address is free to be a wildcard with nothing declared here.
+        let config =
+            config_with_override("[tcp]\naddress = \"0.0.0.0:8090\"\n[cluster]\nenabled = true\n");
+        assert!(config.validate().is_ok());
+    }
+
     #[test]
     fn given_shipped_default_config_when_validating_should_pass() {
         let config: ServerConfig = Figment::new()