hubcio commented on code in PR #3804:
URL: https://github.com/apache/iggy/pull/3804#discussion_r3701945783
##########
core/connectors/runtime/README.md:
##########
@@ -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
Review Comment:
the exposure is not read-only. `PUBLIC_PATHS` in `auth.rs` is only `/` and
`/health`, so `POST /sinks/{key}/configs`, `PUT .../configs/active`, `DELETE
.../configs` and `POST .../restart` all sit behind the same empty key.
`restart_connector()` re-reads the stored config and calls `init_sink()`
with its `plugin_config` and `setup_sink_consumers()` with its `streams`, so
rewrite + restart repoints a sink at an attacker destination and forwards your
topic data using the runtime's own credentials. the stored `path` gets
`dlopen`ed on the next start too.
that changes the decision this doc informs: "local processes can read my
secrets" is acceptable on a trusted network, "anyone reachable can repoint my
sinks" is not. worth naming write and reconfiguration, not just disclosure.
##########
core/connectors/runtime/src/api/mod.rs:
##########
@@ -121,10 +133,247 @@ pub async fn init(config: &HttpConfig, context:
Arc<RuntimeContext>) {
});
}
+/// 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() {
Review Comment:
this is `std::net::ToSocketAddrs`, so a blocking `getaddrinfo` on a tokio
worker inside an `async fn`. every sink and source task is already spawned by
the time `api::init` runs, so a dead resolver parks a worker they are using -
on a single-vcpu pod, the only one.
narrow in practice: the effective default `127.0.0.1:8081` is numeric and
std never calls the resolver, so only hostnames pay. still free to fix with
`tokio::net::lookup_host(&config.address).await`, same pattern as
`core/sdk/src/quic/quic_client.rs:424`. makes the predicate async, so the four
sync tests become `#[tokio::test]`.
one trap: do not bind the resolved `SocketAddr`. tokio's `bind` loops every
resolved address and takes the first that works, so collapsing to one kills the
`localhost` -> `[::1, 127.0.0.1]` fallback on ipv6-disabled hosts. resolve for
classification only, or collect to a `Vec<SocketAddr>` and bind `&addrs[..]`.
##########
core/connectors/runtime/README.md:
##########
@@ -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:
Review Comment:
list is missing every mutating route: `POST /sinks/{key}/restart`, `POST
/sources/{key}/restart`, and `DELETE` on both `configs` routes. `restart`
appears nowhere in this file, and the `DELETE` lines at 127-128 belong to the
config provider section. pre-existing, but a GET/POST/PUT-only list right under
the new notice is what makes the read-only framing look right.
##########
core/connectors/runtime/README.md:
##########
@@ -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.
Review Comment:
true literally, but a browser is a local process. shipped config pairs
`[http.cors] enabled = false` with `allowed_origins = ["*"]`, `configure_cors`
maps that to `AllowOrigin::any()`, and the CORS layer wraps outside auth. flip
`cors.enabled` alone and any page the operator visits reads a config endpoint
cross-origin - simple GET, no preflight, `ACAO: *`.
setting `api_key` closes it (attacker page cannot send the header, gets a
401). chrome's private network access blocks the public-origin case, firefox
and safari do not, and a local-origin page bypasses it everywhere. worth one
clause saying enabling `[http.cors]` voids this containment.
##########
core/connectors/runtime/src/api/mod.rs:
##########
@@ -41,6 +46,13 @@ pub async fn init(config: &HttpConfig, context:
Arc<RuntimeContext>) {
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.",
Review Comment:
same two gaps as the README block. "can read every connector secret"
undersells it - the config POST/PUT/DELETE routes and `/restart` are behind the
same empty key. and "Set http.api_key" omits that the key and the responses
cross in cleartext unless `http.tls.enabled`. message is already long, so maybe
"read or rewrite every connector configuration" plus naming `http.tls`.
##########
core/connectors/runtime/src/api/mod.rs:
##########
@@ -121,10 +133,247 @@ pub async fn init(config: &HttpConfig, context:
Arc<RuntimeContext>) {
});
}
+/// 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() {
Review Comment:
guard reads `config.api_key`, middleware enforces `context.api_key`. same
immutable binding in `main.rs`, no reload path, so nothing diverges today - the
warning just describes a value it does not read.
not free to change though: sourcing from `context` means the four cheap sync
tests have to build a `RuntimeContext` (tempdir + async provider). fine as is,
but the test helper hardcodes an empty context key with no parameter, so a
future `init` test passing a key would get unauthenticated middleware while the
predicate sees the key.
##########
core/connectors/runtime/src/api/mod.rs:
##########
@@ -121,10 +133,247 @@ pub async fn init(config: &HttpConfig, context:
Arc<RuntimeContext>) {
});
}
+/// 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()),
Review Comment:
`all()` on an empty iterator is vacuously true, so `Ok` with zero addresses
stays quiet - opposite of the policy three lines up. unreachable today
(getaddrinfo returns `EAI_NONAME`, and zero addresses cannot bind), so
consistency rather than a hole. the `Vec<SocketAddr>` form above handles it for
free.
unrelated: the match collapses to
`!config.address.to_socket_addrs().is_ok_and(|mut resolved|
resolved.all(|address| address.ip().is_loopback()))`, identical on all three
arms. one edit on this expression though, not two.
##########
core/connectors/runtime/src/api/mod.rs:
##########
@@ -121,10 +133,247 @@ pub async fn init(config: &HttpConfig, context:
Arc<RuntimeContext>) {
});
}
+/// 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<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, OnceLock};
+ use tempfile::TempDir;
+ use tracing::Level;
+ use tracing::field::{Field, Visit};
+ use tracing_subscriber::layer::{Context as LayerContext, SubscriberExt};
+
+ /// Reserved for documentation (RFC 5737), so it is never assignable on a
+ /// real host. Used to reach the warning without binding: any non-loopback
+ /// address that binds successfully would expose a port on every interface
+ /// for the duration of the test.
+ const UNASSIGNABLE_ROUTABLE_ADDRESS: &str = "192.0.2.1:8081";
+
+ fn config(address: &str, api_key: &str) -> HttpConfig {
+ HttpConfig {
+ address: address.to_owned(),
+ api_key: SecretString::from(api_key.to_owned()),
+ ..HttpConfig::default()
+ }
+ }
+
+ fn captured() -> &'static Mutex<Vec<String>> {
+ static WARNINGS: OnceLock<Mutex<Vec<String>>> = 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<S: tracing::Subscriber> tracing_subscriber::Layer<S> 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<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(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")
Review Comment:
bind, read port, drop, let `init` rebind is a race for no gain - lose it and
`init` panics with an unrelated message. window is small and nothing
realistically competes, so this is mostly about deleting code.
the test only asserts nothing warned, so it never needs a known port.
`"127.0.0.1:0"` straight into `config()` works: parses as loopback so the
predicate stays quiet, and the warn interpolates `config.address` verbatim so
`warned_about("127.0.0.1:0")` still discriminates. deletes this helper and the
race.
##########
core/connectors/runtime/src/api/mod.rs:
##########
@@ -121,10 +133,247 @@ pub async fn init(config: &HttpConfig, context:
Arc<RuntimeContext>) {
});
}
+/// 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`.
Review Comment:
this reason is wrong, and it is the justification for the guard's main
design decision so worth correcting rather than dropping.
`HttpConfig::default()` is unreachable in production: `config_provider`
always passes the embedded `config.toml` via `include_str!` as the first
figment layer and merges default -> file -> env, so `[http]` is always
complete. effective default is `127.0.0.1:8081`, which parses as a `SocketAddr`
fine - a parse-based check would not have fired on it.
resolving is still right, just for a different reason: `address` is a
free-form `String` and takes any hostname (`[iggy] address = "localhost:8090"`
in the same file). same claim repeated in the test message at line 345.
##########
core/connectors/runtime/README.md:
##########
@@ -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
Review Comment:
no mention of `http.tls`, which ships disabled. the key then travels as a
cleartext `api-key` header, and so do the responses - the verbatim plugin
configs this block exists to protect. follow this advice exactly (move address,
set key, leave tls alone) and every connector secret goes out in the clear.
worth naming `http.tls` next to `api_key`.
##########
core/connectors/runtime/src/api/mod.rs:
##########
@@ -121,10 +133,247 @@ pub async fn init(config: &HttpConfig, context:
Arc<RuntimeContext>) {
});
}
+/// 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<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, OnceLock};
+ use tempfile::TempDir;
+ use tracing::Level;
+ use tracing::field::{Field, Visit};
+ use tracing_subscriber::layer::{Context as LayerContext, SubscriberExt};
+
+ /// Reserved for documentation (RFC 5737), so it is never assignable on a
+ /// real host. Used to reach the warning without binding: any non-loopback
+ /// address that binds successfully would expose a port on every interface
+ /// for the duration of the test.
+ const UNASSIGNABLE_ROUTABLE_ADDRESS: &str = "192.0.2.1:8081";
Review Comment:
holds on stock hosts but not with `net.ipv4.ip_nonlocal_bind=1`, normal on
keepalived/haproxy vip boxes. there the bind succeeds, `bind_failed` is false,
and the assert fires with a message that is now misleading. loud, not silent,
so it is a contributor-machine hazard rather than a false green.
small correction to the comment: if the bind does succeed, `init` has
already spawned the server, so the listener stays up for the rest of the test
binary, not just this test.
##########
core/connectors/runtime/src/api/mod.rs:
##########
@@ -121,10 +133,247 @@ pub async fn init(config: &HttpConfig, context:
Arc<RuntimeContext>) {
});
}
+/// 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<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, OnceLock};
+ use tempfile::TempDir;
+ use tracing::Level;
+ use tracing::field::{Field, Visit};
+ use tracing_subscriber::layer::{Context as LayerContext, SubscriberExt};
+
+ /// Reserved for documentation (RFC 5737), so it is never assignable on a
+ /// real host. Used to reach the warning without binding: any non-loopback
+ /// address that binds successfully would expose a port on every interface
+ /// for the duration of the test.
+ const UNASSIGNABLE_ROUTABLE_ADDRESS: &str = "192.0.2.1:8081";
+
+ fn config(address: &str, api_key: &str) -> HttpConfig {
+ HttpConfig {
+ address: address.to_owned(),
+ api_key: SecretString::from(api_key.to_owned()),
+ ..HttpConfig::default()
+ }
+ }
+
+ fn captured() -> &'static Mutex<Vec<String>> {
+ static WARNINGS: OnceLock<Mutex<Vec<String>>> = 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<S: tracing::Subscriber> tracing_subscriber::Layer<S> 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<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(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;
Review Comment:
pure negative assertion with no positive control, so nothing proves `init`
reached the guard. it silently depends on `HttpConfig::default().enabled` being
true, which lives in another file and is never asserted - flip that default (a
plausible follow-up to this PR) and `init` early-returns, the test still
passes, and the only in-`init` loopback coverage quietly disappears. the
routable test has `assert!(bind_failed)` as its control; this one could assert
the server came up.
keep the test regardless - it is the only one that kills "init warns
unconditionally". mutate the `if` to `if true` and everything else stays green.
##########
core/connectors/runtime/src/api/mod.rs:
##########
@@ -121,10 +133,247 @@ pub async fn init(config: &HttpConfig, context:
Arc<RuntimeContext>) {
});
}
+/// 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<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, OnceLock};
+ use tempfile::TempDir;
+ use tracing::Level;
+ use tracing::field::{Field, Visit};
+ use tracing_subscriber::layer::{Context as LayerContext, SubscriberExt};
+
+ /// Reserved for documentation (RFC 5737), so it is never assignable on a
+ /// real host. Used to reach the warning without binding: any non-loopback
+ /// address that binds successfully would expose a port on every interface
+ /// for the duration of the test.
+ const UNASSIGNABLE_ROUTABLE_ADDRESS: &str = "192.0.2.1:8081";
+
+ fn config(address: &str, api_key: &str) -> HttpConfig {
+ HttpConfig {
+ address: address.to_owned(),
+ api_key: SecretString::from(api_key.to_owned()),
+ ..HttpConfig::default()
+ }
+ }
+
+ fn captured() -> &'static Mutex<Vec<String>> {
+ static WARNINGS: OnceLock<Mutex<Vec<String>>> = 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);
Review Comment:
`set_global_default` claims the process-wide subscriber slot for the whole
128-test binary. any future test installing its own subscriber (anything going
through `init_logging`, which calls `.init()`) hits the `expect` below. and the
shared `Vec` is never drained, so `warned_about` on a negative assertion is
hostage to warnings from anywhere in the crate.
a per-test `Arc<Mutex<Vec<String>>>` with a `set_default` guard drops both,
plus the address filtering that only exists because the vec is shared.
`benchmark.rs` has the shape (`CaptureLayer` + `FieldVisitor`, no statics),
though its `capture()` takes a sync `FnOnce()` so it is not directly reusable.
separate, worth doing either way: `CaptureWarnings` implements neither
`max_level_hint` nor `enabled`, so the global max level goes to `TRACE` and
every `trace!`/`debug!`/`info!` callsite in the binary stops short-circuiting.
scoping does not fix that - `.with_filter(LevelFilter::WARN)` does, and it
deletes the hand-rolled level check below.
--
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]
