feat(plc4j/opcua): Chunking and encryption of request/response calls.

Update of protocol logic to reflect chunked payloads.

Threading updates within driver itself should allow us to have better control over how results are being processed.. and make error handling more consistent.

Signed-off-by: Łukasz Dywicki <[email protected]>

plc4j/drivers/opcua/pom.xml

@@ -98,6 +98,26 @@
           </excludes>
         </configuration>
       </plugin>
+      <!-- Check if all source files have the required apache license headers -->
+      <plugin>
+        <groupId>org.apache.rat</groupId>
+        <artifactId>apache-rat-plugin</artifactId>
+        <executions>
+          <execution>
+            <id>license-check</id>
+            <phase>verify</phase>
+            <goals>
+              <goal>check</goal>
+            </goals>
+          </execution>
+        </executions>
+        <configuration>
+          <excludes>
+            <!-- exclude static test resources from release checks -->
+            <exclude>src/test/resources/chunk-calculation*.csv</exclude>
+          </excludes>
+        </configuration>
+      </plugin>
     </plugins>
   </build>
 
@@ -127,6 +147,10 @@
       <groupId>org.apache.commons</groupId>
       <artifactId>commons-lang3</artifactId>
     </dependency>
+    <dependency>
+      <groupId>commons-codec</groupId>
+      <artifactId>commons-codec</artifactId>
+    </dependency>
 
     <dependency>
       <groupId>io.vavr</groupId>

plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/config/Limits.java

@@ -0,0 +1,77 @@
+/*
+ * 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.plc4x.java.opcua.config;
+
+import org.apache.plc4x.java.spi.configuration.Configuration;
+import org.apache.plc4x.java.spi.configuration.annotations.ConfigurationParameter;
+
+public class Limits implements Configuration {
+
+    private static final int DEFAULT_RECEIVE_BUFFER_SIZE = 65535;
+    private static final int DEFAULT_SEND_BUFFER_SIZE = 65535;
+    private static final int DEFAULT_MAX_MESSAGE_SIZE = 2097152;
+    private static final int DEFAULT_MAX_CHUNK_COUNT = 64;
+
+    @ConfigurationParameter("receiveBufferSize")
+    private int receiveBufferSize;
+    @ConfigurationParameter("sendBufferSize")
+    private int sendBufferSize;
+    @ConfigurationParameter("maxMessageSize")
+    private int maxMessageSize;
+    @ConfigurationParameter("maxChunkCount")
+    private int maxChunkCount;
+
+    public Limits() {
+        this(DEFAULT_RECEIVE_BUFFER_SIZE, DEFAULT_SEND_BUFFER_SIZE, DEFAULT_MAX_MESSAGE_SIZE, DEFAULT_MAX_CHUNK_COUNT);
+    }
+
+    public Limits(int receiveBufferSize, int sendBufferSize, int maxMessageSize, int maxChunkCount) {
+        this.receiveBufferSize = receiveBufferSize;
+        this.sendBufferSize = sendBufferSize;
+        this.maxMessageSize = maxMessageSize;
+        this.maxChunkCount = maxChunkCount;
+    }
+
+    public int getReceiveBufferSize() {
+        return receiveBufferSize;
+    }
+
+    public int getSendBufferSize() {
+        return sendBufferSize;
+    }
+
+    public int getMaxMessageSize() {
+        return maxMessageSize;
+    }
+
+    public int getMaxChunkCount() {
+        return maxChunkCount;
+    }
+
+    @Override
+    public String toString() {
+        return "Limits{" +
+            " receiveBufferSize=" + receiveBufferSize +
+            ", sendBufferSize=" + sendBufferSize +
+            ", maxMessageSize=" + maxMessageSize +
+            ", maxChunkCount=" + maxChunkCount +
+            '}';
+    }
+}

plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/config/OpcuaConfiguration.java

@@ -18,15 +18,29 @@
  */
 package org.apache.plc4x.java.opcua.config;
 
-import org.apache.plc4x.java.opcua.readwrite.PascalByteString;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.KeyStore;
+import java.security.cert.X509Certificate;
+import org.apache.plc4x.java.opcua.context.SecureChannel;
+import org.apache.plc4x.java.opcua.security.MessageSecurity;
 import org.apache.plc4x.java.opcua.security.SecurityPolicy;
 import org.apache.plc4x.java.spi.configuration.Configuration;
+import org.apache.plc4x.java.spi.configuration.annotations.ComplexConfigurationParameter;
 import org.apache.plc4x.java.spi.configuration.annotations.ConfigurationParameter;
 import org.apache.plc4x.java.spi.configuration.annotations.defaults.BooleanDefaultValue;
-import org.apache.plc4x.java.spi.configuration.annotations.defaults.StringDefaultValue;
 
 public class OpcuaConfiguration implements Configuration {
 
+    public static final long DEFAULT_CHANNEL_LIFETIME = 3600000;
+
+    public static final long DEFAULT_SESSION_TIMEOUT = 120000;
+    public static final long DEFAULT_NEGOTIATION_TIMEOUT = 60000;
+
+    public static final long DEFAULT_REQUEST_TIMEOUT = 30000;
+
     @ConfigurationParameter("protocolCode")
     private String protocolCode;
 
@@ -49,16 +63,47 @@ public class OpcuaConfiguration implements Configuration {
     @ConfigurationParameter("securityPolicy")
     private SecurityPolicy securityPolicy = SecurityPolicy.NONE;
 
+    @ConfigurationParameter("messageSecurity")
+    private MessageSecurity messageSecurity = MessageSecurity.SIGN_ENCRYPT;
+
     @ConfigurationParameter("keyStoreFile")
     private String keyStoreFile;
 
-    @ConfigurationParameter("certDirectory")
-    private String certDirectory;
+    @ConfigurationParameter("keyStoreType")
+    private String keyStoreType = KeyStore.getDefaultType();
 
     @ConfigurationParameter("keyStorePassword")
     private String keyStorePassword;
-    private byte[] senderCertificate;
-    private PascalByteString thumbprint;
+
+    @ConfigurationParameter("serverCertificateFile")
+    private String serverCertificateFile;
+
+    @ConfigurationParameter("trustStoreFile")
+    private String trustStoreFile;
+
+    @ConfigurationParameter("trustStoreType")
+    private String trustStoreType = KeyStore.getDefaultType();
+
+    @ConfigurationParameter("trustStorePassword")
+    private String trustStorePassword;
+
+    // the discovered certificate when discovery is enabled
+    private X509Certificate serverCertificate;
+
+    @ConfigurationParameter("channelLifetime")
+    private long channelLifetime = DEFAULT_CHANNEL_LIFETIME;
+
+    @ConfigurationParameter("sessionTimeout")
+    private long sessionTimeout = DEFAULT_SESSION_TIMEOUT;
+
+    @ConfigurationParameter("negotiationTimeout")
+    private long negotiationTimeout = DEFAULT_NEGOTIATION_TIMEOUT;
+
+    @ConfigurationParameter("requestTimeout")
+    private long requestTimeout = DEFAULT_REQUEST_TIMEOUT;
+
+    @ComplexConfigurationParameter(prefix = "encoding", defaultOverrides = {}, requiredOverrides = {})
+    private Limits limits = new Limits();
 
     public String getProtocolCode() {
         return protocolCode;
@@ -84,20 +129,73 @@ public String getPassword() {
         return password;
     }
 
-    public String getCertDirectory() {
-        return certDirectory;
-    }
-
     public SecurityPolicy getSecurityPolicy() {
         return securityPolicy;
     }
 
+    public MessageSecurity getMessageSecurity() {
+        return messageSecurity;
+    }
+
     public String getKeyStoreFile() {
         return keyStoreFile;
     }
 
-    public String getKeyStorePassword() {
-        return keyStorePassword;
+    public String getKeyStoreType() {
+        return keyStoreType;
+    }
+
+    public char[] getKeyStorePassword() {
+        return keyStorePassword == null ? null : keyStorePassword.toCharArray();
+    }
+
+    public String getTrustStoreFile() {
+        return trustStoreFile;
+    }
+
+    public String getTrustStoreType() {
+        return trustStoreType;
+    }
+
+    public char[] getTrustStorePassword() {
+        return trustStorePassword == null ? null : trustStorePassword.toCharArray();
+    }
+
+    public Limits getEncodingLimits() {
+        return limits;
+    }
+
+    public X509Certificate getServerCertificate() {
+        if (serverCertificate == null && serverCertificateFile != null) {
+            // initialize server certificate from configured file
+            try {
+                byte[] certificateBytes = Files.readAllBytes(Path.of(serverCertificateFile));
+                serverCertificate = SecureChannel.getX509Certificate(certificateBytes);
+            } catch (IOException e) {
+                throw new RuntimeException(e);
+            }
+        }
+        return serverCertificate;
+    }
+
+    public void setServerCertificate(X509Certificate serverCertificate) {
+        this.serverCertificate = serverCertificate;
+    }
+
+    public long getChannelLifetime() {
+        return channelLifetime;
+    }
+
+    public long getSessionTimeout() {
+        return sessionTimeout;
+    }
+
+    public long getRequestTimeout() {
+        return requestTimeout;
+    }
+
+    public long getNegotiationTimeout() {
+        return negotiationTimeout;
     }
 
     @Override
@@ -108,26 +206,9 @@ public String toString() {
             ", password='" + (password != null ? "******" : null) + '\'' +
             ", securityPolicy='" + securityPolicy + '\'' +
             ", keyStoreFile='" + keyStoreFile + '\'' +
-            ", certDirectory='" + certDirectory + '\'' +
             ", keyStorePassword='" + (keyStorePassword != null ? "******" : null) + '\'' +
+            ", limits=" + limits +
             '}';
     }
-
-    public byte[] getSenderCertificate() {
-        return senderCertificate;
-    }
-
-    public void setSenderCertificate(byte[] senderCertificate) {
-        this.senderCertificate = senderCertificate;
-    }
-
-    public PascalByteString getThumbprint() {
-        return this.thumbprint;
-    }
-
-    public void setThumbprint(PascalByteString thumbprint) {
-        this.thumbprint = thumbprint;
-    }
-
 }