This is an automated email from the ASF dual-hosted git repository.
HTHou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 83f6303c253 Add TLS channel failure reporting hooks (#18299)
83f6303c253 is described below
commit 83f6303c253e186a1b44496f185e23cb2cd36edd
Author: Haonan <[email protected]>
AuthorDate: Mon Aug 24 20:34:19 2026 +0800
Add TLS channel failure reporting hooks (#18299)
---
external-service-impl/rest/pom.xml | 4 +
.../java/org/apache/iotdb/rest/RestService.java | 4 +
.../rest/TrustedChannelAuditHandshakeListener.java | 75 +++++++
.../TrustedChannelAuditHandshakeListenerTest.java | 96 +++++++++
.../service/thrift/ConfigNodeRPCService.java | 8 +-
.../iotdb/consensus/config/ConsensusConfig.java | 23 ++-
.../apache/iotdb/consensus/iot/IoTConsensus.java | 4 +-
.../iot/service/IoTConsensusRPCService.java | 16 +-
.../iotdb/consensus/pipe/IoTConsensusV2.java | 4 +-
.../pipe/service/IoTConsensusV2RPCService.java | 16 +-
.../db/consensus/DataRegionConsensusImpl.java | 3 +
.../execution/exchange/MPPDataExchangeService.java | 7 +-
.../db/service/DataNodeInternalRPCService.java | 8 +-
.../iotdb/db/service/ExternalRPCService.java | 14 +-
.../apache/iotdb/commons/i18n/CommonMessages.java | 2 +
.../apache/iotdb/commons/i18n/CommonMessages.java | 2 +
.../iotdb/commons/audit/AbstractAuditLogger.java | 75 +++++++
.../apache/iotdb/commons/audit/AuditEventType.java | 2 +-
...Type.java => TrustedChannelFailureHandler.java} | 42 +---
.../TrustedChannelAuditServerEventHandler.java | 152 ++++++++++++++
.../commons/audit/AbstractAuditLoggerTest.java | 152 ++++++++++++++
.../TrustedChannelAuditServerEventHandlerTest.java | 220 +++++++++++++++++++++
pom.xml | 5 +
23 files changed, 889 insertions(+), 45 deletions(-)
diff --git a/external-service-impl/rest/pom.xml
b/external-service-impl/rest/pom.xml
index f6cedd00796..fc064d28254 100644
--- a/external-service-impl/rest/pom.xml
+++ b/external-service-impl/rest/pom.xml
@@ -98,6 +98,10 @@
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-http</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-io</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.iotdb</groupId>
<artifactId>node-commons</artifactId>
diff --git
a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/RestService.java
b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/RestService.java
index 506facdb4ce..6ecba9aece8 100644
---
a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/RestService.java
+++
b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/RestService.java
@@ -16,6 +16,7 @@
*/
package org.apache.iotdb.rest;
+import org.apache.iotdb.db.audit.DNAuditLogger;
import org.apache.iotdb.db.conf.rest.IoTDBRestServiceConfig;
import org.apache.iotdb.db.conf.rest.IoTDBRestServiceDescriptor;
import org.apache.iotdb.externalservice.api.IExternalService;
@@ -79,6 +80,9 @@ public class RestService implements IExternalService {
new HttpConnectionFactory(httpsConfig));
httpsConnector.setPort(port);
httpsConnector.setIdleTimeout(idleTime);
+ httpsConnector.addBean(
+ new TrustedChannelAuditHandshakeListener(
+
DNAuditLogger.getInstance()::recordTrustedChannelFailureAuditLogIfNecessary));
server.addConnector(httpsConnector);
server.setHandler(constructServletContextHandler());
diff --git
a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/TrustedChannelAuditHandshakeListener.java
b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/TrustedChannelAuditHandshakeListener.java
new file mode 100644
index 00000000000..c4245efab5e
--- /dev/null
+++
b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/TrustedChannelAuditHandshakeListener.java
@@ -0,0 +1,75 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.rest;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.audit.TrustedChannelFailureHandler;
+
+import org.eclipse.jetty.io.EndPoint;
+import org.eclipse.jetty.io.ssl.SslHandshakeListener;
+
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.Objects;
+
+final class TrustedChannelAuditHandshakeListener implements
SslHandshakeListener {
+
+ private final TrustedChannelFailureHandler failureHandler;
+
+ TrustedChannelAuditHandshakeListener(TrustedChannelFailureHandler
failureHandler) {
+ this.failureHandler = Objects.requireNonNull(failureHandler);
+ }
+
+ @Override
+ public void handshakeFailed(Event event, Throwable failure) {
+ recordHandshakeFailure(event.getEndPoint(), failure);
+ }
+
+ void recordHandshakeFailure(EndPoint endPoint, Throwable failure) {
+ if (endPoint == null || failure == null) {
+ return;
+ }
+
+ TEndPoint initiator = toEndPoint(endPoint.getRemoteSocketAddress());
+ TEndPoint target = toEndPoint(endPoint.getLocalSocketAddress());
+ if (initiator == null || target == null) {
+ return;
+ }
+
+ try {
+ failureHandler.onFailure(failure, initiator, target);
+ } catch (RuntimeException auditFailure) {
+ if (auditFailure != failure) {
+ failure.addSuppressed(auditFailure);
+ }
+ }
+ }
+
+ private static TEndPoint toEndPoint(SocketAddress socketAddress) {
+ if (!(socketAddress instanceof InetSocketAddress inetSocketAddress)) {
+ return null;
+ }
+ String host =
+ inetSocketAddress.getAddress() == null
+ ? inetSocketAddress.getHostString()
+ : inetSocketAddress.getAddress().getHostAddress();
+ return new TEndPoint(host, inetSocketAddress.getPort());
+ }
+}
diff --git
a/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/TrustedChannelAuditHandshakeListenerTest.java
b/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/TrustedChannelAuditHandshakeListenerTest.java
new file mode 100644
index 00000000000..fbfdfca1c9d
--- /dev/null
+++
b/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/TrustedChannelAuditHandshakeListenerTest.java
@@ -0,0 +1,96 @@
+/*
+ * 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.
+ */
+package org.apache.iotdb.rest;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+
+import org.eclipse.jetty.io.EndPoint;
+import org.junit.Test;
+
+import javax.net.ssl.SSLHandshakeException;
+
+import java.lang.reflect.Proxy;
+import java.net.InetSocketAddress;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+
+public class TrustedChannelAuditHandshakeListenerTest {
+
+ @Test
+ public void testRecordSocketEndpointsOnHandshakeFailure() {
+ AtomicReference<Throwable> actualFailure = new AtomicReference<>();
+ AtomicReference<TEndPoint> actualInitiator = new AtomicReference<>();
+ AtomicReference<TEndPoint> actualTarget = new AtomicReference<>();
+ TrustedChannelAuditHandshakeListener listener =
+ new TrustedChannelAuditHandshakeListener(
+ (failure, initiator, target) -> {
+ actualFailure.set(failure);
+ actualInitiator.set(initiator);
+ actualTarget.set(target);
+ });
+ SSLHandshakeException failure = new SSLHandshakeException("test");
+ EndPoint endPoint =
+ newEndPoint(
+ new InetSocketAddress("192.0.2.10", 45123), new
InetSocketAddress("192.0.2.20", 18080));
+
+ listener.recordHandshakeFailure(endPoint, failure);
+
+ assertSame(failure, actualFailure.get());
+ assertEquals(new TEndPoint("192.0.2.10", 45123), actualInitiator.get());
+ assertEquals(new TEndPoint("192.0.2.20", 18080), actualTarget.get());
+ }
+
+ @Test
+ public void testReportingFailureDoesNotSelfSuppressHandshakeFailure() {
+ RuntimeException failure = new RuntimeException("test");
+ TrustedChannelAuditHandshakeListener listener =
+ new TrustedChannelAuditHandshakeListener(
+ (ignoredFailure, initiator, target) -> {
+ throw failure;
+ });
+ EndPoint endPoint =
+ newEndPoint(
+ new InetSocketAddress("192.0.2.10", 45123), new
InetSocketAddress("192.0.2.20", 18080));
+
+ listener.recordHandshakeFailure(endPoint, failure);
+
+ assertEquals(0, failure.getSuppressed().length);
+ }
+
+ private static EndPoint newEndPoint(
+ InetSocketAddress remoteAddress, InetSocketAddress localAddress) {
+ return (EndPoint)
+ Proxy.newProxyInstance(
+ EndPoint.class.getClassLoader(),
+ new Class<?>[] {EndPoint.class},
+ (proxy, method, args) -> {
+ switch (method.getName()) {
+ case "getRemoteAddress":
+ case "getRemoteSocketAddress":
+ return remoteAddress;
+ case "getLocalAddress":
+ case "getLocalSocketAddress":
+ return localAddress;
+ default:
+ return null;
+ }
+ });
+ }
+}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCService.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCService.java
index 15be2e90d43..f325da85ab0 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCService.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCService.java
@@ -18,6 +18,7 @@
*/
package org.apache.iotdb.confignode.service.thrift;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.conf.CommonConfig;
import org.apache.iotdb.commons.conf.CommonDescriptor;
@@ -25,6 +26,7 @@ import
org.apache.iotdb.commons.exception.runtime.RPCServiceException;
import org.apache.iotdb.commons.service.ServiceType;
import org.apache.iotdb.commons.service.ThriftService;
import org.apache.iotdb.commons.service.ThriftServiceThread;
+import org.apache.iotdb.commons.service.TrustedChannelAuditServerEventHandler;
import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
@@ -70,7 +72,11 @@ public class ConfigNodeRPCService extends ThriftService
implements ConfigNodeRPC
getBindPort(),
configConf.getCnRpcMaxConcurrentClientNum(),
configConf.getThriftServerAwaitTimeForStopService(),
- new ConfigNodeRPCServiceHandler(),
+ new TrustedChannelAuditServerEventHandler(
+ new ConfigNodeRPCServiceHandler(),
+ new TEndPoint(getBindIP(), getBindPort()),
+
configNodeRPCServiceProcessor.configManager.getAuditLogger()
+ ::recordTrustedChannelFailureAuditLogIfNecessary),
commonConfig.isRpcThriftCompressionEnabled(),
commonConfig.getKeyStorePath(),
commonConfig.getKeyStorePwd(),
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/ConsensusConfig.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/ConsensusConfig.java
index d54299992fe..114a8aed4fc 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/ConsensusConfig.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/ConsensusConfig.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.consensus.config;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.audit.TrustedChannelFailureHandler;
import org.apache.iotdb.commons.disk.strategy.DirectoryStrategyType;
import java.util.List;
@@ -37,6 +38,7 @@ public class ConsensusConfig {
private final IoTConsensusConfig iotConsensusConfig;
private final IoTConsensusV2Config iotConsensusV2Config;
private final DirectoryStrategyType directoryStrategyType;
+ private final TrustedChannelFailureHandler trustedChannelFailureHandler;
private ConsensusConfig(
TEndPoint thisNode,
@@ -47,7 +49,8 @@ public class ConsensusConfig {
RatisConfig ratisConfig,
IoTConsensusConfig iotConsensusConfig,
IoTConsensusV2Config iotConsensusV2Config,
- DirectoryStrategyType directoryStrategyType) {
+ DirectoryStrategyType directoryStrategyType,
+ TrustedChannelFailureHandler trustedChannelFailureHandler) {
this.thisNodeEndPoint = thisNode;
this.thisNodeId = thisNodeId;
this.storageDir = storageDir;
@@ -57,6 +60,7 @@ public class ConsensusConfig {
this.iotConsensusConfig = iotConsensusConfig;
this.iotConsensusV2Config = iotConsensusV2Config;
this.directoryStrategyType = directoryStrategyType;
+ this.trustedChannelFailureHandler = trustedChannelFailureHandler;
}
public TEndPoint getThisNodeEndPoint() {
@@ -95,6 +99,10 @@ public class ConsensusConfig {
return directoryStrategyType;
}
+ public TrustedChannelFailureHandler getTrustedChannelFailureHandler() {
+ return trustedChannelFailureHandler;
+ }
+
public static ConsensusConfig.Builder newBuilder() {
return new ConsensusConfig.Builder();
}
@@ -111,6 +119,8 @@ public class ConsensusConfig {
private IoTConsensusV2Config iotConsensusV2Config;
private DirectoryStrategyType directoryStrategyType =
DirectoryStrategyType.MIN_FOLDER_OCCUPIED_SPACE_FIRST_STRATEGY;
+ private TrustedChannelFailureHandler trustedChannelFailureHandler =
+ TrustedChannelFailureHandler.NO_OP;
public ConsensusConfig build() {
return new ConsensusConfig(
@@ -124,7 +134,8 @@ public class ConsensusConfig {
.orElseGet(() -> IoTConsensusConfig.newBuilder().build()),
Optional.ofNullable(iotConsensusV2Config)
.orElseGet(() -> IoTConsensusV2Config.newBuilder().build()),
- directoryStrategyType);
+ directoryStrategyType,
+ trustedChannelFailureHandler);
}
public Builder setThisNode(TEndPoint thisNode) {
@@ -171,5 +182,13 @@ public class ConsensusConfig {
this.directoryStrategyType = directoryStrategyType;
return this;
}
+
+ public Builder setTrustedChannelFailureHandler(
+ TrustedChannelFailureHandler trustedChannelFailureHandler) {
+ this.trustedChannelFailureHandler =
+ Optional.ofNullable(trustedChannelFailureHandler)
+ .orElse(TrustedChannelFailureHandler.NO_OP);
+ return this;
+ }
}
}
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
index aa7ecbbf4ec..477d8a5cb11 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
@@ -132,7 +132,9 @@ public class IoTConsensus implements IConsensus {
this.recvFolderStrategyType = config.getDirectoryStrategyType();
this.config = config.getIotConsensusConfig();
this.registry = registry;
- this.service = new IoTConsensusRPCService(thisNode,
config.getIotConsensusConfig());
+ this.service =
+ new IoTConsensusRPCService(
+ thisNode, config.getIotConsensusConfig(),
config.getTrustedChannelFailureHandler());
this.clientManager =
new IClientManager.Factory<TEndPoint, AsyncIoTConsensusServiceClient>()
.createClientManager(
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCService.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCService.java
index cfa175b2de5..d7dec0d9782 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCService.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCService.java
@@ -20,11 +20,13 @@
package org.apache.iotdb.consensus.iot.service;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.audit.TrustedChannelFailureHandler;
import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.exception.runtime.RPCServiceException;
import org.apache.iotdb.commons.service.ServiceType;
import org.apache.iotdb.commons.service.ThriftService;
import org.apache.iotdb.commons.service.ThriftServiceThread;
+import org.apache.iotdb.commons.service.TrustedChannelAuditServerEventHandler;
import org.apache.iotdb.consensus.config.IoTConsensusConfig;
import org.apache.iotdb.consensus.iot.thrift.IoTConsensusIService;
import org.apache.iotdb.rpc.ZeroCopyRpcTransportFactory;
@@ -40,11 +42,20 @@ public class IoTConsensusRPCService extends ThriftService
implements IoTConsensu
private final TEndPoint thisNode;
private final IoTConsensusConfig config;
+ private final TrustedChannelFailureHandler trustedChannelFailureHandler;
private IoTConsensusRPCServiceProcessor iotConsensusRPCServiceProcessor;
public IoTConsensusRPCService(TEndPoint thisNode, IoTConsensusConfig config)
{
+ this(thisNode, config, TrustedChannelFailureHandler.NO_OP);
+ }
+
+ public IoTConsensusRPCService(
+ TEndPoint thisNode,
+ IoTConsensusConfig config,
+ TrustedChannelFailureHandler trustedChannelFailureHandler) {
this.thisNode = thisNode;
this.config = config;
+ this.trustedChannelFailureHandler = trustedChannelFailureHandler;
}
@Override
@@ -83,7 +94,10 @@ public class IoTConsensusRPCService extends ThriftService
implements IoTConsensu
getBindPort(),
config.getRpc().getRpcMaxConcurrentClientNum(),
config.getRpc().getThriftServerAwaitTimeForStopService(),
- new
IoTConsensusRPCServiceHandler(iotConsensusRPCServiceProcessor),
+ new TrustedChannelAuditServerEventHandler(
+ new
IoTConsensusRPCServiceHandler(iotConsensusRPCServiceProcessor),
+ thisNode,
+ trustedChannelFailureHandler),
config.getRpc().isRpcThriftCompressionEnabled(),
config.getRpc().getSslKeyStorePath(),
config.getRpc().getSslKeyStorePassword(),
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java
index 4a207331564..7b9c2cc0e00 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java
@@ -108,7 +108,9 @@ public class IoTConsensusV2 implements IConsensus {
this.storageDir = new File(config.getStorageDir());
this.config = config.getIoTConsensusV2Config();
this.registry = registry;
- this.rpcService = new IoTConsensusV2RPCService(thisNode,
config.getIoTConsensusV2Config());
+ this.rpcService =
+ new IoTConsensusV2RPCService(
+ thisNode, config.getIoTConsensusV2Config(),
config.getTrustedChannelFailureHandler());
this.asyncClientManager =
IoTV2GlobalComponentContainer.getInstance().getGlobalAsyncClientManager();
this.syncClientManager =
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/service/IoTConsensusV2RPCService.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/service/IoTConsensusV2RPCService.java
index c218c8051e8..9739ade8e1d 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/service/IoTConsensusV2RPCService.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/service/IoTConsensusV2RPCService.java
@@ -20,11 +20,13 @@
package org.apache.iotdb.consensus.pipe.service;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.audit.TrustedChannelFailureHandler;
import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.exception.runtime.RPCServiceException;
import org.apache.iotdb.commons.service.ServiceType;
import org.apache.iotdb.commons.service.ThriftService;
import org.apache.iotdb.commons.service.ThriftServiceThread;
+import org.apache.iotdb.commons.service.TrustedChannelAuditServerEventHandler;
import org.apache.iotdb.consensus.config.IoTConsensusV2Config;
import org.apache.iotdb.consensus.iotconsensusv2.thrift.IoTConsensusV2IService;
import org.apache.iotdb.rpc.ZeroCopyRpcTransportFactory;
@@ -34,11 +36,20 @@ public class IoTConsensusV2RPCService extends ThriftService
private final TEndPoint thisNode;
private final IoTConsensusV2Config config;
+ private final TrustedChannelFailureHandler trustedChannelFailureHandler;
private IoTConsensusV2RPCServiceProcessor iotConsensusV2RPCServiceProcessor;
public IoTConsensusV2RPCService(TEndPoint thisNode, IoTConsensusV2Config
config) {
+ this(thisNode, config, TrustedChannelFailureHandler.NO_OP);
+ }
+
+ public IoTConsensusV2RPCService(
+ TEndPoint thisNode,
+ IoTConsensusV2Config config,
+ TrustedChannelFailureHandler trustedChannelFailureHandler) {
this.thisNode = thisNode;
this.config = config;
+ this.trustedChannelFailureHandler = trustedChannelFailureHandler;
}
@Override
@@ -71,7 +82,10 @@ public class IoTConsensusV2RPCService extends ThriftService
getBindPort(),
config.getRpc().getRpcMaxConcurrentClientNum(),
config.getRpc().getThriftServerAwaitTimeForStopService(),
- new
IoTConsensusV2RPCServiceHandler(iotConsensusV2RPCServiceProcessor),
+ new TrustedChannelAuditServerEventHandler(
+ new
IoTConsensusV2RPCServiceHandler(iotConsensusV2RPCServiceProcessor),
+ thisNode,
+ trustedChannelFailureHandler),
config.getRpc().isRpcThriftCompressionEnabled(),
config.getRpc().getSslKeyStorePath(),
config.getRpc().getSslKeyStorePassword(),
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java
index a4c8e00f1ad..fbda403b320 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java
@@ -39,6 +39,7 @@ import org.apache.iotdb.consensus.config.IoTConsensusV2Config;
import org.apache.iotdb.consensus.config.IoTConsensusV2Config.ReplicateMode;
import org.apache.iotdb.consensus.config.RatisConfig;
import org.apache.iotdb.consensus.config.RatisConfig.Snapshot;
+import org.apache.iotdb.db.audit.DNAuditLogger;
import org.apache.iotdb.db.conf.DataNodeMemoryConfig;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -140,6 +141,8 @@ public class DataRegionConsensusImpl {
return ConsensusConfig.newBuilder()
.setThisNodeId(CONF.getDataNodeId())
.setThisNode(new TEndPoint(CONF.getInternalAddress(),
CONF.getDataRegionConsensusPort()))
+ .setTrustedChannelFailureHandler(
+
DNAuditLogger.getInstance()::recordTrustedChannelFailureAuditLogIfNecessary)
.setStorageDir(CONF.getDataRegionConsensusDir())
.setRecvSnapshotDirs(Arrays.asList(CONF.getLocalDataDirs()))
// IoTConsensus always balances received snapshot files by least
occupied space,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeService.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeService.java
index a4cef2e4b71..6afcd2a3a91 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeService.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeService.java
@@ -32,7 +32,9 @@ import
org.apache.iotdb.commons.exception.runtime.RPCServiceException;
import org.apache.iotdb.commons.service.ServiceType;
import org.apache.iotdb.commons.service.ThriftService;
import org.apache.iotdb.commons.service.ThriftServiceThread;
+import org.apache.iotdb.commons.service.TrustedChannelAuditServerEventHandler;
import org.apache.iotdb.commons.service.metric.MetricService;
+import org.apache.iotdb.db.audit.DNAuditLogger;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
@@ -103,7 +105,10 @@ public class MPPDataExchangeService extends ThriftService
implements MPPDataExch
getBindPort(),
config.getRpcMaxConcurrentClientNum(),
config.getThriftServerAwaitTimeForStopService(),
- new MPPDataExchangeServiceThriftHandler(),
+ new TrustedChannelAuditServerEventHandler(
+ new MPPDataExchangeServiceThriftHandler(),
+ new TEndPoint(getBindIP(), getBindPort()),
+
DNAuditLogger.getInstance()::recordTrustedChannelFailureAuditLogIfNecessary),
config.isRpcThriftCompressionEnable(),
commonConfig.getKeyStorePath(),
commonConfig.getKeyStorePwd(),
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java
index b58d647fcd7..f9b9c6e0101 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.service;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.conf.CommonConfig;
import org.apache.iotdb.commons.conf.CommonDescriptor;
@@ -26,7 +27,9 @@ import
org.apache.iotdb.commons.exception.runtime.RPCServiceException;
import org.apache.iotdb.commons.service.ServiceType;
import org.apache.iotdb.commons.service.ThriftService;
import org.apache.iotdb.commons.service.ThriftServiceThread;
+import org.apache.iotdb.commons.service.TrustedChannelAuditServerEventHandler;
import org.apache.iotdb.commons.service.metric.MetricService;
+import org.apache.iotdb.db.audit.DNAuditLogger;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import
org.apache.iotdb.db.protocol.thrift.handler.InternalServiceThriftHandler;
@@ -74,7 +77,10 @@ public class DataNodeInternalRPCService extends ThriftService
getBindPort(),
config.getRpcMaxConcurrentClientNum(),
config.getThriftServerAwaitTimeForStopService(),
- new InternalServiceThriftHandler(),
+ new TrustedChannelAuditServerEventHandler(
+ new InternalServiceThriftHandler(),
+ new TEndPoint(getBindIP(), getBindPort()),
+
DNAuditLogger.getInstance()::recordTrustedChannelFailureAuditLogIfNecessary),
config.isRpcThriftCompressionEnable(),
commonConfig.getKeyStorePath(),
commonConfig.getKeyStorePwd(),
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/ExternalRPCService.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/ExternalRPCService.java
index f9b34dd9e58..782788547bf 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/ExternalRPCService.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/ExternalRPCService.java
@@ -18,6 +18,7 @@
*/
package org.apache.iotdb.db.service;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.conf.CommonConfig;
import org.apache.iotdb.commons.conf.CommonDescriptor;
@@ -25,7 +26,9 @@ import
org.apache.iotdb.commons.exception.runtime.RPCServiceException;
import org.apache.iotdb.commons.service.ServiceType;
import org.apache.iotdb.commons.service.ThriftService;
import org.apache.iotdb.commons.service.ThriftServiceThread;
+import org.apache.iotdb.commons.service.TrustedChannelAuditServerEventHandler;
import org.apache.iotdb.commons.service.metric.MetricService;
+import org.apache.iotdb.db.audit.DNAuditLogger;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
@@ -94,7 +97,7 @@ public class ExternalRPCService extends ThriftService
implements ExternalRPCServ
getBindPort(),
config.getRpcMaxConcurrentClientNum(),
config.getThriftServerAwaitTimeForStopService(),
- new RPCServiceThriftHandler(impl),
+ newTrustedChannelAuditHandler(),
config.isRpcThriftCompressionEnable(),
commonConfig.getKeyStorePath(),
commonConfig.getKeyStorePwd(),
@@ -111,7 +114,7 @@ public class ExternalRPCService extends ThriftService
implements ExternalRPCServ
getBindPort(),
config.getRpcMaxConcurrentClientNum(),
config.getThriftServerAwaitTimeForStopService(),
- new RPCServiceThriftHandler(impl),
+ newTrustedChannelAuditHandler(),
config.isRpcThriftCompressionEnable(),
commonConfig.getKeyStorePath(),
commonConfig.getKeyStorePwd(),
@@ -150,6 +153,13 @@ public class ExternalRPCService extends ThriftService
implements ExternalRPCServ
return value != null && !value.trim().isEmpty();
}
+ private TrustedChannelAuditServerEventHandler
newTrustedChannelAuditHandler() {
+ return new TrustedChannelAuditServerEventHandler(
+ new RPCServiceThriftHandler(impl),
+ new TEndPoint(getBindIP(), getBindPort()),
+
DNAuditLogger.getInstance()::recordTrustedChannelFailureAuditLogIfNecessary);
+ }
+
private static class RPCServiceHolder {
private static final ExternalRPCService INSTANCE = new
ExternalRPCService();
diff --git
a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java
b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java
index ae97d09ddc2..94194a086b4 100644
---
a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java
+++
b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java
@@ -222,5 +222,7 @@ public final class CommonMessages {
public static final String
EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_DATA_ARGUMENT_MUST_CONTAIN_EXACTLY_THE_TIME_COLUMN_SPECIFIED_BY_THE_TIMECOL_ARGUMENT_4375BAE9
= "The ORDER BY clause of the DATA argument must contain exactly the time
column specified by the TIMECOL argument.";
public static final String EXCEPTION_UNSUPPORTED_M4_VALUE_TYPE_AF0EF286 =
"Unsupported M4 value type: ";
public static final String
EXCEPTION_DISK_SPACE_WARNING_THRESHOLD_MUST_BE_IN_0_1_BUT_WAS_7B345766 =
"disk_space_warning_threshold must be in [0, 1), but was ";
+ public static final String
LOG_TRUSTED_CHANNEL_FUNCTION_FAILED_INITIATOR_ARG_TARGET_ARG_E4C28443 =
+ "Trusted channel function failed: initiator=%s, target=%s";
}
diff --git
a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java
b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java
index 71c45ccaf3f..07d886a44f3 100644
---
a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java
+++
b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java
@@ -215,5 +215,7 @@ public final class CommonMessages {
public static final String
EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_DATA_ARGUMENT_MUST_CONTAIN_EXACTLY_THE_TIME_COLUMN_SPECIFIED_BY_THE_TIMECOL_ARGUMENT_4375BAE9
= "DATA 参数的 ORDER BY 子句必须仅包含 TIMECOL 参数指定的时间列。";
public static final String EXCEPTION_UNSUPPORTED_M4_VALUE_TYPE_AF0EF286 =
"不支持的 M4 值类型：";
public static final String
EXCEPTION_DISK_SPACE_WARNING_THRESHOLD_MUST_BE_IN_0_1_BUT_WAS_7B345766 =
"disk_space_warning_threshold 必须在 [0, 1) 范围内，但实际为 ";
+ public static final String
LOG_TRUSTED_CHANNEL_FUNCTION_FAILED_INITIATOR_ARG_TARGET_ARG_E4C28443 =
+ "可信信道功能失效：发起者=%s，目标端=%s";
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AbstractAuditLogger.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AbstractAuditLogger.java
index d7c00c9128b..d4b94265f23 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AbstractAuditLogger.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AbstractAuditLogger.java
@@ -19,12 +19,23 @@
package org.apache.iotdb.commons.audit;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.auth.entity.PrivilegeType;
+import org.apache.iotdb.commons.auth.entity.User;
import org.apache.iotdb.commons.conf.CommonConfig;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.i18n.CommonMessages;
+import org.apache.iotdb.commons.utils.NodeUrlUtils;
+
+import javax.net.ssl.SSLException;
import java.util.function.Supplier;
public abstract class AbstractAuditLogger {
+ private static final long INTERNAL_AUDIT_LOG_USER_ID = 4;
+ private static final ThreadLocal<Boolean> RECORDING_TRUSTED_CHANNEL_FAILURE =
+ ThreadLocal.withInitial(() -> false);
+
public static final String OBJECT_AUTHENTICATION_AUDIT_STR =
"User %s (ID=%d) requests authority on object %s with result %s";
public static final String AUDIT_LOG_NODE_ID = "node_id";
@@ -61,4 +72,68 @@ public abstract class AbstractAuditLogger {
auditObject.get(),
auditEntity.getResult()));
}
+
+ /**
+ * Records a failure of the trusted-channel function.
+ *
+ * <p>The caller determines the channel direction and supplies the actual
initiator and target
+ * identifiers. This keeps the audit hook independent of any concrete
SSL/TLS implementation.
+ */
+ public void recordTrustedChannelFailureAuditLog(
+ final IAuditEntity auditEntity,
+ final Supplier<String> initiator,
+ final Supplier<String> target) {
+ log(
+ auditEntity
+ .setAuditEventType(AuditEventType.TRUSTED_CHANNEL_FUNCTION_FAILURE)
+ .setAuditLogOperation(AuditLogOperation.CONTROL)
+ .setResult(false),
+ () ->
+ String.format(
+ CommonMessages
+
.LOG_TRUSTED_CHANNEL_FUNCTION_FAILED_INITIATOR_ARG_TARGET_ARG_E4C28443,
+ initiator.get(),
+ target.get()));
+ }
+
+ public static boolean isSslFailure(Throwable failure) {
+ Throwable cause = failure;
+ while (cause != null) {
+ if (cause instanceof SSLException) {
+ return true;
+ }
+ cause = cause.getCause();
+ }
+ return false;
+ }
+
+ public void recordTrustedChannelFailureAuditLogIfNecessary(
+ Throwable failure, TEndPoint initiator, TEndPoint target) {
+ if (Boolean.TRUE.equals(RECORDING_TRUSTED_CHANNEL_FAILURE.get())
+ || !isSslFailure(failure)
+ || initiator == null
+ || target == null) {
+ return;
+ }
+
+ final String initiatorIdentifier =
NodeUrlUtils.convertTEndPointUrl(initiator);
+ final String targetIdentifier = NodeUrlUtils.convertTEndPointUrl(target);
+ RECORDING_TRUSTED_CHANNEL_FAILURE.set(true);
+ try {
+ recordTrustedChannelFailureAuditLog(
+ new UserEntity(
+ INTERNAL_AUDIT_LOG_USER_ID,
+ User.BUILTIN_INTERNAL_AUDIT_LOG_USERNAME,
+ initiatorIdentifier)
+ .setPrivilegeType(PrivilegeType.SECURITY),
+ () -> initiatorIdentifier,
+ () -> targetIdentifier);
+ } catch (RuntimeException auditFailure) {
+ if (auditFailure != failure) {
+ failure.addSuppressed(auditFailure);
+ }
+ } finally {
+ RECORDING_TRUSTED_CHANNEL_FAILURE.remove();
+ }
+ }
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AuditEventType.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AuditEventType.java
index 58e6f752b60..d9e2a1f5c30 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AuditEventType.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AuditEventType.java
@@ -45,7 +45,7 @@ public enum AuditEventType {
LOGIN_EXCEED_LIMIT,
SESSION_TIME_EXCEEDED,
LOGIN_REJECT_IP,
- SESSION_ENCRYPT_FAILED,
+ TRUSTED_CHANNEL_FUNCTION_FAILURE,
SYSTEM_OPERATION,
DN_SHUTDOWN;
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AuditEventType.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/TrustedChannelFailureHandler.java
similarity index 55%
copy from
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AuditEventType.java
copy to
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/TrustedChannelFailureHandler.java
index 58e6f752b60..b14cb894400 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AuditEventType.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/TrustedChannelFailureHandler.java
@@ -19,39 +19,15 @@
package org.apache.iotdb.commons.audit;
-public enum AuditEventType {
- CHANGE_AUDIT_OPTION,
- AUDIT_STORAGE_FULL,
- GENERATE_KEY,
- DESTROY_KEY,
- EXECUTE_ENCRYPT,
- OBJECT_AUTHENTICATION,
- LBAC_AUTHENTICATION,
- EXPORT_DATA_WITH_LABEL,
- IMPORT_DATA_WITH_LABEL,
- INTEGRITY_CHECK,
- LOGIN_FAIL_MAX_TIMES,
- MODIFY_PASSWD,
- LOGIN,
- LOGOUT,
- LOGIN_FINAL,
- MODIFY_SECURITY_OPTIONS,
- MODIFY_DEFAULT_SECURITY_VALUES,
- MODIFY_ROLE_MEMBERSHIP,
- REVOKE_FAILED,
- GRANT_ROLE_FAILED,
- LOGIN_RESOURCE_RESTRICT,
- LOGIN_FAILED_TRIES,
- LOGIN_EXCEED_LIMIT,
- SESSION_TIME_EXCEEDED,
- LOGIN_REJECT_IP,
- SESSION_ENCRYPT_FAILED,
- SYSTEM_OPERATION,
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
- DN_SHUTDOWN;
+@FunctionalInterface
+public interface TrustedChannelFailureHandler {
- @Override
- public String toString() {
- return name();
- }
+ TrustedChannelFailureHandler NO_OP =
+ (failure, initiator, target) -> {
+ // Do nothing.
+ };
+
+ void onFailure(Throwable failure, TEndPoint initiator, TEndPoint target);
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/TrustedChannelAuditServerEventHandler.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/TrustedChannelAuditServerEventHandler.java
new file mode 100644
index 00000000000..f99e6df4cfe
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/TrustedChannelAuditServerEventHandler.java
@@ -0,0 +1,152 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.commons.service;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.audit.TrustedChannelFailureHandler;
+import org.apache.iotdb.rpc.TElasticFramedTransport;
+
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.server.ServerContext;
+import org.apache.thrift.server.TServerEventHandler;
+import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransport;
+
+import javax.net.ssl.SSLSocket;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.SocketAddress;
+import java.util.Objects;
+
+/**
+ * Performs the server-side TLS handshake before the first Thrift request is
processed and reports
+ * handshake failures together with the raw socket peer and the local service
endpoint.
+ */
+public class TrustedChannelAuditServerEventHandler implements
TServerEventHandler {
+
+ private final TServerEventHandler delegate;
+ private final TEndPoint target;
+ private final TrustedChannelFailureHandler failureHandler;
+
+ public TrustedChannelAuditServerEventHandler(
+ TServerEventHandler delegate, TEndPoint target,
TrustedChannelFailureHandler failureHandler) {
+ this.delegate = Objects.requireNonNull(delegate);
+ this.target = Objects.requireNonNull(target);
+ this.failureHandler = Objects.requireNonNull(failureHandler);
+ }
+
+ @Override
+ public void preServe() {
+ delegate.preServe();
+ }
+
+ @Override
+ public ServerContext createContext(TProtocol input, TProtocol output) {
+ startHandshakeIfNecessary(output);
+ try {
+ return delegate.createContext(input, output);
+ } catch (RuntimeException | Error contextFailure) {
+ // A delegate may have already allocated connection state before
createContext fails. The
+ // Thrift server subsequently invokes this wrapper with a null context,
which is deliberately
+ // ignored by deleteContext below, so clean up the partially created
delegate context here.
+ try {
+ delegate.deleteContext(null, input, output);
+ } catch (RuntimeException | Error cleanupFailure) {
+ if (cleanupFailure != contextFailure) {
+ contextFailure.addSuppressed(cleanupFailure);
+ }
+ }
+ throw contextFailure;
+ }
+ }
+
+ @Override
+ public void deleteContext(ServerContext serverContext, TProtocol input,
TProtocol output) {
+ if (serverContext != null) {
+ delegate.deleteContext(serverContext, input, output);
+ }
+ }
+
+ @Override
+ public void processContext(
+ ServerContext serverContext, TTransport inputTransport, TTransport
outputTransport) {
+ delegate.processContext(serverContext, inputTransport, outputTransport);
+ }
+
+ private void startHandshakeIfNecessary(TProtocol output) {
+ Socket socket = getSocket(output);
+ if (!(socket instanceof SSLSocket)) {
+ return;
+ }
+
+ try {
+ ((SSLSocket) socket).startHandshake();
+ } catch (IOException e) {
+ notifyFailure(e, socket.getRemoteSocketAddress(),
socket.getLocalSocketAddress());
+ try {
+ socket.close();
+ } catch (IOException closeFailure) {
+ if (closeFailure != e) {
+ e.addSuppressed(closeFailure);
+ }
+ }
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ private void notifyFailure(
+ Throwable failure, SocketAddress remoteAddress, SocketAddress
localAddress) {
+ TEndPoint initiator = toEndPoint(remoteAddress);
+ if (initiator == null) {
+ return;
+ }
+ TEndPoint actualTarget = toEndPoint(localAddress);
+ try {
+ failureHandler.onFailure(failure, initiator, actualTarget == null ?
target : actualTarget);
+ } catch (RuntimeException auditFailure) {
+ if (auditFailure != failure) {
+ failure.addSuppressed(auditFailure);
+ }
+ }
+ }
+
+ private static Socket getSocket(TProtocol protocol) {
+ if (protocol == null
+ || !(protocol.getTransport() instanceof TElasticFramedTransport
framedTransport)) {
+ return null;
+ }
+ TTransport socketTransport = framedTransport.getSocket();
+ return socketTransport instanceof TSocket socket ? socket.getSocket() :
null;
+ }
+
+ private static TEndPoint toEndPoint(SocketAddress socketAddress) {
+ if (!(socketAddress instanceof InetSocketAddress inetSocketAddress)) {
+ return null;
+ }
+ String host =
+ inetSocketAddress.getAddress() == null
+ ? inetSocketAddress.getHostString()
+ : inetSocketAddress.getAddress().getHostAddress();
+ return new TEndPoint(host, inetSocketAddress.getPort());
+ }
+}
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/audit/AbstractAuditLoggerTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/audit/AbstractAuditLoggerTest.java
new file mode 100644
index 00000000000..fcb3dbf6d1c
--- /dev/null
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/audit/AbstractAuditLoggerTest.java
@@ -0,0 +1,152 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.commons.audit;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.i18n.CommonMessages;
+
+import org.apache.thrift.TException;
+import org.junit.Test;
+
+import javax.net.ssl.SSLException;
+import javax.net.ssl.SSLHandshakeException;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+public class AbstractAuditLoggerTest {
+
+ @Test
+ public void testRecordTrustedChannelFailureAuditLog() {
+ TestAuditLogger auditLogger = new TestAuditLogger();
+ UserEntity auditEntity = new UserEntity(0, "system", "127.0.0.1");
+ AtomicInteger identifierEvaluationCount = new AtomicInteger();
+
+ auditLogger.recordTrustedChannelFailureAuditLog(
+ auditEntity,
+ () -> {
+ identifierEvaluationCount.incrementAndGet();
+ return "[email protected]:10730";
+ },
+ () -> {
+ identifierEvaluationCount.incrementAndGet();
+ return "[email protected]:10730";
+ });
+
+ assertSame(auditEntity, auditLogger.auditEntity);
+ assertEquals(
+ AuditEventType.TRUSTED_CHANNEL_FUNCTION_FAILURE,
+ auditLogger.auditEntity.getAuditEventType());
+ assertEquals(AuditLogOperation.CONTROL,
auditLogger.auditEntity.getAuditLogOperation());
+ assertFalse(auditLogger.auditEntity.getResult());
+ assertEquals(0, identifierEvaluationCount.get());
+ assertEquals(
+ String.format(
+
CommonMessages.LOG_TRUSTED_CHANNEL_FUNCTION_FAILED_INITIATOR_ARG_TARGET_ARG_E4C28443,
+ "[email protected]:10730",
+ "[email protected]:10730"),
+ auditLogger.auditLog.get());
+ assertEquals(2, identifierEvaluationCount.get());
+ }
+
+ @Test
+ public void testIsSslFailure() {
+ assertTrue(AbstractAuditLogger.isSslFailure(new SSLException("ssl
failure")));
+ assertTrue(
+ AbstractAuditLogger.isSslFailure(
+ new TException(new IOException(new
SSLHandshakeException("handshake failure")))));
+ assertFalse(
+ AbstractAuditLogger.isSslFailure(new TException(new
IOException("network failure"))));
+ assertFalse(AbstractAuditLogger.isSslFailure(null));
+ }
+
+ @Test
+ public void testRecordTrustedChannelFailureAuditLogIfNecessary() {
+ TestAuditLogger auditLogger = new TestAuditLogger();
+ TEndPoint initiator = new TEndPoint("10.0.0.1", 10730);
+ TEndPoint target = new TEndPoint("10.0.0.2", 10730);
+
+ auditLogger.recordTrustedChannelFailureAuditLogIfNecessary(
+ new IOException("network failure"), initiator, target);
+ assertNull(auditLogger.auditEntity);
+
+ auditLogger.recordTrustedChannelFailureAuditLogIfNecessary(
+ new SSLHandshakeException("handshake failure"), initiator, target);
+ assertEquals(
+ AuditEventType.TRUSTED_CHANNEL_FUNCTION_FAILURE,
+ auditLogger.auditEntity.getAuditEventType());
+ assertEquals(
+ String.format(
+
CommonMessages.LOG_TRUSTED_CHANNEL_FUNCTION_FAILED_INITIATOR_ARG_TARGET_ARG_E4C28443,
+ "10.0.0.1:10730",
+ "10.0.0.2:10730"),
+ auditLogger.auditLog.get());
+ }
+
+ @Test
+ public void testAuditFailureShouldNotReplaceTrustedChannelFailure() {
+ final RuntimeException auditFailure = new RuntimeException("audit
failure");
+ final ThrowingAuditLogger auditLogger = new
ThrowingAuditLogger(auditFailure);
+ final SSLHandshakeException channelFailure = new
SSLHandshakeException("handshake failure");
+ final TEndPoint initiator = new TEndPoint("10.0.0.1", 10730);
+ final TEndPoint target = new TEndPoint("10.0.0.2", 10730);
+
+ auditLogger.recordTrustedChannelFailureAuditLogIfNecessary(channelFailure,
initiator, target);
+
+ assertEquals(1, auditLogger.invocationCount.get());
+ assertEquals(1, channelFailure.getSuppressed().length);
+ assertSame(auditFailure, channelFailure.getSuppressed()[0]);
+ }
+
+ private static class TestAuditLogger extends AbstractAuditLogger {
+
+ private IAuditEntity auditEntity;
+ private Supplier<String> auditLog;
+
+ @Override
+ public void log(IAuditEntity auditLogFields, Supplier<String> log) {
+ auditEntity = auditLogFields;
+ auditLog = log;
+ }
+ }
+
+ private static class ThrowingAuditLogger extends AbstractAuditLogger {
+
+ private final RuntimeException auditFailure;
+ private final AtomicInteger invocationCount = new AtomicInteger();
+
+ private ThrowingAuditLogger(final RuntimeException auditFailure) {
+ this.auditFailure = auditFailure;
+ }
+
+ @Override
+ public void log(final IAuditEntity auditLogFields, final Supplier<String>
log) {
+ invocationCount.incrementAndGet();
+ throw auditFailure;
+ }
+ }
+}
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/TrustedChannelAuditServerEventHandlerTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/TrustedChannelAuditServerEventHandlerTest.java
new file mode 100644
index 00000000000..f231ef18d2a
--- /dev/null
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/TrustedChannelAuditServerEventHandlerTest.java
@@ -0,0 +1,220 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.commons.service;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.audit.TrustedChannelFailureHandler;
+import org.apache.iotdb.rpc.TElasticFramedTransport;
+
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.server.ServerContext;
+import org.apache.thrift.server.TServerEventHandler;
+import org.apache.thrift.transport.TSocket;
+import org.junit.Test;
+import org.mockito.InOrder;
+
+import javax.net.ssl.SSLHandshakeException;
+import javax.net.ssl.SSLSocket;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.net.InetSocketAddress;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class TrustedChannelAuditServerEventHandlerTest {
+
+ @Test
+ public void testHandshakeBeforeCreatingDelegateContext() throws Exception {
+ TServerEventHandler delegate = mock(TServerEventHandler.class);
+ ServerContext context = mock(ServerContext.class);
+ TProtocol input = mock(TProtocol.class);
+ SSLSocket socket = mock(SSLSocket.class);
+ TProtocol output = createProtocol(socket);
+ TEndPoint target = new TEndPoint("10.0.0.2", 10730);
+ when(delegate.createContext(input, output)).thenReturn(context);
+
+ TrustedChannelAuditServerEventHandler handler =
+ new TrustedChannelAuditServerEventHandler(
+ delegate, target, TrustedChannelFailureHandler.NO_OP);
+
+ assertSame(context, handler.createContext(input, output));
+ InOrder inOrder = inOrder(socket, delegate);
+ inOrder.verify(socket).startHandshake();
+ inOrder.verify(delegate).createContext(input, output);
+ }
+
+ @Test
+ public void testHandshakeFailureReportsPeerAndSkipsDelegate() throws
Exception {
+ TServerEventHandler delegate = mock(TServerEventHandler.class);
+ TProtocol input = mock(TProtocol.class);
+ SSLSocket socket = mock(SSLSocket.class);
+ TProtocol output = createProtocol(socket);
+ SSLHandshakeException failure = new SSLHandshakeException("handshake
failure");
+ InetSocketAddress remoteAddress = new InetSocketAddress("192.0.2.10",
45123);
+ TEndPoint target = new TEndPoint("10.0.0.2", 10730);
+ AtomicReference<Throwable> reportedFailure = new AtomicReference<>();
+ AtomicReference<TEndPoint> reportedInitiator = new AtomicReference<>();
+ AtomicReference<TEndPoint> reportedTarget = new AtomicReference<>();
+ when(socket.getRemoteSocketAddress()).thenReturn(remoteAddress);
+ org.mockito.Mockito.doThrow(failure).when(socket).startHandshake();
+
+ TrustedChannelAuditServerEventHandler handler =
+ new TrustedChannelAuditServerEventHandler(
+ delegate,
+ target,
+ (throwable, initiator, endpoint) -> {
+ reportedFailure.set(throwable);
+ reportedInitiator.set(initiator);
+ reportedTarget.set(endpoint);
+ });
+
+ UncheckedIOException thrown =
+ assertThrows(UncheckedIOException.class, () ->
handler.createContext(input, output));
+
+ assertSame(failure, thrown.getCause());
+ assertSame(failure, reportedFailure.get());
+ assertEquals(new TEndPoint("192.0.2.10", 45123), reportedInitiator.get());
+ assertSame(target, reportedTarget.get());
+ verify(socket).close();
+ verify(delegate, never()).createContext(any(), any());
+
+ handler.deleteContext(null, input, output);
+ verify(delegate, never()).deleteContext(isNull(), any(), any());
+ }
+
+ @Test
+ public void testWrappedSslFailureReachesCallback() throws Exception {
+ TServerEventHandler delegate = mock(TServerEventHandler.class);
+ TProtocol input = mock(TProtocol.class);
+ SSLSocket socket = mock(SSLSocket.class);
+ TProtocol output = createProtocol(socket);
+ IOException failure =
+ new IOException(
+ "wrapped handshake failure", new SSLHandshakeException("handshake
failure"));
+ AtomicReference<Throwable> reportedFailure = new AtomicReference<>();
+ when(socket.getRemoteSocketAddress()).thenReturn(new
InetSocketAddress("192.0.2.10", 45123));
+ org.mockito.Mockito.doThrow(failure).when(socket).startHandshake();
+
+ TrustedChannelAuditServerEventHandler handler =
+ new TrustedChannelAuditServerEventHandler(
+ delegate,
+ new TEndPoint("10.0.0.2", 10730),
+ (throwable, initiator, endpoint) ->
reportedFailure.set(throwable));
+
+ UncheckedIOException thrown =
+ assertThrows(UncheckedIOException.class, () ->
handler.createContext(input, output));
+
+ assertSame(failure, thrown.getCause());
+ assertSame(failure, reportedFailure.get());
+ verify(delegate, never()).createContext(any(), any());
+ }
+
+ @Test
+ public void testHandshakeFailurePrefersActualLocalEndpoint() throws
Exception {
+ TServerEventHandler delegate = mock(TServerEventHandler.class);
+ TProtocol input = mock(TProtocol.class);
+ SSLSocket socket = mock(SSLSocket.class);
+ TProtocol output = createProtocol(socket);
+ SSLHandshakeException failure = new SSLHandshakeException("handshake
failure");
+ AtomicReference<TEndPoint> reportedTarget = new AtomicReference<>();
+ when(socket.getRemoteSocketAddress()).thenReturn(new
InetSocketAddress("192.0.2.10", 45123));
+ when(socket.getLocalSocketAddress()).thenReturn(new
InetSocketAddress("10.0.0.9", 10730));
+ org.mockito.Mockito.doThrow(failure).when(socket).startHandshake();
+
+ TrustedChannelAuditServerEventHandler handler =
+ new TrustedChannelAuditServerEventHandler(
+ delegate,
+ new TEndPoint("0.0.0.0", 10730),
+ (throwable, initiator, endpoint) -> reportedTarget.set(endpoint));
+
+ assertThrows(UncheckedIOException.class, () ->
handler.createContext(input, output));
+
+ assertEquals(new TEndPoint("10.0.0.9", 10730), reportedTarget.get());
+ }
+
+ @Test
+ public void testCloseFailureDoesNotSelfSuppressHandshakeFailure() throws
Exception {
+ TServerEventHandler delegate = mock(TServerEventHandler.class);
+ TProtocol input = mock(TProtocol.class);
+ SSLSocket socket = mock(SSLSocket.class);
+ TProtocol output = createProtocol(socket);
+ SSLHandshakeException failure = new SSLHandshakeException("handshake
failure");
+ when(socket.getRemoteSocketAddress()).thenReturn(new
InetSocketAddress("192.0.2.10", 45123));
+ org.mockito.Mockito.doThrow(failure).when(socket).startHandshake();
+ org.mockito.Mockito.doThrow(failure).when(socket).close();
+
+ TrustedChannelAuditServerEventHandler handler =
+ new TrustedChannelAuditServerEventHandler(
+ delegate, new TEndPoint("10.0.0.2", 10730),
TrustedChannelFailureHandler.NO_OP);
+
+ UncheckedIOException thrown =
+ assertThrows(UncheckedIOException.class, () ->
handler.createContext(input, output));
+
+ assertSame(failure, thrown.getCause());
+ assertEquals(0, failure.getSuppressed().length);
+ verify(delegate, never()).createContext(any(), any());
+ }
+
+ @Test
+ public void testDelegateContextFailureCleansUpExactlyOnce() {
+ TServerEventHandler delegate = mock(TServerEventHandler.class);
+ TProtocol input = mock(TProtocol.class);
+ TProtocol output = mock(TProtocol.class);
+ RuntimeException contextFailure = new RuntimeException("context failure");
+
org.mockito.Mockito.doThrow(contextFailure).when(delegate).createContext(input,
output);
+
+ TrustedChannelAuditServerEventHandler handler =
+ new TrustedChannelAuditServerEventHandler(
+ delegate, new TEndPoint("10.0.0.2", 10730),
TrustedChannelFailureHandler.NO_OP);
+
+ assertSame(
+ contextFailure,
+ assertThrows(RuntimeException.class, () ->
handler.createContext(input, output)));
+
+ // TThreadPoolServer invokes deleteContext with the null context after
createContext fails.
+ handler.deleteContext(null, input, output);
+
+ InOrder inOrder = inOrder(delegate);
+ inOrder.verify(delegate).createContext(input, output);
+ inOrder.verify(delegate).deleteContext(null, input, output);
+ verify(delegate).deleteContext(null, input, output);
+ }
+
+ private static TProtocol createProtocol(SSLSocket socket) {
+ TProtocol protocol = mock(TProtocol.class);
+ TElasticFramedTransport framedTransport =
mock(TElasticFramedTransport.class);
+ TSocket socketTransport = mock(TSocket.class);
+ when(protocol.getTransport()).thenReturn(framedTransport);
+ when(framedTransport.getSocket()).thenReturn(socketTransport);
+ when(socketTransport.getSocket()).thenReturn(socket);
+ return protocol;
+ }
+}
diff --git a/pom.xml b/pom.xml
index 331a4524613..e427d8680a7 100644
--- a/pom.xml
+++ b/pom.xml
@@ -539,6 +539,11 @@
<artifactId>jetty-http</artifactId>
<version>${jetty.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-io</artifactId>
+ <version>${jetty.version}</version>
+ </dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-util</artifactId>
