You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "mlevkov (via GitHub)" <gi...@apache.org> on 2026/08/02 21:34:56 UTC
[PR] fix(connectors): warn when the runtime API is exposed without a key [iggy]
mlevkov opened a new pull request, #3804:
URL: https://github.com/apache/iggy/pull/3804
Addresses suggestions **2 and 3** of #3802. Deliberately does **not** close it —
suggestion 1 (redacting or gating the config responses) needs a maintainer
decision, explained at the bottom.
## The gap
Two defaults compose badly. `api_key = ""` means `resolve_api_key` waves every
request through:
```rust
if context.api_key.expose_secret().is_empty() {
return Ok(next.run(request).await);
}
```
and the configuration routes return plugin configuration exactly as parsed from
TOML — a database connection string, an S3 secret key, a webhook signing secret.
There is no redaction anywhere in `core/connectors/runtime/src`.
The shipped `address = "127.0.0.1:8081"` confines that to local processes, which
is a defensible posture for an admin API and is why this is a hardening gap
rather than a disclosure. The problem is what happens next: changing `address` to
`0.0.0.0` to reach the API from outside a container is an ordinary thing to do,
and it silently turns "local processes can read the credentials" into "the
network can", with no signal at any layer.
## What this changes
**A startup warning** when the API is enabled, `api_key` is empty, and the
address resolves beyond loopback. Warns rather than refuses to start — refusing
would break deployments that are exposed today, and that is the maintainers'
call, not mine.
It **resolves** the address rather than parsing it, which matters more than it
looks: `HttpConfig::default()` is `localhost:8081`, which is loopback but is not
a `SocketAddr`. A parse-based check would fire on the default config and train
operators to ignore the warning. An address that cannot resolve counts as
exposed, since it is about to fail the bind regardless.
Four unit tests cover the decision table (loopback / IPv6 loopback / `localhost`
/ `0.0.0.0` / a routable IP / key configured / unresolvable). These are the first
tests in `src/api/`, so they add a `mod tests` there. Mutation-checked: dropping
the `api_key` early return fails the configured-key case.
**Documentation.** The endpoint list said what each route returns but never that
the config routes return credentials, so a reader had no way to know exposing the
port exposes their secrets. Added an admonition under `## HTTP API`, and the
`api_key` comment in `config.toml` now says that empty disables authentication
rather than just calling the key "optional" — plus a note on why the default
address is loopback, which otherwise reads as an arbitrary default rather than
the control doing the confining.
## Why suggestion 1 is not here
"Redact credential-bearing fields in the config responses" cannot be done
faithfully at this layer: `plugin_config` is an opaque `serde_json::Value`, so the
runtime has no way to know which keys are credentials. A key-name heuristic
(`*password*`, `*token*`, `*secret*`, `*key*`) would both miss fields and redact
innocent ones, and silently returning altered config from an API operators may
read programmatically is its own hazard.
The alternative in the issue — requiring `api_key` for those specific routes
regardless of the global default — is a clean fix but a breaking behaviour change
for anyone consuming those routes unauthenticated today.
Both are defensible; picking between them is a maintainer decision, so I have
left #3802 open rather than guess. Happy to implement either.
Related: #3801, which is why the plugin-side `SecretString` annotations do not
help here — the runtime never routes through them.
## Verification
`cargo fmt --all --check`, `cargo sort --check --no-format --workspace`,
`cargo clippy -p iggy-connectors --all-features --all-targets -- -D warnings`,
`cargo test -p iggy-connectors` (126 pass), `taplo fmt --check`, `hawkeye check`,
`typos`, `markdownlint` — all exit 0.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): warn when the runtime API is exposed without a key [iggy]
Posted by "mlevkov (via GitHub)" <gi...@apache.org>.
mlevkov commented on PR #3804:
URL: https://github.com/apache/iggy/pull/3804#issuecomment-5228212998
/ready
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): warn when the runtime API is exposed without a key [iggy]
Posted by "mlevkov (via GitHub)" <gi...@apache.org>.
mlevkov commented on PR #3804:
URL: https://github.com/apache/iggy/pull/3804#issuecomment-5161229387
Added the test, so the 3 lines are no longer uncovered.
Reaching the warning requires a non-loopback address, and any non-loopback
address that *binds* would open a port on every interface for the length of the
test (on macOS that also trips the "accept incoming connections" prompt for
unsigned test binaries). So the test uses a documentation-range address,
`192.0.2.1:8081` (RFC 5737), which is never assignable on a real host: `init`
emits the warning and then fails the bind.
That constraint turns out to give the stronger assertion. The warning is only
observable at all if it precedes the bind, which is exactly what an operator
whose bind subsequently fails depends on. Both mutations fail the test:
- deleting the `if is_unauthenticated_beyond_loopback(config)` block from `init`
- moving it below the `TcpListener::bind`
The second is the one a plain "does it warn" test would have missed.
The loopback case is covered too, asserting the shipped default produces no
warning. Warning on the default posture would be worse than not warning, since
operators would learn to ignore it.
Capture is via a global subscriber installed once per test binary, because a
`warn!` is invisible to a test without one. Each test filters the captured lines
by its own address, so events from tests running in parallel cannot be mistaken
for each other.
No production code changed in this commit: `git diff` on it is 152 insertions and
0 deletions, all inside `#[cfg(test)]`.
`cargo test -p iggy-connectors` is 128 passing (was 126). fmt, sort, clippy
`-D warnings`, taplo, hawkeye, typos all exit 0.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): warn when the runtime API is exposed without a key [iggy]
Posted by "mlevkov (via GitHub)" <gi...@apache.org>.
mlevkov commented on PR #3804:
URL: https://github.com/apache/iggy/pull/3804#issuecomment-5229123116
Ran a self-review over this branch before handing it back, and it caught something I had introduced in the previous commit rather than inherited. Pushed `e1a3a8ae9`.
**The fix documented three exposure paths and warned about one of them.**
That commit added the README block naming three ways the containment goes away, and the guard only covered the first. So:
- `[http.cors] enabled = true` on the shipped loopback default, no key: silent. Loopback does not contain it, since a browser is a local process and the layer wraps outside auth, which is the whole point of the note you left. That is the case the warning most needed to catch and it was the one it missed.
- `address = "0.0.0.0:8081"`, `api_key` set, `http.tls` disabled: silent, because the guard returned early on a non-empty key. Having a key says nothing about whether the key and the credential-bearing responses cross the wire in clear.
`warn_on_weak_containment` now emits one warning per path. Separate rather than combined because they compose independently, and an operator who closes one has not closed the others. The address check drops its key test and becomes `resolves_beyond_loopback`, which is all it was ever classifying.
That also resolves the remediation wording you flagged, from the other direction than I first took it. Conditioning TLS advice on `config.tls.enabled` means it is a separate warning that only fires when TLS is actually off, so it can no longer tell an operator to enable something they already enabled, and the awkward "unless ... may cross in cleartext" clause is gone.
**`example_config/config.toml` was still carrying the old comments.** I had updated the embedded default and the README and missed the file operators actually copy, which defeated the point of the documentation half. Fixed, and the two no longer diverge.
**Tests.** Added the CORS path, the key-without-TLS path, and `enabled = false` (nothing should warn about an API that is not listening; without it, hoisting the guard above the enabled check would go unnoticed). The loopback assertion was matching "no WARN containing this address", which goes vacuous the moment the message is reworded, so it now asserts no warning at all, which the per-test capture makes safe. 129 passing.
All four mutation-checked: disabling either new warning block fails exactly its own test.
**Deliberately not changed, in case you would rather I did:**
- The address is resolved twice at startup, once here and once by `bind`. That is the direct cost of your "do not bind what you resolve" note, and I would rather pay it than lose the `localhost` to `[::1, 127.0.0.1]` fallback. Recorded in the doc comment as the trade rather than left implicit.
- Still the `Vec<SocketAddr>` form you suggested rather than a `peekable` variant.
- The ordering test still detects "warned before binding" via the bind panic. It now also asserts no `Started` event was captured, so the ordering claim rests on two observations instead of one.
Separately, while re-checking a claim in the README block I filed #3848: `restart_connector` asks the provider for `get_{sink,source}_config(key, None)`, which the local provider resolves to the highest version while the HTTP provider resolves to the active one, so a rollback via `PUT /configs/active` survives a process start and is undone by `POST /restart`. Not touched here.
Gate: fmt, sort, taplo, hawkeye, typos, markdownlint, clippy `-D warnings`, `cargo test -p iggy-connectors` 129 passing, all exit 0.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): warn when the runtime API is exposed without a key [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): warn when the runtime API is exposed without a key [iggy]
Posted by "mlevkov (via GitHub)" <gi...@apache.org>.
mlevkov commented on PR #3804:
URL: https://github.com/apache/iggy/pull/3804#issuecomment-5228212163
All ten addressed in `a6ebf76d1`. The first one changed the shape of the PR, so taking them in that order.
**The exposure is not read-only.** Correct, and I had the threat model wrong, not just the wording. I traced it: `PUBLIC_PATHS` is `["/", "/health"]`, `restart_connector` calls `get_sink_config(key, None)` then `start_connector(key, &config, ..)`, so write-then-restart is a repoint using the runtime's own Iggy credentials. The warning and the README now say read *and* rewrite, name `restart` and `DELETE`, and spell out the config-write plus activate plus restart chain rather than leaving it as "disclosure".
**CORS.** Confirmed the layer order: `.layer(auth)` then `.layer(cors)` puts CORS outermost, and shipped `allowed_origins = ["*"]` maps to `AllowOrigin::any()`. Added as one of three named ways the loopback containment goes away, with the point that a browser is a local process and that setting `api_key` closes it.
**TLS.** Also correct, and it made the old advice actively harmful: move the address, set the key, leave `http.tls` alone, and the key plus the credential-bearing responses all go out in clear. Named next to `api_key` in both the README and the warning.
**The endpoint list.** Added `DELETE /{sinks,sources}/{key}/configs` and `POST /{sinks,sources}/{key}/restart`. You were right that this was what made the read-only framing look correct.
**Blocking `getaddrinfo`.** Switched to `tokio::net::lookup_host`, same as `quic_client.rs:424`. Kept it classification-only and did not bind the result, for the `localhost` to `[::1, 127.0.0.1]` reason you flagged. The four predicate tests are now `#[tokio::test]`.
**Empty iterator.** Collecting to `Vec<SocketAddr>` and testing `is_empty()` first, so a resolve-to-nothing reports exposed rather than inheriting the vacuous `all`. Took the collapse to `is_ok_and` as far as the let-else form, which reads better now the empty case needs naming.
**The wrong rationale.** This one I would not have caught: `HttpConfig::default()` really is unreachable because the embedded `config.toml` is the first figment layer, so `127.0.0.1:8081` is the effective default and would have parsed. The doc comment now gives the real reason, `address` being free-form and accepting hostnames, and the test message that repeated the claim is fixed too.
**`config.api_key` vs `context.api_key`.** Left reading `config`, as you suggested, but the helper now takes the key as a parameter so a later test cannot set one and not the other.
**`free_port`.** Deleted, `127.0.0.1:0` straight into `config()`. You were right that nothing needed the port.
**Global subscriber.** Now per-test: `Arc<Mutex<Vec<(Level, String)>>>` behind a `set_default` guard, no statics, no shared buffer, and the address filtering that only existed because of sharing is gone. Added `.with_filter(LevelFilter::INFO)`, which fixes the `max_level_hint` problem you noted separately and deletes the hand-rolled level check. The tests document that they rely on `#[tokio::test]`'s current-thread runtime for the spawned task to see the thread-local subscriber, and that a multi-thread flavour would fail them rather than pass silently.
**`ip_nonlocal_bind`.** Comment corrected on both counts, including that a successful bind leaves a listener for the rest of the binary rather than the test.
**The missing positive control.** The best catch of the set. The loopback test asserted only an absence, so it would have kept passing if `enabled` ever defaulted to false, taking the only in-`init` loopback coverage with it. It now asserts `init` reached the listener first. Verified by mutating `if !config.enabled` to `if true`: the control fails with "init must reach the listener, or the assertion below proves nothing", where previously that mutation left the test green.
Re-mutated the wiring after the harness change, since I had rewritten the capture path: removing the guard from `init` still fails with "init must consult the guard and name the address it is exposing".
Gate: fmt, sort, taplo, hawkeye, typos, markdownlint, clippy `-D warnings`, `cargo test -p iggy-connectors` 128 passing, all exit 0.
Still open for your call, unchanged: whether #3802's redaction should be a key-name heuristic over opaque `plugin_config` JSON or a hard `api_key` requirement on the config routes. This PR does not close that issue.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): warn when the runtime API is exposed without a key [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio merged PR #3804:
URL: https://github.com/apache/iggy/pull/3804
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): warn when the runtime API is exposed without a key [iggy]
Posted by "mlevkov (via GitHub)" <gi...@apache.org>.
mlevkov commented on PR #3804:
URL: https://github.com/apache/iggy/pull/3804#issuecomment-5229123199
/ready
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): warn when the runtime API is exposed without a key [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): warn when the runtime API is exposed without a key [iggy]
Posted by "mlevkov (via GitHub)" <gi...@apache.org>.
mlevkov commented on PR #3804:
URL: https://github.com/apache/iggy/pull/3804#issuecomment-5161865157
/request-review @hubcio
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): warn when the runtime API is exposed without a key [iggy]
Posted by "mlevkov (via GitHub)" <gi...@apache.org>.
mlevkov commented on PR #3804:
URL: https://github.com/apache/iggy/pull/3804#issuecomment-5229179669
Rebased onto `01a64b2e0` (`b84590a2d`). All five of my open PRs were 35 commits behind, not just this one; the other four are rebased too. No conflicts anywhere, and the full local gate passes on each.
**A correction to my earlier comment on this PR.** I explained the `-59.11%` project figure as:
> Those languages therefore upload no coverage for this head commit, and Codecov scores all of their lines as misses against a fully-uploaded base.
That mechanism is wrong. `codecov.yml` has `flag_management.default_rules.carryforward: true`, and it was already set at the base commit I was describing, so untouched flags should carry their previous report forward rather than count as misses. I found this while reading #3822, whose rationale states the same thing.
The observation was real (head project 17.18% against a 75.72% base, hits 120684 to 23540 on a 40-line diff) and `codecov/project` passed throughout, so nothing about the PR changed. But I inferred the cause rather than verifying it, and I would rather flag that than leave a confident wrong explanation in the thread for someone to build on. I do not have a confirmed replacement mechanism and am not going to guess at a second one.
Two things from the rebase worth knowing:
- #3694 fixed the bash-4 builtins in `scripts/ci/license-headers.sh`, so it now runs on macOS bash 3.2. It passes on all five branches. I had been substituting `hawkeye check` for it locally.
- #3822 changed how the master coverage baseline is produced. Post-rebase the patch figure here is unchanged at 0 missed lines.
Gate on this branch after the rebase: fmt, sort, clippy `-D warnings`, `cargo test -p iggy-connectors` 129 passing, license-headers, typos, taplo, all exit 0.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): warn when the runtime API is exposed without a key [iggy]
Posted by "mlevkov (via GitHub)" <gi...@apache.org>.
mlevkov commented on PR #3804:
URL: https://github.com/apache/iggy/pull/3804#issuecomment-5229126579
/request-review @hubcio
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): warn when the runtime API is exposed without a key [iggy]
Posted by "mlevkov (via GitHub)" <gi...@apache.org>.
mlevkov commented on PR #3804:
URL: https://github.com/apache/iggy/pull/3804#issuecomment-5444563723
/request-review @hubcio
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): warn when the runtime API is exposed without a key [iggy]
Posted by "mlevkov (via GitHub)" <gi...@apache.org>.
mlevkov commented on PR #3804:
URL: https://github.com/apache/iggy/pull/3804#issuecomment-5444563482
/ready
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): warn when the runtime API is exposed without a key [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #3804:
URL: https://github.com/apache/iggy/pull/3804#issuecomment-5160478528
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/3804?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `38.00000%` with `31 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 16.62%. Comparing base ([`1453114`](https://app.codecov.io/gh/apache/iggy/commit/1453114da046107a54927d620050e497e4c06c87?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`009e1e0`](https://app.codecov.io/gh/apache/iggy/commit/009e1e05021610552b9b49a833c6c44aed470909?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/3804?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [core/connectors/runtime/src/api/mod.rs](https://app.codecov.io/gh/apache/iggy/pull/3804?src=pr&el=tree&filepath=core%2Fconnectors%2Fruntime%2Fsrc%2Fapi%2Fmod.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb25uZWN0b3JzL3J1bnRpbWUvc3JjL2FwaS9tb2QucnM=) | 38.00% | [31 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/3804?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
:x: Your patch check has failed because the patch coverage (38.00%) is below the target coverage (50.00%). You can increase the patch coverage or adjust the [target](https://docs.codecov.com/docs/commit-status#target) coverage.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #3804 +/- ##
=============================================
- Coverage 75.72% 16.62% -59.11%
Complexity 969 969
=============================================
Files 1322 1320 -2
Lines 159363 136997 -22366
Branches 132746 110380 -22366
=============================================
- Hits 120684 22778 -97906
- Misses 35041 113780 +78739
+ Partials 3638 439 -3199
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/3804/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/3804/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `1.33% <38.00%> (-74.37%)` | :arrow_down: |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/3804/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `62.71% <ø> (ø)` | |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/3804/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `72.29% <ø> (ø)` | |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/3804/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `93.10% <ø> (ø)` | |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/3804/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.52% <ø> (ø)` | |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/3804/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `95.23% <ø> (ø)` | |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/3804/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `43.08% <ø> (ø)` | |
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/3804?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [core/connectors/runtime/src/api/mod.rs](https://app.codecov.io/gh/apache/iggy/pull/3804?src=pr&el=tree&filepath=core%2Fconnectors%2Fruntime%2Fsrc%2Fapi%2Fmod.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb25uZWN0b3JzL3J1bnRpbWUvc3JjL2FwaS9tb2QucnM=) | `15.44% <38.00%> (-35.24%)` | :arrow_down: |
... and [787 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/3804/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): warn when the runtime API is exposed without a key [iggy]
Posted by "mlevkov (via GitHub)" <gi...@apache.org>.
mlevkov commented on PR #3804:
URL: https://github.com/apache/iggy/pull/3804#issuecomment-5161178811
### On the Codecov comment above
It is a snapshot taken before this PR's Rust coverage upload landed, and Codecov
never re-posted it, so the red X is stuck on the page. Current state from the
Codecov API (`api.codecov.io/api/v2/github/apache/repos/iggy/pulls/3804/`):
```
patch: 94.0% 47 hits / 3 misses / 0 partials
```
All 22 `codecov/*` checks pass, including `codecov/patch`. The comment was written
at 21:43:36 with the head commit at 7 upload sessions; the 8th landed at 21:43:39
and took the patch figure from 38% to 94%.
**The 3 genuinely uncovered added lines** are the call site in `init()`:
```rust
49 if is_unauthenticated_beyond_loopback(config) {
50 warn!(
54 }
```
The predicate itself has four tests covering the decision table (loopback, IPv6
loopback, `localhost`, `0.0.0.0`, a routable IP, key configured, unresolvable),
and dropping its `api_key` early return fails one of them. What is untested is
that `init()` consults it. Covering that means constructing a `RuntimeContext` and
binding real listeners to assert a log line was emitted, and it would still not
catch `init()` dropping the call, since a test of the predicate passes either way.
Happy to add it if you would rather have the line covered.
**The `-59.11%` project figure is an upload artifact, not a regression.** This PR
touches only Rust connectors plus docs, so the path-filtered language jobs resolve
to `noop` in `_detect.yml` and skip. Those languages therefore upload no coverage
for this head commit, and Codecov scores all of their lines as misses against a
fully-uploaded base: hits go 120684 to 23540 for a 40-line diff. `codecov/project`
passes for the same reason.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): warn when the runtime API is exposed without a key [iggy]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #3804:
URL: https://github.com/apache/iggy/pull/3804#issuecomment-5160447407
<!-- iggy-pr-triage-welcome -->
Thanks for the PR. It is labeled `S-waiting-on-review` and queued for review.
Slash commands (own line, regular comment) move it around the queue:
- `/ready` - back to `S-waiting-on-review` after addressing feedback
- `/author` - flip to `S-waiting-on-author` while you finish changes
- `/request-review @user-or-team` - request a reviewer
See [CONTRIBUTING.md](https://github.com/apache/iggy/blob/master/CONTRIBUTING.md#pr-triage-commands) for details.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): warn when the runtime API is exposed without a key [iggy]
Posted by "mlevkov (via GitHub)" <gi...@apache.org>.
mlevkov commented on PR #3804:
URL: https://github.com/apache/iggy/pull/3804#issuecomment-5444641080
Pushed `4a5b254`, and a correction to what I said above.
I wrote that `configure_cors` panicking on bad CORS config "is consistent with
bad config being fatal in this runtime". That is true of the runtime and I should
have checked the rest of the repo before offering it as a rationale, because
`core/server` already made the opposite call for the identically named field.
`core/server/src/http.rs` rejects a misplaced wildcard with
`IggyError::InvalidConfiguration` rather than letting `AllowOrigin::list` panic,
and `core/server/config.toml` documents the position rule right above
`allowed_origins`. So porting that guard is not a new design question, it is
catching the connectors copy up.
It is also the worse of the two: `api::init` runs at `main.rs:248`, after "All
sources and sinks spawned", so the panic takes down a runtime whose connectors
are already running. Until the guard is ported, the three shipped TOMLs now carry
the sentence the server's config has had all along:
```toml
# "*" is honored only as the first entry; anywhere else it panics at startup.
```
Happy to port `parse_cors_values` and the rejecting arm in a follow-up, or in
this PR if you would rather it not ship with only a comment.
One related note so the ordering does not bite later. The server trims each entry
before comparing, and the two predicates here deliberately do not, because
`configure_cors` does not either: `[" *"]` currently becomes a list entry no
`Origin` matches, so it allows nothing and there is nothing to warn about.
Trimming has to land in the predicate and the mapping together. Trimming only the
predicate warns about a closed config, and trimming only the mapping opens one
without a warning. That reasoning is now a comment on `allows_any_origin` so
whoever ports the server's handling sees it.
Also in this push: the `null` test was the fourth warning-emitting test to skip
the pointer assertion the others carry, which is the same omission I fixed for
the TLS test earlier in the round.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): warn when the runtime API is exposed without a key [iggy]
Posted by "mlevkov (via GitHub)" <gi...@apache.org>.
mlevkov commented on PR #3804:
URL: https://github.com/apache/iggy/pull/3804#issuecomment-5444560821
Rebased onto master first, which was 24 commits ahead. That turned out to matter:
`RuntimeContext::state_path` became a `StateStorageFactory` while this branch sat,
so the API test context no longer compiled. Fixed as its own commit before
anything else, so the rest of the diff is only about your comments.
Of the 23 inline comments, 13 are from the first round and 12 of those were
already answered by the three follow-up commits. I checked each against the code
rather than against my own commit messages before deciding not to touch it. The
thirteenth is the `ip_nonlocal_bind` one, which your later comment partly
retracts, so it is answered below with the newer version.
## The ten from this round
**`warn_on_weak_containment` ignores `allowed_origins`.** Correct, and the
covering test locked it in by leaving the list empty. The predicate now lives on
`HttpCorsConfig` and `configure_cors` calls it too, so the warning cannot drift
from the mapping it describes. The test sets `["*"]`, and a new one asserts a
pinned list produces no warning.
While fixing the false positive I found the false negative: the predicate also
has to catch `null`, which is what a browser sends from a sandboxed iframe, a
`data:` URL and a `file://` page. `AllowOrigin::list` echoes any listed value on
a match, so pinning `null` buys nothing an attacker cannot produce at will. It is
a narrower set than `*`, just as worthless to pin. That could not widen
`allows_any_origin`, since `configure_cors` still has to build a one-entry list
for it, so the warning gets `allows_unowned_origin`, defined in terms of the
other so the two cannot disagree about `*`.
**CORS allows writes, not just reads.** Fixed in the warn string and in the
README, with the mechanism named: the preflight is answered inside the CORS layer
before `resolve_api_key` runs. The README also now says the rewrite half needs
the method on `allowed_methods` and `content-type` on `allowed_headers`, both of
which the shipped block grants. I kept the warn string at "read and rewrite"
rather than hedging it, because that is true of the shipped block and hedging it
cost more characters than your next comment allows.
**`POST .../restart` is CORS-simple.** Documented as its own item, separate from
the three edits, because it is true of the defaults rather than of a change an
operator made. No startup warning covers it for that reason, and the README says
so. Also raised on #3802, whose first suggested fix would close it.
**`dlopen` reads like restart does it.** Split into its own paragraph:
`start_connector` reuses the container loaded at boot and only re-runs the
plugin's init, so a hostile `path` waits for the next process start. Deferred
rather than immediate.
**Restart uses the highest stored version.** Both endpoint rows now say so and
point at #3848. The notice paragraph carries the same qualifier, since it only
holds on the local provider.
**`config.toml` is weaker than the example.** All three copies now carry the
three-line version, byte-identical. The README's own TOML sample was the third
copy and had the same gap. I also put a caution above `[http.cors] enabled`, in
all three, since flipping that one flag is what voids the containment the
`[http]` comment leans on.
**Disabled-API test has no positive control.** It now asserts the
"HTTP API is disabled" line landed first.
**"Outlives the run" is wrong, and `ip_nonlocal_bind`.** Clause dropped from the
assertion message and from the const doc. Both tests pre-flight the bind, and on
a host that binds a documentation-range address anyway only the two
bind-dependent assertions stand down. They do not skip: a skipped test reports
`ok` with no signal, and the warning itself is required on every host. The test
at the second site was discarding its join result entirely, so it now asserts
the same thing as the first.
**Three warnings of 246/208/221 characters.** Now 197/212/206, one sentence each
plus a pointer, and each names the exact key to change. The pointer is a URL
rather than a repo path, because the published image carries the binary and its
licences and no source tree, so the reader most likely to see these lines has
nowhere to look one up. That URL is 76 characters, which is why the CORS warning
lands 4 over the 208 you measured while the other two drop by 49 and 15. Say the
word if you would rather have the four characters back than a fetchable link.
Two of them still interpolate the same address, which I left deliberately: they
fire independently on different runs, and dropping the address from either makes
that one useless on its own.
**Capture guard installed before the fixture.** Moved after `context()` in every
test.
## Found while doing the above, not by you
`Ipv6Addr::is_loopback` matches only `::1`, so `[::ffff:127.0.0.1]` bound to
127.0.0.1 and still warned. The classifier unmaps first now.
The README claimed "no redaction layer anywhere in the runtime", which one grep
disproves. The true and narrower claim is that nothing redacts plugin
configuration; the runtime does redact its own secrets. Scoped accordingly.
The `DELETE .../configs` rows I added in the first round said "configuration
versions". Each call removes exactly one, defaulting to the active version, which
is the more dangerous misreading of the two.
The TLS warning is the one that survives setting a key, and it was the one
warning the README never mentioned.
## Left alone on purpose
`configure_cors` panics on a wildcard anywhere but first, on an unparseable
origin or header, on an unknown method, and on `allow_credentials = true` with
`["*"]`. That is consistent with bad config being fatal in this runtime, and
making it recoverable is a signature change on a function with existing callers.
Happy to take it as a follow-up if you would rather it produced an error.
## What a second review round changed
Worth saying plainly, because it caught things I would have shipped. A mutation
run found three conjuncts in `warn_on_weak_containment` with no negative
coverage at all: dropping `unauthenticated`, `cors.enabled` or `!tls.enabled`
each left the whole suite green. The middle one is the shipped block, a wildcard
origin list that is not switched on, so losing that conjunct would have warned
on every stock deployment, which is the exact false positive your comment was
about. All three are covered now, and `configure_cors` has its first tests: the
shapes the predicates classify all build, and a wildcard past the first position
panics rather than allowing nothing, which is what the doc comment claims.
The same round found the README claiming `["null"]` is "no narrower than `*`"
(it is narrower, it is just as worthless to pin), the redaction sentence
asserting more than the code does, and the DELETE rows describing the local
provider's fallback as if every provider had it.
195 unit tests and the 8 `connectors::api` integration tests pass. fmt, sort,
clippy `--all-features --all-targets`, and the `RUSTDOCFLAGS=-D warnings` gate
from #3925 are all clean. Every guard was mutation-checked before the commit
that introduced it.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org