From b0f2c85825cd5bf1270b85f907936188f5ac602a Mon Sep 17 00:00:00 2001 From: mlevkov Date: Sun, 2 Aug 2026 14:34:17 -0700 Subject: [PATCH 1/3] fix(connectors): warn when the runtime API is exposed without a key MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Authentication is off whenever `api_key` is empty, and the shipped `address` is loopback, so the default posture is "any local process may read every connector credential" — defensible for an admin API. What nothing catches is an operator moving `address` to reach the API from outside a container and getting an unauthenticated endpoint serving credentials, with no signal at any layer. Warns rather than refuses to start: refusing would break deployments that are exposed today, and that call is the maintainers' to make. Resolves the address rather than parsing it. The default is `localhost:8081`, which is loopback but is not a `SocketAddr`, so a parse check would warn on the shipped config and teach operators to ignore the warning. An address that cannot resolve counts as exposed — it is about to fail the bind anyway. --- core/connectors/runtime/src/api/mod.rs | 102 ++++++++++++++++++++++++- 1 file changed, 100 insertions(+), 2 deletions(-) diff --git a/core/connectors/runtime/src/api/mod.rs b/core/connectors/runtime/src/api/mod.rs index 38cd8f6ed1..0e36f15796 100644 --- a/core/connectors/runtime/src/api/mod.rs +++ b/core/connectors/runtime/src/api/mod.rs @@ -22,9 +22,14 @@ use axum::{Json, Router, extract::State, middleware, routing::get}; use axum_server::tls_rustls::RustlsConfig; use config::{HttpConfig, configure_cors}; use iggy_connector_sdk::api::ConnectorRuntimeStats; -use std::{net::SocketAddr, path::PathBuf, sync::Arc}; +use secrecy::ExposeSecret; +use std::{ + net::{SocketAddr, ToSocketAddrs}, + path::PathBuf, + sync::Arc, +}; use tokio::spawn; -use tracing::{error, info}; +use tracing::{error, info, warn}; mod auth; pub mod config; @@ -41,6 +46,13 @@ pub async fn init(config: &HttpConfig, context: Arc) { return; } + if is_unauthenticated_beyond_loopback(config) { + warn!( + "{NAME} HTTP API is enabled on {} with no api_key configured. Its configuration endpoints return plugin configuration verbatim, credentials included, so anyone able to reach that address can read every connector secret. Set http.api_key, or bind the API to loopback.", + config.address + ); + } + let mut system_router = Router::new().route("/stats", get(get_stats)); if config.metrics.enabled { @@ -121,6 +133,30 @@ pub async fn init(config: &HttpConfig, context: Arc) { }); } +/// Whether the API would answer beyond loopback with no key required. +/// +/// The configuration endpoints return plugin configuration verbatim, so an +/// unauthenticated listener on a routable address hands out every credential an +/// operator put in their TOML. Loopback with no key is the shipped default and +/// a defensible posture for an admin API; moving only the address is the +/// combination no other layer catches. +/// +/// Resolves rather than parses, because `address` accepts a hostname and the +/// default is `localhost:8081` - which is loopback but is not a `SocketAddr`. +/// The bind that follows resolves the same string, so this classifies what will +/// actually be listened on. An address that cannot resolve counts as exposed: +/// it is about to fail the bind anyway, and staying quiet about an address we +/// could not classify is the wrong direction to be wrong in. +fn is_unauthenticated_beyond_loopback(config: &HttpConfig) -> bool { + if !config.api_key.expose_secret().is_empty() { + return false; + } + match config.address.to_socket_addrs() { + Ok(mut resolved) => !resolved.all(|address| address.ip().is_loopback()), + Err(_) => true, + } +} + async fn get_metrics(State(context): State>) -> String { context.metrics.get_formatted_output() } @@ -128,3 +164,65 @@ async fn get_metrics(State(context): State>) -> String { async fn get_stats(State(context): State>) -> Json { Json(stats::get_runtime_stats(&context).await) } + +#[cfg(test)] +mod tests { + use super::*; + use secrecy::SecretString; + + fn config(address: &str, api_key: &str) -> HttpConfig { + HttpConfig { + address: address.to_owned(), + api_key: SecretString::from(api_key.to_owned()), + ..HttpConfig::default() + } + } + + #[test] + fn given_loopback_address_and_no_key_when_checked_should_stay_quiet() { + // The shipped posture. Warning here would train operators to ignore it. + assert!(!is_unauthenticated_beyond_loopback(&config( + "127.0.0.1:8081", + "" + ))); + assert!(!is_unauthenticated_beyond_loopback(&config( + "[::1]:8081", + "" + ))); + assert!( + !is_unauthenticated_beyond_loopback(&config("localhost:8081", "")), + "the default address is a hostname, so parsing alone would misjudge it" + ); + } + + #[test] + fn given_routable_address_and_no_key_when_checked_should_report_it() { + assert!( + is_unauthenticated_beyond_loopback(&config("0.0.0.0:8081", "")), + "binding every interface to reach the API from outside a container \ + is the case this exists to catch" + ); + assert!(is_unauthenticated_beyond_loopback(&config( + "192.0.2.10:8081", + "" + ))); + } + + #[test] + fn given_configured_key_when_checked_should_stay_quiet_on_any_address() { + assert!(!is_unauthenticated_beyond_loopback(&config( + "0.0.0.0:8081", + "secret" + ))); + } + + #[test] + fn given_unresolvable_address_when_checked_should_report_it() { + // About to fail the bind regardless, so the warning costs nothing and + // the alternative is silence about an address we cannot classify. + assert!(is_unauthenticated_beyond_loopback(&config( + "not a valid address", + "" + ))); + } +} From 009e1e05021610552b9b49a833c6c44aed470909 Mon Sep 17 00:00:00 2001 From: mlevkov Date: Sun, 2 Aug 2026 14:34:17 -0700 Subject: [PATCH 2/3] docs(connectors): document the runtime control API as privileged MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The endpoint list said what each route returns but not that the configuration routes return plugin configuration verbatim, credentials included, with no redaction layer anywhere in the runtime. An operator reading it had no way to know that exposing the port exposes every secret in their TOML. The `api_key` comment also described the key as optional without saying that leaving it empty disables authentication outright, and nothing explained why the default address is loopback — which made it look like an arbitrary default rather than the control that confines the exposure. --- core/connectors/runtime/README.md | 17 ++++++++++++++++- core/connectors/runtime/config.toml | 4 +++- 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/core/connectors/runtime/README.md b/core/connectors/runtime/README.md index 1c1339f49c..c3d71bd7af 100644 --- a/core/connectors/runtime/README.md +++ b/core/connectors/runtime/README.md @@ -136,8 +136,10 @@ Connector runtime has an optional HTTP API that can be enabled by setting the `e ```toml [http] # Optional HTTP API configuration enabled = true +# Loopback on purpose: the configuration endpoints return plugin credentials in +# plaintext. Set api_key in the same edit if you move this off loopback. address = "127.0.0.1:8081" -api_key = "" # Optional API key for authentication to be passed as `api-key` header +api_key = "" # Optional API key for authentication to be passed as `api-key` header; empty disables authentication [http.cors] # Optional CORS configuration for HTTP API enabled = false @@ -158,6 +160,19 @@ cert_file = "core/certs/iggy_cert.pem" key_file = "core/certs/iggy_key.pem" ``` +> [!IMPORTANT] +> **Treat this API as privileged.** The configuration endpoints return plugin +> configuration exactly as it was parsed from TOML, credentials included - a +> database connection string, an S3 secret key, a webhook signing secret. There +> is no redaction layer. `api_key` is empty by default, which means +> authentication is **off** by default; the loopback default `address` is what +> confines that to local processes. +> +> If you change `address` to reach the API from outside a container, set +> `api_key` in the same edit. The runtime logs a warning at startup when the +> address resolves beyond loopback with no key configured, but nothing prevents +> it. + Currently, it does expose the following endpoints: - `GET /`: welcome message. diff --git a/core/connectors/runtime/config.toml b/core/connectors/runtime/config.toml index 247a67e6b7..90e855d455 100644 --- a/core/connectors/runtime/config.toml +++ b/core/connectors/runtime/config.toml @@ -17,8 +17,10 @@ [http] # Optional HTTP API configuration enabled = true +# Loopback on purpose: the configuration endpoints return plugin credentials in +# plaintext. Set api_key in the same edit if you move this off loopback. address = "127.0.0.1:8081" -api_key = "" # Optional API key for authentication to be passed as `api-key` header +api_key = "" # Optional API key for authentication to be passed as `api-key` header; empty disables authentication [http.cors] # Optional CORS configuration for HTTP API enabled = false From 5c28f902a193cee83b90cc5096fce003a0aa2e9d Mon Sep 17 00:00:00 2001 From: mlevkov Date: Sun, 2 Aug 2026 17:54:38 -0700 Subject: [PATCH 3/3] test(connectors): cover the warning wiring in the API init The four existing tests cover the guard's decision table but not that `init` consults it, so deleting the call left the suite green. These two drive the real `init` and close that. Reaching the warning needs a non-loopback address, and any such address that binds would open a port on every interface for the length of the test, which on macOS also trips the firewall prompt. The test uses a documentation-range address instead: `init` warns, then fails the bind. That turns the awkward constraint into the stronger assertion, because the warning is only observable if it precedes the bind, which is what an operator whose bind then fails depends on. Both mutations were checked: removing the call and moving it after the bind each fail the test. Captured through a global subscriber, since a warning is invisible to a test without one. Tests filter the captured lines by their own address so that events from tests running in parallel cannot be confused. The loopback case is here too. Warning on the shipped default would be worse than not warning at all, because operators learn to ignore it. --- core/connectors/runtime/src/api/mod.rs | 151 +++++++++++++++++++++++++ 1 file changed, 151 insertions(+) diff --git a/core/connectors/runtime/src/api/mod.rs b/core/connectors/runtime/src/api/mod.rs index 0e36f15796..5d161a64c9 100644 --- a/core/connectors/runtime/src/api/mod.rs +++ b/core/connectors/runtime/src/api/mod.rs @@ -168,7 +168,26 @@ async fn get_stats(State(context): State>) -> Json HttpConfig { HttpConfig { @@ -178,6 +197,138 @@ mod tests { } } + fn captured() -> &'static Mutex> { + static WARNINGS: OnceLock>> = OnceLock::new(); + WARNINGS.get_or_init(|| Mutex::new(Vec::new())) + } + + /// Installs the capture once for the whole test binary, since a global + /// subscriber can only be set once. Every test filters the captured lines + /// by its own address, so events from tests running in parallel cannot be + /// mistaken for each other. + fn capture_warnings() { + static INSTALLED: OnceLock<()> = OnceLock::new(); + INSTALLED.get_or_init(|| { + let subscriber = tracing_subscriber::registry().with(CaptureWarnings); + tracing::subscriber::set_global_default(subscriber) + .expect("no other test in this binary installs a subscriber"); + }); + } + + fn warned_about(address: &str) -> bool { + captured() + .lock() + .expect("the capture mutex is only held to push a line") + .iter() + .any(|warning| warning.contains(address)) + } + + struct CaptureWarnings; + + impl tracing_subscriber::Layer for CaptureWarnings { + fn on_event(&self, event: &tracing::Event<'_>, _context: LayerContext<'_, S>) { + if *event.metadata().level() != Level::WARN { + return; + } + let mut recorded = Recorded(String::new()); + event.record(&mut recorded); + captured() + .lock() + .expect("the capture mutex is only held to push a line") + .push(recorded.0); + } + } + + /// Every field the event carried, rendered into one line. + /// + /// Unconditional on purpose. These tests only ask whether a warning + /// mentioned a given address, so singling out the `message` field would add + /// a branch to the scaffolding whose other side nothing here would ever + /// take. `record_str` needs no impl either: it forwards here by default, + /// and a formatted `warn!` message arrives as `fmt::Arguments` regardless. + struct Recorded(String); + + impl Visit for Recorded { + fn record_debug(&mut self, _field: &Field, value: &dyn std::fmt::Debug) { + self.0.push_str(&format!("{value:?} ")); + } + } + + /// The cheapest context `init` will accept. Nothing here reaches Iggy: the + /// clients are never connected, and the warning is decided from the config + /// alone. + async fn context() -> (Arc, TempDir) { + let directory = tempfile::tempdir().expect("a temp dir must be available"); + let config_provider = + create_connectors_config_provider(&ConnectorsConfig::Local(LocalConnectorsConfig { + config_dir: directory.path().display().to_string(), + })) + .await + .expect("an empty config dir must initialize with no connectors"); + + let context = RuntimeContext { + sinks: SinkManager::new(vec![]), + sources: SourceManager::new(vec![]), + api_key: SecretString::from(String::new()), + config_provider: Arc::from(config_provider), + metrics: Arc::new(Metrics::init()), + start_time: IggyTimestamp::now(), + iggy_clients: Arc::new(IggyClients { + producer: IggyClient::default(), + consumer: IggyClient::default(), + }), + state_path: directory.path().display().to_string(), + }; + (Arc::new(context), directory) + } + + fn free_port() -> u16 { + std::net::TcpListener::bind("127.0.0.1:0") + .expect("the loopback interface must offer a port") + .local_addr() + .expect("a bound listener has an address") + .port() + } + + #[tokio::test] + async fn given_no_key_and_a_routable_address_when_initialized_should_warn_before_binding() { + capture_warnings(); + let (context, _directory) = context().await; + let config = config(UNASSIGNABLE_ROUTABLE_ADDRESS, ""); + + // `init` panics when the bind fails, which is what makes this the + // ordering test: the warning has to already be out by then, or an + // operator whose bind fails never learns the API was unauthenticated. + let bind_failed = tokio::spawn(async move { init(&config, context).await }) + .await + .is_err(); + + assert!( + bind_failed, + "a documentation-range address must not be bindable, or this test \ + would be exposing a port instead of exercising the warning" + ); + assert!( + warned_about(UNASSIGNABLE_ROUTABLE_ADDRESS), + "init must consult the guard and name the address it is exposing" + ); + } + + #[tokio::test] + async fn given_loopback_address_when_initialized_should_not_warn() { + capture_warnings(); + let address = format!("127.0.0.1:{}", free_port()); + let (context, _directory) = context().await; + + init(&config(&address, ""), context).await; + + assert!( + !warned_about(&address), + "the shipped posture is loopback with no key; warning about it \ + would teach operators to ignore the one that matters" + ); + } + #[test] fn given_loopback_address_and_no_key_when_checked_should_stay_quiet() { // The shipped posture. Warning here would train operators to ignore it.