This is an automated email from the ASF dual-hosted git repository.
piotr 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 5a717d91b feat(connectors,mcp): implement TCP TLS connection to Iggy
(#2330)
5a717d91b is described below
commit 5a717d91b9f3603984899b8c369c7126931f9aa8
Author: Maciej Modzelewski <[email protected]>
AuthorDate: Tue Nov 11 09:18:41 2025 +0100
feat(connectors,mcp): implement TCP TLS connection to Iggy (#2330)
resolves #2319
---
core/ai/mcp/README.md | 5 ++++
core/ai/mcp/config.toml | 5 ++++
core/ai/mcp/src/configs.rs | 27 ++++++++++++++++--
core/ai/mcp/src/error.rs | 5 +++-
core/ai/mcp/src/stream.rs | 52 +++++++++++++++++++++++-----------
core/connectors/runtime/README.md | 5 ++++
core/connectors/runtime/config.toml | 5 ++++
core/connectors/runtime/src/configs.rs | 22 +++++++++++++-
core/connectors/runtime/src/error.rs | 5 +++-
core/connectors/runtime/src/stream.rs | 52 +++++++++++++++++++++++-----------
core/server/src/configs/displays.rs | 7 +++--
11 files changed, 149 insertions(+), 41 deletions(-)
diff --git a/core/ai/mcp/README.md b/core/ai/mcp/README.md
index 999ac7e84..ae5537180 100644
--- a/core/ai/mcp/README.md
+++ b/core/ai/mcp/README.md
@@ -18,6 +18,11 @@ password = "iggy"
# token = "secret" # Personal Access Token (PAT) can be used instead of
username and password
# consumer = "iggy-mcp" # Optional consumer name
+[iggy.tls] # Optional TLS configuration for Iggy TCP connection
+enabled = false
+ca_file = "core/certs/iggy_cert.pem"
+domain = "" # Optional domain for TLS connection
+
[http] # Optional HTTP API configuration
address = "127.0.0.1:8082"
path = "/mcp"
diff --git a/core/ai/mcp/config.toml b/core/ai/mcp/config.toml
index 41f5b5725..69ff5f349 100644
--- a/core/ai/mcp/config.toml
+++ b/core/ai/mcp/config.toml
@@ -42,6 +42,11 @@ password = "iggy"
token = "" # Personal Access Token (PAT) can be used instead of username and
password
consumer = "iggy-mcp" # Optional consumer name
+[iggy.tls] # Optional TLS configuration for Iggy TCP connection
+enabled = false
+ca_file = "core/certs/iggy_cert.pem"
+domain = "" # Optional domain for TLS connection
+
[permissions]
create = true
read = true
diff --git a/core/ai/mcp/src/configs.rs b/core/ai/mcp/src/configs.rs
index 14aaf6694..e29b33092 100644
--- a/core/ai/mcp/src/configs.rs
+++ b/core/ai/mcp/src/configs.rs
@@ -1,4 +1,5 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
+/*
+ * 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
@@ -45,6 +46,14 @@ pub struct IggyConfig {
pub password: String,
pub token: String,
pub consumer: String,
+ pub tls: IggyTlsConfig,
+}
+
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+pub struct IggyTlsConfig {
+ pub enabled: bool,
+ pub ca_file: String,
+ pub domain: Option<String>,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
@@ -111,6 +120,7 @@ impl Default for IggyConfig {
password: DEFAULT_ROOT_PASSWORD.to_owned(),
token: "".to_owned(),
consumer: "iggy-mcp".to_owned(),
+ tls: IggyTlsConfig::default(),
}
}
}
@@ -189,7 +199,7 @@ impl std::fmt::Display for IggyConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
- "{{ address: {}, username: {}, password: {}, token: {}, consumer:
{} }}",
+ "{{ address: {}, username: {}, password: {}, token: {}, consumer:
{}, tls: {} }}",
self.address,
self.username,
if !self.password.is_empty() {
@@ -198,7 +208,18 @@ impl std::fmt::Display for IggyConfig {
""
},
if !self.token.is_empty() { "****" } else { "" },
- self.consumer
+ self.consumer,
+ self.tls
+ )
+ }
+}
+
+impl std::fmt::Display for IggyTlsConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ enabled: {}, ca_file: {:?}, domain: {:?} }}",
+ self.enabled, self.ca_file, self.domain
)
}
}
diff --git a/core/ai/mcp/src/error.rs b/core/ai/mcp/src/error.rs
index efdcb606b..950056020 100644
--- a/core/ai/mcp/src/error.rs
+++ b/core/ai/mcp/src/error.rs
@@ -1,4 +1,5 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
+/*
+ * 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
@@ -30,6 +31,8 @@ pub enum McpRuntimeError {
IggyError(#[from] iggy::prelude::IggyError),
#[error("Missing Iggy credentials")]
MissingIggyCredentials,
+ #[error("Missing TLS certificate file")]
+ MissingTlsCertificateFile,
#[error("Failed to create Iggy consumer ID")]
FailedToCreateConsumerId,
#[error("Invalid API path")]
diff --git a/core/ai/mcp/src/stream.rs b/core/ai/mcp/src/stream.rs
index 4b56df035..9ef9be6f4 100644
--- a/core/ai/mcp/src/stream.rs
+++ b/core/ai/mcp/src/stream.rs
@@ -1,19 +1,21 @@
-// 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.
+/*
+ * 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 crate::{configs::IggyConfig, error::McpRuntimeError};
use iggy::prelude::{Client, IggyClient, IggyClientBuilder};
@@ -49,6 +51,24 @@ pub async fn init(config: IggyConfig) -> Result<IggyClient,
McpRuntimeError> {
format!("iggy://{username}:{password}@{address}")
};
+ let connection_string = if config.tls.enabled {
+ let ca_file = &config.tls.ca_file;
+ if ca_file.is_empty() {
+ error!("TLS CA file must be provided when TLS is enabled.");
+ return Err(McpRuntimeError::MissingTlsCertificateFile);
+ }
+ let domain = config
+ .tls
+ .domain
+ .clone()
+ .filter(|domain| !domain.is_empty())
+ .map(|domain| format!("&tls_domain={domain}"))
+ .unwrap_or_default();
+ format!("{connection_string}?tls=true&tls_ca_file={ca_file}{domain}")
+ } else {
+ connection_string
+ };
+
let client =
IggyClientBuilder::from_connection_string(&connection_string)?.build()?;
client.connect().await?;
Ok(client)
diff --git a/core/connectors/runtime/README.md
b/core/connectors/runtime/README.md
index 1de35950d..ef810688a 100644
--- a/core/connectors/runtime/README.md
+++ b/core/connectors/runtime/README.md
@@ -23,6 +23,11 @@ username = "iggy"
password = "iggy"
token = "" # Personal Access Token (PAT) can be used instead of username and
password
+[iggy.tls] # Optional TLS configuration for Iggy TCP connection
+enabled = false
+ca_file = "core/certs/iggy_cert.pem"
+domain = "" # Optional domain for TLS connection
+
[state]
path = "local_state"
diff --git a/core/connectors/runtime/config.toml
b/core/connectors/runtime/config.toml
index 291065c65..0ef577bdf 100644
--- a/core/connectors/runtime/config.toml
+++ b/core/connectors/runtime/config.toml
@@ -40,6 +40,11 @@ username = "iggy"
password = "iggy"
token = "" # Personal Access Token (PAT) can be used instead of username and
password
+[iggy.tls] # Optional TLS configuration for Iggy TCP connection
+enabled = false
+ca_file = "core/certs/iggy_cert.pem"
+domain = "" # Optional domain for TLS connection
+
[state]
path = "local_state"
diff --git a/core/connectors/runtime/src/configs.rs
b/core/connectors/runtime/src/configs.rs
index 8afc44ba8..137731f52 100644
--- a/core/connectors/runtime/src/configs.rs
+++ b/core/connectors/runtime/src/configs.rs
@@ -69,6 +69,14 @@ pub struct IggyConfig {
pub username: String,
pub password: String,
pub token: String,
+ pub tls: IggyTlsConfig,
+}
+
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+pub struct IggyTlsConfig {
+ pub enabled: bool,
+ pub ca_file: String,
+ pub domain: Option<String>,
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
@@ -167,7 +175,7 @@ impl std::fmt::Display for IggyConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
- "{{ address: {}, username: {}, password: {}, token: {} }}",
+ "{{ address: {}, username: {}, password: {}, token: {}, tls: {}
}}",
self.address,
self.username,
if !self.password.is_empty() {
@@ -176,6 +184,17 @@ impl std::fmt::Display for IggyConfig {
""
},
if !self.token.is_empty() { "****" } else { "" },
+ self.tls
+ )
+ }
+}
+
+impl std::fmt::Display for IggyTlsConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ enabled: {}, ca_file: {:?}, domain: {:?} }}",
+ self.enabled, self.ca_file, self.domain
)
}
}
@@ -286,6 +305,7 @@ impl Default for IggyConfig {
username: DEFAULT_ROOT_USERNAME.to_owned(),
password: DEFAULT_ROOT_PASSWORD.to_owned(),
token: "".to_owned(),
+ tls: IggyTlsConfig::default(),
}
}
}
diff --git a/core/connectors/runtime/src/error.rs
b/core/connectors/runtime/src/error.rs
index 227a4c5ec..640400bef 100644
--- a/core/connectors/runtime/src/error.rs
+++ b/core/connectors/runtime/src/error.rs
@@ -1,4 +1,5 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
+/*
+ * 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
@@ -38,6 +39,8 @@ pub enum RuntimeError {
IggyError(#[from] iggy::prelude::IggyError),
#[error("Missing Iggy credentials")]
MissingIggyCredentials,
+ #[error("Missing TLS certificate file")]
+ MissingTlsCertificateFile,
#[error("JSON error")]
JsonError(#[from] serde_json::Error),
#[error("Sink not found with key: {0}")]
diff --git a/core/connectors/runtime/src/stream.rs
b/core/connectors/runtime/src/stream.rs
index 51943ae1f..0b86f2f16 100644
--- a/core/connectors/runtime/src/stream.rs
+++ b/core/connectors/runtime/src/stream.rs
@@ -1,19 +1,21 @@
-// 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.
+/*
+ * 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 iggy::prelude::{Client, IggyClient, IggyClientBuilder};
use tracing::{error, info};
@@ -62,6 +64,24 @@ async fn create_client(config: &IggyConfig) ->
Result<IggyClient, RuntimeError>
format!("iggy://{username}:{password}@{address}")
};
+ let connection_string = if config.tls.enabled {
+ let ca_file = &config.tls.ca_file;
+ if ca_file.is_empty() {
+ error!("TLS CA file must be provided when TLS is enabled.");
+ return Err(RuntimeError::MissingTlsCertificateFile);
+ }
+ let domain = config
+ .tls
+ .domain
+ .clone()
+ .filter(|domain| !domain.is_empty())
+ .map(|domain| format!("&tls_domain={domain}"))
+ .unwrap_or_default();
+ format!("{connection_string}?tls=true&tls_ca_file={ca_file}{domain}")
+ } else {
+ connection_string
+ };
+
let client =
IggyClientBuilder::from_connection_string(&connection_string)?.build()?;
client.connect().await?;
Ok(client)
diff --git a/core/server/src/configs/displays.rs
b/core/server/src/configs/displays.rs
index 506e6eafe..f0749eece 100644
--- a/core/server/src/configs/displays.rs
+++ b/core/server/src/configs/displays.rs
@@ -1,4 +1,5 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
+/*
+ * 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
@@ -270,8 +271,8 @@ impl Display for TcpTlsConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
- "{{ enabled: {}, cert_file: {}, key_file: {} }}",
- self.enabled, self.cert_file, self.key_file
+ "{{ enabled: {}, self_signed: {}, cert_file: {}, key_file: {} }}",
+ self.enabled, self.self_signed, self.cert_file, self.key_file
)
}
}
