diff --git a/Cargo.lock b/Cargo.lock index d9e4d32066c7..abada3806e9d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2981,6 +2981,7 @@ dependencies = [ "log", "lru 0.16.4", "mio 1.2.2", + "nix 0.30.1", "notify", "nuid", "once_cell", diff --git a/docs/fern/pages/reference/components/runtime-configuration.mdx b/docs/fern/pages/reference/components/runtime-configuration.mdx index bbe54c7f5070..84c1d30fb608 100644 --- a/docs/fern/pages/reference/components/runtime-configuration.mdx +++ b/docs/fern/pages/reference/components/runtime-configuration.mdx @@ -143,11 +143,48 @@ Unless a field is marked environment-only, it has both a CLI flag and an environ See [Frontend Configuration](frontend-configuration.mdx) for `DYN_HTTP_BODY_LIMIT_MB`. + + Host for the TCP request-plane server. Accepts IPv4 and IPv6 literals, bracketed IPv6 literals, + wildcards, and interface names, including aliases such as `eth0:1`. If unset or empty, Dynamo + selects the first usable non-loopback IPv4 address, then IPv6, then IPv4 loopback, then IPv6 + loopback. On Unix, automatic selection excludes interfaces that are down. Dynamo trims surrounding + whitespace from configured hosts; a value containing only whitespace still fails server startup. + + For a named interface, Dynamo uses the same address selection order, restricted to that interface. + It selects the first usable address in each category in operating-system enumeration order. An + interface with only loopback addresses uses IPv4 loopback, then IPv6 loopback. Explicit interface + lookup does not filter by interface state. + + A wildcard uses the same automatic interface inventory to select a usable non-loopback address in + its requested family for advertisement. If only the other family has one, Dynamo switches the bind + wildcard to that family. If neither family has one, Dynamo prefers loopback in the requested family + before trying the other family. If no + loopback address is found, it uses the requested family's standard loopback address. The bind + wildcard uses the selected address family. Wildcards are never advertised. IPv4 link-local + addresses are accepted. Multicast, broadcast, and IPv6 link-local addresses are rejected. + + IPv4-mapped IPv6 addresses use IPv4 rules. For example, `::ffff:0.0.0.0` uses the `0.0.0.0` + wildcard rules and advertises a concrete address. + + The server resolves the address once at startup. A loopback fallback persists until restart, and + an interface-enumeration error fails server startup. + + + + Port used by the TCP request-plane server. Accepted values are integers from `0` to `65535`. + When the value is unset, empty, or `0`, the operating system assigns a free port. Dynamo does not + trim whitespace. An invalid value, including a value with surrounding whitespace, silently falls + back to an operating-system-assigned port. + + - IPv4 or IPv6 address, or exact network interface name, used to bind and advertise the TCP response - stream server. If an interface name occurs more than once in the operating system's interface - list, Dynamo uses the last address. Dynamo trims whitespace and treats an empty value as unset. - Unspecified addresses such as `0.0.0.0` and `::` are invalid. + Host for the TCP response-stream server. It accepts the same address syntax and uses the same + address selection and wildcard-family rules as `DYN_TCP_RPC_HOST`. Dynamo trims whitespace and + treats an empty value as unset. With a wildcard host, the TCP listener binds the selected wildcard, + while the QUIC response listener binds only the selected concrete address. + + The server resolves the address once at startup. A loopback fallback persists until restart, and + an interface-enumeration error fails server startup. @@ -167,13 +204,17 @@ Unless a field is marked environment-only, it has both a CLI flag and an environ IPv4 or IPv6 address, or exact network interface name, advertised by direct ZMQ event publishers. - If an interface name occurs more than once in the operating system's interface list, Dynamo uses - the last address. Dynamo trims whitespace and treats an empty value as unset. Unspecified addresses - such as `0.0.0.0` and `::` are invalid. + For an interface name, Dynamo selects the first usable IPv4 address, then IPv6. Dynamo trims + whitespace and treats an empty value as unset. Unspecified addresses such as `0.0.0.0` and `::` are + invalid, including IPv4-mapped wildcards such as `::ffff:0.0.0.0`. When unset, automatic + advertisement stays in the IPv4 bind family and, on Unix, excludes down interfaces. If no eligible + interface has a usable non-loopback IPv4 address, automatic selection advertises IPv4 loopback + (`127.0.0.1` when no IPv4 loopback address is found), and remote subscribers cannot reach the publisher. This variable changes the discovery address, not the listener address. Direct ZMQ continues to bind - an IPv4 wildcard, so an IPv6 value does not enable IPv6 listening. This variable does not affect - NATS or ZMQ broker mode. + an IPv4 wildcard. For remote subscribers, an explicit override must identify a reachable IPv4 + address served by that listener. An IPv6 value does not enable IPv6 listening. This variable does + not affect NATS or ZMQ broker mode. diff --git a/lib/backend-common/src/rl.rs b/lib/backend-common/src/rl.rs index 404e9eb40c68..042bd501a9e1 100644 --- a/lib/backend-common/src/rl.rs +++ b/lib/backend-common/src/rl.rs @@ -90,13 +90,7 @@ pub(crate) async fn serve_endpoint( fn self_host_base_url(drt: &dynamo_runtime::DistributedRuntime) -> Option { let info = drt.system_status_server_info()?; - let socket_addr = info.socket_addr; - if socket_addr.ip().is_unspecified() { - let host = dynamo_runtime::utils::local_ip_for_advertise(); - Some(format!("http://{host}:{}", socket_addr.port())) - } else { - Some(format!("http://{socket_addr}")) - } + Some(format!("http://{}", info.advertised_socket_addr())) } fn resolve_endpoint_name(primary_name: &str) -> anyhow::Result { diff --git a/lib/bindings/kvbm/Cargo.lock b/lib/bindings/kvbm/Cargo.lock index ecb50844c826..fe4fffdb500f 100644 --- a/lib/bindings/kvbm/Cargo.lock +++ b/lib/bindings/kvbm/Cargo.lock @@ -2394,6 +2394,7 @@ dependencies = [ "log", "lru", "mio 1.2.2", + "nix 0.30.1", "notify", "nuid", "once_cell", @@ -4865,6 +4866,7 @@ dependencies = [ "cfg-if", "cfg_aliases", "libc", + "memoffset 0.9.1", ] [[package]] diff --git a/lib/bindings/python/Cargo.lock b/lib/bindings/python/Cargo.lock index fe3b163c7f0c..cd5cf3a79e46 100644 --- a/lib/bindings/python/Cargo.lock +++ b/lib/bindings/python/Cargo.lock @@ -2501,6 +2501,7 @@ dependencies = [ "log", "lru", "mio 1.2.2", + "nix 0.30.1", "notify", "nuid", "once_cell", @@ -5280,6 +5281,7 @@ dependencies = [ "cfg-if", "cfg_aliases", "libc", + "memoffset 0.9.1", ] [[package]] diff --git a/lib/bindings/python/rust/lib.rs b/lib/bindings/python/rust/lib.rs index 7ae9539e9fb8..354a374f2563 100644 --- a/lib/bindings/python/rust/lib.rs +++ b/lib/bindings/python/rust/lib.rs @@ -1470,15 +1470,9 @@ impl DistributedRuntime { /// Workers use this in their RL request-plane route descriptor so the /// frontend does not need to derive worker system URLs from static env vars. fn system_status_server_url(&self) -> Option { - self.inner.system_status_server_info().map(|info| { - let socket_addr = info.socket_addr; - if socket_addr.ip().is_unspecified() { - let host = dynamo_runtime::utils::ip_resolver::local_ip_for_advertise(); - format!("http://{host}:{}", socket_addr.port()) - } else { - format!("http://{socket_addr}") - } - }) + self.inner + .system_status_server_info() + .map(|info| format!("http://{}", info.advertised_socket_addr())) } /// Register an async Python callback for /engine/{route_name} diff --git a/lib/llm/src/local_model.rs b/lib/llm/src/local_model.rs index 4f0d95bdd7a1..63b3e5dfe986 100644 --- a/lib/llm/src/local_model.rs +++ b/lib/llm/src/local_model.rs @@ -888,15 +888,11 @@ pub(crate) fn self_host_base_url( return Ok(None); }; - let configured = dynamo_runtime::RuntimeConfig::from_settings() - .unwrap_or_default() - .system_host; - let host = match configured.as_str() { - "0.0.0.0" | "::" | "[::]" => dynamo_runtime::utils::local_ip_for_advertise(), - _ => configured, - }; + Ok(Some(system_status_base_url(&info))) +} - Ok(Some(format!("http://{host}:{}", info.port()))) +fn system_status_base_url(info: &dynamo_runtime::SystemStatusServerInfo) -> String { + format!("http://{}", info.advertised_socket_addr()) } /// Scan `model_dir` for files to advertise alongside the typed MDC slots. @@ -951,6 +947,13 @@ mod self_host_metadata_default_tests { assert!(self_host_metadata_default(Some("garbage"))); // unrecognized assert!(!self_host_metadata_default(Some("false"))); // explicit opt-out } + + #[test] + fn self_host_url_uses_the_bound_status_address() { + let info = dynamo_runtime::SystemStatusServerInfo::new("[::1]:8080".parse().unwrap(), None); + + assert_eq!(system_status_base_url(&info), "http://[::1]:8080"); + } } #[cfg(test)] diff --git a/lib/runtime/Cargo.toml b/lib/runtime/Cargo.toml index 5bf28152d53e..e935d0311997 100644 --- a/lib/runtime/Cargo.toml +++ b/lib/runtime/Cargo.toml @@ -121,6 +121,9 @@ tokio-rayon = { version = "2.1" } kube = { version = "2.0.1", default-features = false, features = ["runtime", "derive", "client", "rustls-tls", "aws-lc-rs"] } k8s-openapi = { version = "0.26.0", features = ["v1_32"] } +[target.'cfg(unix)'.dependencies] +nix = { version = "0.30.1", default-features = false, features = ["net"] } + [dev-dependencies] criterion = { version = "0.5", features = ["async_tokio"] } rcgen = { workspace = true } diff --git a/lib/runtime/examples/Cargo.lock b/lib/runtime/examples/Cargo.lock index 23f376c5d65b..659923239bfe 100644 --- a/lib/runtime/examples/Cargo.lock +++ b/lib/runtime/examples/Cargo.lock @@ -817,6 +817,7 @@ dependencies = [ "log", "lru", "mio 1.2.2", + "nix", "notify", "nuid", "once_cell", @@ -1976,6 +1977,15 @@ version = "2.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" +[[package]] +name = "memoffset" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "488016bfae457b036d996092f6cb448677611ce4449e970ceaf42695203f218a" +dependencies = [ + "autocfg", +] + [[package]] name = "mime" version = "0.3.17" @@ -2051,6 +2061,19 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "nix" +version = "0.30.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "74523f3a35e05aba87a1d978330aef40f67b0304ac79c1c00b294c9830543db6" +dependencies = [ + "bitflags 2.11.1", + "cfg-if", + "cfg_aliases", + "libc", + "memoffset", +] + [[package]] name = "nkeys" version = "0.4.5" diff --git a/lib/runtime/src/component/endpoint.rs b/lib/runtime/src/component/endpoint.rs index 99ca80b88a52..0d6cac4937d0 100644 --- a/lib/runtime/src/component/endpoint.rs +++ b/lib/runtime/src/component/endpoint.rs @@ -1,6 +1,7 @@ // SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 +use std::net::SocketAddr; use std::sync::Arc; use anyhow::Result; @@ -327,25 +328,8 @@ fn build_transport_type_inner( ) -> Result { match mode { RequestPlaneMode::Tcp => { - let tcp_host = crate::utils::tcp_rpc_host_from_env(); - // If a fixed port is explicitly configured, use it directly (no init ordering dependency). - // Otherwise, use the actual bound port (set by TCP server after binding when port 0 is used). - let tcp_port = std::env::var("DYN_TCP_RPC_PORT") - .ok() - .and_then(|p| p.parse::().ok()) - .filter(|&p| p != 0) - .unwrap_or(crate::pipeline::network::manager::get_actual_tcp_rpc_port()?); - - // Include instance_id and endpoint name for proper TCP routing. - // Format: host:port/instance_id_hex/endpoint_name - // This ensures each worker has a unique routing key when multiple workers - // share the same TCP server (e.g., --num-workers > 1). - let tcp_endpoint = format!( - "{}:{}/{:x}/{}", - tcp_host, tcp_port, connection_id, endpoint_id.name - ); - - Ok(TransportType::Tcp(tcp_endpoint)) + let address = crate::pipeline::network::manager::get_actual_tcp_rpc_address()?; + Ok(tcp_transport_type(address, endpoint_id, connection_id)) } RequestPlaneMode::Nats => Ok(TransportType::Nats(nats::instance_subject( endpoint_id, @@ -354,11 +338,19 @@ fn build_transport_type_inner( } } +fn tcp_transport_type( + address: SocketAddr, + endpoint_id: &EndpointId, + connection_id: u64, +) -> TransportType { + TransportType::Tcp(format!("{address}/{connection_id:x}/{}", endpoint_id.name)) +} + /// Build transport type, ensuring TCP server is initialized when needed. /// -/// In TCP mode with an OS-assigned port (`DYN_TCP_RPC_PORT` unset or invalid), the server must bind -/// before we can construct a correct transport address. This helper ensures that initialization -/// occurs, then delegates to the internal builder. +/// In TCP mode the server must bind before discovery can publish the concrete +/// advertised address. This helper ensures that initialization occurs, then +/// delegates to the internal builder. pub async fn build_transport_type( endpoint: &Endpoint, endpoint_id: &EndpointId, @@ -366,19 +358,7 @@ pub async fn build_transport_type( ) -> Result { let mode = endpoint.drt().request_plane(); - // For TCP with OS-assigned ports, we must ensure the server is initialized - // (bound to a port) before we can construct a correct transport address. - let has_fixed_port = match mode { - RequestPlaneMode::Tcp => std::env::var("DYN_TCP_RPC_PORT") - .ok() - .and_then(|p| p.parse::().ok()) - .filter(|&p| p != 0) - .is_some(), - RequestPlaneMode::Nats => true, // NATS doesn't need port init - }; - - if !has_fixed_port { - // Ensure request plane server is initialized before building transport. + if mode == RequestPlaneMode::Tcp { let _ = endpoint.drt().request_plane_server().await?; } @@ -473,3 +453,27 @@ impl Endpoint { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn tcp_transport_uses_concrete_ipv4_and_bracketed_ipv6_addresses() { + let endpoint_id = EndpointId { + namespace: "ns".to_string(), + component: "worker".to_string(), + name: "generate".to_string(), + }; + + for (address, expected) in [ + ("192.0.2.10:1234", "192.0.2.10:1234/2a/generate"), + ("[2001:db8::10]:1234", "[2001:db8::10]:1234/2a/generate"), + ] { + let transport = tcp_transport_type(address.parse().unwrap(), &endpoint_id, 0x2a); + assert_eq!(transport.address(), expected); + assert!(!transport.address().starts_with("0.0.0.0")); + assert!(!transport.address().starts_with("[::]")); + } + } +} diff --git a/lib/runtime/src/config/environment_names.rs b/lib/runtime/src/config/environment_names.rs index 3c2eaa457a66..c6fd04d8a8aa 100644 --- a/lib/runtime/src/config/environment_names.rs +++ b/lib/runtime/src/config/environment_names.rs @@ -798,6 +798,15 @@ pub mod request_plane { /// Buffer size above which the TCP decoder shrinks an empty buffer, in bytes. pub const DYN_TCP_SHRINK_MESSAGE_SIZE: &str = "DYN_TCP_SHRINK_MESSAGE_SIZE"; + + /// Host or interface for the TCP request-plane server. + /// The server resolves the value once at startup. A loopback fallback + /// persists until restart, and an enumeration error fails server startup. + pub const DYN_TCP_RPC_HOST: &str = "DYN_TCP_RPC_HOST"; + + /// Port for the TCP request-plane server. + /// If unset, the OS assigns a free ephemeral port. + pub const DYN_TCP_RPC_PORT: &str = "DYN_TCP_RPC_PORT"; } /// Response plane transport configuration. @@ -813,10 +822,17 @@ pub mod tcp_response_stream { /// If unset or 0, the OS assigns a free ephemeral port. pub const DYN_TCP_RESPONSE_STREAM_PORT: &str = "DYN_TCP_RESPONSE_STREAM_PORT"; - /// IP address or exact interface shared by the TCP request callback and QUIC response - /// listeners. - /// Unspecified addresses are rejected. - /// If unset, the server auto-detects a routable local IP. + /// Host or interface for the TCP response stream server and QUIC response listener. + /// + /// Accepts IPv4 and IPv6 literals, bracketed IPv6 literals, IPv4 or IPv6 + /// wildcards, and interface names. Interface aliases such as `eth0:1` are + /// also accepted. If unset, the server selects the first usable + /// non-loopback IPv4 address, then IPv6, then IPv4 loopback, then IPv6 + /// loopback. A wildcard uses a reachable address in its requested family. + /// If only the other family has a usable non-loopback address, the server + /// switches the bind wildcard to that family. The server resolves the value + /// once at startup. A loopback fallback persists until restart, and an + /// enumeration error fails server startup. pub const DYN_TCP_RESPONSE_STREAM_HOST: &str = "DYN_TCP_RESPONSE_STREAM_HOST"; /// TCP request-plane TLS configuration @@ -1142,6 +1158,8 @@ mod tests { response_plane::DYN_RESPONSE_PLANE, request_plane::DYN_TCP_MAX_MESSAGE_SIZE, request_plane::DYN_TCP_SHRINK_MESSAGE_SIZE, + request_plane::DYN_TCP_RPC_HOST, + request_plane::DYN_TCP_RPC_PORT, // TCP Response Stream tcp_response_stream::DYN_TCP_RESPONSE_STREAM_PORT, tcp_response_stream::DYN_TCP_RESPONSE_STREAM_HOST, diff --git a/lib/runtime/src/pipeline/network/manager.rs b/lib/runtime/src/pipeline/network/manager.rs index af55dc3f2280..92978d843994 100644 --- a/lib/runtime/src/pipeline/network/manager.rs +++ b/lib/runtime/src/pipeline/network/manager.rs @@ -16,23 +16,26 @@ use super::egress::unified_client::RequestPlaneClient; use super::ingress::shared_tcp_endpoint::SharedTcpServer; use super::ingress::unified_server::RequestPlaneServer; +use crate::config::environment_names::request_plane; use crate::distributed::RequestPlaneMode; +use crate::utils::ip_resolver::{DefaultIpResolver, resolve_host_or_interface, resolve_local_host}; use anyhow::Result; use async_once_cell::OnceCell; +use std::net::SocketAddr; use std::sync::Arc; use std::sync::OnceLock; use tokio_util::sync::CancellationToken; -/// Global storage for the actual TCP RPC port after binding. -/// Uses OnceLock since the port is set once when the server binds and never changes. -static ACTUAL_TCP_RPC_PORT: OnceLock = OnceLock::new(); +/// Global storage for the advertised TCP RPC address after binding. +/// Uses OnceLock since the shared server binds once and never changes. +static ACTUAL_TCP_RPC_ADDRESS: OnceLock = OnceLock::new(); /// Global storage for the shared TCP server instance. /// /// When multiple workers run in the same process, they must share a single TCP server /// to ensure all endpoints are registered on the same server. Without this, each worker /// would create its own server on a different port, but all would publish the same port -/// (from ACTUAL_TCP_RPC_PORT) to discovery, causing "No handler found" errors. +/// (from ACTUAL_TCP_RPC_ADDRESS) to discovery, causing "No handler found" errors. /// /// Uses `tokio::sync::OnceCell` to support async initialization (binding the TCP socket). static GLOBAL_TCP_SERVER: tokio::sync::OnceCell> = @@ -48,23 +51,26 @@ static GLOBAL_TCP_SERVER_TOKEN: std::sync::LazyLock = /// Get the actual TCP RPC port that the server is listening on. pub fn get_actual_tcp_rpc_port() -> anyhow::Result { - ACTUAL_TCP_RPC_PORT.get().copied().ok_or_else(|| { + Ok(get_actual_tcp_rpc_address()?.port()) +} + +/// Get the concrete TCP RPC address published to discovery. +pub(crate) fn get_actual_tcp_rpc_address() -> anyhow::Result { + ACTUAL_TCP_RPC_ADDRESS.get().copied().ok_or_else(|| { tracing::error!( - "TCP RPC port not set - request_plane_server() must be called before get_actual_tcp_rpc_port()" + "TCP RPC address not set - request_plane_server() must be called before get_actual_tcp_rpc_address()" ); - anyhow::anyhow!( - "TCP RPC port not initialized. This is not expected." - ) + anyhow::anyhow!("TCP RPC address not initialized. This is not expected.") }) } -/// Set the actual TCP RPC port (called internally after server binds). -fn set_actual_tcp_rpc_port(port: u16) { - if let Err(existing) = ACTUAL_TCP_RPC_PORT.set(port) { +/// Set the address published to discovery after the shared server binds. +fn set_actual_tcp_rpc_address(address: SocketAddr) { + if let Err(existing) = ACTUAL_TCP_RPC_ADDRESS.set(address) { tracing::warn!( - existing_port = existing, - new_port = port, - "TCP RPC port already set, ignoring new value" + %existing, + new_address = %address, + "TCP RPC address already set, ignoring new value" ); } } @@ -73,7 +79,7 @@ fn set_actual_tcp_rpc_port(port: u16) { #[derive(Clone)] struct NetworkConfig { // TCP server configuration - tcp_host: String, + tcp_host: Option, /// TCP port to bind to. If None, the OS will assign a free port. tcp_port: Option, @@ -92,8 +98,10 @@ impl NetworkConfig { Self { // TCP server configuration // If DYN_TCP_RPC_PORT is set, use that port; otherwise None means OS will assign a free port - tcp_host: crate::utils::tcp_rpc_host_from_env(), - tcp_port: std::env::var("DYN_TCP_RPC_PORT") + tcp_host: std::env::var(request_plane::DYN_TCP_RPC_HOST) + .ok() + .filter(|host| !host.is_empty()), + tcp_port: std::env::var(request_plane::DYN_TCP_RPC_PORT) .ok() .and_then(|p| p.parse().ok()), @@ -175,7 +183,7 @@ impl NetworkManager { .unwrap_or_else(|| "OS-assigned".to_string()); tracing::info!( %mode, - host = %config.tcp_host, + host = config.tcp_host.as_deref().unwrap_or("auto"), port = %port_display, "Initializing NetworkManager with TCP request plane" ); @@ -267,13 +275,20 @@ impl NetworkManager { .get_or_try_init(|| async { // Use configured port if specified, otherwise use port 0 (OS assigns free port) let port = self.config.tcp_port.unwrap_or(0); - let bind_addr = format!("{}:{}", self.config.tcp_host, port) - .parse() - .map_err(|e| anyhow::anyhow!("Invalid TCP bind address: {}", e))?; + let resolver = DefaultIpResolver; + let resolved_host = match self.config.tcp_host.as_deref() { + Some(host) => resolve_host_or_interface(host, &resolver).map_err(|error| { + anyhow::anyhow!("Failed to resolve configured TCP RPC host '{host}': {error}") + })?, + None => resolve_local_host(&resolver).map_err(|error| { + anyhow::anyhow!("Failed to resolve local TCP RPC host: {error}") + })?, + }; + let bind_addr = SocketAddr::new(resolved_host.bind_ip(), port); tracing::info!( bind_addr = %bind_addr, - port_source = if self.config.tcp_port.is_some() { "DYN_TCP_RPC_PORT" } else { "OS-assigned" }, + port_source = if self.config.tcp_port.is_some() { request_plane::DYN_TCP_RPC_PORT } else { "OS-assigned" }, "Creating TCP request plane server" ); @@ -282,12 +297,13 @@ impl NetworkManager { // Bind and start server, getting the actual bound address let actual_addr = server.clone().bind_and_start().await?; - // Store the actual bound port globally so build_transport_type() can access it - set_actual_tcp_rpc_port(actual_addr.port()); + let advertised_addr = + SocketAddr::new(resolved_host.advertise_ip(), actual_addr.port()); + set_actual_tcp_rpc_address(advertised_addr); tracing::info!( actual_addr = %actual_addr, - actual_port = actual_addr.port(), + %advertised_addr, "TCP request plane server started" ); diff --git a/lib/runtime/src/pipeline/network/tcp/server.rs b/lib/runtime/src/pipeline/network/tcp/server.rs index ed0847f73615..7233686bbba4 100644 --- a/lib/runtime/src/pipeline/network/tcp/server.rs +++ b/lib/runtime/src/pipeline/network/tcp/server.rs @@ -4,7 +4,7 @@ use socket2::{Domain, SockAddr, SockRef, Socket, Type}; use std::{ collections::{HashMap, HashSet}, - net::{IpAddr, SocketAddr, TcpListener}, + net::{IpAddr, SocketAddr}, os::fd::{AsFd, FromRawFd}, sync::Arc, time::Duration, @@ -22,7 +22,6 @@ const TOMBSTONE_TTL: Duration = Duration::from_secs(5); use bytes::Bytes; use derive_builder::Builder; use futures::{SinkExt, StreamExt}; -use local_ip_address::Error; use parking_lot::Mutex; use serde::{Deserialize, Serialize}; @@ -47,7 +46,7 @@ use crate::pipeline::{ tcp::StreamType, }, }; -use crate::utils::ip_resolver::resolve_host_or_interface; +use crate::utils::ip_resolver::{resolve_host_or_interface, resolve_local_host}; use anyhow::{Result, anyhow as error}; pub use crate::utils::ip_resolver::{DefaultIpResolver, IpResolver}; @@ -60,8 +59,9 @@ pub struct ServerOptions { #[builder(default = "0")] pub port: u16, - /// IP literal or exact network interface name used to bind and advertise the server. + /// IP literal, bracketed IPv6 literal, wildcard, or network interface name. /// When unset, Dynamo selects a local address automatically. + /// The field name is retained for source compatibility. #[builder(default)] pub interface: Option, } @@ -76,8 +76,7 @@ impl ServerOptions { /// A Response connection is a connection that is established by a client with the intention of sending /// specific data back to the server. pub struct TcpStreamServer { - local_ip: IpAddr, - local_port: u16, + address: String, state: Arc>, } @@ -160,7 +159,7 @@ fn prune_tombstones(tombstones: &mut HashMap, now: impl TcpStreamServer { pub fn local_address(&self) -> Result { - Ok(SocketAddr::new(self.local_ip, self.local_port)) + Ok(self.address.parse()?) } pub fn options_builder() -> ServerOptionsBuilder { @@ -175,39 +174,26 @@ impl TcpStreamServer { options: ServerOptions, resolver: R, ) -> Result, PipelineError> { - let local_ip = match options.interface.as_deref() { - Some(host) => resolve_host_or_interface(host, &resolver).map_err(|error| { + let resolved_host = if let Some(host_or_interface) = options.interface.as_deref() { + resolve_host_or_interface(host_or_interface, &resolver).map_err(|error| { PipelineError::Generic(format!( - "Failed to resolve configured TCP host '{host}': {error}" + "Failed to resolve configured TCP host '{host_or_interface}': {error}" )) - })?, - None => { - let resolved_ip = resolver.local_ip().or_else(|err| match err { - Error::LocalIpAddressNotFound => resolver.local_ipv6(), - _ => Err(err), - }); - - match resolved_ip { - Ok(addr) => addr, - // Only fall back to loopback when no routable IP exists at all; - // propagate other resolver errors (I/O, platform) so - // misconfigured hosts fail fast instead of silently binding - // to 127.0.0.1. - Err(Error::LocalIpAddressNotFound) => { - tracing::warn!( - "No routable local IP address found; falling back to 127.0.0.1" - ); - IpAddr::from([127, 0, 0, 1]) - } - Err(err) => { - return Err(PipelineError::Generic(format!( - "Failed to resolve local IP address: {err}" - ))); - } - } - } + })? + } else { + resolve_local_host(&resolver).map_err(|error| { + PipelineError::Generic(format!("Failed to resolve local IP address: {error}")) + })? }; + if resolved_host.used_loopback_fallback() { + tracing::warn!( + bind_ip = %resolved_host.bind_ip(), + advertise_ip = %resolved_host.advertise_ip(), + "No usable local IP address found; advertising loopback" + ); + } + let state = Arc::new(Mutex::new(State::default())); // Build TLS acceptor from environment if cert+key paths are configured. @@ -215,20 +201,19 @@ impl TcpStreamServer { PipelineError::Generic(format!("Failed to build TCP TLS acceptor: {}", e)) })?; - let local_port = Self::start(local_ip, options.port, state.clone(), tls_acceptor) + let bind_address = SocketAddr::new(resolved_host.bind_ip(), options.port); + let local_address = Self::start(bind_address, state.clone(), tls_acceptor) .await - .map_err(|e| { - PipelineError::Generic(format!("Failed to start TcpStreamServer: {}", e)) + .map_err(|error| { + PipelineError::Generic(format!("Failed to start TcpStreamServer: {error}")) })?; + let advertised_address = + SocketAddr::new(resolved_host.advertise_ip(), local_address.port()); + let address = advertised_address.to_string(); - let local_addr = SocketAddr::new(local_ip, local_port); - tracing::debug!("tcp transport service on {local_addr}"); + tracing::debug!(%local_address, %advertised_address, "tcp transport service started"); - Ok(Arc::new(Self { - local_ip, - local_port, - state, - })) + Ok(Arc::new(Self { address, state })) } /// Associate one or both halves of a registration with a backend instance. @@ -391,28 +376,25 @@ impl TcpStreamServer { } async fn start( - local_ip: IpAddr, - local_port: u16, + bind_address: SocketAddr, state: Arc>, tls_acceptor: Option, - ) -> Result { - let addr = SocketAddr::new(local_ip, local_port); + ) -> Result { let state_clone = state.clone(); - let (ready_tx, ready_rx) = tokio::sync::oneshot::channel::>(); + let (ready_tx, ready_rx) = tokio::sync::oneshot::channel::>(); { let mut guard = state.lock(); if guard.handle.is_some() { panic!("TcpStreamServer already started"); } guard.handle = Some(tokio::spawn(tcp_listener( - addr, + bind_address, state_clone, tls_acceptor, ready_tx, ))); } - let local_port = ready_rx.await??; - Ok(local_port) + ready_rx.await? } fn insert_request_stream(&self, subject: String, connection: RequestedSendConnection) { @@ -481,7 +463,7 @@ impl ResponseService for TcpStreamServer { async fn register(&self, options: StreamOptions) -> PendingConnections { // oneshot channels to pass back the sender and receiver objects - let address = SocketAddr::new(self.local_ip, self.local_port).to_string(); + let address = self.address.clone(); tracing::debug!("Registering new TcpStream on {address}"); let send_stream = if options.enable_request_stream { @@ -776,7 +758,7 @@ async fn tcp_listener( addr: SocketAddr, state: Arc>, tls_acceptor: Option, - read_tx: tokio::sync::oneshot::Sender>, + read_tx: tokio::sync::oneshot::Sender>, ) -> Result<()> { let listener = tokio::net::TcpListener::bind(addr) .await @@ -789,9 +771,7 @@ async fn tcp_listener( .map_err(|e| anyhow::anyhow!("Failed get SocketAddr: {:?}", e)) .unwrap(); - read_tx - .send(Ok(addr.port())) - .expect("Failed to send ready signal"); + read_tx.send(Ok(addr)).expect("Failed to send ready signal"); listener } @@ -1345,6 +1325,7 @@ mod tests { use crate::pipeline::network::DEFAULT_SEND_BUFFER_COUNT; use crate::pipeline::network::tcp::client::TcpClient; use crate::tls_utils::test_certs::self_signed_pair; + use crate::utils::ip_resolver::test_support::{ProbeOutcome, StubResolver}; use tokio::io::{AsyncWriteExt, ReadHalf, WriteHalf}; use tokio::net::TcpStream; @@ -1432,17 +1413,99 @@ mod tests { ); } - // Mock resolver that always fails to simulate the fallback scenario - struct FailingIpResolver; + async fn registered_tcp_info(server: &TcpStreamServer) -> TcpStreamConnectionInfo { + let context = Context::new(()); + let stream_options = StreamOptions::builder() + .context(context.context()) + .enable_request_stream(false) + .enable_response_stream(true) + .build() + .unwrap(); + + let pending_connection = server.register(stream_options).await; + let connection_info = pending_connection + .recv_stream + .as_ref() + .unwrap() + .connection_info + .clone(); + connection_info.try_into().unwrap() + } + + #[tokio::test] + async fn wildcard_bind_advertises_concrete_ipv4_with_bound_port() { + for host in ["0.0.0.0", "::ffff:0.0.0.0", " [::ffff:0.0.0.0] "] { + let mut resolver = StubResolver::not_found(); + resolver + .interfaces + .push(("eth0", "192.0.2.20".parse().unwrap())); + let options = ServerOptions::builder() + .port(0) + .interface(Some(host.to_string())) + .build() + .unwrap(); + let server = TcpStreamServer::new_with_resolver(options, resolver) + .await + .unwrap(); - impl IpResolver for FailingIpResolver { - fn local_ip(&self) -> Result { - Err(Error::LocalIpAddressNotFound) + let tcp_info = registered_tcp_info(&server).await; + let socket_addr: SocketAddr = tcp_info.address.parse().unwrap(); + assert_eq!(socket_addr.ip(), "192.0.2.20".parse::().unwrap()); + assert_ne!(socket_addr.port(), 0); + assert!(!socket_addr.ip().is_unspecified()); + // The advertised port must belong to the real wildcard listener. + let _connection = tokio::net::TcpStream::connect(SocketAddr::new( + std::net::Ipv4Addr::LOCALHOST.into(), + socket_addr.port(), + )) + .await + .unwrap(); } + } - fn local_ipv6(&self) -> Result { - Err(Error::LocalIpAddressNotFound) + #[tokio::test] + async fn real_bracketed_ipv6_host_binds_and_registers() { + if let Err(error) = std::net::TcpListener::bind("[::1]:0") { + eprintln!("Skipping IPv6 loopback bind test: {error}"); + return; } + + let options = ServerOptions::builder() + .port(0) + .interface(Some("[::1]".to_string())) + .build() + .unwrap(); + let server = TcpStreamServer::new_with_resolver(options, StubResolver::not_found()) + .await + .unwrap(); + + let tcp_info = registered_tcp_info(&server).await; + let socket_addr: SocketAddr = tcp_info.address.parse().unwrap(); + assert_eq!(socket_addr.ip(), IpAddr::V6(std::net::Ipv6Addr::LOCALHOST)); + assert_ne!(socket_addr.port(), 0); + assert!( + tcp_info.address.starts_with("[::1]:"), + "{}", + tcp_info.address + ); + } + + #[tokio::test] + async fn tcp_server_preserves_interface_enumeration_error() { + let mut resolver = StubResolver::not_found(); + resolver.interface_error = Some(ProbeOutcome::Platform("test-platform")); + let options = ServerOptions::builder().port(0).build().unwrap(); + let error = TcpStreamServer::new_with_resolver(options, resolver) + .await + .err() + .expect("an interface enumeration failure must be returned"); + + let error = error.to_string(); + assert!( + error.contains("failed to enumerate network interfaces"), + "{error}" + ); + assert!(error.contains("test-platform"), "{error}"); } #[tokio::test] @@ -1459,26 +1522,7 @@ mod tests { let server = result.unwrap(); - // Verify the server can be used by registering a stream - let context = Context::new(()); - let stream_options = StreamOptions::builder() - .context(context.context()) - .enable_request_stream(false) - .enable_response_stream(true) - .build() - .unwrap(); - - let pending_connection = server.register(stream_options).await; - - // Verify connection info is available and valid - let connection_info = pending_connection - .recv_stream - .as_ref() - .unwrap() - .connection_info - .clone(); - - let tcp_info: TcpStreamConnectionInfo = connection_info.try_into().unwrap(); + let tcp_info = registered_tcp_info(&server).await; let socket_addr = tcp_info.address.parse::().unwrap(); // Should have a valid port assigned @@ -1552,7 +1596,7 @@ mod tests { let options = ServerOptions::builder().port(0).build().unwrap(); // Use the failing resolver to force the fallback - let result = TcpStreamServer::new_with_resolver(options, FailingIpResolver).await; + let result = TcpStreamServer::new_with_resolver(options, StubResolver::not_found()).await; assert!( result.is_ok(), "Server creation should succeed with fallback even when IP detection fails" @@ -1560,24 +1604,7 @@ mod tests { let server = result.unwrap(); - // Get the actual bound address by registering a stream - let context = Context::new(()); - let stream_options = StreamOptions::builder() - .context(context.context()) - .enable_request_stream(false) - .enable_response_stream(true) - .build() - .unwrap(); - - let pending_connection = server.register(stream_options).await; - let connection_info = pending_connection - .recv_stream - .as_ref() - .unwrap() - .connection_info - .clone(); - - let tcp_info: TcpStreamConnectionInfo = connection_info.try_into().unwrap(); + let tcp_info = registered_tcp_info(&server).await; let socket_addr = tcp_info.address.parse::().unwrap(); // With the failing resolver, fallback should ALWAYS be used @@ -1603,23 +1630,32 @@ mod tests { #[tokio::test] async fn configured_ip_literals_bind_and_format_addresses() { - let ipv6_available = TcpListener::bind("[::1]:0").is_ok(); + let ipv6_available = std::net::TcpListener::bind("[::1]:0").is_ok(); for (host, expected_ip) in [ ("127.0.0.1", "127.0.0.1".parse::().unwrap()), ("::1", "::1".parse::().unwrap()), + (" 127.0.0.1 ", "127.0.0.1".parse::().unwrap()), + (" [::1]\t", "::1".parse::().unwrap()), + ("::ffff:127.0.0.1", "127.0.0.1".parse::().unwrap()), + (" lo ", "127.0.0.1".parse::().unwrap()), + (" lo:1 ", "127.0.0.1".parse::().unwrap()), ] { if expected_ip.is_ipv6() && !ipv6_available { eprintln!("skipping IPv6 bind because this host does not support IPv6 loopback"); continue; } + let mut resolver = StubResolver::not_found(); + resolver + .interfaces + .push(("lo:1", "127.0.0.1".parse().unwrap())); let server = TcpStreamServer::new_with_resolver( ServerOptions { port: 0, interface: Some(host.to_string()), }, - FailingIpResolver, + resolver, ) .await .unwrap(); @@ -1638,6 +1674,7 @@ mod tests { let tcp_info: TcpStreamConnectionInfo = connection_info.try_into().unwrap(); let address = tcp_info.address.parse::().unwrap(); + assert_eq!(address, server.local_address().unwrap()); assert_eq!(address.ip(), expected_ip); assert_ne!(address.port(), 0); if expected_ip.is_ipv6() { @@ -1650,7 +1687,7 @@ mod tests { async fn test_server() -> Arc { TcpStreamServer::new_with_resolver( ServerOptions::builder().port(0).build().unwrap(), - FailingIpResolver, + StubResolver::not_found(), ) .await .unwrap() @@ -2208,7 +2245,7 @@ mod tests { /// framed reader/writer along with the receiver. async fn open_registered_response_stream() -> TestResponseStream { let options = ServerOptions::builder().port(0).build().unwrap(); - let server = TcpStreamServer::new_with_resolver(options, FailingIpResolver) + let server = TcpStreamServer::new_with_resolver(options, StubResolver::not_found()) .await .unwrap(); let context = Context::new(()); @@ -2346,7 +2383,7 @@ mod tests { #[tokio::test] async fn test_tcp_stream_server_returns_error_on_invalid_prologue() { let options = ServerOptions::builder().port(0).build().unwrap(); - let server = TcpStreamServer::new_with_resolver(options, FailingIpResolver) + let server = TcpStreamServer::new_with_resolver(options, StubResolver::not_found()) .await .unwrap(); let context = Context::new(()); @@ -2403,7 +2440,7 @@ mod tests { #[tokio::test] async fn test_unknown_typed_error_preserves_the_legacy_prologue_error() { let options = ServerOptions::builder().port(0).build().unwrap(); - let server = TcpStreamServer::new_with_resolver(options, FailingIpResolver) + let server = TcpStreamServer::new_with_resolver(options, StubResolver::not_found()) .await .unwrap(); let context = Context::new(()); diff --git a/lib/runtime/src/system_status_server.rs b/lib/runtime/src/system_status_server.rs index 0d46ca2aca19..97d744a9a810 100644 --- a/lib/runtime/src/system_status_server.rs +++ b/lib/runtime/src/system_status_server.rs @@ -11,6 +11,9 @@ use crate::config::environment_names::runtime::system as env_system; use crate::logging::make_system_request_span; use crate::metrics::MetricsHierarchy; use crate::traits::DistributedRuntimeProvider; +use crate::utils::ip_resolver::{ + DefaultIpResolver, IpResolutionError, IpResolver, resolve_advertise_ip_for_bind, +}; use axum::{ Router, body::Bytes, @@ -23,6 +26,7 @@ use futures::StreamExt; use serde::{Deserialize, Serialize}; use serde_json::json; use std::collections::HashMap; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use std::sync::{Arc, OnceLock}; use std::time::Instant; use tokio::{net::TcpListener, task::JoinHandle}; @@ -32,12 +36,12 @@ use tower_http::trace::TraceLayer; /// System status server information containing socket address and handle #[derive(Debug)] pub struct SystemStatusServerInfo { - pub socket_addr: std::net::SocketAddr, + pub socket_addr: SocketAddr, pub handle: Option>>, } impl SystemStatusServerInfo { - pub fn new(socket_addr: std::net::SocketAddr, handle: Option>) -> Self { + pub fn new(socket_addr: SocketAddr, handle: Option>) -> Self { Self { socket_addr, handle: handle.map(Arc::new), @@ -55,6 +59,37 @@ impl SystemStatusServerInfo { pub fn port(&self) -> u16 { self.socket_addr.port() } + + /// Return an address that is served by the bound socket. + /// + /// A wildcard bind is replaced with a usable address from the same address + /// family. If interface enumeration fails, the same-family loopback is + /// used so IPv4 and IPv6 are never combined. + pub fn advertised_socket_addr(&self) -> SocketAddr { + match advertised_socket_addr(self.socket_addr, &DefaultIpResolver) { + Ok(address) => address, + Err(error) => { + let fallback = match self.socket_addr.ip() { + IpAddr::V4(_) => IpAddr::V4(Ipv4Addr::LOCALHOST), + IpAddr::V6(_) => IpAddr::V6(Ipv6Addr::LOCALHOST), + }; + tracing::warn!( + %error, + %fallback, + "Failed to resolve the system status advertisement address; using loopback" + ); + SocketAddr::new(fallback, self.socket_addr.port()) + } + } + } +} + +fn advertised_socket_addr( + bound: SocketAddr, + resolver: &R, +) -> Result { + let advertise_ip = resolve_advertise_ip_for_bind(bound.ip(), resolver)?; + Ok(SocketAddr::new(advertise_ip, bound.port())) } impl Clone for SystemStatusServerInfo { @@ -710,8 +745,23 @@ async fn engine_route_handler( #[cfg(test)] mod tests { use super::*; + use crate::utils::ip_resolver::test_support::StubResolver; use tokio::time::Duration; + #[test] + fn advertised_address_stays_in_the_bound_family() { + let mut resolver = StubResolver::not_found(); + resolver.interfaces = vec![ + ("lo", "127.0.0.1".parse().unwrap()), + ("eth0", "2001:db8::20".parse().unwrap()), + ]; + + let advertised = advertised_socket_addr("0.0.0.0:8080".parse().unwrap(), &resolver) + .expect("wildcard advertisement should resolve"); + + assert_eq!(advertised, "127.0.0.1:8080".parse().unwrap()); + } + // This is a basic test to verify the HTTP server is working before testing other more complicated tests #[tokio::test] async fn test_http_server_lifecycle() { diff --git a/lib/runtime/src/transports/event_plane/mod.rs b/lib/runtime/src/transports/event_plane/mod.rs index 2a8126f054f1..2127e2ac8531 100644 --- a/lib/runtime/src/transports/event_plane/mod.rs +++ b/lib/runtime/src/transports/event_plane/mod.rs @@ -50,8 +50,8 @@ use crate::discovery::{ use crate::protocols::EndpointId; use crate::traits::DistributedRuntimeProvider; use crate::utils::ip_resolver::{ - DefaultIpResolver, IpResolver, host_override_from_env, resolve_host_or_interface, - resolve_local_host, + DefaultIpResolver, IpResolver, host_override_from_env, resolve_advertise_ip_for_bind, + resolve_host_or_interface, }; fn event_plane_host_from_env() -> Result { @@ -60,11 +60,20 @@ fn event_plane_host_from_env() -> Result { fn event_plane_host_from_env_with_resolver(resolver: &R) -> Result { let Some(host) = host_override_from_env(DYN_EVENT_PLANE_HOST)? else { - return Ok(resolve_local_host(resolver)); + return Ok(resolve_advertise_ip_for_bind( + std::net::Ipv4Addr::UNSPECIFIED.into(), + resolver, + )?); }; resolve_host_or_interface(&host, resolver) .map_err(|error| anyhow::anyhow!("Invalid {DYN_EVENT_PLANE_HOST} value '{host}': {error}")) + .and_then(|resolved| { + if resolved.bind_ip().to_canonical().is_unspecified() { + anyhow::bail!("Invalid {DYN_EVENT_PLANE_HOST} value '{host}': unspecified IP addresses cannot be advertised"); + } + Ok(resolved.advertise_ip()) + }) } fn direct_zmq_public_endpoint(advertised_ip: IpAddr, actual_bind_endpoint: &str) -> Result { @@ -949,6 +958,34 @@ fn current_timestamp_ms() -> u64 { mod tests { use super::*; use crate::config::environment_names::zmq_broker as broker_env; + use crate::utils::ip_resolver::test_support::StubResolver; + + #[test] + fn direct_zmq_advertisement_matches_bound_family() { + let mut resolver = StubResolver::not_found(); + resolver.interfaces = vec![ + ("lo", "127.0.0.1".parse().unwrap()), + ("lo", "::1".parse().unwrap()), + ("eth0", "2001:db8::20".parse().unwrap()), + ]; + + assert_eq!( + direct_zmq_public_endpoint( + resolve_advertise_ip_for_bind("0.0.0.0".parse().unwrap(), &resolver).unwrap(), + "tcp://0.0.0.0:4321" + ) + .unwrap(), + "tcp://127.0.0.1:4321" + ); + assert_eq!( + direct_zmq_public_endpoint( + resolve_advertise_ip_for_bind("::".parse().unwrap(), &resolver).unwrap(), + "tcp://[::]:4321" + ) + .unwrap(), + "tcp://[2001:db8::20]:4321" + ); + } struct EventPlaneHostResolver { ipv4: Option, @@ -1009,28 +1046,28 @@ mod tests { } #[test] - fn direct_zmq_advertise_host_preserves_ipv6_fallback_and_rejects_wildcards() { + fn direct_zmq_advertise_host_keeps_ipv4_bind_family_and_rejects_wildcards() { let resolver = EventPlaneHostResolver { ipv4: None, ipv6: Some("2001:db8::1".parse().unwrap()), - interfaces: Vec::new(), + interfaces: vec![("eth0".to_string(), "2001:db8::1".parse().unwrap())], }; assert_eq!( temp_env::with_vars([(DYN_EVENT_PLANE_HOST, None::<&str>)], || { event_plane_host_from_env_with_resolver(&resolver) }) .unwrap(), - "2001:db8::1".parse::().unwrap() + "127.0.0.1".parse::().unwrap() ); assert_eq!( temp_env::with_vars([(DYN_EVENT_PLANE_HOST, Some(" \t"))], || { event_plane_host_from_env_with_resolver(&resolver) }) .unwrap(), - "2001:db8::1".parse::().unwrap() + "127.0.0.1".parse::().unwrap() ); - for host in ["0.0.0.0", "::"] { + for host in ["0.0.0.0", "::", "::ffff:0.0.0.0", "[::ffff:0.0.0.0]"] { let error = temp_env::with_vars([(DYN_EVENT_PLANE_HOST, Some(host))], || { event_plane_host_from_env_with_resolver(&resolver) }) diff --git a/lib/runtime/src/utils.rs b/lib/runtime/src/utils.rs index 74a6408a3fbf..58e4663077b6 100644 --- a/lib/runtime/src/utils.rs +++ b/lib/runtime/src/utils.rs @@ -12,4 +12,5 @@ pub mod tasks; pub mod typed_prefix_watcher; pub use graceful_shutdown::{GracefulShutdownTracker, GracefulTaskGuard}; +#[allow(deprecated)] pub use ip_resolver::{local_ip_for_advertise, tcp_rpc_host_from_env}; diff --git a/lib/runtime/src/utils/ip_resolver.rs b/lib/runtime/src/utils/ip_resolver.rs index bb098c2400a7..024ee2ba42fd 100644 --- a/lib/runtime/src/utils/ip_resolver.rs +++ b/lib/runtime/src/utils/ip_resolver.rs @@ -3,13 +3,39 @@ //! Local IP address resolution for advertising endpoints. -use anyhow::{Result, bail}; use local_ip_address::{Error, list_afinet_netifas, local_ip, local_ipv6}; -use std::{ffi::OsString, net::IpAddr}; +use std::ffi::OsString; +use std::{ + net::{IpAddr, Ipv4Addr, Ipv6Addr}, + sync::OnceLock, +}; -const FALLBACK: IpAddr = IpAddr::V4(std::net::Ipv4Addr::LOCALHOST); +const DEFAULT_LOOPBACK: IpAddr = IpAddr::V4(Ipv4Addr::LOCALHOST); +static LOCAL_IP_FOR_ADVERTISE: OnceLock = OnceLock::new(); -/// IP address resolution interface used by networking components and tests. +/// Read and normalize a host override from the environment. +pub(crate) fn host_override_from_env(name: &str) -> anyhow::Result> { + host_override_from_lookup(name, |key| std::env::var_os(key)) +} + +fn host_override_from_lookup( + name: &str, + mut get_env: impl FnMut(&str) -> Option, +) -> anyhow::Result> { + let Some(value) = get_env(name) else { + return Ok(None); + }; + let value = value + .into_string() + .map_err(|_| anyhow::anyhow!("{name} must contain valid Unicode"))?; + let value = value.trim(); + Ok((!value.is_empty()).then(|| value.to_string())) +} + +/// IP address operations used by the runtime. +/// +/// This trait allows address resolution and interface enumeration to be +/// controlled in tests. pub trait IpResolver { fn local_ip(&self) -> Result; fn local_ipv6(&self) -> Result; @@ -17,9 +43,17 @@ pub trait IpResolver { fn list_afinet_netifas(&self) -> Result, Error> { list_afinet_netifas() } + + /// List addresses eligible for automatic selection. + /// + /// Custom resolvers can override this to exclude down interfaces. The + /// default trusts their existing interface inventory for compatibility. + fn list_up_afinet_netifas(&self) -> Result, Error> { + self.list_afinet_netifas() + } } -/// Default resolver backed by the local network interfaces. +/// The system IP resolver. pub struct DefaultIpResolver; impl IpResolver for DefaultIpResolver { @@ -30,237 +64,861 @@ impl IpResolver for DefaultIpResolver { fn local_ipv6(&self) -> Result { local_ipv6() } + + #[cfg(unix)] + fn list_up_afinet_netifas(&self) -> Result, Error> { + let interfaces = nix::ifaddrs::getifaddrs() + .map_err(|error| Error::StrategyError(format!("getifaddrs failed: {error}")))?; + Ok(up_ip_interfaces(interfaces)) + } } -/// Resolve the local IP for advertising endpoints, falling back to 127.0.0.1. -/// -/// IPv6 addresses are bracketed (e.g. `[::1]`) so the result is safe to -/// interpolate into a `host:port` URL. -pub fn local_ip_for_advertise() -> String { - resolve(DefaultIpResolver) +#[cfg(unix)] +fn up_ip_interfaces( + interfaces: impl IntoIterator, +) -> Vec<(String, IpAddr)> { + use nix::net::if_::InterfaceFlags; + + interfaces + .into_iter() + .filter(|interface| interface.flags.contains(InterfaceFlags::IFF_UP)) + .filter_map(|interface| { + let address = interface.address?; + let ip = if let Some(address) = address.as_sockaddr_in() { + IpAddr::V4(address.ip()) + } else { + IpAddr::V6(address.as_sockaddr_in6()?.ip()) + }; + Some((interface.interface_name, ip)) + }) + .collect() } -/// TCP RPC host: `DYN_TCP_RPC_HOST` if set, otherwise the resolved local IP. -pub fn tcp_rpc_host_from_env() -> String { - std::env::var("DYN_TCP_RPC_HOST").unwrap_or_else(|_| local_ip_for_advertise()) +#[derive(Debug, thiserror::Error)] +pub(crate) enum IpResolutionError { + #[error("failed to enumerate network interfaces: {0}")] + InterfaceEnumeration(#[source] Error), + + #[error("interface not found: {0}")] + InterfaceNotFound(String), + + #[error("interface has no usable IP address: {0}")] + NoUsableInterfaceAddress(String), + + #[error("IP address is not usable: {0}")] + UnusableAddress(IpAddr), + + #[error("invalid IP literal: {0}")] + InvalidLiteral(String), } -/// Read and normalize a host override from the environment. -pub(crate) fn host_override_from_env(name: &str) -> Result> { - host_override_from_lookup(name, |key| std::env::var_os(key)) +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) struct ResolvedHost { + bind_ip: IpAddr, + advertise_ip: IpAddr, + used_loopback_fallback: bool, } -fn host_override_from_lookup( - name: &str, - mut get_env: impl FnMut(&str) -> Option, -) -> Result> { - let Some(value) = get_env(name) else { - return Ok(None); - }; - let value = value - .into_string() - .map_err(|_| anyhow::anyhow!("{name} must contain valid Unicode"))?; - let value = value.trim(); - Ok((!value.is_empty()).then(|| value.to_string())) +impl ResolvedHost { + pub(crate) fn same_address(address: IpAddr) -> Self { + Self { + bind_ip: address, + advertise_ip: address, + used_loopback_fallback: false, + } + } + + pub(crate) fn loopback_fallback(address: IpAddr) -> Self { + Self { + bind_ip: address, + advertise_ip: address, + used_loopback_fallback: true, + } + } + + pub(crate) fn bind_ip(self) -> IpAddr { + self.bind_ip + } + + pub(crate) fn advertise_ip(self) -> IpAddr { + self.advertise_ip + } + + pub(crate) fn used_loopback_fallback(self) -> bool { + self.used_loopback_fallback + } } -/// Resolve a configured host as an IP literal or exact network interface name. +#[derive(Default)] +struct AddressCandidates { + ipv4: Option, + ipv6: Option, + ipv4_loopback: Option, + ipv6_loopback: Option, +} + +impl AddressCandidates { + fn consider(&mut self, address: IpAddr) { + let address = address.to_canonical(); + if !is_usable(address) { + return; + } + + match address { + IpAddr::V4(address) if address.is_loopback() => { + self.ipv4_loopback.get_or_insert(address); + } + IpAddr::V4(address) => { + self.ipv4.get_or_insert(address); + } + IpAddr::V6(address) if address.is_loopback() => { + self.ipv6_loopback.get_or_insert(address); + } + IpAddr::V6(address) => { + self.ipv6.get_or_insert(address); + } + } + } + + fn preferred_non_loopback(&self) -> Option { + self.ipv4 + .map(IpAddr::V4) + .or_else(|| self.ipv6.map(IpAddr::V6)) + } + + fn preferred_loopback(&self) -> Option { + self.ipv4_loopback + .map(IpAddr::V4) + .or_else(|| self.ipv6_loopback.map(IpAddr::V6)) + } + + fn non_loopback_for(&self, family: IpAddr) -> Option { + match family { + IpAddr::V4(_) => self.ipv4.map(IpAddr::V4), + IpAddr::V6(_) => self.ipv6.map(IpAddr::V6), + } + } + + fn loopback_for(&self, family: IpAddr) -> Option { + match family { + IpAddr::V4(_) => self.ipv4_loopback.map(IpAddr::V4), + IpAddr::V6(_) => self.ipv6_loopback.map(IpAddr::V6), + } + } + + fn other_non_loopback(&self, family: IpAddr) -> Option { + match family { + IpAddr::V4(_) => self.ipv6.map(IpAddr::V6), + IpAddr::V6(_) => self.ipv4.map(IpAddr::V4), + } + } +} + +fn local_candidates(resolver: &R) -> Result { + let interfaces = resolver + .list_up_afinet_netifas() + .map_err(IpResolutionError::InterfaceEnumeration)?; + let mut candidates = AddressCandidates::default(); + for (_, address) in interfaces { + candidates.consider(address); + } + Ok(candidates) +} + +/// Resolve the preferred local bind and advertisement address from one +/// interface snapshot. The system resolver excludes down interfaces on Unix. +pub(crate) fn resolve_local_host( + resolver: &R, +) -> Result { + let candidates = local_candidates(resolver)?; + if let Some(address) = candidates.preferred_non_loopback() { + return Ok(ResolvedHost::same_address(address)); + } + + Ok(ResolvedHost::loopback_fallback( + candidates.preferred_loopback().unwrap_or(DEFAULT_LOOPBACK), + )) +} + +/// Resolve a configured host value as an IP literal or interface name. /// -/// Interface lookup preserves the existing TCP response-stream behavior: when -/// an interface name occurs more than once, the last enumerated address wins. +/// Named dual-stack interfaces prefer the first usable IPv4 address in OS +/// enumeration order. IPv6 is used when no usable IPv4 address exists. +/// Surrounding whitespace is trimmed, and IPv4-mapped addresses use IPv4 rules. pub(crate) fn resolve_host_or_interface( host_or_interface: &str, resolver: &R, -) -> Result { +) -> Result { let host_or_interface = host_or_interface.trim(); - let ip = match host_or_interface.parse::() { - Ok(ip) => ip, - Err(_) => resolver - .list_afinet_netifas()? - .into_iter() - .filter_map(|(name, ip)| (name == host_or_interface).then_some(ip)) - .next_back() - .ok_or_else(|| { - anyhow::anyhow!( - "'{host_or_interface}' is not a valid IP address and no network interface with that name was found" - ) - })?, - }; - if ip.to_canonical().is_unspecified() { - bail!("unspecified IP addresses cannot be advertised"); + if let Some(address) = parse_ip_literal(host_or_interface)? { + if address.is_unspecified() { + return resolve_wildcard(address, &local_candidates(resolver)?); + } + + validate_configured_address(address)?; + return Ok(ResolvedHost::same_address(address)); + } + + let interfaces = resolver + .list_afinet_netifas() + .map_err(IpResolutionError::InterfaceEnumeration)?; + let mut interface_found = false; + let mut candidates = AddressCandidates::default(); + + for (name, address) in interfaces { + if name != host_or_interface { + continue; + } + + interface_found = true; + candidates.consider(address); } - Ok(ip) + + if !interface_found { + return Err(IpResolutionError::InterfaceNotFound( + host_or_interface.to_string(), + )); + } + + candidates + .preferred_non_loopback() + .or_else(|| candidates.preferred_loopback()) + .map(ResolvedHost::same_address) + .ok_or_else(|| IpResolutionError::NoUsableInterfaceAddress(host_or_interface.to_string())) } -pub(crate) fn resolve_local_host(resolver: &R) -> IpAddr { - resolver - .local_ip() - .or_else(|err| match err { - Error::LocalIpAddressNotFound => resolver.local_ipv6(), - _ => Err(err), - }) - .unwrap_or(FALLBACK) +fn resolve_wildcard( + wildcard: IpAddr, + candidates: &AddressCandidates, +) -> Result { + if let Some(advertise_ip) = candidates.non_loopback_for(wildcard) { + return Ok(ResolvedHost { + bind_ip: wildcard, + advertise_ip, + used_loopback_fallback: false, + }); + } + + if let Some(advertise_ip) = candidates.other_non_loopback(wildcard) { + return Ok(ResolvedHost { + bind_ip: unspecified_for(advertise_ip), + advertise_ip, + used_loopback_fallback: false, + }); + } + + let advertise_ip = candidates + .loopback_for(wildcard) + .or_else(|| candidates.preferred_loopback()) + .unwrap_or_else(|| loopback_for(wildcard)); + Ok(ResolvedHost { + bind_ip: unspecified_for(advertise_ip), + advertise_ip, + used_loopback_fallback: true, + }) +} + +/// Resolve an advertisement address that is served by an existing listener. +/// Wildcard listeners stay in their bound family. +pub(crate) fn resolve_advertise_ip_for_bind( + bind_ip: IpAddr, + resolver: &R, +) -> Result { + let bind_ip = bind_ip.to_canonical(); + if !bind_ip.is_unspecified() { + validate_configured_address(bind_ip)?; + return Ok(bind_ip); + } + + let candidates = local_candidates(resolver)?; + Ok(candidates + .non_loopback_for(bind_ip) + .or_else(|| candidates.loopback_for(bind_ip)) + .unwrap_or_else(|| loopback_for(bind_ip))) +} + +fn unspecified_for(address: IpAddr) -> IpAddr { + match address { + IpAddr::V4(_) => IpAddr::V4(Ipv4Addr::UNSPECIFIED), + IpAddr::V6(_) => IpAddr::V6(Ipv6Addr::UNSPECIFIED), + } +} + +fn loopback_for(address: IpAddr) -> IpAddr { + match address { + IpAddr::V4(_) => IpAddr::V4(Ipv4Addr::LOCALHOST), + IpAddr::V6(_) => IpAddr::V6(Ipv6Addr::LOCALHOST), + } +} + +fn parse_ip_literal(value: &str) -> Result, IpResolutionError> { + let has_open_bracket = value.starts_with('['); + let has_close_bracket = value.ends_with(']'); + + if has_open_bracket || has_close_bracket { + if !(has_open_bracket && has_close_bracket) { + return Err(IpResolutionError::InvalidLiteral(value.to_string())); + } + + let inner = &value[1..value.len() - 1]; + return inner + .parse::() + .map(|address| Some(IpAddr::V6(address).to_canonical())) + .map_err(|_| IpResolutionError::InvalidLiteral(value.to_string())); + } + + match value.parse::() { + Ok(address) => Ok(Some(address.to_canonical())), + Err(_) if looks_like_ip_literal(value) => { + Err(IpResolutionError::InvalidLiteral(value.to_string())) + } + Err(_) => Ok(None), + } +} + +fn looks_like_ip_literal(value: &str) -> bool { + value.bytes().filter(|byte| *byte == b':').count() >= 2 + || (value.contains('.') + && value + .chars() + .all(|character| character.is_ascii_digit() || character == '.')) +} + +fn validate_configured_address(address: IpAddr) -> Result<(), IpResolutionError> { + if is_usable(address) { + Ok(()) + } else { + Err(IpResolutionError::UnusableAddress(address)) + } +} + +fn is_usable(address: IpAddr) -> bool { + match address { + IpAddr::V4(address) => { + !address.is_unspecified() && !address.is_multicast() && !address.is_broadcast() + } + IpAddr::V6(address) => { + !address.is_unspecified() && !address.is_multicast() && !address.is_unicast_link_local() + } + } +} + +/// Resolve the local IP for advertising endpoints, with loopback fallback. +/// +/// IPv6 addresses are bracketed (for example, `[::1]`) so the result is safe +/// to interpolate into a `host:port` URL. Resolution is cached for the process +/// lifetime only after a non-loopback address is found. Enumeration failures +/// and loopback fallbacks are retried on the next call. +pub fn local_ip_for_advertise() -> String { + cached_local_ip_for_advertise(&LOCAL_IP_FOR_ADVERTISE, &DefaultIpResolver) +} + +/// TCP RPC host: `DYN_TCP_RPC_HOST` if set, otherwise the resolved local IP. +#[deprecated(note = "DYN_TCP_RPC_HOST is resolved by the request-plane server at startup")] +pub fn tcp_rpc_host_from_env() -> String { + std::env::var(crate::config::environment_names::request_plane::DYN_TCP_RPC_HOST) + .unwrap_or_else(|_| local_ip_for_advertise()) } -fn resolve(resolver: R) -> String { - let ip = resolve_local_host(&resolver); +fn cached_local_ip_for_advertise(cache: &OnceLock, resolver: &R) -> String { + if let Some(address) = cache.get() { + return format_host(*address); + } + + match resolve_local_host(resolver) { + Ok(resolved) if !resolved.used_loopback_fallback() => { + let address = resolved.advertise_ip(); + if cache.set(address).is_err() { + return format_host(*cache.get().expect("advertisement cache was initialized")); + } + format_host(address) + } + Ok(resolved) => { + let loopback = resolved.advertise_ip(); + tracing::warn!( + %loopback, + "No usable non-loopback IP address found; advertising loopback" + ); + format_host(loopback) + } + Err(error) => { + tracing::warn!( + %error, + loopback = %DEFAULT_LOOPBACK, + "Failed to resolve a usable local IP address; advertising loopback" + ); + format_host(DEFAULT_LOOPBACK) + } + } +} - match ip { - IpAddr::V6(_) => format!("[{ip}]"), - IpAddr::V4(_) => ip.to_string(), +fn format_host(address: IpAddr) -> String { + match address { + IpAddr::V6(_) => format!("[{address}]"), + IpAddr::V4(_) => address.to_string(), } } #[cfg(test)] -mod tests { +pub(crate) mod test_support { use super::*; + use std::cell::Cell; + + #[derive(Clone, Copy)] + pub(crate) enum ProbeOutcome { + NotFound, + Strategy(&'static str), + Platform(&'static str), + } + + impl ProbeOutcome { + fn result(self) -> Result { + match self { + Self::NotFound => Err(Error::LocalIpAddressNotFound), + Self::Strategy(message) => Err(Error::StrategyError(message.to_string())), + Self::Platform(platform) => Err(Error::PlatformNotSupported(platform.to_string())), + } + } + + fn error(self) -> Error { + self.result().unwrap_err() + } + } - struct MockIpResolver { - v4: Result, - v6: Result, - interfaces: Vec<(String, IpAddr)>, + pub(crate) struct StubResolver { + pub(crate) ipv4: ProbeOutcome, + pub(crate) ipv6: ProbeOutcome, + pub(crate) interfaces: Vec<(&'static str, IpAddr)>, + pub(crate) down_interfaces: Vec<&'static str>, + pub(crate) interface_error: Option, + pub(crate) ipv4_calls: Cell, + pub(crate) ipv6_calls: Cell, + pub(crate) interface_calls: Cell, } - impl IpResolver for MockIpResolver { + impl StubResolver { + pub(crate) fn new(ipv4: ProbeOutcome, ipv6: ProbeOutcome) -> Self { + Self { + ipv4, + ipv6, + interfaces: vec![ + ("lo", IpAddr::V4(Ipv4Addr::LOCALHOST)), + ("lo", IpAddr::V6(Ipv6Addr::LOCALHOST)), + ], + down_interfaces: Vec::new(), + interface_error: None, + ipv4_calls: Cell::new(0), + ipv6_calls: Cell::new(0), + interface_calls: Cell::new(0), + } + } + + pub(crate) fn not_found() -> Self { + Self::new(ProbeOutcome::NotFound, ProbeOutcome::NotFound) + } + } + + impl IpResolver for StubResolver { fn local_ip(&self) -> Result { - self.v4 - .as_ref() - .copied() - .map_err(|_| Error::LocalIpAddressNotFound) + self.ipv4_calls.set(self.ipv4_calls.get() + 1); + self.ipv4.result() } fn local_ipv6(&self) -> Result { - self.v6 - .as_ref() - .copied() - .map_err(|_| Error::LocalIpAddressNotFound) + self.ipv6_calls.set(self.ipv6_calls.get() + 1); + self.ipv6.result() } fn list_afinet_netifas(&self) -> Result, Error> { - Ok(self.interfaces.clone()) + self.interface_calls.set(self.interface_calls.get() + 1); + if let Some(error) = self.interface_error { + return Err(error.error()); + } + + Ok(self + .interfaces + .iter() + .map(|(name, address)| ((*name).to_string(), *address)) + .collect()) + } + + fn list_up_afinet_netifas(&self) -> Result, Error> { + Ok(self + .list_afinet_netifas()? + .into_iter() + .filter(|(name, _)| !self.down_interfaces.contains(&name.as_str())) + .collect()) } } +} - #[test] - fn ipv4_returned_unbracketed() { - let r = MockIpResolver { - v4: Ok(IpAddr::from([192, 168, 1, 100])), - v6: Err(Error::LocalIpAddressNotFound), - interfaces: Vec::new(), - }; - assert_eq!(resolve(r), "192.168.1.100"); +#[cfg(test)] +mod tests { + use super::test_support::{ProbeOutcome, StubResolver}; + use super::*; + use ProbeOutcome::{NotFound, Platform, Strategy}; + + fn ip(address: &str) -> IpAddr { + address.parse().unwrap() } - #[test] - fn ipv6_fallback_is_bracketed() { - let r = MockIpResolver { - v4: Err(Error::LocalIpAddressNotFound), - v6: Ok(IpAddr::from([0x2001, 0xdb8, 0, 0, 0, 0, 0, 1])), - interfaces: Vec::new(), - }; - assert_eq!(resolve(r), "[2001:db8::1]"); + fn interface(name: &'static str, addresses: &[&str]) -> Vec<(&'static str, IpAddr)> { + addresses + .iter() + .map(|address| (name, ip(address))) + .collect() } #[test] - fn both_fail_uses_localhost() { - let r = MockIpResolver { - v4: Err(Error::LocalIpAddressNotFound), - v6: Err(Error::LocalIpAddressNotFound), - interfaces: Vec::new(), - }; - assert_eq!(resolve(r), "127.0.0.1"); + fn ipv6_only_inventory_uses_global_ipv6_without_family_probes() { + let mut resolver = StubResolver::new(Strategy("IPv4 failed"), NotFound); + resolver.interfaces = vec![ + ("lo", ip("127.0.0.1")), + ("lo", ip("::1")), + ("eth0", ip("2001:db8::5")), + ]; + + let resolved = resolve_local_host(&resolver).unwrap(); + assert_eq!(resolved.bind_ip(), ip("2001:db8::5")); + assert!(!resolved.used_loopback_fallback()); + assert_eq!(resolver.ipv4_calls.get(), 0); + assert_eq!(resolver.ipv6_calls.get(), 0); + assert_eq!(resolver.interface_calls.get(), 1); } + #[cfg(unix)] #[test] - fn explicit_ip_literals_are_used_directly_and_trimmed() { - let resolver = MockIpResolver { - v4: Err(Error::LocalIpAddressNotFound), - v6: Err(Error::LocalIpAddressNotFound), - interfaces: Vec::new(), + fn system_inventory_keeps_only_up_ip_addresses() { + use nix::{ + ifaddrs::InterfaceAddress, net::if_::InterfaceFlags, sys::socket::SockaddrStorage, }; + use std::{net::SocketAddr, os::fd::AsRawFd, os::unix::net::UnixDatagram}; + + let socket = UnixDatagram::unbound().unwrap(); + let unsupported = + nix::sys::socket::getsockname::(socket.as_raw_fd()).unwrap(); + let up = InterfaceFlags::IFF_UP; + let entries = [ + ( + "stale0", + InterfaceFlags::empty(), + Some(SocketAddr::new(ip("198.51.100.1"), 0).into()), + ), + ("eth0", up, Some(SocketAddr::new(ip("192.0.2.1"), 0).into())), + ( + "eth0", + up, + Some(SocketAddr::new(ip("2001:db8::1"), 0).into()), + ), + ( + "lo", + up | InterfaceFlags::IFF_LOOPBACK, + Some(SocketAddr::new(ip("127.0.0.1"), 0).into()), + ), + ("unsupported", up, Some(unsupported)), + ("no-address", up, None), + ]; + let interfaces = entries + .into_iter() + .map(|(name, flags, address)| InterfaceAddress { + interface_name: name.to_string(), + flags, + address, + netmask: None, + broadcast: None, + destination: None, + }); assert_eq!( - resolve_host_or_interface(" 172.16.0.87 ", &resolver).unwrap(), - "172.16.0.87".parse::().unwrap() - ); - assert_eq!( - resolve_host_or_interface("2001:db8::1", &resolver).unwrap(), - "2001:db8::1".parse::().unwrap() + up_ip_interfaces(interfaces), + vec![ + ("eth0".to_string(), ip("192.0.2.1")), + ("eth0".to_string(), ip("2001:db8::1")), + ("lo".to_string(), ip("127.0.0.1")), + ] ); } #[test] - fn explicit_host_rejects_unspecified_addresses() { - let resolver = MockIpResolver { - v4: Err(Error::LocalIpAddressNotFound), - v6: Err(Error::LocalIpAddressNotFound), - interfaces: vec![("wildcard0".to_string(), "::ffff:0.0.0.0".parse().unwrap())], - }; - - for host in ["0.0.0.0", "::", "::ffff:0.0.0.0", "wildcard0"] { + fn automatic_selection_excludes_down_interfaces_but_explicit_lookup_keeps_them() { + for active in [Some("192.0.2.1"), Some("2001:db8::1"), None] { + let mut resolver = StubResolver::not_found(); + resolver + .interfaces + .insert(0, ("stale0", ip("198.51.100.1"))); + resolver.down_interfaces.push("stale0"); + if let Some(address) = active { + resolver.interfaces.push(("eth0", ip(address))); + } + let expected = active.map(ip).unwrap_or(DEFAULT_LOOPBACK); assert_eq!( - resolve_host_or_interface(host, &resolver) - .unwrap_err() - .to_string(), - "unspecified IP addresses cannot be advertised" + resolve_local_host(&resolver).unwrap().advertise_ip(), + expected + ); + assert_eq!( + resolve_host_or_interface("0.0.0.0", &resolver) + .unwrap() + .advertise_ip(), + expected + ); + assert_eq!( + resolve_advertise_ip_for_bind(unspecified_for(expected), &resolver).unwrap(), + expected + ); + assert_eq!( + resolve_host_or_interface("stale0", &resolver) + .unwrap() + .advertise_ip(), + ip("198.51.100.1") ); } } #[test] - fn exact_interface_name_supports_ipv4_and_ipv6() { - let resolver = MockIpResolver { - v4: Err(Error::LocalIpAddressNotFound), - v6: Err(Error::LocalIpAddressNotFound), - interfaces: vec![ - ("ib0".to_string(), "172.16.0.87".parse().unwrap()), - ("eth0".to_string(), "2001:db8::20".parse().unwrap()), - ], - }; + fn configured_literals_are_parsed_and_validated() { + let resolver = StubResolver::not_found(); + + for (literal, expected) in [ + ("192.0.2.10", "192.0.2.10"), + ("169.254.1.1", "169.254.1.1"), + ("2001:db8::2", "2001:db8::2"), + ("[2001:db8::2]", "2001:db8::2"), + ("fd00::2", "fd00::2"), + (" 127.0.0.1 ", "127.0.0.1"), + ("\t[::1] ", "::1"), + ("::ffff:192.0.2.10", "192.0.2.10"), + ("[::ffff:127.0.0.1]", "127.0.0.1"), + ] { + let resolved = resolve_host_or_interface(literal, &resolver).unwrap(); + assert_eq!(resolved.bind_ip(), ip(expected)); + assert_eq!(resolved.advertise_ip(), resolved.bind_ip()); + } + + for literal in ["[::1", "::1]", "[not-an-ip]", "192.0.2.999"] { + let error = resolve_host_or_interface(literal, &resolver) + .unwrap_err() + .to_string(); + assert!(error.contains("invalid IP literal"), "{error}"); + } + + for literal in [ + "224.0.0.1", + "255.255.255.255", + "fe80::1", + "ff02::1", + "::ffff:224.0.0.1", + "[::ffff:255.255.255.255]", + ] { + let error = resolve_host_or_interface(literal, &resolver).unwrap_err(); + assert!(matches!(error, IpResolutionError::UnusableAddress(_))); + } + } + + #[test] + fn wildcard_switches_bind_family_before_using_loopback() { + let mut resolver = StubResolver::new(NotFound, NotFound); + resolver.interfaces = interface("eth0", &["192.0.2.20"]); + let cases = [ + ("0.0.0.0", ip("0.0.0.0"), ip("192.0.2.20")), + ("::", ip("0.0.0.0"), ip("192.0.2.20")), + ("[::]", ip("0.0.0.0"), ip("192.0.2.20")), + ("::ffff:0.0.0.0", ip("0.0.0.0"), ip("192.0.2.20")), + (" [::ffff:0.0.0.0] ", ip("0.0.0.0"), ip("192.0.2.20")), + ]; + + for (literal, bind_ip, advertise_ip) in cases { + let resolved = resolve_host_or_interface(literal, &resolver).unwrap(); + assert_eq!(resolved.bind_ip(), bind_ip); + assert_eq!(resolved.advertise_ip(), advertise_ip); + } + } + + #[test] + fn wildcard_prefers_loopback_in_requested_family() { + let mut resolver = StubResolver::not_found(); + resolver.interfaces = vec![("lo", ip("127.0.0.1")), ("lo", ip("::1"))]; + + for (literal, expected_bind, expected_advertise) in [ + ("0.0.0.0", ip("0.0.0.0"), ip("127.0.0.1")), + ("::", ip("::"), ip("::1")), + ("[::]", ip("::"), ip("::1")), + ("::ffff:0.0.0.0", ip("0.0.0.0"), ip("127.0.0.1")), + ] { + let resolved = resolve_host_or_interface(literal, &resolver).unwrap(); + assert_eq!(resolved.bind_ip(), expected_bind); + assert_eq!(resolved.advertise_ip(), expected_advertise); + } + + resolver.interfaces = vec![("lo", ip("127.0.0.1"))]; + let resolved = resolve_host_or_interface("::", &resolver).unwrap(); + assert_eq!(resolved.bind_ip(), ip("0.0.0.0")); + assert_eq!(resolved.advertise_ip(), ip("127.0.0.1")); + + resolver.interfaces.clear(); + let resolved = resolve_host_or_interface("::", &resolver).unwrap(); + assert_eq!(resolved.bind_ip(), ip("::")); + assert_eq!(resolved.advertise_ip(), ip("::1")); + } + + #[test] + fn interface_alias_preserves_order_and_accepts_ipv4_link_local() { + let mut resolver = StubResolver::not_found(); + resolver.interfaces = interface("eth0:1", &["2001:db8::2", "169.254.10.5", "192.0.2.10"]); + + for name in ["eth0:1", " eth0:1\t"] { + let resolved = resolve_host_or_interface(name, &resolver).unwrap(); + assert_eq!(resolved.advertise_ip(), ip("169.254.10.5")); + } + } + + #[test] + fn interface_errors_retain_context() { + let mut resolver = StubResolver::not_found(); + resolver.interfaces = vec![("eth0", ip("fe80::1"))]; + + for (name, expected) in [ + ("missing", "interface not found: missing"), + ("", "interface not found: "), + (" \t", "interface not found: "), + ("eth0", "interface has no usable IP address: eth0"), + ] { + let error = resolve_host_or_interface(name, &resolver) + .unwrap_err() + .to_string(); + assert!(error.contains(expected), "{error}"); + } + + resolver.interface_error = Some(Platform("test-platform")); + let error = resolve_host_or_interface("eth0", &resolver) + .unwrap_err() + .to_string(); + assert!( + error.contains("failed to enumerate network interfaces"), + "{error}" + ); + assert!(error.contains("test-platform"), "{error}"); + } + + #[test] + fn loopback_only_host_preserves_ipv4_compatibility() { + let mut resolver = StubResolver::not_found(); + resolver.interfaces = vec![ + ("lo", IpAddr::V4(Ipv4Addr::LOCALHOST)), + ("lo", IpAddr::V6(Ipv6Addr::LOCALHOST)), + ]; + + let resolved = resolve_local_host(&resolver).unwrap(); + assert_eq!(resolved.advertise_ip(), DEFAULT_LOOPBACK); + assert!(resolved.used_loopback_fallback()); + } + + #[test] + fn bound_wildcard_advertisement_stays_in_bound_family() { + let mut resolver = StubResolver::not_found(); + resolver.interfaces = vec![ + ("lo", ip("127.0.0.1")), + ("lo", ip("::1")), + ("eth0", ip("2001:db8::20")), + ]; assert_eq!( - resolve_host_or_interface(" ib0 ", &resolver).unwrap(), - "172.16.0.87".parse::().unwrap() + resolve_advertise_ip_for_bind(ip("0.0.0.0"), &resolver).unwrap(), + ip("127.0.0.1") + ); + assert_eq!( + resolve_advertise_ip_for_bind(ip("::"), &resolver).unwrap(), + ip("2001:db8::20") + ); + assert_eq!( + resolve_advertise_ip_for_bind(ip("::ffff:0.0.0.0"), &resolver).unwrap(), + ip("127.0.0.1") ); assert_eq!( - resolve_host_or_interface("eth0", &resolver).unwrap(), - "2001:db8::20".parse::().unwrap() + resolve_advertise_ip_for_bind(ip("::ffff:127.0.0.1"), &resolver).unwrap(), + ip("127.0.0.1") ); } #[test] - fn interface_lookup_preserves_existing_last_address_behavior() { - let resolver = MockIpResolver { - v4: Err(Error::LocalIpAddressNotFound), - v6: Err(Error::LocalIpAddressNotFound), - interfaces: vec![ - ("ib0".to_string(), "172.16.0.10".parse().unwrap()), - ("ib0".to_string(), "172.16.0.20".parse().unwrap()), + fn mapped_interface_addresses_use_ipv4_classification() { + let mut resolver = StubResolver::not_found(); + resolver.interfaces = interface( + "eth0", + &[ + "::ffff:0.0.0.0", + "::ffff:224.0.0.1", + "::ffff:255.255.255.255", + "::ffff:127.0.0.1", + "2001:db8::20", ], - }; - + ); + assert_eq!( + resolve_local_host(&resolver).unwrap().advertise_ip(), + ip("2001:db8::20") + ); + resolver.interfaces.push(("eth0", ip("::ffff:192.0.2.1"))); assert_eq!( - resolve_host_or_interface("ib0", &resolver).unwrap(), - "172.16.0.20".parse::().unwrap() + resolve_local_host(&resolver).unwrap().advertise_ip(), + ip("192.0.2.1") + ); + assert_eq!( + resolve_host_or_interface("eth0", &resolver) + .unwrap() + .advertise_ip(), + ip("192.0.2.1") ); } #[test] - fn missing_interface_reports_the_accepted_input_forms() { - let resolver = MockIpResolver { - v4: Err(Error::LocalIpAddressNotFound), - v6: Err(Error::LocalIpAddressNotFound), - interfaces: Vec::new(), - }; + fn transient_fallbacks_are_not_cached_but_success_is() { + let cache = OnceLock::new(); + let mut enumeration_error = StubResolver::not_found(); + enumeration_error.interface_error = Some(Platform("transient")); + assert_eq!( + cached_local_ip_for_advertise(&cache, &enumeration_error), + "127.0.0.1" + ); + assert!(cache.get().is_none()); + let loopback = StubResolver::not_found(); assert_eq!( - resolve_host_or_interface("missing0", &resolver) - .unwrap_err() - .to_string(), - "'missing0' is not a valid IP address and no network interface with that name was found" + cached_local_ip_for_advertise(&cache, &loopback), + "127.0.0.1" + ); + assert!(cache.get().is_none()); + + let mut success = StubResolver::not_found(); + success.interfaces.push(("eth0", ip("2001:db8::10"))); + assert_eq!( + cached_local_ip_for_advertise(&cache, &success), + "[2001:db8::10]" + ); + assert_eq!(cache.get(), Some(&ip("2001:db8::10"))); + + assert_eq!( + cached_local_ip_for_advertise(&cache, &loopback), + "[2001:db8::10]" ); } + #[test] + fn legacy_resolver_implementation_remains_source_compatible() { + struct LegacyResolver; + + impl IpResolver for LegacyResolver { + fn local_ip(&self) -> Result { + Err(Error::LocalIpAddressNotFound) + } + + fn local_ipv6(&self) -> Result { + Err(Error::LocalIpAddressNotFound) + } + } + + fn assert_resolver() {} + assert_resolver::(); + } + #[test] fn host_override_is_trimmed_and_empty_is_unset() { assert_eq!(