Skip to content

Commit efa0fe4

Browse files
CaideyipiCaideyipi
committed
bone
1 parent 109a3fc commit efa0fe4

File tree

6 files changed

+526
-2
lines changed

6 files changed

+526
-2
lines changed

‎iotdb-core/datanode/pom.xml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -191,6 +191,11 @@
191191
<groupId>org.eclipse.milo</groupId>
192192
<artifactId>stack-server</artifactId>
193193
</dependency>
194+
<dependency>
195+
<groupId>org.eclipse.milo</groupId>
196+
<artifactId>sdk-client</artifactId>
197+
<version>${milo.version}</version>
198+
</dependency>
194199
<dependency>
195200
<groupId>org.eclipse.jetty</groupId>
196201
<artifactId>jetty-http</artifactId>

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java‎

Lines changed: 75 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.apache.iotdb.db.conf.IoTDBConfig;
2525
import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
2626
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
27+
import org.apache.iotdb.db.pipe.sink.protocol.opcua.client.IoTDBOpcUaClient;
2728
import org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace;
2829
import org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaServerBuilder;
2930
import org.apache.iotdb.db.storageengine.StorageEngine;
@@ -40,7 +41,11 @@
4041

4142
import org.apache.tsfile.utils.Pair;
4243
import org.apache.tsfile.write.record.Tablet;
44+
import org.eclipse.milo.opcua.sdk.client.api.identity.AnonymousProvider;
45+
import org.eclipse.milo.opcua.sdk.client.api.identity.IdentityProvider;
46+
import org.eclipse.milo.opcua.sdk.client.api.identity.UsernameProvider;
4347
import org.eclipse.milo.opcua.sdk.server.OpcUaServer;
48+
import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
4449
import org.slf4j.Logger;
4550
import org.slf4j.LoggerFactory;
4651

@@ -66,12 +71,20 @@
6671
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_MODEL_DEFAULT_VALUE;
6772
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_MODEL_KEY;
6873
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_MODEL_PUB_SUB_VALUE;
74+
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_NODE_URL_KEY;
6975
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_PLACEHOLDER_DEFAULT_VALUE;
7076
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_PLACEHOLDER_KEY;
7177
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_QUALITY_NAME_DEFAULT_VALUE;
7278
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_QUALITY_NAME_KEY;
79+
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_QUALITY_SECURITY_POLICY_AES128_SHA256_RSAOAEP_VALUE;
80+
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_QUALITY_SECURITY_POLICY_AES256_SHA256_RSAPSS_VALUE;
81+
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_QUALITY_SECURITY_POLICY_BASIC_128_RSA_15_VALUE;
82+
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_QUALITY_SECURITY_POLICY_BASIC_256_SHA_256_VALUE;
83+
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_QUALITY_SECURITY_POLICY_BASIC_256_VALUE;
84+
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_QUALITY_SECURITY_POLICY_NONE_VALUE;
7385
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_SECURITY_DIR_DEFAULT_VALUE;
7486
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_SECURITY_DIR_KEY;
87+
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_SECURITY_POLICY_KEY;
7588
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_TCP_BIND_PORT_DEFAULT_VALUE;
7689
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_TCP_BIND_PORT_KEY;
7790
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_VALUE_NAME_DEFAULT_VALUE;
@@ -84,9 +97,11 @@
8497
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_ENABLE_ANONYMOUS_ACCESS_KEY;
8598
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_HTTPS_BIND_PORT_KEY;
8699
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_MODEL_KEY;
100+
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_NODE_URL_KEY;
87101
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_PLACEHOLDER_KEY;
88102
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_QUALITY_NAME_KEY;
89103
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_SECURITY_DIR_KEY;
104+
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_SECURITY_POLICY_KEY;
90105
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_TCP_BIND_PORT_KEY;
91106
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_VALUE_NAME_KEY;
92107
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_WITH_QUALITY_KEY;
@@ -114,7 +129,12 @@ public class OpcUaSink implements PipeConnector {
114129
String placeHolder;
115130
@Nullable String valueName;
116131
@Nullable String qualityName;
117-
private OpcUaNameSpace nameSpace;
132+
133+
// Inner server
134+
private @Nullable OpcUaNameSpace nameSpace;
135+
136+
// Outer server
137+
private @Nullable IoTDBOpcUaClient client;
118138

119139
@Override
120140
public void validate(final PipeParameterValidator validator) throws Exception {
@@ -139,6 +159,17 @@ public void validate(final PipeParameterValidator validator) throws Exception {
139159
public void customize(
140160
final PipeParameters parameters, final PipeConnectorRuntimeConfiguration configuration)
141161
throws Exception {
162+
final String nodeUrl =
163+
parameters.getStringByKeys(CONNECTOR_OPC_UA_NODE_URL_KEY, SINK_OPC_UA_NODE_URL_KEY);
164+
if (Objects.isNull(nodeUrl)) {
165+
customizeServer(parameters, configuration);
166+
} else {
167+
customizeClient(nodeUrl, parameters);
168+
}
169+
}
170+
171+
private void customizeServer(
172+
final PipeParameters parameters, final PipeConnectorRuntimeConfiguration configuration) {
142173
final int tcpBindPort =
143174
parameters.getIntOrDefault(
144175
Arrays.asList(CONNECTOR_OPC_UA_TCP_BIND_PORT_KEY, SINK_OPC_UA_TCP_BIND_PORT_KEY),
@@ -250,6 +281,48 @@ public void customize(
250281
}
251282
}
252283

284+
private void customizeClient(final String nodeUrl, final PipeParameters parameters) {
285+
final SecurityPolicy policy;
286+
switch (parameters
287+
.getStringOrDefault(
288+
Arrays.asList(CONNECTOR_OPC_UA_SECURITY_POLICY_KEY, SINK_OPC_UA_SECURITY_POLICY_KEY),
289+
CONNECTOR_OPC_UA_QUALITY_SECURITY_POLICY_BASIC_256_SHA_256_VALUE)
290+
.toUpperCase()) {
291+
case CONNECTOR_OPC_UA_QUALITY_SECURITY_POLICY_NONE_VALUE:
292+
policy = SecurityPolicy.None;
293+
break;
294+
case CONNECTOR_OPC_UA_QUALITY_SECURITY_POLICY_BASIC_128_RSA_15_VALUE:
295+
policy = SecurityPolicy.Basic128Rsa15;
296+
break;
297+
case CONNECTOR_OPC_UA_QUALITY_SECURITY_POLICY_BASIC_256_VALUE:
298+
policy = SecurityPolicy.Basic256;
299+
break;
300+
case CONNECTOR_OPC_UA_QUALITY_SECURITY_POLICY_BASIC_256_SHA_256_VALUE:
301+
policy = SecurityPolicy.Basic256Sha256;
302+
break;
303+
case CONNECTOR_OPC_UA_QUALITY_SECURITY_POLICY_AES128_SHA256_RSAOAEP_VALUE:
304+
policy = SecurityPolicy.Aes128_Sha256_RsaOaep;
305+
break;
306+
case CONNECTOR_OPC_UA_QUALITY_SECURITY_POLICY_AES256_SHA256_RSAPSS_VALUE:
307+
policy = SecurityPolicy.Aes256_Sha256_RsaPss;
308+
break;
309+
default:
310+
policy = null;
311+
break;
312+
}
313+
314+
final IdentityProvider provider;
315+
final String userName =
316+
parameters.getStringByKeys(CONNECTOR_IOTDB_USER_KEY, SINK_IOTDB_USER_KEY);
317+
provider =
318+
Objects.nonNull(userName)
319+
? new UsernameProvider(
320+
userName,
321+
parameters.getStringByKeys(CONNECTOR_IOTDB_PASSWORD_KEY, SINK_IOTDB_PASSWORD_KEY))
322+
: new AnonymousProvider();
323+
client = new IoTDBOpcUaClient(nodeUrl, policy, provider);
324+
}
325+
253326
@Override
254327
public void handshake() throws Exception {
255328
// Server side, do nothing
@@ -359,7 +432,7 @@ public void close() throws Exception {
359432
}
360433
}
361434

362-
// Getter
435+
/////////////////////////////// Getter ///////////////////////////////
363436

364437
public boolean isClientServerModel() {
365438
return isClientServerModel;

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunner.java‎

Lines changed: 147 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,147 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iotdb.db.pipe.sink.protocol.opcua.client;
21+
22+
import org.bouncycastle.jce.provider.BouncyCastleProvider;
23+
import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
24+
import org.eclipse.milo.opcua.stack.client.security.DefaultClientCertificateValidator;
25+
import org.eclipse.milo.opcua.stack.core.Stack;
26+
import org.eclipse.milo.opcua.stack.core.security.DefaultTrustListManager;
27+
import org.eclipse.milo.opcua.stack.core.types.builtin.LocalizedText;
28+
import org.slf4j.LoggerFactory;
29+
30+
import java.io.File;
31+
import java.nio.file.Files;
32+
import java.nio.file.Path;
33+
import java.nio.file.Paths;
34+
import java.security.Security;
35+
import java.util.concurrent.CompletableFuture;
36+
import java.util.concurrent.ExecutionException;
37+
import java.util.concurrent.TimeUnit;
38+
39+
import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;
40+
41+
public class ClientRunner {
42+
43+
static {
44+
// Required for SecurityPolicy.Aes256_Sha256_RsaPss
45+
Security.addProvider(new BouncyCastleProvider());
46+
}
47+
48+
private final CompletableFuture<OpcUaClient> future = new CompletableFuture<>();
49+
50+
private final IoTDBOpcUaClient configurableUaClient;
51+
52+
public ClientRunner(IoTDBOpcUaClient configurableUaClient) {
53+
this.configurableUaClient = configurableUaClient;
54+
}
55+
56+
private OpcUaClient createClient() throws Exception {
57+
final Path securityTempDir =
58+
Paths.get(System.getProperty("java.io.tmpdir"), "client", "security");
59+
Files.createDirectories(securityTempDir);
60+
if (!Files.exists(securityTempDir)) {
61+
throw new Exception("unable to create security dir: " + securityTempDir);
62+
}
63+
64+
final File pkiDir = securityTempDir.resolve("pki").toFile();
65+
66+
System.out.println("security dir: " + securityTempDir.toAbsolutePath());
67+
LoggerFactory.getLogger(getClass()).info("security pki dir: {}", pkiDir.getAbsolutePath());
68+
69+
final IoTDBKeyStoreLoaderClient loader = new IoTDBKeyStoreLoaderClient().load(securityTempDir);
70+
71+
final DefaultTrustListManager trustListManager = new DefaultTrustListManager(pkiDir);
72+
73+
final DefaultClientCertificateValidator certificateValidator =
74+
new DefaultClientCertificateValidator(trustListManager);
75+
76+
return OpcUaClient.create(
77+
configurableUaClient.getNodeUrl(),
78+
endpoints -> endpoints.stream().filter(configurableUaClient.endpointFilter()).findFirst(),
79+
configBuilder ->
80+
configBuilder
81+
.setApplicationName(LocalizedText.english("eclipse milo opc-ua client"))
82+
.setApplicationUri("urn:eclipse:milo:examples:client")
83+
.setKeyPair(loader.getClientKeyPair())
84+
.setCertificate(loader.getClientCertificate())
85+
.setCertificateChain(loader.getClientCertificateChain())
86+
.setCertificateValidator(certificateValidator)
87+
.setIdentityProvider(configurableUaClient.getIdentityProvider())
88+
.setRequestTimeout(uint(5000))
89+
.build());
90+
}
91+
92+
public void run() {
93+
try {
94+
final OpcUaClient client = createClient();
95+
96+
future.whenCompleteAsync(
97+
(c, ex) -> {
98+
if (ex != null) {
99+
System.out.println("Error running example: " + ex.getMessage());
100+
}
101+
102+
try {
103+
client.disconnect().get();
104+
Stack.releaseSharedResources();
105+
} catch (InterruptedException | ExecutionException e) {
106+
Thread.currentThread().interrupt();
107+
System.out.println("Error disconnecting: {}" + e.getMessage());
108+
}
109+
110+
try {
111+
Thread.sleep(1000);
112+
System.exit(0);
113+
} catch (InterruptedException e) {
114+
Thread.currentThread().interrupt();
115+
e.printStackTrace();
116+
}
117+
});
118+
119+
try {
120+
configurableUaClient.run(client);
121+
future.get(100000, TimeUnit.SECONDS);
122+
} catch (Throwable t) {
123+
System.out.println("Error running client example: " + t.getMessage() + t);
124+
future.completeExceptionally(t);
125+
}
126+
} catch (Throwable t) {
127+
System.out.println("Error getting client: {}" + t.getMessage());
128+
129+
future.completeExceptionally(t);
130+
131+
try {
132+
Thread.sleep(1000);
133+
System.exit(0);
134+
} catch (InterruptedException e) {
135+
Thread.currentThread().interrupt();
136+
e.printStackTrace();
137+
}
138+
}
139+
140+
try {
141+
Thread.sleep(999_999_999);
142+
} catch (InterruptedException e) {
143+
Thread.currentThread().interrupt();
144+
e.printStackTrace();
145+
}
146+
}
147+
}

0 commit comments

Comments
 (0)