This is an automated email from the ASF dual-hosted git repository.
spetz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new ab734f17f fix(connectors): validate connector key before it becomes a
path (#4083)
ab734f17f is described below
commit ab734f17f849497d828667b5fba51820c3c3b531
Author: Ryan Huang <[email protected]>
AuthorDate: Mon Sep 7 22:57:18 2026 +0800
fix(connectors): validate connector key before it becomes a path (#4083)
Closes #4058
---
core/connectors/runtime/README.md | 15 ++
core/connectors/runtime/src/api/error.rs | 15 +-
core/connectors/runtime/src/api/key.rs | 42 +++++
core/connectors/runtime/src/api/mod.rs | 1 +
core/connectors/runtime/src/api/sink.rs | 44 +++--
core/connectors/runtime/src/api/source.rs | 44 +++--
core/connectors/runtime/src/configs/connectors.rs | 182 +++++++++++++++++++--
.../src/configs/connectors/http_provider.rs | 12 +-
.../src/configs/connectors/local_provider.rs | 50 ++++--
core/connectors/runtime/src/error.rs | 6 +
core/connectors/runtime/src/main.rs | 30 +++-
core/integration/tests/connectors/api/endpoints.rs | 104 ++++++++++++
.../tests/connectors/api/key_validation.toml | 31 ++++
13 files changed, 505 insertions(+), 71 deletions(-)
diff --git a/core/connectors/runtime/README.md
b/core/connectors/runtime/README.md
index 23803704d..ec824abda 100644
--- a/core/connectors/runtime/README.md
+++ b/core/connectors/runtime/README.md
@@ -326,6 +326,21 @@ Currently, it does expose the following endpoints:
- `POST /sources/{key}/restart`: stop the source and start it again from its
highest stored configuration version, which on the local provider is not
necessarily the active one
([#3848](https://github.com/apache/iggy/issues/3848)).
- `GET /sources/{key}/transforms`: source transforms to be applied to the
fields.
+`{key}` is the connector key: at most 128 bytes of ASCII letters, digits, `-`,
+`_` and `.`, starting with a letter or digit. A decoded segment outside that
+rule, such as `..%2F..%2Fpwned`, is answered with `400 Bad Request` and the
+error code `invalid_connector_key` before the request reaches the configuration
+provider, because on the local provider the key becomes part of a filename
under
+`config_dir`, and on the HTTP provider part of a URL. An unencoded `/` splits
+the path and matches no route, so it is a `404`.
+
+Keys loaded from configuration files or the HTTP provider are not rejected, so
+existing deployments keep starting, but a key outside the rule is logged at
+startup and cannot be addressed through the API. Keys are case-sensitive to the
+runtime while the local provider maps them to filenames, so on a
+case-insensitive filesystem two keys that differ only by case share one file;
+prefer lowercase.
+
## Telemetry
The connector runtime supports OpenTelemetry for logs and traces. To enable
telemetry, add the following configuration:
diff --git a/core/connectors/runtime/src/api/error.rs
b/core/connectors/runtime/src/api/error.rs
index e22070c17..77125bad8 100644
--- a/core/connectors/runtime/src/api/error.rs
+++ b/core/connectors/runtime/src/api/error.rs
@@ -16,7 +16,7 @@
// under the License.
use crate::error::RuntimeError;
-use axum::{Json, http::StatusCode, response::IntoResponse};
+use axum::{Json, extract::rejection::PathRejection, http::StatusCode,
response::IntoResponse};
use serde::Serialize;
use thiserror::Error;
use tracing::error;
@@ -27,6 +27,8 @@ pub enum ApiError {
Error(#[from] RuntimeError),
#[error(transparent)]
JsonError(#[from] serde_json::Error),
+ #[error(transparent)]
+ PathRejection(#[from] PathRejection),
}
#[derive(Debug, Serialize)]
@@ -43,6 +45,7 @@ impl IntoResponse for ApiError {
let status_code = match error {
RuntimeError::MissingIggyCredentials =>
StatusCode::BAD_REQUEST,
RuntimeError::InvalidConfiguration(_) =>
StatusCode::BAD_REQUEST,
+ RuntimeError::InvalidConnectorKey(_) =>
StatusCode::BAD_REQUEST,
RuntimeError::CannotConvertConfiguration =>
StatusCode::BAD_REQUEST,
RuntimeError::SinkNotFound(_) => StatusCode::NOT_FOUND,
RuntimeError::SourceNotFound(_) => StatusCode::NOT_FOUND,
@@ -66,6 +69,16 @@ impl IntoResponse for ApiError {
}),
)
}
+ ApiError::PathRejection(rejection) => {
+ error!("There was a path error: {rejection}");
+ (
+ rejection.status(),
+ Json(ErrorResponse {
+ code: "invalid_path".to_owned(),
+ reason: rejection.body_text(),
+ }),
+ )
+ }
}
.into_response()
}
diff --git a/core/connectors/runtime/src/api/key.rs
b/core/connectors/runtime/src/api/key.rs
new file mode 100644
index 000000000..a6438c2a7
--- /dev/null
+++ b/core/connectors/runtime/src/api/key.rs
@@ -0,0 +1,42 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::error::ApiError;
+use crate::configs::connectors::ConnectorKey;
+use axum::extract::{FromRequestParts, Path};
+use axum::http::request::Parts;
+use serde::Deserialize;
+
+/// The `{key}` route segment parsed as a `ConnectorKey`. Deserializing
+/// `Path<ConnectorKey>` directly would answer a bad key with axum's plain-text
+/// rejection; going through `ApiError` keeps the `{code, reason}` envelope the
+/// rest of the API returns.
+pub struct KeyPath(pub ConnectorKey);
+
+#[derive(Deserialize)]
+struct KeyParam {
+ key: String,
+}
+
+impl<S: Send + Sync> FromRequestParts<S> for KeyPath {
+ type Rejection = ApiError;
+
+ async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<Self,
Self::Rejection> {
+ let Path(KeyParam { key }) = Path::from_request_parts(parts,
state).await?;
+ Ok(Self(ConnectorKey::try_from(key)?))
+ }
+}
diff --git a/core/connectors/runtime/src/api/mod.rs
b/core/connectors/runtime/src/api/mod.rs
index 62950fa6e..5669ede29 100644
--- a/core/connectors/runtime/src/api/mod.rs
+++ b/core/connectors/runtime/src/api/mod.rs
@@ -34,6 +34,7 @@ use tracing::{error, info, warn};
mod auth;
pub mod config;
mod error;
+mod key;
mod models;
mod sink;
mod source;
diff --git a/core/connectors/runtime/src/api/sink.rs
b/core/connectors/runtime/src/api/sink.rs
index c70f85a57..d99e89cf1 100644
--- a/core/connectors/runtime/src/api/sink.rs
+++ b/core/connectors/runtime/src/api/sink.rs
@@ -18,6 +18,7 @@
use super::{
config::map_connector_config,
error::ApiError,
+ key::KeyPath,
models::{SinkDetailsResponse, SinkInfoResponse, TransformResponse},
};
use crate::api::models::SinkConfigResponse;
@@ -69,10 +70,10 @@ async fn get_sinks(
async fn get_sink(
State(context): State<Arc<RuntimeContext>>,
- Path(key): Path<String>,
+ KeyPath(key): KeyPath,
) -> Result<Json<SinkDetailsResponse>, ApiError> {
let Some(sink) = context.sinks.get(&key).await else {
- return Err(ApiError::Error(RuntimeError::SinkNotFound(key)));
+ return Err(ApiError::Error(RuntimeError::SinkNotFound(key.into())));
};
let sink = sink.lock().await;
Ok(Json(SinkDetailsResponse {
@@ -83,11 +84,11 @@ async fn get_sink(
async fn get_sink_plugin_config(
State(context): State<Arc<RuntimeContext>>,
- Path(key): Path<String>,
+ KeyPath(key): KeyPath,
Query(query): Query<GetSinkConfig>,
) -> Result<impl IntoResponse, ApiError> {
let Some(sink) = context.sinks.get(&key).await else {
- return Err(ApiError::Error(RuntimeError::SinkNotFound(key)));
+ return Err(ApiError::Error(RuntimeError::SinkNotFound(key.into())));
};
let sink = sink.lock().await;
let Some(config) = sink.config.plugin_config.as_ref() else {
@@ -110,10 +111,10 @@ struct GetSinkConfig {
async fn get_sink_transforms(
State(context): State<Arc<RuntimeContext>>,
- Path(key): Path<String>,
+ KeyPath(key): KeyPath,
) -> Result<Json<Vec<TransformResponse>>, ApiError> {
let Some(sink) = context.sinks.get(&key).await else {
- return Err(ApiError::Error(RuntimeError::SinkNotFound(key)));
+ return Err(ApiError::Error(RuntimeError::SinkNotFound(key.into())));
};
let sink = sink.lock().await;
let Some(transforms) = sink.config.transforms.as_ref() else {
@@ -134,13 +135,13 @@ async fn get_sink_transforms(
async fn get_sink_configs(
State(context): State<Arc<RuntimeContext>>,
- Path((key,)): Path<(String,)>,
+ KeyPath(key): KeyPath,
) -> Result<Json<Vec<SinkConfigResponse>>, ApiError> {
let active_config = context
.sinks
.get_config(&key)
.await
- .ok_or(ApiError::Error(RuntimeError::SinkNotFound(key.clone())))?;
+ .ok_or_else(||
ApiError::Error(RuntimeError::SinkNotFound(key.to_string())))?;
let configs = context.config_provider.get_sink_configs(&key).await?;
let configs = configs
.into_iter()
@@ -154,12 +155,12 @@ async fn get_sink_configs(
async fn create_sink_config(
State(context): State<Arc<RuntimeContext>>,
- Path((key,)): Path<(String,)>,
+ KeyPath(key): KeyPath,
Json(config): Json<CreateSinkConfig>,
) -> Result<Json<SinkConfigResponse>, ApiError> {
let created_config = context
.config_provider
- .create_sink_config(&key, config.clone())
+ .create_sink_config(&key, config)
.await?;
Ok(Json(SinkConfigResponse {
@@ -168,15 +169,21 @@ async fn create_sink_config(
}))
}
+#[derive(Debug, Deserialize)]
+struct ConfigVersion {
+ version: u64,
+}
+
async fn get_sink_config(
State(context): State<Arc<RuntimeContext>>,
- Path((key, version)): Path<(String, u64)>,
+ KeyPath(key): KeyPath,
+ Path(ConfigVersion { version }): Path<ConfigVersion>,
) -> Result<Json<SinkConfigResponse>, ApiError> {
let active_config = context
.sinks
.get_config(&key)
.await
- .ok_or(ApiError::Error(RuntimeError::SinkNotFound(key.clone())))?;
+ .ok_or_else(||
ApiError::Error(RuntimeError::SinkNotFound(key.to_string())))?;
let config = context
.config_provider
@@ -192,20 +199,21 @@ async fn get_sink_config(
}))
}
None => Err(ApiError::Error(RuntimeError::SinkConfigNotFound(
- key, version,
+ key.into(),
+ version,
))),
}
}
async fn get_sink_active_config(
State(context): State<Arc<RuntimeContext>>,
- Path((key,)): Path<(String,)>,
+ KeyPath(key): KeyPath,
) -> Result<Json<SinkConfigResponse>, ApiError> {
let config = context
.sinks
.get_config(&key)
.await
- .ok_or(ApiError::Error(RuntimeError::SinkNotFound(key)))?;
+ .ok_or(ApiError::Error(RuntimeError::SinkNotFound(key.into())))?;
Ok(Json(SinkConfigResponse {
config,
active: true,
@@ -219,7 +227,7 @@ struct UpdateSinkActiveConfig {
async fn update_sink_active_config(
State(context): State<Arc<RuntimeContext>>,
- Path((key,)): Path<(String,)>,
+ KeyPath(key): KeyPath,
Json(update): Json<UpdateSinkActiveConfig>,
) -> Result<StatusCode, ApiError> {
context
@@ -236,7 +244,7 @@ struct DeleteSinkConfig {
async fn delete_sink_config(
State(context): State<Arc<RuntimeContext>>,
- Path((key,)): Path<(String,)>,
+ KeyPath(key): KeyPath,
Query(query): Query<DeleteSinkConfig>,
) -> Result<StatusCode, ApiError> {
context
@@ -248,7 +256,7 @@ async fn delete_sink_config(
async fn restart_sink(
State(context): State<Arc<RuntimeContext>>,
- Path(key): Path<String>,
+ KeyPath(key): KeyPath,
) -> Result<StatusCode, ApiError> {
context
.sinks
diff --git a/core/connectors/runtime/src/api/source.rs
b/core/connectors/runtime/src/api/source.rs
index a1c700edf..480c7dffa 100644
--- a/core/connectors/runtime/src/api/source.rs
+++ b/core/connectors/runtime/src/api/source.rs
@@ -18,6 +18,7 @@
use super::{
config::map_connector_config,
error::ApiError,
+ key::KeyPath,
models::{SourceDetailsResponse, SourceInfoResponse, TransformResponse},
};
use crate::api::models::SourceConfigResponse;
@@ -72,10 +73,10 @@ async fn get_sources(
async fn get_source(
State(context): State<Arc<RuntimeContext>>,
- Path(key): Path<String>,
+ KeyPath(key): KeyPath,
) -> Result<Json<SourceDetailsResponse>, ApiError> {
let Some(source) = context.sources.get(&key).await else {
- return Err(ApiError::Error(RuntimeError::SourceNotFound(key)));
+ return Err(ApiError::Error(RuntimeError::SourceNotFound(key.into())));
};
let source = source.lock().await;
Ok(Json(SourceDetailsResponse {
@@ -86,11 +87,11 @@ async fn get_source(
async fn get_source_plugin_config(
State(context): State<Arc<RuntimeContext>>,
- Path(key): Path<String>,
+ KeyPath(key): KeyPath,
Query(query): Query<GetSourceConfig>,
) -> Result<impl IntoResponse, ApiError> {
let Some(source) = context.sources.get(&key).await else {
- return Err(ApiError::Error(RuntimeError::SourceNotFound(key)));
+ return Err(ApiError::Error(RuntimeError::SourceNotFound(key.into())));
};
let source = source.lock().await;
let Some(config) = source.config.plugin_config.as_ref() else {
@@ -113,10 +114,10 @@ struct GetSourceConfig {
async fn get_source_transforms(
State(context): State<Arc<RuntimeContext>>,
- Path(key): Path<String>,
+ KeyPath(key): KeyPath,
) -> Result<Json<Vec<TransformResponse>>, ApiError> {
let Some(source) = context.sources.get(&key).await else {
- return Err(ApiError::Error(RuntimeError::SourceNotFound(key)));
+ return Err(ApiError::Error(RuntimeError::SourceNotFound(key.into())));
};
let source = source.lock().await;
let Some(transforms) = source.config.transforms.as_ref() else {
@@ -137,13 +138,13 @@ async fn get_source_transforms(
async fn get_source_configs(
State(context): State<Arc<RuntimeContext>>,
- Path((key,)): Path<(String,)>,
+ KeyPath(key): KeyPath,
) -> Result<Json<Vec<SourceConfigResponse>>, ApiError> {
let active_config = context
.sources
.get_config(&key)
.await
- .ok_or(ApiError::Error(RuntimeError::SourceNotFound(key.clone())))?;
+ .ok_or_else(||
ApiError::Error(RuntimeError::SourceNotFound(key.to_string())))?;
let configs = context.config_provider.get_source_configs(&key).await?;
let configs = configs
.into_iter()
@@ -157,12 +158,12 @@ async fn get_source_configs(
async fn create_source_config(
State(context): State<Arc<RuntimeContext>>,
- Path((key,)): Path<(String,)>,
+ KeyPath(key): KeyPath,
Json(config): Json<CreateSourceConfig>,
) -> Result<Json<SourceConfigResponse>, ApiError> {
let created_config = context
.config_provider
- .create_source_config(&key, config.clone())
+ .create_source_config(&key, config)
.await?;
Ok(Json(SourceConfigResponse {
@@ -171,15 +172,21 @@ async fn create_source_config(
}))
}
+#[derive(Debug, Deserialize)]
+struct ConfigVersion {
+ version: u64,
+}
+
async fn get_source_config(
State(context): State<Arc<RuntimeContext>>,
- Path((key, version)): Path<(String, u64)>,
+ KeyPath(key): KeyPath,
+ Path(ConfigVersion { version }): Path<ConfigVersion>,
) -> Result<Json<SourceConfigResponse>, ApiError> {
let active_config = context
.sources
.get_config(&key)
.await
- .ok_or(ApiError::Error(RuntimeError::SourceNotFound(key.clone())))?;
+ .ok_or_else(||
ApiError::Error(RuntimeError::SourceNotFound(key.to_string())))?;
let config = context
.config_provider
@@ -195,20 +202,21 @@ async fn get_source_config(
}))
}
None => Err(ApiError::Error(RuntimeError::SourceConfigNotFound(
- key, version,
+ key.into(),
+ version,
))),
}
}
async fn get_source_active_config(
State(context): State<Arc<RuntimeContext>>,
- Path((key,)): Path<(String,)>,
+ KeyPath(key): KeyPath,
) -> Result<Json<SourceConfigResponse>, ApiError> {
let config = context
.sources
.get_config(&key)
.await
- .ok_or(ApiError::Error(RuntimeError::SourceNotFound(key)))?;
+ .ok_or(ApiError::Error(RuntimeError::SourceNotFound(key.into())))?;
Ok(Json(SourceConfigResponse {
config,
active: true,
@@ -222,7 +230,7 @@ struct UpdateSourceActiveConfig {
async fn update_source_active_config(
State(context): State<Arc<RuntimeContext>>,
- Path((key,)): Path<(String,)>,
+ KeyPath(key): KeyPath,
Json(update): Json<UpdateSourceActiveConfig>,
) -> Result<StatusCode, ApiError> {
context
@@ -239,7 +247,7 @@ struct DeleteSourceConfig {
async fn delete_source_config(
State(context): State<Arc<RuntimeContext>>,
- Path((key,)): Path<(String,)>,
+ KeyPath(key): KeyPath,
Query(query): Query<DeleteSourceConfig>,
) -> Result<StatusCode, ApiError> {
context
@@ -251,7 +259,7 @@ async fn delete_source_config(
async fn restart_source(
State(context): State<Arc<RuntimeContext>>,
- Path(key): Path<String>,
+ KeyPath(key): KeyPath,
) -> Result<StatusCode, ApiError> {
context
.sources
diff --git a/core/connectors/runtime/src/configs/connectors.rs
b/core/connectors/runtime/src/configs/connectors.rs
index 24647e870..ad6a6c435 100644
--- a/core/connectors/runtime/src/configs/connectors.rs
+++ b/core/connectors/runtime/src/configs/connectors.rs
@@ -30,7 +30,9 @@ use iggy_connector_sdk::transforms::TransformType;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::fmt::Formatter;
+use std::ops::Deref;
use std::path::PathBuf;
+use std::str::FromStr;
use strum::Display;
#[derive(
@@ -49,6 +51,74 @@ pub enum ConfigFormat {
Text,
}
+/// A connector key becomes part of a filename under the local provider's
+/// `config_dir` and of a URL under the HTTP provider, so it must stay a single
+/// path component. Requiring a leading letter or digit is what rules out `.`,
+/// `..` and hidden-file names outright, instead of relying on the `sink_` /
+/// `source_` filename prefix to neutralize them.
+#[derive(Debug)]
+pub struct ConnectorKey(String);
+
+impl ConnectorKey {
+ /// Leaves room for the `source_` prefix, the version suffix and the
+ /// `.toml` extension inside a 255-byte filename limit.
+ pub const MAX_LENGTH: usize = 128;
+
+ pub fn as_str(&self) -> &str {
+ &self.0
+ }
+
+ fn is_valid(key: &str) -> bool {
+ key.len() <= Self::MAX_LENGTH
+ && key.as_bytes().split_first().is_some_and(|(first, rest)| {
+ first.is_ascii_alphanumeric()
+ && rest.iter().all(|byte| {
+ byte.is_ascii_alphanumeric() || matches!(*byte, b'-' |
b'_' | b'.')
+ })
+ })
+ }
+}
+
+impl TryFrom<String> for ConnectorKey {
+ type Error = RuntimeError;
+
+ fn try_from(key: String) -> Result<Self, Self::Error> {
+ if Self::is_valid(&key) {
+ Ok(Self(key))
+ } else {
+ Err(RuntimeError::InvalidConnectorKey(key))
+ }
+ }
+}
+
+impl FromStr for ConnectorKey {
+ type Err = RuntimeError;
+
+ fn from_str(key: &str) -> Result<Self, Self::Err> {
+ Self::try_from(key.to_owned())
+ }
+}
+
+impl Deref for ConnectorKey {
+ type Target = str;
+
+ fn deref(&self) -> &Self::Target {
+ &self.0
+ }
+}
+
+impl From<ConnectorKey> for String {
+ fn from(key: ConnectorKey) -> Self {
+ key.0
+ }
+}
+
+impl std::fmt::Display for ConnectorKey {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.write_str(&self.0)
+ }
+}
+
#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(tag = "type", rename_all = "lowercase")]
pub enum ConnectorConfig {
@@ -87,17 +157,17 @@ pub struct CreateSinkConfig {
}
impl CreateSinkConfig {
- fn to_sink_config(&self, key: &str, version: u64) -> SinkConfig {
+ fn into_sink_config(self, key: &ConnectorKey, version: u64) -> SinkConfig {
SinkConfig {
- key: key.to_owned(),
+ key: key.to_string(),
enabled: self.enabled,
version,
- name: self.name.clone(),
- path: self.path.clone(),
- transforms: self.transforms.clone(),
- streams: self.streams.clone(),
+ name: self.name,
+ path: self.path,
+ transforms: self.transforms,
+ streams: self.streams,
plugin_config_format: self.plugin_config_format,
- plugin_config: self.plugin_config.clone(),
+ plugin_config: self.plugin_config,
verbose: self.verbose,
benchmark: self.benchmark,
}
@@ -140,17 +210,17 @@ pub struct CreateSourceConfig {
}
impl CreateSourceConfig {
- fn to_source_config(&self, key: &str, version: u64) -> SourceConfig {
+ fn into_source_config(self, key: &ConnectorKey, version: u64) ->
SourceConfig {
SourceConfig {
- key: key.to_owned(),
+ key: key.to_string(),
enabled: self.enabled,
version,
- name: self.name.clone(),
- path: self.path.clone(),
- transforms: self.transforms.clone(),
- streams: self.streams.clone(),
+ name: self.name,
+ path: self.path,
+ transforms: self.transforms,
+ streams: self.streams,
plugin_config_format: self.plugin_config_format,
- plugin_config: self.plugin_config.clone(),
+ plugin_config: self.plugin_config,
verbose: self.verbose,
benchmark: self.benchmark,
}
@@ -222,16 +292,19 @@ pub struct ConnectorConfigVersions {
pub sources: HashMap<String, ConnectorConfigVersionInfo>,
}
+/// Only the two `create_*` methods take a parsed key: they are where the local
+/// provider turns the key into a filename, so the type carries the proof that
+/// the API boundary already validated it. The other methods only compare keys.
#[async_trait]
pub trait ConnectorsConfigProvider: Send + Sync {
async fn create_sink_config(
&self,
- key: &str,
+ key: &ConnectorKey,
config: CreateSinkConfig,
) -> Result<SinkConfig, RuntimeError>;
async fn create_source_config(
&self,
- key: &str,
+ key: &ConnectorKey,
config: CreateSourceConfig,
) -> Result<SourceConfig, RuntimeError>;
async fn get_active_configs(&self) -> Result<ConnectorsConfig,
RuntimeError>;
@@ -411,3 +484,80 @@ impl ConnectorsConfig {
&self.sources
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn given_single_component_key_when_parsed_should_succeed() {
+ for key in ["postgres", "es-sink.v2_1", "A1", "9lives", "a.b-c_d"] {
+ let parsed: ConnectorKey = key
+ .parse()
+ .unwrap_or_else(|error| panic!("key {key:?} should be
accepted, got: {error}"));
+ assert_eq!(parsed.as_str(), key);
+ assert_eq!(parsed.to_string(), key);
+ }
+ }
+
+ #[test]
+ fn given_key_at_the_length_limit_when_parsed_should_succeed() {
+ let key = "k".repeat(ConnectorKey::MAX_LENGTH);
+ assert_eq!(key.parse::<ConnectorKey>().unwrap().as_str(), key);
+ }
+
+ #[test]
+ fn given_key_over_the_length_limit_when_parsed_should_fail() {
+ assert_rejected(&"k".repeat(ConnectorKey::MAX_LENGTH + 1));
+ }
+
+ #[test]
+ fn given_key_with_path_separator_when_parsed_should_fail() {
+ for key in ["../../pwned", "x/../../../tmp/pwn", "a/b", "a\\b",
"/abs"] {
+ assert_rejected(key);
+ }
+ }
+
+ #[test]
+ fn
given_key_that_is_a_dot_segment_or_hidden_name_when_parsed_should_fail() {
+ for key in [".", "..", "..evil", ".hidden"] {
+ assert_rejected(key);
+ }
+ }
+
+ #[test]
+ fn given_key_with_characters_outside_the_charset_when_parsed_should_fail()
{
+ for key in [
+ "",
+ "-leading-dash",
+ "_leading_underscore",
+ "with space",
+ "k\0ey",
+ "k\ney",
+ "ключ",
+ "a#b",
+ ] {
+ assert_rejected(key);
+ }
+ }
+
+ #[test]
+ fn given_owned_key_when_converted_should_apply_the_same_rule() {
+ let accepted = ConnectorKey::try_from("random".to_owned()).unwrap();
+ assert_eq!(accepted.as_str(), "random");
+
+ let rejected =
ConnectorKey::try_from("../pwned".to_owned()).unwrap_err();
+ assert!(
+ matches!(&rejected, RuntimeError::InvalidConnectorKey(key) if key
== "../pwned"),
+ "unexpected error: {rejected}"
+ );
+ }
+
+ fn assert_rejected(key: &str) {
+ let result = key.parse::<ConnectorKey>();
+ assert!(
+ matches!(&result, Err(RuntimeError::InvalidConnectorKey(rejected))
if rejected == key),
+ "key {key:?} should be rejected, got: {result:?}"
+ );
+ }
+}
diff --git a/core/connectors/runtime/src/configs/connectors/http_provider.rs
b/core/connectors/runtime/src/configs/connectors/http_provider.rs
index ea49e5e0d..8133719e9 100644
--- a/core/connectors/runtime/src/configs/connectors/http_provider.rs
+++ b/core/connectors/runtime/src/configs/connectors/http_provider.rs
@@ -21,8 +21,8 @@ mod url_builder;
use
crate::configs::connectors::http_provider::response_extractor::ResponseExtractor;
use crate::configs::connectors::http_provider::url_builder::{TemplateKeys,
UrlBuilder};
use crate::configs::connectors::{
- ConnectorConfigVersions, ConnectorsConfig, ConnectorsConfigProvider,
CreateSinkConfig,
- CreateSourceConfig, SinkConfig, SourceConfig,
+ ConnectorConfigVersions, ConnectorKey, ConnectorsConfig,
ConnectorsConfigProvider,
+ CreateSinkConfig, CreateSourceConfig, SinkConfig, SourceConfig,
};
use crate::configs::runtime::{ResponseConfig, RetryConfig};
use crate::error::RuntimeError;
@@ -131,11 +131,11 @@ impl HttpConnectorsConfigProvider {
impl ConnectorsConfigProvider for HttpConnectorsConfigProvider {
async fn create_sink_config(
&self,
- key: &str,
+ key: &ConnectorKey,
config: CreateSinkConfig,
) -> Result<SinkConfig, RuntimeError> {
let mut vars = HashMap::new();
- vars.insert("key", key);
+ vars.insert("key", key.as_str());
let url = self.url_builder.build(TemplateKeys::CREATE_SINK, &vars);
let response = self
@@ -157,11 +157,11 @@ impl ConnectorsConfigProvider for
HttpConnectorsConfigProvider {
async fn create_source_config(
&self,
- key: &str,
+ key: &ConnectorKey,
config: CreateSourceConfig,
) -> Result<SourceConfig, RuntimeError> {
let mut vars = HashMap::new();
- vars.insert("key", key);
+ vars.insert("key", key.as_str());
let url = self.url_builder.build(TemplateKeys::CREATE_SOURCE, &vars);
let response = self
diff --git a/core/connectors/runtime/src/configs/connectors/local_provider.rs
b/core/connectors/runtime/src/configs/connectors/local_provider.rs
index 566ca48b5..5df18e238 100644
--- a/core/connectors/runtime/src/configs/connectors/local_provider.rs
+++ b/core/connectors/runtime/src/configs/connectors/local_provider.rs
@@ -16,8 +16,9 @@
// under the License.
use crate::configs::connectors::{
- ConnectorConfig, ConnectorConfigVersionInfo, ConnectorConfigVersions,
ConnectorsConfig,
- ConnectorsConfigProvider, CreateSinkConfig, CreateSourceConfig,
SinkConfig, SourceConfig,
+ ConnectorConfig, ConnectorConfigVersionInfo, ConnectorConfigVersions,
ConnectorKey,
+ ConnectorsConfig, ConnectorsConfigProvider, CreateSinkConfig,
CreateSourceConfig, SinkConfig,
+ SourceConfig,
};
use crate::error::RuntimeError;
use ::configs::{ConfigProvider, FileConfigProvider, TypedEnvProvider};
@@ -392,18 +393,18 @@ impl BaseConnectorConfig {
impl ConnectorsConfigProvider for LocalConnectorsConfigProvider<Initialized> {
async fn create_sink_config(
&self,
- key: &str,
+ key: &ConnectorKey,
cmd: CreateSinkConfig,
) -> Result<SinkConfig, RuntimeError> {
let sinks = self.state.connectors_config.sinks();
let next_version = sinks
.iter()
- .filter(|entry| entry.key().key == key)
+ .filter(|entry| entry.key().key == key.as_str())
.max_by_key(|entry| entry.config.version)
.map(|entry| entry.config.version + 1)
.unwrap_or(0);
- let config = cmd.to_sink_config(key, next_version);
+ let config = cmd.into_sink_config(key, next_version);
let connector_config = ConnectorConfig::Sink(config.clone());
let connector_id: ConnectorId = (&connector_config).into();
@@ -418,7 +419,7 @@ impl ConnectorsConfigProvider for
LocalConnectorsConfigProvider<Initialized> {
SinkConfigFile {
config: config.clone(),
created_at: Utc::now(),
- path: path.clone(),
+ path,
},
);
@@ -427,18 +428,18 @@ impl ConnectorsConfigProvider for
LocalConnectorsConfigProvider<Initialized> {
async fn create_source_config(
&self,
- key: &str,
+ key: &ConnectorKey,
cmd: CreateSourceConfig,
) -> Result<SourceConfig, RuntimeError> {
let sources = &self.state.connectors_config.sources;
let next_version = sources
.iter()
- .filter(|entry| entry.key().key == key)
+ .filter(|entry| entry.key().key == key.as_str())
.max_by_key(|entry| entry.config.version)
.map(|entry| entry.config.version + 1)
.unwrap_or(0);
- let config = cmd.to_source_config(key, next_version);
+ let config = cmd.into_source_config(key, next_version);
let connector_config = ConnectorConfig::Source(config.clone());
let connector_id: ConnectorId = (&connector_config).into();
@@ -453,7 +454,7 @@ impl ConnectorsConfigProvider for
LocalConnectorsConfigProvider<Initialized> {
SourceConfigFile {
config: config.clone(),
created_at: Utc::now(),
- path: path.clone(),
+ path,
},
);
@@ -851,3 +852,32 @@ impl Provider for ConnectorEnvProvider {
}
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use tempfile::TempDir;
+
+ #[tokio::test]
+ async fn
given_valid_key_when_creating_source_config_should_write_prefixed_file() {
+ let dir = TempDir::new().unwrap();
+ let provider =
LocalConnectorsConfigProvider::new(dir.path().to_str().unwrap())
+ .init()
+ .await
+ .unwrap();
+ let key: ConnectorKey = "random".parse().unwrap();
+
+ let config = provider
+ .create_source_config(&key, CreateSourceConfig::default())
+ .await
+ .unwrap();
+
+ assert_eq!(config.key, "random");
+ assert_eq!(config.version, 0);
+ let entries: Vec<String> = std::fs::read_dir(dir.path())
+ .unwrap()
+ .map(|entry|
entry.unwrap().file_name().to_string_lossy().into_owned())
+ .collect();
+ assert_eq!(entries, vec!["source_random_0.toml"]);
+ }
+}
diff --git a/core/connectors/runtime/src/error.rs
b/core/connectors/runtime/src/error.rs
index 35be3bb17..f0f5d11b0 100644
--- a/core/connectors/runtime/src/error.rs
+++ b/core/connectors/runtime/src/error.rs
@@ -21,6 +21,11 @@ use thiserror::Error;
pub enum RuntimeError {
#[error("Invalid configuration: {0}")]
InvalidConfiguration(String),
+ #[error(
+ "Invalid connector key {0:?}: expected at most {max_length} bytes of
ASCII letters, digits, '-', '_' or '.', starting with a letter or digit",
+ max_length = crate::configs::connectors::ConnectorKey::MAX_LENGTH
+ )]
+ InvalidConnectorKey(String),
#[error("Failed to serialize topic metadata")]
FailedToSerializeTopicMetadata,
#[error("Failed to serialize messages metadata")]
@@ -79,6 +84,7 @@ impl RuntimeError {
RuntimeError::SourceConfigNotFound(_, _) =>
"source_config_not_found",
RuntimeError::MissingIggyCredentials => "invalid_configuration",
RuntimeError::InvalidConfiguration(_) => "invalid_configuration",
+ RuntimeError::InvalidConnectorKey(_) => "invalid_connector_key",
RuntimeError::HttpRequestFailed(_) => "http_request_failed",
RuntimeError::StateLoadFailed { .. } => "state_load_failed",
RuntimeError::TokenFileNotFound(_) => "invalid_configuration",
diff --git a/core/connectors/runtime/src/main.rs
b/core/connectors/runtime/src/main.rs
index 08311d7d1..fcf8cafa9 100644
--- a/core/connectors/runtime/src/main.rs
+++ b/core/connectors/runtime/src/main.rs
@@ -15,7 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-use crate::configs::connectors::{ConnectorsConfigProvider,
create_connectors_config_provider};
+use crate::configs::connectors::{
+ ConnectorKey, ConnectorsConfig, ConnectorsConfigProvider,
create_connectors_config_provider,
+};
use ::configs::ConfigProvider;
use clap::Parser;
use configs::connectors::ConfigFormat;
@@ -40,7 +42,7 @@ use std::{
sync::{Arc, atomic::AtomicU32},
};
use system_stats::capture_allowed_cpus;
-use tracing::{error, info};
+use tracing::{error, info, warn};
mod api;
mod benchmark;
@@ -155,6 +157,7 @@ async fn main() -> Result<(), RuntimeError> {
connectors_config.sources().len(),
connectors_config.sinks().len()
);
+ warn_on_unaddressable_keys(&connectors_config);
let sources_config = connectors_config.sources();
let (sources, failed_sources) = source::init(
sources_config.clone(),
@@ -306,6 +309,29 @@ async fn main() -> Result<(), RuntimeError> {
Ok(())
}
+/// Keys loaded from a provider are not held to `ConnectorKey`, so existing
+/// deployments keep starting, but the control API only routes keys that pass
+/// it. Say so at startup instead of letting the operator discover a 400.
+fn warn_on_unaddressable_keys(connectors_config: &ConnectorsConfig) {
+ let keys = connectors_config
+ .sinks()
+ .keys()
+ .map(|key| ("sink", key))
+ .chain(
+ connectors_config
+ .sources()
+ .keys()
+ .map(|key| ("source", key)),
+ );
+ for (connector_type, key) in keys {
+ if let Err(error) = key.parse::<ConnectorKey>() {
+ warn!(
+ "Loaded {connector_type} connector with key {key:?} that the
control API cannot address: {error}"
+ );
+ }
+ }
+}
+
/// Resolves a plugin shared library path from the connector config `path`
field.
///
/// Accepts both `plugin.so` and `plugin` (OS-specific extension appended if
missing).
diff --git a/core/integration/tests/connectors/api/endpoints.rs
b/core/integration/tests/connectors/api/endpoints.rs
index 97ebd72e4..d1d8c75ab 100644
--- a/core/integration/tests/connectors/api/endpoints.rs
+++ b/core/integration/tests/connectors/api/endpoints.rs
@@ -21,8 +21,15 @@ use iggy_connector_sdk::api::{
use integration::harness::seeds;
use integration::iggy_harness;
use reqwest::Client;
+use serde_json::{Value, json};
+use std::fs;
const API_KEY: &str = "test-api-key";
+/// `config_dir` of `key_validation.toml`, relative to the crate root, which is
+/// the working directory of both the test process and the spawned runtime. It
+/// lives under the gitignored `test_logs/` so a regression that writes a
config
+/// file cannot land in the source tree or be loaded by another test.
+const CONNECTORS_CONFIG_DIR: &str =
"../../test_logs/connectors_api_key_validation";
#[iggy_harness(
server(connectors_runtime(config_path =
"tests/connectors/api/config.toml")),
@@ -243,3 +250,100 @@ async fn
api_key_authentication_rejected_with_invalid_key(harness: &TestHarness)
assert_eq!(response.status(), 401);
}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path =
"tests/connectors/api/key_validation.toml")),
+ seed = seeds::connector_stream
+)]
+async fn key_endpoints_reject_key_that_is_not_a_single_path_component(harness:
&TestHarness) {
+ let api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
+ let client = Client::new();
+ let config = json!({
+ "enabled": false,
+ "name": "x",
+ "path": "/tmp/evil.so",
+ "streams": []
+ });
+ let config_dir_before = config_dir_entries();
+
+ // Each entry is a percent-encoded path segment: axum decodes it before
+ // handing it to the handler, so `..%2F..%2Fpwned` arrives as
`../../pwned`.
+ // The charset itself is covered by unit tests; these are the two probes
+ // from the issue report plus a hidden-file name.
+ let keys = ["..%2F..%2Fpwned", "x%2F..%2F..%2F..%2Ftmp%2Fpwn", ".hidden"];
+ for key in keys {
+ for kind in ["sources", "sinks"] {
+ let response = client
+ .post(format!("{api_address}/{kind}/{key}/configs"))
+ .header("api-key", API_KEY)
+ .json(&config)
+ .send()
+ .await
+ .unwrap();
+ assert_eq!(response.status(), 400, "POST /{kind}/{key}/configs");
+ let body: Value = response.json().await.unwrap();
+ assert_eq!(
+ body["code"], "invalid_connector_key",
+ "POST /{kind}/{key}/configs body: {body}"
+ );
+
+ let response = client
+ .get(format!("{api_address}/{kind}/{key}"))
+ .header("api-key", API_KEY)
+ .send()
+ .await
+ .unwrap();
+ assert_eq!(response.status(), 400, "GET /{kind}/{key}");
+ let body: Value = response.json().await.unwrap();
+ assert_eq!(
+ body["code"], "invalid_connector_key",
+ "GET /{kind}/{key} body: {body}"
+ );
+ }
+ }
+
+ assert_eq!(
+ config_dir_entries(),
+ config_dir_before,
+ "no config file should have been written"
+ );
+}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path =
"tests/connectors/api/config.toml")),
+ seed = seeds::connector_stream
+)]
+async fn key_endpoints_accept_single_component_key(harness: &TestHarness) {
+ let api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
+ let client = Client::new();
+
+ for (kind, code) in [("sources", "source_not_found"), ("sinks",
"sink_not_found")] {
+ let response = client
+ .get(format!("{api_address}/{kind}/postgres-cdc.v2_1"))
+ .header("api-key", API_KEY)
+ .send()
+ .await
+ .unwrap();
+ assert_eq!(response.status(), 404, "GET /{kind}/postgres-cdc.v2_1");
+ let body: Value = response.json().await.unwrap();
+ assert_eq!(
+ body["code"], code,
+ "GET /{kind}/postgres-cdc.v2_1 body: {body}"
+ );
+ }
+}
+
+fn config_dir_entries() -> Vec<String> {
+ let mut entries: Vec<String> = fs::read_dir(CONNECTORS_CONFIG_DIR)
+ .unwrap()
+ .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
+ .collect();
+ entries.sort();
+ entries
+}
diff --git a/core/integration/tests/connectors/api/key_validation.toml
b/core/integration/tests/connectors/api/key_validation.toml
new file mode 100644
index 000000000..72c2fa69a
--- /dev/null
+++ b/core/integration/tests/connectors/api/key_validation.toml
@@ -0,0 +1,31 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[http]
+enabled = true
+address = "0.0.0.0:0"
+api_key = "test-api-key"
+
+[http.metrics]
+enabled = true
+endpoint = "/metrics"
+
+[connectors]
+config_type = "local"
+# Gitignored: the key-validation test snapshots this directory and must
+# never leave a config file in the source tree.
+config_dir = "../../test_logs/connectors_api_key_validation"
