hubcio commented on code in PR #3804:
URL: https://github.com/apache/iggy/pull/3804#discussion_r3850833486
##########
core/connectors/runtime/README.md:
##########
@@ -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.
Review Comment:
`dlopen`ed on the next start is literally right but sitting at the end of
this paragraph it reads like restart does it. `start_connector` reuses the
container loaded at boot and only re-runs `init_sink` with the new
plugin_config - the only `Container::load` is in `sink.rs` / `source.rs` off
the boot path in `main.rs`. a hostile `path` sits there until the next runtime
process start.
worth spelling out, it's the difference between immediate code execution and
deferred.
##########
core/connectors/runtime/README.md:
##########
@@ -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
Review Comment:
this is framed as read-only but the shipped CORS block also allows writes.
`allowed_methods` is `["GET", "POST", "PUT", "DELETE"]` and `allowed_headers`
is `["content-type"]`, and tower-http answers the OPTIONS preflight inside the
CORS layer before auth ever runs, so a hostile page gets a green preflight for
`POST /sinks/{key}/configs` with a json body and the real post then walks past
the empty key.
that's the full publish-then-restart repoint you describe a few lines up,
not just disclosure. same gap in the warn string in `warn_on_weak_containment`
- both should say read and rewrite.
##########
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
Review Comment:
weaker than the same comment you added in `example_config/config.toml` -
that one also says the endpoints accept writes and points at `http.tls`. this
is the file that gets `include_str!`ed as the default config layer and it's
also `DEFAULT_CONFIG_PATH`, so it's the one people actually read and edit. use
the three-line version from the example in both.
##########
core/connectors/runtime/src/api/mod.rs:
##########
@@ -121,10 +124,341 @@ pub async fn init(config: &HttpConfig, context:
Arc<RuntimeContext>) {
});
}
+/// 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 {
Review Comment:
this ignores `allowed_origins`, so it warns on configs that expose nothing.
`configure_cors` maps an empty list to `AllowOrigin::default()`, which is an
empty list and emits no ACAO header at all, and a pinned list to
`AllowOrigin::list`, which a non-matching origin fails. only `first() == "*"`
becomes `any()`.
the shipped default is `["*"]` so a stock deployment gets warned correctly -
the false positive lands on whoever pinned their origins properly. gate on
`config.cors.allowed_origins.first().is_some_and(|origin| origin == "*")` to
match what `configure_cors` itself does.
the covering test leaves `HttpCorsConfig::default()`, so `allowed_origins`
is empty there - it asserts the warning fires in exactly the case where nothing
is allowed, which locks the false positive into the suite.
##########
core/connectors/runtime/README.md:
##########
@@ -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.
Review Comment:
also line 223. restart doesn't use the active configuration.
`restart_connector` calls `get_sink_config(key, None)` and the local provider
resolves `None` to `max_by_key(version)`, never touching
`.active_versions.toml` - only the http provider maps `None` to the active url,
and `config_type = "local"` is the default.
latest happens to equal active until someone pins an older version with `PUT
.../configs/active`. say highest stored configuration version here. boot and
restart disagreeing after a pin is pre-existing, worth its own issue.
##########
core/connectors/runtime/src/api/mod.rs:
##########
@@ -121,10 +124,341 @@ pub async fn init(config: &HttpConfig, context:
Arc<RuntimeContext>) {
});
}
+/// 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<SocketAddr> = 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<Arc<RuntimeContext>>) -> String {
context.metrics.get_formatted_output()
}
async fn get_stats(State(context): State<Arc<RuntimeContext>>) ->
Json<ConnectorRuntimeStats> {
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<Mutex<Vec<(Level, String)>>>;
+
+ 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<String> {
+ 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<S: tracing::Subscriber> tracing_subscriber::Layer<S> 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<RuntimeContext>, 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 \
Review Comment:
my earlier note was wrong and it ended up in the comment - a leaked listener
doesn't outlive the run. `#[tokio::test]` builds the runtime as a temporary in
its generated `return`, so it drops at end of statement and takes the spawned
task with it. drop that clause here and in the const doc at 215-217.
the `ip_nonlocal_bind=1` fragility is still real though: the bind panics
before the spawn, so on such a host `bind_failed` is false and this fails
loudly. pre-flight `std::net::TcpListener::bind(UNASSIGNABLE_ROUTABLE_ADDRESS)`
in the test body and bail early if it works. line 430 throws the join result
away entirely, so that one would stay green on the same host - same guard there.
##########
core/connectors/runtime/src/api/mod.rs:
##########
@@ -121,10 +124,341 @@ pub async fn init(config: &HttpConfig, context:
Arc<RuntimeContext>) {
});
}
+/// 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<SocketAddr> = 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<Arc<RuntimeContext>>) -> String {
context.metrics.get_formatted_output()
}
async fn get_stats(State(context): State<Arc<RuntimeContext>>) ->
Json<ConnectorRuntimeStats> {
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<Mutex<Vec<(Level, String)>>>;
+
+ 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<String> {
+ 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<S: tracing::Subscriber> tracing_subscriber::Layer<S> 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<RuntimeContext>, 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!(
Review Comment:
no positive control here, so this passes for any reason `init` returns
early. the loopback test above got one. `init` logs "HTTP API is disabled" at
info on this path and the capture layer admits info, so assert that line landed
before asserting nothing warned.
##########
core/connectors/runtime/src/api/mod.rs:
##########
@@ -121,10 +124,341 @@ pub async fn init(config: &HttpConfig, context:
Arc<RuntimeContext>) {
});
}
+/// 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.",
Review Comment:
246 / 208 / 221 chars across the three. keyless + routable + no-tls fires
two of them naming the same address, three with CORS on. one sentence each plus
a pointer to the runtime readme carries the same signal.
##########
core/connectors/runtime/README.md:
##########
@@ -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.
Review Comment:
one more way in, and CORS doesn't gate it. `POST .../restart` has no body
and no content-type, so it's a CORS-simple request - a `mode:'no-cors'` fetch
gets sent whatever `[http.cors]` says, since CORS gates reading the response,
not issuing the request. so on the shipped keyless loopback default, with CORS
off, any page the operator visits can restart any connector.
chrome's private network access blocks the public-origin case, firefox and
safari don't, and a local-origin page bypasses it everywhere. this list pins
browser reach on CORS alone, which isn't true for side-effect-only routes.
##########
core/connectors/runtime/src/api/mod.rs:
##########
@@ -121,10 +124,341 @@ pub async fn init(config: &HttpConfig, context:
Arc<RuntimeContext>) {
});
}
+/// 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<SocketAddr> = 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<Arc<RuntimeContext>>) -> String {
context.metrics.get_formatted_output()
}
async fn get_stats(State(context): State<Arc<RuntimeContext>>) ->
Json<ConnectorRuntimeStats> {
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<Mutex<Vec<(Level, String)>>>;
+
+ 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<String> {
+ 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<S: tracing::Subscriber> tracing_subscriber::Layer<S> 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<RuntimeContext>, 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();
Review Comment:
guard goes in before the fixture, so `create_connectors_config_provider`
logs into the same buffer. `warnings()` filters to warn so it's fine today, but
any future warn on the provider-construction path fails both negative tests for
a reason that has nothing to do with the guard under test. install the capture
after `context()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
