From 1e2e59bddceedd1aaca3c858d899c10d0e01f562 Mon Sep 17 00:00:00 2001 From: lingniu Date: Tue, 30 Jun 2026 01:01:39 +0800 Subject: [PATCH] fix: support yutong mqtt tls hostname override --- .../src/main/resources/application.yml | 1 + .../inbound/mqtt/client/MqttEndpointManager.java | 11 ++++++++++- .../ingest/inbound/mqtt/client/MqttTlsSupport.java | 3 +++ .../inbound/mqtt/config/MqttInboundProperties.java | 3 +++ .../ingest/inbound/mqtt/MqttTlsSupportTest.java | 14 ++++++++++++++ 5 files changed, 31 insertions(+), 1 deletion(-) diff --git a/modules/apps/yutong-mqtt-app/src/main/resources/application.yml b/modules/apps/yutong-mqtt-app/src/main/resources/application.yml index 5b84ccb0..a6efb018 100644 --- a/modules/apps/yutong-mqtt-app/src/main/resources/application.yml +++ b/modules/apps/yutong-mqtt-app/src/main/resources/application.yml @@ -47,6 +47,7 @@ lingniu: ca-pem: ${YUTONG_MQTT_TLS_CA_PEM:} client-pem: ${YUTONG_MQTT_TLS_CLIENT_PEM:} client-key: ${YUTONG_MQTT_TLS_CLIENT_KEY:} + hostname-verification-enabled: ${YUTONG_MQTT_TLS_HOSTNAME_VERIFICATION_ENABLED:true} pipeline: disruptor: ring-buffer-size: ${PIPELINE_RING_BUFFER_SIZE:131072} diff --git a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttEndpointManager.java b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttEndpointManager.java index 282a911b..ab9c1e5f 100644 --- a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttEndpointManager.java +++ b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttEndpointManager.java @@ -3,6 +3,7 @@ package com.lingniu.ingest.inbound.mqtt.client; import com.hivemq.client.mqtt.datatypes.MqttQos; import com.hivemq.client.mqtt.mqtt3.Mqtt3AsyncClient; import com.hivemq.client.mqtt.mqtt3.Mqtt3Client; +import com.hivemq.client.mqtt.mqtt3.message.connect.connack.Mqtt3ConnAck; import com.lingniu.ingest.api.ProtocolId; import com.lingniu.ingest.api.pipeline.RawFrame; import com.lingniu.ingest.core.dispatcher.Dispatcher; @@ -25,6 +26,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; /** @@ -62,10 +64,14 @@ public final class MqttEndpointManager implements AutoCloseable { } public void start() { + log.info("mqtt endpoint manager starting, endpoints={}", props.getEndpoints().size()); for (MqttInboundProperties.Endpoint ep : props.getEndpoints()) { try { // 多 endpoint 互不影响：单个连接初始化失败会派发 operationalError，但不会阻止其他 endpoint。 + log.info("mqtt endpoint [{}] initializing uri={} topic={} qos={} tls={}", + ep.getName(), ep.getUri(), ep.getTopic(), ep.getQos(), hasCustomTls(ep.getTls())); Mqtt3AsyncClient client = buildClient(ep); + log.info("mqtt endpoint [{}] client built", ep.getName()); clients.add(client); connectAndSubscribe(client, ep); } catch (Exception e) { @@ -116,7 +122,10 @@ public final class MqttEndpointManager implements AutoCloseable { .password(ep.getPassword() == null ? new byte[0] : ep.getPassword().getBytes()) .applySimpleAuth(); } - connectBuilder.send().whenComplete((ack, err) -> { + log.info("mqtt endpoint [{}] connect sending", ep.getName()); + CompletableFuture timeoutProbe = connectBuilder.send() + .orTimeout(Math.max(15, ep.getConnectionTimeoutSeconds() + 5L), TimeUnit.SECONDS); + timeoutProbe.whenComplete((ack, err) -> { if (err != null) { log.error("mqtt endpoint [{}] connect failed", ep.getName(), err); publishOperationalError(ep, "connect", err.getMessage() == null ? err.getClass().getSimpleName() : err.getMessage()); diff --git a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttTlsSupport.java b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttTlsSupport.java index 383ac9f3..0135da2b 100644 --- a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttTlsSupport.java +++ b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttTlsSupport.java @@ -34,6 +34,9 @@ public final class MqttTlsSupport { if (tls != null && hasText(tls.getClientPem()) && hasText(tls.getClientKey())) { builder.keyManagerFactory(keyManagerFactory(Path.of(tls.getClientPem()), Path.of(tls.getClientKey()))); } + if (tls != null && !tls.isHostnameVerificationEnabled()) { + builder.hostnameVerifier((hostname, session) -> true); + } return builder.build(); } catch (Exception e) { throw new IllegalStateException("mqtt tls config build failed", e); diff --git a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/config/MqttInboundProperties.java b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/config/MqttInboundProperties.java index e39e646c..d5241893 100644 --- a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/config/MqttInboundProperties.java +++ b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/config/MqttInboundProperties.java @@ -74,11 +74,14 @@ public class MqttInboundProperties { private String caPem; private String clientPem; private String clientKey; + private boolean hostnameVerificationEnabled = true; public String getCaPem() { return caPem; } public void setCaPem(String caPem) { this.caPem = caPem; } public String getClientPem() { return clientPem; } public void setClientPem(String clientPem) { this.clientPem = clientPem; } public String getClientKey() { return clientKey; } public void setClientKey(String clientKey) { this.clientKey = clientKey; } + public boolean isHostnameVerificationEnabled() { return hostnameVerificationEnabled; } + public void setHostnameVerificationEnabled(boolean hostnameVerificationEnabled) { this.hostnameVerificationEnabled = hostnameVerificationEnabled; } } } diff --git a/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttTlsSupportTest.java b/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttTlsSupportTest.java index 7012e012..ad41d63b 100644 --- a/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttTlsSupportTest.java +++ b/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttTlsSupportTest.java @@ -46,6 +46,20 @@ class MqttTlsSupportTest { assertThat(config.getTrustManagerFactory()).isPresent(); assertThat(config.getKeyManagerFactory()).isPresent(); + assertThat(config.getHostnameVerifier()).isEmpty(); + } + + @Test + void canDisableHostnameVerificationForLegacyMutualTlsBrokers() throws Exception { + Path ca = tempDir.resolve("ca.pem"); + Files.writeString(ca, CERT); + MqttInboundProperties.Tls tls = new MqttInboundProperties.Tls(); + tls.setCaPem(ca.toString()); + tls.setHostnameVerificationEnabled(false); + + MqttClientSslConfig config = MqttTlsSupport.build(tls); + + assertThat(config.getHostnameVerifier()).isPresent(); } private static final String CERT = """