This is an automated email from the ASF dual-hosted git repository.
jt2594838 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 b8ff9eeaea0 [Pipe] Prevent unexpected OPC UA endpoint redirects
(#18259)
b8ff9eeaea0 is described below
commit b8ff9eeaea0d2e5afd79446e8bb4cf654c8c8038
Author: Caideyipi <[email protected]>
AuthorDate: Tue Jul 21 19:04:43 2026 +0800
[Pipe] Prevent unexpected OPC UA endpoint redirects (#18259)
* [Pipe] Prevent unexpected OPC UA endpoint redirects
* [Pipe] Add advertised host for internal OPC UA server
---
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 11 ++
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 10 ++
.../db/pipe/sink/protocol/opcua/OpcUaSink.java | 29 +++-
.../sink/protocol/opcua/client/ClientRunner.java | 89 ++++++++++-
.../protocol/opcua/client/IoTDBOpcUaClient.java | 12 +-
.../protocol/opcua/server/OpcUaKeyStoreLoader.java | 21 ++-
.../sink/protocol/opcua/server/OpcUaNameSpace.java | 2 +
.../protocol/opcua/server/OpcUaServerBuilder.java | 143 ++++++++++++++++--
.../protocol/opcua/client/ClientRunnerTest.java | 147 ++++++++++++++++++
.../opcua/server/OpcUaServerBuilderTest.java | 164 +++++++++++++++++++++
.../pipe/config/constant/PipeSinkConstant.java | 9 ++
11 files changed, 603 insertions(+), 34 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 40b499410aa..f13bca9a4a8 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -945,6 +945,9 @@ public final class DataNodePipeMessages {
"Network failed to receive tsFile %s, status: %s";
public static final String SECURITY_DIR = "security dir: {}";
public static final String SECURITY_PKI_DIR = "security pki dir: {}";
+ public static final String
+
LOG_OPC_UA_ENDPOINT_SELECTED_CONFIGURED_ARG_ADVERTISED_ARG_EFFECTIVE_ARG_ALLOWENDPOINTREDIRECT_ARG_4FE076CB
=
+ "OPC UA endpoint selected: configured={}, advertised={},
effective={}, allowEndpointRedirect={}.";
public static final String SSL_TRUST_STORE_PAIR_REQUIRED_WHEN_SSL_ENABLED =
"When %s or %s is true, specify a complete trust-store pair under the
same "
+ "alias: %s and %s, %s and %s, or %s and %s";
@@ -1535,6 +1538,14 @@ public final class DataNodePipeMessages {
public static final String UNABLE_CREATE_SECURITY_DIR = "Unable to create
security dir: ";
public static final String OPC_UA_SECURITY_DIR = "Security dir: {}";
public static final String OPC_UA_SECURITY_PKI_DIR = "Security pki dir: {}";
+ public static final String
+
EXCEPTION_THE_ADVERTISED_HOST_MUST_BE_A_HOSTNAME_OR_IP_ADDRESS_WITHOUT_A_SCHEME_PORT_OR_PATH_6857C67A
=
+ "The advertised host must be a hostname or IP address without a
scheme, port, or path.";
+ public static final String
+
LOG_ADVERTISED_HOST_ARG_IS_NOT_PRESENT_IN_THE_LOADED_OPC_UA_SERVER_CERTIFICATE_SUBJECT_ALTERNATIVE_NAMES_SECURED_CLIENTS_MAY_REJECT_IT_REPLACE_OR_REGENERATE_THE_CERTIFICATE_AND_ESTABLISH_TRUST_AGAIN_912358AF
=
+ "Advertised host {} is not present in the loaded OPC UA server
certificate subject "
+ + "alternative names. Secured clients may reject it; replace or
regenerate the "
+ + "certificate and establish trust again.";
//
---------------------------------------------------------------------------
// pipe – PipeDataNodePluginAgent
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 85a8ae289cf..17dd4c6d836 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -892,6 +892,9 @@ public final class DataNodePipeMessages {
"网络接收 TsFile %s 失败，状态：%s";
public static final String SECURITY_DIR = "security 目录：{}";
public static final String SECURITY_PKI_DIR = "security pki 目录：{}";
+ public static final String
+
LOG_OPC_UA_ENDPOINT_SELECTED_CONFIGURED_ARG_ADVERTISED_ARG_EFFECTIVE_ARG_ALLOWENDPOINTREDIRECT_ARG_4FE076CB
=
+ "已选择 OPC UA
endpoint：configured={}，advertised={}，effective={}，allowEndpointRedirect={}。";
public static final String SSL_TRUST_STORE_PAIR_REQUIRED_WHEN_SSL_ENABLED =
"当 %s 或 %s 为 true 时，请在同一别名下指定完整的 trust-store 参数对：%s 和 %s、%s 和 %s，或 %s 和
%s";
public static final String
SSL_KEY_STORE_PATH_AND_PASSWORD_MUST_BE_SPECIFIED_TOGETHER =
@@ -1439,6 +1442,13 @@ public final class DataNodePipeMessages {
"安全目录：{}";
public static final String OPC_UA_SECURITY_PKI_DIR =
"安全 PKI 目录：{}";
+ public static final String
+
EXCEPTION_THE_ADVERTISED_HOST_MUST_BE_A_HOSTNAME_OR_IP_ADDRESS_WITHOUT_A_SCHEME_PORT_OR_PATH_6857C67A
=
+ "advertised host 必须是不带 scheme、port 或 path 的 hostname 或 IP 地址。";
+ public static final String
+
LOG_ADVERTISED_HOST_ARG_IS_NOT_PRESENT_IN_THE_LOADED_OPC_UA_SERVER_CERTIFICATE_SUBJECT_ALTERNATIVE_NAMES_SECURED_CLIENTS_MAY_REJECT_IT_REPLACE_OR_REGENERATE_THE_CERTIFICATE_AND_ESTABLISH_TRUST_AGAIN_912358AF
=
+ "advertised host {} 不在已加载的 OPC UA server 证书 subject alternative
names 中。安全客户端可能拒绝该证书；"
+ + "请替换或重新生成证书并重新建立信任。";
//
---------------------------------------------------------------------------
// pipe – PipeDataNodePluginAgent
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java
index ea32d6263aa..2eb4e7e8136 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java
@@ -70,6 +70,9 @@ import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CON
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_USERNAME_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_USER_DEFAULT_VALUE;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_USER_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_ADVERTISED_HOST_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_ALLOW_ENDPOINT_REDIRECT_DEFAULT_VALUE;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_ALLOW_ENDPOINT_REDIRECT_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_DEBOUNCE_TIME_MS_DEFAULT_VALUE;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_DEBOUNCE_TIME_MS_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_DEFAULT_QUALITY_BAD_VALUE;
@@ -112,6 +115,8 @@ import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CON
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_PASSWORD_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_USERNAME_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_USER_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_ADVERTISED_HOST_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_ALLOW_ENDPOINT_REDIRECT_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_DEBOUNCE_TIME_MS_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_DEFAULT_QUALITY_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_ENABLE_ANONYMOUS_ACCESS_KEY;
@@ -268,6 +273,9 @@ public class OpcUaSink implements PipeConnector {
parameters.getIntOrDefault(
Arrays.asList(CONNECTOR_OPC_UA_HTTPS_BIND_PORT_KEY,
SINK_OPC_UA_HTTPS_BIND_PORT_KEY),
CONNECTOR_OPC_UA_HTTPS_BIND_PORT_DEFAULT_VALUE);
+ final String advertisedHost =
+ parameters.getStringByKeys(
+ CONNECTOR_OPC_UA_ADVERTISED_HOST_KEY,
SINK_OPC_UA_ADVERTISED_HOST_KEY);
final String user =
parameters.getStringOrDefault(
@@ -330,6 +338,7 @@ public class OpcUaSink implements PipeConnector {
new OpcUaServerBuilder()
.setTcpBindPort(tcpBindPort)
.setHttpsBindPort(httpsBindPort)
+ .setAdvertisedHost(advertisedHost)
.setUser(user)
.setPassword(password)
.setSecurityDir(securityDir)
@@ -345,6 +354,7 @@ public class OpcUaSink implements PipeConnector {
oldValue
.getRight()
.checkEquals(
+ advertisedHost,
user,
password,
securityDir,
@@ -399,6 +409,12 @@ public class OpcUaSink implements PipeConnector {
parameters.getLongOrDefault(
Arrays.asList(CONNECTOR_OPC_UA_TIMEOUT_SECONDS_KEY,
SINK_OPC_UA_TIMEOUT_SECONDS_KEY),
CONNECTOR_OPC_UA_TIMEOUT_SECONDS_DEFAULT_VALUE);
+ final boolean allowEndpointRedirect =
+ parameters.getBooleanOrDefault(
+ Arrays.asList(
+ CONNECTOR_OPC_UA_ALLOW_ENDPOINT_REDIRECT_KEY,
+ SINK_OPC_UA_ALLOW_ENDPOINT_REDIRECT_KEY),
+ CONNECTOR_OPC_UA_ALLOW_ENDPOINT_REDIRECT_DEFAULT_VALUE);
synchronized (CLIENT_KEY_TO_REFERENCE_COUNT_AND_CLIENT_MAP) {
client =
@@ -418,11 +434,20 @@ public class OpcUaSink implements PipeConnector {
SINK_OPC_UA_HISTORIZING_KEY),
CONNECTOR_OPC_UA_HISTORIZING_DEFAULT_VALUE));
final ClientRunner runner =
- new ClientRunner(result, securityDir, password,
userName, timeoutSeconds);
+ new ClientRunner(
+ result,
+ securityDir,
+ password,
+ userName,
+ timeoutSeconds,
+ allowEndpointRedirect);
runner.run();
return new Pair<>(new AtomicInteger(0), result);
}
- oldValue.getRight().checkEquals(userName, password,
securityDir, policy);
+ oldValue
+ .getRight()
+ .checkEquals(
+ userName, password, securityDir, policy,
allowEndpointRedirect);
return oldValue;
})
.getRight();
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunner.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunner.java
index dc68e0d2b96..6cbc5c3d61d 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunner.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunner.java
@@ -27,7 +27,10 @@ import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import
org.eclipse.milo.opcua.stack.client.security.DefaultClientCertificateValidator;
import org.eclipse.milo.opcua.stack.core.security.DefaultTrustListManager;
import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
+import org.eclipse.milo.opcua.stack.core.transport.TransportProfile;
import org.eclipse.milo.opcua.stack.core.types.builtin.LocalizedText;
+import org.eclipse.milo.opcua.stack.core.types.structured.EndpointDescription;
+import org.eclipse.milo.opcua.stack.core.util.EndpointUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -37,7 +40,10 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.security.Security;
+import java.util.List;
+import java.util.Locale;
import java.util.Objects;
+import java.util.Optional;
import static
org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;
@@ -54,6 +60,7 @@ public class ClientRunner {
private final Path securityDir;
private final String password;
private final long timeoutSeconds;
+ private final boolean allowEndpointRedirect;
// For conflict checking
private final String user;
@@ -64,11 +71,22 @@ public class ClientRunner {
final String password,
final String user,
final long timeoutSeconds) {
+ this(configurableUaClient, securityDir, password, user, timeoutSeconds,
false);
+ }
+
+ public ClientRunner(
+ final IoTDBOpcUaClient configurableUaClient,
+ final String securityDir,
+ final String password,
+ final String user,
+ final long timeoutSeconds,
+ final boolean allowEndpointRedirect) {
this.configurableUaClient = configurableUaClient;
this.securityDir = Paths.get(securityDir);
this.password = password;
this.user = user;
this.timeoutSeconds = timeoutSeconds;
+ this.allowEndpointRedirect = allowEndpointRedirect;
configurableUaClient.setRunner(this);
}
@@ -93,7 +111,12 @@ public class ClientRunner {
return OpcUaClient.create(
configurableUaClient.getNodeUrl(),
- endpoints ->
endpoints.stream().filter(configurableUaClient.endpointFilter()).findFirst(),
+ endpoints ->
+ selectEndpoint(
+ endpoints,
+ configurableUaClient.getNodeUrl(),
+ configurableUaClient.getSecurityPolicy(),
+ allowEndpointRedirect),
configBuilder ->
configBuilder
.setApplicationName(LocalizedText.english("Apache IoTDB OPC UA
client"))
@@ -109,6 +132,66 @@ public class ClientRunner {
.build());
}
+ static Optional<EndpointDescription> selectEndpoint(
+ final List<EndpointDescription> endpoints,
+ final String configuredNodeUrl,
+ final SecurityPolicy securityPolicy,
+ final boolean allowEndpointRedirect) {
+ final String configuredScheme =
normalizeScheme(EndpointUtil.getScheme(configuredNodeUrl));
+
+ return endpoints.stream()
+ .filter(endpoint ->
securityPolicy.getUri().equals(endpoint.getSecurityPolicyUri()))
+ .filter(endpoint -> matchesConfiguredTransport(endpoint,
configuredScheme))
+ .findFirst()
+ .map(
+ advertisedEndpoint -> {
+ final EndpointDescription effectiveEndpoint =
+ allowEndpointRedirect
+ ? advertisedEndpoint
+ : EndpointUtil.updateUrl(
+ advertisedEndpoint,
+ EndpointUtil.getHost(configuredNodeUrl),
+ EndpointUtil.getPort(configuredNodeUrl));
+ logger.info(
+ DataNodePipeMessages
+
.LOG_OPC_UA_ENDPOINT_SELECTED_CONFIGURED_ARG_ADVERTISED_ARG_EFFECTIVE_ARG_ALLOWENDPOINTREDIRECT_ARG_4FE076CB,
+ configuredNodeUrl,
+ advertisedEndpoint.getEndpointUrl(),
+ effectiveEndpoint.getEndpointUrl(),
+ allowEndpointRedirect);
+ return effectiveEndpoint;
+ });
+ }
+
+ private static boolean matchesConfiguredTransport(
+ final EndpointDescription endpoint, final String configuredScheme) {
+ if (Objects.isNull(configuredScheme)
+ || !Objects.equals(
+ configuredScheme,
normalizeScheme(EndpointUtil.getScheme(endpoint.getEndpointUrl())))) {
+ return false;
+ }
+
+ final String transportProfileUri = endpoint.getTransportProfileUri();
+ if (Objects.isNull(transportProfileUri)) {
+ return true;
+ }
+ try {
+ return Objects.equals(
+ configuredScheme,
+
normalizeScheme(TransportProfile.fromUri(transportProfileUri).getScheme()));
+ } catch (final IllegalArgumentException ignored) {
+ // Preserve compatibility with servers that advertise a custom transport
profile. The
+ // endpoint URL scheme still has to match the configured URL.
+ return true;
+ }
+ }
+
+ private static String normalizeScheme(final String scheme) {
+ return Objects.nonNull(scheme) && scheme.equalsIgnoreCase("opc.https")
+ ? "https"
+ : Objects.isNull(scheme) ? null : scheme.toLowerCase(Locale.ROOT);
+ }
+
public void run() {
try {
final OpcUaClient client = createClient();
@@ -143,7 +226,8 @@ public class ClientRunner {
final String user,
final String password,
final Path securityDir,
- final SecurityPolicy securityPolicy) {
+ final SecurityPolicy securityPolicy,
+ final boolean allowEndpointRedirect) {
checkEquals("user", this.user, user);
checkEquals("password", this.password, password);
checkEquals(
@@ -151,6 +235,7 @@ public class ClientRunner {
FileSystems.getDefault().getPath(this.securityDir.toAbsolutePath().toString()),
FileSystems.getDefault().getPath(securityDir.toAbsolutePath().toString()));
checkEquals("securityPolicy", configurableUaClient.getSecurityPolicy(),
securityPolicy);
+ checkEquals("allow endpoint redirect", this.allowEndpointRedirect,
allowEndpointRedirect);
}
private void checkEquals(final String attrName, Object thisAttr, Object
thatAttr) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java
index 62e489b2e1f..3dd2141fad5 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java
@@ -52,7 +52,6 @@ import
org.eclipse.milo.opcua.stack.core.types.enumerated.NodeClass;
import org.eclipse.milo.opcua.stack.core.types.structured.AddNodesItem;
import org.eclipse.milo.opcua.stack.core.types.structured.AddNodesResponse;
import org.eclipse.milo.opcua.stack.core.types.structured.AddNodesResult;
-import org.eclipse.milo.opcua.stack.core.types.structured.EndpointDescription;
import org.eclipse.milo.opcua.stack.core.types.structured.ObjectAttributes;
import org.eclipse.milo.opcua.stack.core.types.structured.VariableAttributes;
import org.slf4j.Logger;
@@ -65,7 +64,6 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ExecutionException;
-import java.util.function.Predicate;
import static
org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace.convertToOpcDataType;
import static
org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace.timestampToUtc;
@@ -307,10 +305,6 @@ public class IoTDBOpcUaClient {
return nodeUrl;
}
- Predicate<EndpointDescription> endpointFilter() {
- return e -> getSecurityPolicy().getUri().equals(e.getSecurityPolicyUri());
- }
-
SecurityPolicy getSecurityPolicy() {
return securityPolicy;
}
@@ -365,7 +359,9 @@ public class IoTDBOpcUaClient {
final String user,
final String password,
final String securityDir,
- final SecurityPolicy securityPolicy) {
- runner.checkEquals(user, password, Paths.get(securityDir), securityPolicy);
+ final SecurityPolicy securityPolicy,
+ final boolean allowEndpointRedirect) {
+ runner.checkEquals(
+ user, password, Paths.get(securityDir), securityPolicy,
allowEndpointRedirect);
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaKeyStoreLoader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaKeyStoreLoader.java
index 26fa2084430..2e27ffc2277 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaKeyStoreLoader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaKeyStoreLoader.java
@@ -22,7 +22,7 @@ package org.apache.iotdb.db.pipe.sink.protocol.opcua.server;
import org.apache.iotdb.commons.utils.FileUtils;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
-import com.google.common.collect.Sets;
+import com.google.common.net.InetAddresses;
import org.eclipse.milo.opcua.sdk.server.util.HostnameUtil;
import org.eclipse.milo.opcua.stack.core.util.SelfSignedCertificateBuilder;
import org.eclipse.milo.opcua.stack.core.util.SelfSignedCertificateGenerator;
@@ -41,22 +41,21 @@ import java.security.KeyStore;
import java.security.PrivateKey;
import java.security.PublicKey;
import java.security.cert.X509Certificate;
+import java.util.LinkedHashSet;
import java.util.Set;
import java.util.UUID;
-import java.util.regex.Pattern;
class OpcUaKeyStoreLoader {
private static final Logger LOGGER =
LoggerFactory.getLogger(OpcUaKeyStoreLoader.class);
- private static final Pattern IP_ADDR_PATTERN =
-
Pattern.compile("^(([01]?\\d\\d?|2[0-4]\\d|25[0-5])\\.){3}([01]?\\d\\d?|2[0-4]\\d|25[0-5])$");
-
private static final String SERVER_ALIAS = "server-ai";
private X509Certificate serverCertificate;
private KeyPair serverKeyPair;
- OpcUaKeyStoreLoader load(final Path baseDir, final char[] password) throws
Exception {
+ OpcUaKeyStoreLoader load(
+ final Path baseDir, final char[] password, final Set<String>
advertisedHostnames)
+ throws Exception {
final KeyStore keyStore = KeyStore.getInstance("PKCS12");
final File serverKeyStore = baseDir.resolve("iotdb-server.pfx").toFile();
@@ -90,14 +89,14 @@ class OpcUaKeyStoreLoader {
.setApplicationUri(applicationUri);
// Get as many hostnames and IP addresses as we can list in the
certificate.
- final Set<String> hostnames =
- Sets.union(
- Sets.newHashSet(HostnameUtil.getHostname()),
- HostnameUtil.getHostnames("0.0.0.0", false));
+ final Set<String> hostnames = new LinkedHashSet<>();
+ hostnames.add(HostnameUtil.getHostname());
+ hostnames.addAll(HostnameUtil.getHostnames("0.0.0.0", false));
+ hostnames.addAll(advertisedHostnames);
hostnames.forEach(
hostname -> {
- if (IP_ADDR_PATTERN.matcher(hostname).matches()) {
+ if (InetAddresses.isInetAddress(hostname)) {
builder.addIpAddress(hostname);
} else {
builder.addDnsName(hostname);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
index f95e8ed245b..86c52149719 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
@@ -703,6 +703,7 @@ public class OpcUaNameSpace extends
ManagedNamespaceWithLifecycle {
/////////////////////////////// Conflict detection
///////////////////////////////
public void checkEquals(
+ final String advertisedHost,
final String user,
final String password,
final String securityDir,
@@ -710,6 +711,7 @@ public class OpcUaNameSpace extends
ManagedNamespaceWithLifecycle {
final Set<SecurityPolicy> securityPolicies,
final long debounceTimeMs) {
builder.checkEquals(
+ advertisedHost,
user,
password,
Paths.get(securityDir),
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaServerBuilder.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaServerBuilder.java
index 6ac50c959f5..687c1533519 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaServerBuilder.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaServerBuilder.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.pipe.sink.protocol.opcua.server;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.pipe.api.exception.PipeException;
+import com.google.common.net.InetAddresses;
import org.eclipse.milo.opcua.sdk.server.OpcUaServer;
import org.eclipse.milo.opcua.sdk.server.api.config.OpcUaServerConfig;
import org.eclipse.milo.opcua.sdk.server.identity.CompositeValidator;
@@ -78,9 +79,11 @@ public class OpcUaServerBuilder implements Closeable {
private static final Logger LOGGER =
LoggerFactory.getLogger(OpcUaServerBuilder.class);
private static final String WILD_CARD_ADDRESS = "0.0.0.0";
+ private static final int BACKSLASH = 0x5c;
private int tcpBindPort;
private int httpsBindPort;
+ private String advertisedHost;
private String user;
private String password;
private Path securityDir;
@@ -99,6 +102,54 @@ public class OpcUaServerBuilder implements Closeable {
return this;
}
+ /** Configures the host published in endpoint URLs without changing the
wildcard bind address. */
+ public OpcUaServerBuilder setAdvertisedHost(final String advertisedHost) {
+ this.advertisedHost = normalizeAdvertisedHost(advertisedHost);
+ return this;
+ }
+
+ private static String normalizeAdvertisedHost(final String advertisedHost) {
+ if (Objects.isNull(advertisedHost)) {
+ return null;
+ }
+
+ String normalizedAdvertisedHost = advertisedHost.trim();
+ if (normalizedAdvertisedHost.isEmpty()) {
+ throw invalidAdvertisedHost();
+ }
+ final boolean bracketed =
+ normalizedAdvertisedHost.startsWith("[") ||
normalizedAdvertisedHost.endsWith("]");
+ if (bracketed) {
+ if (!normalizedAdvertisedHost.startsWith("[") ||
!normalizedAdvertisedHost.endsWith("]")) {
+ throw invalidAdvertisedHost();
+ }
+ normalizedAdvertisedHost =
+ normalizedAdvertisedHost.substring(1,
normalizedAdvertisedHost.length() - 1);
+ }
+
+ final boolean isIpAddress =
InetAddresses.isInetAddress(normalizedAdvertisedHost);
+ if (normalizedAdvertisedHost.isEmpty()
+ || (bracketed && !isIpAddress)
+ || normalizedAdvertisedHost.chars().anyMatch(Character::isWhitespace)
+ || normalizedAdvertisedHost.contains("/")
+ || normalizedAdvertisedHost.indexOf(BACKSLASH) >= 0
+ || normalizedAdvertisedHost.contains("?")
+ || normalizedAdvertisedHost.contains("#")
+ || normalizedAdvertisedHost.contains("@")
+ || normalizedAdvertisedHost.contains("[")
+ || normalizedAdvertisedHost.contains("]")
+ || (!isIpAddress && normalizedAdvertisedHost.contains(":"))) {
+ throw invalidAdvertisedHost();
+ }
+ return normalizedAdvertisedHost;
+ }
+
+ private static IllegalArgumentException invalidAdvertisedHost() {
+ return new IllegalArgumentException(
+ DataNodePipeMessages
+
.EXCEPTION_THE_ADVERTISED_HOST_MUST_BE_A_HOSTNAME_OR_IP_ADDRESS_WITHOUT_A_SCHEME_PORT_OR_PATH_6857C67A);
+ }
+
public OpcUaServerBuilder setUser(final String user) {
this.user = user;
return this;
@@ -147,8 +198,10 @@ public class OpcUaServerBuilder implements Closeable {
LoggerFactory.getLogger(OpcUaServerBuilder.class)
.info(DataNodePipeMessages.OPC_UA_SECURITY_PKI_DIR,
pkiDir.getAbsolutePath());
+ final Set<String> endpointHostnames = getEndpointHostnames();
+ final Set<String> certificateHostnames =
getCertificateHostnames(endpointHostnames);
final OpcUaKeyStoreLoader loader =
- new OpcUaKeyStoreLoader().load(securityDir, password.toCharArray());
+ new OpcUaKeyStoreLoader().load(securityDir, password.toCharArray(),
certificateHostnames);
final DefaultCertificateManager certificateManager =
new DefaultCertificateManager(loader.getServerKeyPair(),
loader.getServerCertificate());
@@ -165,8 +218,15 @@ public class OpcUaServerBuilder implements Closeable {
final SelfSignedHttpsCertificateBuilder httpsCertificateBuilder =
new SelfSignedHttpsCertificateBuilder(httpsKeyPair);
- httpsCertificateBuilder.setCommonName(HostnameUtil.getHostname());
-
HostnameUtil.getHostnames(WILD_CARD_ADDRESS).forEach(httpsCertificateBuilder::addDnsName);
+
httpsCertificateBuilder.setCommonName(certificateHostnames.iterator().next());
+ certificateHostnames.forEach(
+ hostname -> {
+ if (InetAddresses.isInetAddress(hostname)) {
+ httpsCertificateBuilder.addIpAddress(hostname);
+ } else {
+ httpsCertificateBuilder.addDnsName(hostname);
+ }
+ });
final X509Certificate httpsCertificate = httpsCertificateBuilder.build();
final DefaultServerCertificateValidator certificateValidator =
@@ -190,6 +250,13 @@ public class OpcUaServerBuilder implements Closeable {
StatusCodes.Bad_ConfigurationError,
DataNodePipeMessages.NO_CERTIFICATE_FOUND));
+ if (Objects.nonNull(advertisedHost) &&
!isAdvertisedHostInCertificate(certificate)) {
+ LOGGER.warn(
+ DataNodePipeMessages
+
.LOG_ADVERTISED_HOST_ARG_IS_NOT_PRESENT_IN_THE_LOADED_OPC_UA_SERVER_CERTIFICATE_SUBJECT_ALTERNATIVE_NAMES_SECURED_CLIENTS_MAY_REJECT_IT_REPLACE_OR_REGENERATE_THE_CERTIFICATE_AND_ESTABLISH_TRUST_AGAIN_912358AF,
+ advertisedHost);
+ }
+
final String applicationUri =
CertificateUtil.getSanUri(certificate)
.orElseThrow(
@@ -199,7 +266,7 @@ public class OpcUaServerBuilder implements Closeable {
DataNodePipeMessages.CERTIFICATE_MISSING_APPLICATION_URI));
final Set<EndpointConfiguration> endpointConfigurations =
- createEndpointConfigurations(certificate, tcpBindPort, httpsBindPort);
+ createEndpointConfigurations(certificate, tcpBindPort, httpsBindPort,
endpointHostnames);
serverConfig =
OpcUaServerConfig.builder()
@@ -233,19 +300,71 @@ public class OpcUaServerBuilder implements Closeable {
return server;
}
- private Set<EndpointConfiguration> createEndpointConfigurations(
- final X509Certificate certificate, final int tcpBindPort, final int
httpsBindPort) {
+ private Set<String> getEndpointHostnames() {
+ if (Objects.nonNull(advertisedHost)) {
+ final Set<String> hostnames = new LinkedHashSet<>();
+ hostnames.add(toEndpointHostname(advertisedHost));
+ return hostnames;
+ }
+ final Set<String> hostnames = new LinkedHashSet<>();
+ hostnames.add(toEndpointHostname(HostnameUtil.getHostname()));
+ HostnameUtil.getHostnames(WILD_CARD_ADDRESS).stream()
+ .map(OpcUaServerBuilder::toEndpointHostname)
+ .forEach(hostnames::add);
+ return hostnames;
+ }
+
+ private static Set<String> getCertificateHostnames(final Set<String>
endpointHostnames) {
+ final Set<String> certificateHostnames = new LinkedHashSet<>();
+ endpointHostnames.stream()
+ .map(OpcUaServerBuilder::removeIpv6Brackets)
+ .forEach(certificateHostnames::add);
+ return certificateHostnames;
+ }
+
+ private static String toEndpointHostname(final String hostname) {
+ return InetAddresses.isInetAddress(hostname) && hostname.indexOf(':') >= 0
+ ? '[' + hostname + ']'
+ : hostname;
+ }
+
+ private static String removeIpv6Brackets(final String hostname) {
+ return hostname.startsWith("[") && hostname.endsWith("]")
+ ? hostname.substring(1, hostname.length() - 1)
+ : hostname;
+ }
+
+ private boolean isAdvertisedHostInCertificate(final X509Certificate
certificate) {
+ if (InetAddresses.isInetAddress(advertisedHost)) {
+ return CertificateUtil.getSanIpAddresses(certificate).stream()
+ .filter(InetAddresses::isInetAddress)
+ .map(InetAddresses::forString)
+ .anyMatch(InetAddresses.forString(advertisedHost)::equals);
+ }
+ return CertificateUtil.getSanDnsNames(certificate).stream()
+ .anyMatch(hostname -> hostname.equalsIgnoreCase(advertisedHost));
+ }
+
+ Set<EndpointConfiguration> createEndpointConfigurations(
+ final X509Certificate certificate,
+ final int tcpBindPort,
+ final int httpsBindPort,
+ final Set<String> hostnames) {
final Set<EndpointConfiguration> endpointConfigurations = new
LinkedHashSet<>();
+ final Set<String> effectiveHostnames = new LinkedHashSet<>();
+ if (Objects.nonNull(advertisedHost)) {
+ effectiveHostnames.add(toEndpointHostname(advertisedHost));
+ } else {
+ hostnames.stream()
+ .map(OpcUaServerBuilder::toEndpointHostname)
+ .forEach(effectiveHostnames::add);
+ }
final List<String> bindAddresses = newArrayList();
bindAddresses.add(WILD_CARD_ADDRESS);
- final Set<String> hostnames = new LinkedHashSet<>();
- hostnames.add(HostnameUtil.getHostname());
- hostnames.addAll(HostnameUtil.getHostnames(WILD_CARD_ADDRESS));
-
for (final String bindAddress : bindAddresses) {
- for (final String hostname : hostnames) {
+ for (final String hostname : effectiveHostnames) {
final EndpointConfiguration.Builder builder =
EndpointConfiguration.newBuilder()
.setBindAddress(bindAddress)
@@ -322,12 +441,14 @@ public class OpcUaServerBuilder implements Closeable {
/////////////////////////////// Conflict detection
///////////////////////////////
void checkEquals(
+ final String advertisedHost,
final String user,
final String password,
final Path securityDir,
final boolean enableAnonymousAccess,
final Set<SecurityPolicy> securityPolicies,
final long debounceTimeMs) {
+ checkEquals("advertised host", this.advertisedHost,
normalizeAdvertisedHost(advertisedHost));
checkEquals("user", this.user, user);
checkEquals("password", this.password, password);
checkEquals(
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunnerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunnerTest.java
new file mode 100644
index 00000000000..c760f661a71
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunnerTest.java
@@ -0,0 +1,147 @@
+/*
+ * 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.db.pipe.sink.protocol.opcua.client;
+
+import org.apache.iotdb.pipe.api.exception.PipeException;
+
+import org.eclipse.milo.opcua.sdk.client.api.identity.AnonymousProvider;
+import org.eclipse.milo.opcua.stack.core.Stack;
+import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
+import org.eclipse.milo.opcua.stack.core.types.builtin.ByteString;
+import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned;
+import org.eclipse.milo.opcua.stack.core.types.enumerated.MessageSecurityMode;
+import org.eclipse.milo.opcua.stack.core.types.structured.EndpointDescription;
+import org.eclipse.milo.opcua.stack.core.types.structured.UserTokenPolicy;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.nio.file.Paths;
+import java.util.Arrays;
+
+public class ClientRunnerTest {
+
+ private static final String CONFIGURED_NODE_URL =
"opc.tcp://10.60.80.65:12686/iotdb";
+ private static final SecurityPolicy SECURITY_POLICY =
SecurityPolicy.Basic256Sha256;
+
+ @Test
+ public void testConfiguredHostAndPortAreUsedByDefault() {
+ final EndpointDescription advertisedEndpoint =
+ createEndpoint(
+ "opc.tcp://fwq03-15:4840/server-path",
+ SECURITY_POLICY,
+ Stack.TCP_UASC_UABINARY_TRANSPORT_URI);
+
+ final EndpointDescription effectiveEndpoint =
+ ClientRunner.selectEndpoint(
+ Arrays.asList(advertisedEndpoint), CONFIGURED_NODE_URL,
SECURITY_POLICY, false)
+ .orElseThrow(AssertionError::new);
+
+ Assert.assertEquals(
+ "opc.tcp://10.60.80.65:12686/server-path",
effectiveEndpoint.getEndpointUrl());
+ Assert.assertNotSame(advertisedEndpoint, effectiveEndpoint);
+ Assert.assertSame(
+ advertisedEndpoint.getServerCertificate(),
effectiveEndpoint.getServerCertificate());
+ Assert.assertEquals(advertisedEndpoint.getSecurityMode(),
effectiveEndpoint.getSecurityMode());
+ Assert.assertEquals(
+ advertisedEndpoint.getSecurityPolicyUri(),
effectiveEndpoint.getSecurityPolicyUri());
+ Assert.assertSame(
+ advertisedEndpoint.getUserIdentityTokens(),
effectiveEndpoint.getUserIdentityTokens());
+ Assert.assertEquals(
+ advertisedEndpoint.getTransportProfileUri(),
effectiveEndpoint.getTransportProfileUri());
+ }
+
+ @Test
+ public void testAdvertisedEndpointIsUsedWhenRedirectIsAllowed() {
+ final EndpointDescription advertisedEndpoint =
+ createEndpoint(
+ "opc.tcp://fwq03-15:4840/server-path",
+ SECURITY_POLICY,
+ Stack.TCP_UASC_UABINARY_TRANSPORT_URI);
+
+ final EndpointDescription effectiveEndpoint =
+ ClientRunner.selectEndpoint(
+ Arrays.asList(advertisedEndpoint), CONFIGURED_NODE_URL,
SECURITY_POLICY, true)
+ .orElseThrow(AssertionError::new);
+
+ Assert.assertSame(advertisedEndpoint, effectiveEndpoint);
+ }
+
+ @Test
+ public void testEndpointSelectionMatchesConfiguredTransport() {
+ final EndpointDescription wrongSchemeEndpoint =
+ createEndpoint(
+ "https://wrong-scheme:12686/iotdb";,
+ SECURITY_POLICY,
+ Stack.HTTPS_UABINARY_TRANSPORT_URI);
+ final EndpointDescription wrongTransportEndpoint =
+ createEndpoint(
+ "opc.tcp://wrong-transport:12686/iotdb",
+ SECURITY_POLICY,
+ Stack.HTTPS_UABINARY_TRANSPORT_URI);
+ final EndpointDescription matchingEndpoint =
+ createEndpoint(
+ "opc.tcp://matching:12686/iotdb",
+ SECURITY_POLICY,
+ Stack.TCP_UASC_UABINARY_TRANSPORT_URI);
+
+ final EndpointDescription selectedEndpoint =
+ ClientRunner.selectEndpoint(
+ Arrays.asList(wrongSchemeEndpoint, wrongTransportEndpoint,
matchingEndpoint),
+ CONFIGURED_NODE_URL,
+ SECURITY_POLICY,
+ true)
+ .orElseThrow(AssertionError::new);
+
+ Assert.assertSame(matchingEndpoint, selectedEndpoint);
+ }
+
+ @Test
+ public void testAllowEndpointRedirectParticipatesInConflictDetection() {
+ final String securityDir = "target/opcua-client-runner-test";
+ final IoTDBOpcUaClient client =
+ new IoTDBOpcUaClient(
+ CONFIGURED_NODE_URL, SECURITY_POLICY, AnonymousProvider.INSTANCE,
false);
+ final ClientRunner runner = new ClientRunner(client, securityDir,
"password", null, 10, false);
+
+ runner.checkEquals(null, "password", Paths.get(securityDir),
SECURITY_POLICY, false);
+ final PipeException exception =
+ Assert.assertThrows(
+ PipeException.class,
+ () ->
+ runner.checkEquals(
+ null, "password", Paths.get(securityDir), SECURITY_POLICY,
true));
+ Assert.assertTrue(exception.getMessage().contains("allow endpoint
redirect"));
+ }
+
+ private static EndpointDescription createEndpoint(
+ final String endpointUrl,
+ final SecurityPolicy securityPolicy,
+ final String transportProfileUri) {
+ return new EndpointDescription(
+ endpointUrl,
+ null,
+ ByteString.of(new byte[] {1, 2, 3}),
+ MessageSecurityMode.SignAndEncrypt,
+ securityPolicy.getUri(),
+ new UserTokenPolicy[0],
+ transportProfileUri,
+ Unsigned.ubyte(1));
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaServerBuilderTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaServerBuilderTest.java
new file mode 100644
index 00000000000..97938f56d09
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaServerBuilderTest.java
@@ -0,0 +1,164 @@
+/*
+ * 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.db.pipe.sink.protocol.opcua.server;
+
+import org.apache.iotdb.pipe.api.exception.PipeException;
+
+import org.eclipse.milo.opcua.sdk.server.OpcUaServer;
+import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
+import org.eclipse.milo.opcua.stack.core.util.CertificateUtil;
+import org.eclipse.milo.opcua.stack.server.EndpointConfiguration;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class OpcUaServerBuilderTest {
+
+ @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Test
+ public void testDetectedHostsArePublishedByDefault() {
+ final Set<SecurityPolicy> securityPolicies =
Collections.singleton(SecurityPolicy.None);
+ final Set<String> detectedHostnames =
+ new LinkedHashSet<>(Arrays.asList("opc-server", "10.60.80.65"));
+ final OpcUaServerBuilder builder =
+ new OpcUaServerBuilder().setSecurityPolicies(securityPolicies);
+
+ final Set<EndpointConfiguration> endpoints =
+ builder.createEndpointConfigurations(null, 12686, 8443,
detectedHostnames);
+
+ for (final String hostname : detectedHostnames) {
+ Assert.assertEquals(
+ 2,
+ endpoints.stream()
+ .filter(endpoint -> endpoint.getPath().equals("/iotdb"))
+ .filter(endpoint -> endpoint.getHostname().equals(hostname))
+ .filter(endpoint -> endpoint.getSecurityPolicy() ==
SecurityPolicy.None)
+ .count());
+ }
+ Assert.assertEquals(Collections.singleton(SecurityPolicy.None),
securityPolicies);
+ }
+
+ @Test
+ public void testOnlyExplicitAdvertisedHostIsPublished() {
+ final Set<String> detectedHostnames =
+ new LinkedHashSet<>(Arrays.asList("opc-server", "10.60.80.65"));
+ final OpcUaServerBuilder builder =
+ new OpcUaServerBuilder()
+ .setAdvertisedHost("opc.example.com")
+ .setSecurityPolicies(Collections.singleton(SecurityPolicy.None));
+
+ final Set<EndpointConfiguration> endpoints =
+ builder.createEndpointConfigurations(null, 12686, 8443,
detectedHostnames);
+
+ Assert.assertEquals(
+ Collections.singleton("opc.example.com"),
+
endpoints.stream().map(EndpointConfiguration::getHostname).collect(Collectors.toSet()));
+ Assert.assertTrue(
+ endpoints.stream().allMatch(endpoint ->
"0.0.0.0".equals(endpoint.getBindAddress())));
+ }
+
+ @Test
+ public void testIpv6AdvertisedHostIsNormalizedAndEndpointUrlIsRejected() {
+ final OpcUaServerBuilder builder =
+ new OpcUaServerBuilder()
+ .setAdvertisedHost("[2001:db8::1]")
+ .setSecurityPolicies(Collections.singleton(SecurityPolicy.None));
+
+ final Set<EndpointConfiguration> endpoints =
+ builder.createEndpointConfigurations(
+ null, 12686, 8443, Collections.singleton("opc-server"));
+
+ Assert.assertEquals(
+ Collections.singleton("[2001:db8::1]"),
+
endpoints.stream().map(EndpointConfiguration::getHostname).collect(Collectors.toSet()));
+ Assert.assertTrue(
+ endpoints.stream()
+ .map(EndpointConfiguration::getEndpointUrl)
+ .allMatch(endpointUrl ->
endpointUrl.contains("://[2001:db8::1]:")));
+ Assert.assertThrows(
+ IllegalArgumentException.class,
+ () ->
builder.setAdvertisedHost("opc.tcp://opc.example.com:12686/iotdb"));
+ }
+
+ @Test
+ public void testNewCertificateContainsAdvertisedHost() throws Exception {
+ final String advertisedHost = "opc.example.com";
+ final Path securityDir = temporaryFolder.newFolder("security").toPath();
+
+ try (final OpcUaServerBuilder builder =
+ new OpcUaServerBuilder()
+ .setTcpBindPort(12686)
+ .setHttpsBindPort(8443)
+ .setAdvertisedHost(advertisedHost)
+ .setUser("root")
+ .setPassword("root")
+ .setSecurityDir(securityDir.toString())
+ .setEnableAnonymousAccess(true)
+ .setSecurityPolicies(Collections.singleton(SecurityPolicy.None))
+ .setDebounceTimeMs(50)) {
+ final OpcUaServer server = builder.build();
+ final Set<EndpointConfiguration> endpoints =
server.getConfig().getEndpoints();
+
+ Assert.assertEquals(
+ Collections.singleton(advertisedHost),
+
endpoints.stream().map(EndpointConfiguration::getHostname).collect(Collectors.toSet()));
+ Assert.assertTrue(
+ endpoints.stream()
+ .map(EndpointConfiguration::getCertificate)
+ .allMatch(
+ certificate ->
+
CertificateUtil.getSanDnsNames(certificate).contains(advertisedHost)));
+ }
+ }
+
+ @Test
+ public void testAdvertisedHostParticipatesInConflictDetection() {
+ final Path securityDir = temporaryFolder.getRoot().toPath();
+ final Set<SecurityPolicy> securityPolicies =
Collections.singleton(SecurityPolicy.None);
+ final OpcUaServerBuilder builder =
+ new OpcUaServerBuilder()
+ .setAdvertisedHost("opc.example.com")
+ .setUser("root")
+ .setPassword("root")
+ .setSecurityDir(securityDir.toString())
+ .setEnableAnonymousAccess(true)
+ .setSecurityPolicies(securityPolicies)
+ .setDebounceTimeMs(50);
+
+ builder.checkEquals("opc.example.com", "root", "root", securityDir, true,
securityPolicies, 50);
+ final PipeException exception =
+ Assert.assertThrows(
+ PipeException.class,
+ () ->
+ builder.checkEquals(
+ "other.example.com", "root", "root", securityDir, true,
securityPolicies, 50));
+
+ Assert.assertTrue(exception.getMessage().contains("advertised host"));
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java
index 25eb4c8bb21..2db1bd0c06b 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java
@@ -207,6 +207,10 @@ public class PipeSinkConstant {
public static final String SINK_OPC_UA_HTTPS_BIND_PORT_KEY =
"sink.opcua.https.port";
public static final int CONNECTOR_OPC_UA_HTTPS_BIND_PORT_DEFAULT_VALUE =
8443;
+ public static final String CONNECTOR_OPC_UA_ADVERTISED_HOST_KEY =
+ "connector.opcua.advertised-host";
+ public static final String SINK_OPC_UA_ADVERTISED_HOST_KEY =
"sink.opcua.advertised-host";
+
public static final String CONNECTOR_OPC_UA_SECURITY_DIR_KEY =
"connector.opcua.security.dir";
public static final String SINK_OPC_UA_SECURITY_DIR_KEY =
"sink.opcua.security.dir";
public static final String CONNECTOR_OPC_UA_SECURITY_DIR_DEFAULT_VALUE =
@@ -245,6 +249,11 @@ public class PipeSinkConstant {
public static final String CONNECTOR_OPC_UA_NODE_URL_KEY =
"connector.opcua.node-url";
public static final String SINK_OPC_UA_NODE_URL_KEY = "sink.opcua.node-url";
+ public static final String CONNECTOR_OPC_UA_ALLOW_ENDPOINT_REDIRECT_KEY =
+ "connector.opcua.allow-endpoint-redirect";
+ public static final String SINK_OPC_UA_ALLOW_ENDPOINT_REDIRECT_KEY =
+ "sink.opcua.allow-endpoint-redirect";
+ public static final boolean
CONNECTOR_OPC_UA_ALLOW_ENDPOINT_REDIRECT_DEFAULT_VALUE = false;
public static final String CONNECTOR_OPC_UA_SECURITY_POLICY_KEY =
"connector.opcua.security-policy";
