diff --git a/core/connectors/runtime/README.md b/core/connectors/runtime/README.md index 1c1339f49c..c828ef61f3 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,40 @@ cert_file = "core/certs/iggy_cert.pem" key_file = "core/certs/iggy_key.pem" ``` +> [!IMPORTANT] +> **Treat this API as privileged. It reads and it writes.** +> +> The configuration endpoints return plugin configuration exactly as stored, +> credentials included - a database connection string, an S3 secret key, a +> webhook signing secret. There is no redaction layer anywhere in the runtime. +> +> The exposure is not limited to disclosure. Publishing a configuration with +> `POST /{sinks,sources}/{key}/configs` and then calling `POST .../restart` is +> enough to repoint a connector at a destination of the caller's choosing, +> because `restart` re-reads the stored configuration and starts the connector +> from it. The runtime then forwards your topic data using its own Iggy +> credentials, and the stored plugin `path` is `dlopen`ed on the next start. +> `PUT .../configs/active` and `DELETE .../configs` sit behind the same key. +> +> `api_key` is empty by default, which means authentication is **off** by +> default. Only `/` and `/health` are exempt once it is set, so everything above +> sits behind that one empty string, and the loopback default `address` is what +> confines it to local processes. +> +> Three ways that containment goes away: +> +> - **Moving `address` off loopback.** Set `api_key` in the same edit. The +> runtime warns at startup when the address resolves beyond loopback with no +> key configured, but nothing prevents it. +> - **Enabling `[http.cors]`.** It ships `allowed_origins = ["*"]`, which becomes +> `AllowOrigin::any()`, and the CORS layer wraps *outside* authentication. A +> browser is a local process, so with CORS enabled and no key, any page the +> operator visits can read the configuration endpoints cross-origin. Setting +> `api_key` closes it, since an attacker's page cannot supply the header. +> - **Leaving `http.tls.enabled = false`.** It ships disabled, so the `api-key` +> header and the responses carrying your credentials both travel in cleartext. +> Enable TLS alongside `api_key` whenever this API leaves loopback. + Currently, it does expose the following endpoints: - `GET /`: welcome message. @@ -168,19 +204,23 @@ Currently, it does expose the following endpoints: - `GET /sinks/{key}`: sink details. - `GET /sinks/{key}/configs`: list of configuration versions for the sink. - `POST /sinks/{key}/configs`: add a new configuration version for the sink. +- `DELETE /sinks/{key}/configs`: delete configuration versions for the sink. - `GET /sinks/{key}/configs/{version}`: configuration details for a specific version. - `GET /sinks/{key}/configs/active`: active configuration details. - `PUT /sinks/{key}/configs/active`: activate a specific configuration version for the sink. - `GET /sinks/{key}/configs/plugin`: sink plugin config, including the optional `format` query parameter to specify the config format. +- `POST /sinks/{key}/restart`: stop the sink and start it again from its stored active configuration. - `GET /sinks/{key}/transforms`: sink transforms to be applied to the fields. - `GET /sources`: list of sources. - `GET /sources/{key}`: source details. - `GET /sources/{key}/configs`: list of configuration versions for the source. - `POST /sources/{key}/configs`: add a new configuration version for the source. +- `DELETE /sources/{key}/configs`: delete configuration versions for the source. - `GET /sources/{key}/configs/{version}`: configuration details for a specific version. - `GET /sources/{key}/configs/active`: active configuration details. - `PUT /sources/{key}/configs/active`: activate a specific configuration version for the source. - `GET /sources/{key}/configs/plugin`: source plugin config, including the optional `format` query parameter to specify the config format. +- `POST /sources/{key}/restart`: stop the source and start it again from its stored active configuration. - `GET /sources/{key}/transforms`: source transforms to be applied to the fields. ## Telemetry 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 diff --git a/core/connectors/runtime/example_config/config.toml b/core/connectors/runtime/example_config/config.toml index 4fad587f22..ce84c8d0c7 100644 --- a/core/connectors/runtime/example_config/config.toml +++ b/core/connectors/runtime/example_config/config.toml @@ -17,8 +17,11 @@ [http] # Optional HTTP API configuration enabled = true +# Loopback on purpose: the configuration endpoints return plugin credentials in +# plaintext and also accept writes. Set api_key in the same edit if you move +# this off loopback, and http.tls unless cleartext is acceptable. 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 diff --git a/core/connectors/runtime/src/api/mod.rs b/core/connectors/runtime/src/api/mod.rs index 38cd8f6ed1..01c5b1fd51 100644 --- a/core/connectors/runtime/src/api/mod.rs +++ b/core/connectors/runtime/src/api/mod.rs @@ -22,9 +22,10 @@ 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 secrecy::ExposeSecret; use std::{net::SocketAddr, path::PathBuf, sync::Arc}; -use tokio::spawn; -use tracing::{error, info}; +use tokio::{net::lookup_host, spawn}; +use tracing::{error, info, warn}; mod auth; pub mod config; @@ -41,6 +42,8 @@ pub async fn init(config: &HttpConfig, context: Arc) { return; } + warn_on_weak_containment(config).await; + let mut system_router = Router::new().route("/stats", get(get_stats)); if config.metrics.enabled { @@ -121,6 +124,65 @@ pub async fn init(config: &HttpConfig, context: Arc) { }); } +/// Warns once for each way this API is less contained than its defaults look. +/// +/// Separate warnings rather than one, because the three compose independently +/// and an operator who closes one has not necessarily closed the others. All +/// three are the paths the runtime README documents. +async fn warn_on_weak_containment(config: &HttpConfig) { + let unauthenticated = config.api_key.expose_secret().is_empty(); + let beyond_loopback = resolves_beyond_loopback(&config.address).await; + + if unauthenticated && beyond_loopback { + warn!( + "{NAME} HTTP API is enabled on {} with no api_key configured. Anyone able to reach that address can read or rewrite every connector configuration, credentials included, and restart connectors from it. Set http.api_key, or bind the API to loopback.", + config.address + ); + } + + // Loopback does not contain this one. A browser is a local process, and the + // CORS layer wraps outside authentication, so the shipped + // `allowed_origins = ["*"]` lets any page the operator visits read these + // endpoints cross-origin. + if unauthenticated && config.cors.enabled { + warn!( + "{NAME} HTTP API has http.cors enabled with no api_key configured. Any page the operator visits can read the configuration endpoints, credentials included, cross-origin. Set http.api_key, or disable http.cors." + ); + } + + if beyond_loopback && !config.tls.enabled { + warn!( + "{NAME} HTTP API is enabled on {} with http.tls disabled. The api-key header and the configuration responses carrying connector credentials both cross the network in cleartext. Enable http.tls, or bind the API to loopback.", + config.address + ); + } +} + +/// Whether `address` resolves to anything outside loopback. +/// +/// Resolves rather than parses because `address` is a free-form `String` that +/// takes a hostname, as `[iggy] address` does in the same file. Not because the +/// default needs it: the embedded `config.toml` is the first figment layer, so +/// the effective default is `127.0.0.1:8081` and would parse. An address that +/// cannot resolve counts as exposed, since it is about to fail the bind anyway +/// and staying quiet about one we could not classify is the wrong direction to +/// be wrong in. +/// +/// Classification only. Do not bind what this resolves: `TcpListener::bind` +/// walks every resolved address and takes the first that works, so collapsing +/// to one would drop the `localhost` -> `[::1, 127.0.0.1]` fallback on hosts +/// with IPv6 disabled. The cost is resolving twice at startup, which is the +/// trade for keeping that fallback. +async fn resolves_beyond_loopback(address: &str) -> bool { + let Ok(resolved) = lookup_host(address).await else { + return true; + }; + let addresses: Vec = resolved.collect(); + // Empty is reported as exposed rather than confined: `all` over nothing is + // vacuously true, which would quietly invert the policy above. + addresses.is_empty() || !addresses.iter().all(|address| address.ip().is_loopback()) +} + async fn get_metrics(State(context): State>) -> String { context.metrics.get_formatted_output() } @@ -128,3 +190,275 @@ 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 crate::configs::connectors::create_connectors_config_provider; + use crate::configs::runtime::{ConnectorsConfig, LocalConnectorsConfig}; + use crate::manager::sink::SinkManager; + use crate::manager::source::SourceManager; + use crate::metrics::Metrics; + use crate::stream::IggyClients; + use iggy::prelude::IggyClient; + use iggy_common::IggyTimestamp; + use secrecy::SecretString; + use std::sync::Mutex; + use tempfile::TempDir; + use tracing::Level; + use tracing::field::{Field, Visit}; + use tracing::subscriber::DefaultGuard; + use tracing_subscriber::Layer as _; + use tracing_subscriber::filter::LevelFilter; + use tracing_subscriber::layer::{Context as LayerContext, SubscriberExt}; + + /// Reserved for documentation by RFC 5737, so the bind fails and the test + /// reaches the warning without listening anywhere. A routable address that + /// binds would put a port on every interface for the life of the binary. + const UNASSIGNABLE_ROUTABLE_ADDRESS: &str = "192.0.2.1:8081"; + const EPHEMERAL_LOOPBACK_ADDRESS: &str = "127.0.0.1:0"; + + type Captured = Arc>>; + + fn config(address: &str, api_key: &str) -> HttpConfig { + HttpConfig { + address: address.to_owned(), + api_key: SecretString::from(api_key.to_owned()), + ..HttpConfig::default() + } + } + + /// Captures events for the current thread only, for as long as the guard + /// lives. Not a global subscriber: that slot is process-wide, and a shared + /// buffer would leave negative assertions hostage to the rest of the binary. + /// + /// `#[tokio::test]` builds a current-thread runtime, so a task spawned by + /// the test body sees this subscriber. Under a multi-thread flavour the + /// capture would come back empty and these tests would fail, not pass. + /// + /// Filtered rather than checking the level in `on_event`: a layer with no + /// filter reports no `max_level_hint`, which pushes the global max level to + /// TRACE and stops every callsite in the binary short-circuiting. + fn capture_events() -> (DefaultGuard, Captured) { + let captured: Captured = Arc::new(Mutex::new(Vec::new())); + let layer = CaptureEvents { + captured: Arc::clone(&captured), + } + .with_filter(LevelFilter::INFO); + let guard = tracing::subscriber::set_default(tracing_subscriber::registry().with(layer)); + (guard, captured) + } + + fn warnings(captured: &Captured) -> Vec { + captured + .lock() + .expect("the capture mutex is only held to push a line") + .iter() + .filter(|(level, _)| *level == Level::WARN) + .map(|(_, message)| message.clone()) + .collect() + } + + /// Whether `init` got as far as serving. The positive control for tests + /// whose real assertion is that something was not warned about. + fn started_serving(captured: &Captured) -> bool { + captured + .lock() + .expect("the capture mutex is only held to push a line") + .iter() + .any(|(level, message)| *level == Level::INFO && message.contains("Started")) + } + + struct CaptureEvents { + captured: Captured, + } + + impl tracing_subscriber::Layer for CaptureEvents { + fn on_event(&self, event: &tracing::Event<'_>, _context: LayerContext<'_, S>) { + let mut recorded = Recorded(String::new()); + event.record(&mut recorded); + self.captured + .lock() + .expect("the capture mutex is only held to push a line") + .push((*event.metadata().level(), recorded.0)); + } + } + + /// Unconditional on purpose: singling out the `message` field would add a + /// branch whose other side nothing here takes. + 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. + /// + /// `api_key` is a parameter because the guard reads `config.api_key` while + /// the middleware enforces `context.api_key`; a test that set only one + /// would exercise a state the runtime cannot reach. + async fn context(api_key: &str) -> (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(api_key.to_owned()), + 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) + } + + #[tokio::test] + async fn given_loopback_addresses_when_classified_should_report_contained() { + assert!(!resolves_beyond_loopback("127.0.0.1:8081").await); + assert!(!resolves_beyond_loopback("[::1]:8081").await); + assert!( + !resolves_beyond_loopback("localhost:8081").await, + "`address` accepts a hostname, and parsing alone would misjudge one" + ); + } + + #[tokio::test] + async fn given_routable_or_unresolvable_addresses_when_classified_should_report_exposed() { + assert!( + resolves_beyond_loopback("0.0.0.0:8081").await, + "binding every interface to reach the API from outside a container \ + is the case this exists to catch" + ); + assert!(resolves_beyond_loopback("192.0.2.10:8081").await); + // About to fail the bind regardless, so staying quiet about an address + // we cannot classify is the wrong direction to be wrong in. + assert!(resolves_beyond_loopback("not a valid address").await); + } + + #[tokio::test] + async fn given_no_key_and_a_routable_address_when_initialized_should_warn_before_binding() { + let (_capture, captured) = capture_events(); + 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 is expected to be unbindable. A host with \ + net.ipv4.ip_nonlocal_bind=1, which keepalived and haproxy boxes set, binds \ + it instead, and this test has then started a listener that outlives the run" + ); + assert!( + !started_serving(&captured), + "the bind must not have completed, or this proves nothing about ordering" + ); + assert!( + warnings(&captured) + .iter() + .any(|warning| warning.contains(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_serve_without_warning() { + let (_capture, captured) = capture_events(); + let (context, _directory) = context("").await; + + init(&config(EPHEMERAL_LOOPBACK_ADDRESS, ""), context).await; + + // Positive control first: without it the assertion below passes for any + // reason `init` returns early, including `enabled` ever defaulting to + // false, and the only in-`init` loopback coverage disappears silently. + assert!( + started_serving(&captured), + "init must reach the listener, or the assertion below proves nothing" + ); + // Any warning at all, not one matching this address: matching on the + // address goes vacuous the moment the message is reworded. + assert!( + warnings(&captured).is_empty(), + "the shipped posture is loopback with no key; warning about it would \ + teach operators to ignore the ones that matter: {:?}", + warnings(&captured) + ); + } + + #[tokio::test] + async fn given_cors_enabled_and_no_key_when_initialized_should_warn_despite_loopback() { + let (_capture, captured) = capture_events(); + let (context, _directory) = context("").await; + let mut config = config(EPHEMERAL_LOOPBACK_ADDRESS, ""); + config.cors.enabled = true; + + init(&config, context).await; + + assert!(started_serving(&captured)); + assert!( + warnings(&captured) + .iter() + .any(|warning| warning.contains("http.cors")), + "loopback does not contain CORS: a browser is a local process and the \ + layer wraps outside authentication" + ); + } + + #[tokio::test] + async fn given_a_key_but_no_tls_beyond_loopback_when_initialized_should_still_warn() { + let (_capture, captured) = capture_events(); + let (context, _directory) = context("configured").await; + let config = config(UNASSIGNABLE_ROUTABLE_ADDRESS, "configured"); + + let _ = tokio::spawn(async move { init(&config, context).await }).await; + + let warnings = warnings(&captured); + assert!( + warnings.iter().any(|warning| warning.contains("http.tls")), + "setting a key does not stop the key and the credential-bearing \ + responses crossing the network in cleartext" + ); + assert!( + !warnings + .iter() + .any(|warning| warning.contains("no api_key")), + "and the key that was set must not still be reported as missing" + ); + } + + #[tokio::test] + async fn given_a_disabled_api_when_initialized_should_warn_about_nothing() { + let (_capture, captured) = capture_events(); + let (context, _directory) = context("").await; + // Routable, keyless and untrusting in every direction, but switched off. + let mut config = config(UNASSIGNABLE_ROUTABLE_ADDRESS, ""); + config.enabled = false; + config.cors.enabled = true; + + init(&config, context).await; + + assert!( + warnings(&captured).is_empty(), + "an API that is not listening exposes nothing, and warning about one \ + is the false positive that teaches operators to ignore the rest: {:?}", + warnings(&captured) + ); + } +}