feat(#3710): Centrally manage OPC-UA certificates (#3720)

* feat(#3710): Add initial support for centrally managed OPC-UA certificates

* Modify imports

* Add certificate details dialog, improve exception messages

* Add missing headers

* Fix checkstyle

Author: dominikriemer

streampipes-client-api/src/main/java/org/apache/streampipes/client/api/ICustomRequestApi.java

@@ -18,6 +18,7 @@
 
 package org.apache.streampipes.client.api;
 
+import java.util.List;
 import java.util.Map;
 
 public interface ICustomRequestApi {
@@ -26,4 +27,6 @@ public interface ICustomRequestApi {
   <T> T sendGet(String apiPath, Class<T> responseClass);
 
   <T> T sendGet(String apiPath, Map<String, String> queryParameters, Class<T> responseClass);
+
+  <T> List<T> getList(String apiPath, Class<T> response);
 }

streampipes-client/src/main/java/org/apache/streampipes/client/api/AbstractClientApi.java

@@ -25,6 +25,7 @@
 import org.apache.streampipes.client.http.PostRequestWithoutPayloadResponse;
 import org.apache.streampipes.client.http.PutRequest;
 import org.apache.streampipes.client.model.StreamPipesClientConfig;
+import org.apache.streampipes.client.serializer.ListSerializer;
 import org.apache.streampipes.client.serializer.ObjectSerializer;
 import org.apache.streampipes.client.serializer.Serializer;
 import org.apache.streampipes.client.util.StreamPipesApiPath;
@@ -33,6 +34,7 @@
 
 import org.apache.http.HttpStatus;
 
+import java.util.List;
 import java.util.Optional;
 
 public class AbstractClientApi {
@@ -86,6 +88,11 @@ protected <T> T getSingle(StreamPipesApiPath apiPath, Class<T> targetClass) thro
     return new GetRequest<>(clientConfig, apiPath, targetClass, serializer).executeRequest();
   }
 
+  protected <T> List<T> getList(StreamPipesApiPath apiPath, Class<T> targetClass) throws SpRuntimeException {
+    ListSerializer<Void, T> serializer = new ListSerializer<>();
+    return new GetRequest<>(clientConfig, apiPath, targetClass, serializer).executeRequest();
+  }
+
   protected <T> Optional<T> getSingleOpt(StreamPipesApiPath apiPath,
                                          Class<T> targetClass) throws SpRuntimeException {
     try {

streampipes-client/src/main/java/org/apache/streampipes/client/api/CustomRequestApi.java

@@ -20,6 +20,7 @@
 import org.apache.streampipes.client.model.StreamPipesClientConfig;
 import org.apache.streampipes.client.util.StreamPipesApiPath;
 
+import java.util.List;
 import java.util.Map;
 
 public class CustomRequestApi extends AbstractClientApi implements ICustomRequestApi {
@@ -46,4 +47,11 @@ public <T> T sendGet(String apiPath, Map<String, String> queryParameters, Class<
         responseClass);
   }
 
+  @Override
+  public <T> List<T> getList(String apiPath, Class<T> responseClass) {
+    return getList(
+        StreamPipesApiPath.fromStreamPipesBasePath(apiPath), responseClass
+    );
+  }
+
 }

streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaAdapter.java

@@ -204,7 +204,10 @@ public void onAdapterStarted(IAdapterParameterExtractor extractor,
                                IEventCollector collector,
                                IAdapterRuntimeContext adapterRuntimeContext) throws AdapterException {
     this.opcUaAdapterConfig =
-        SpOpcUaConfigExtractor.extractAdapterConfig(extractor.getStaticPropertyExtractor());
+        SpOpcUaConfigExtractor.extractAdapterConfig(
+            extractor.getStaticPropertyExtractor(),
+            adapterRuntimeContext.getStreamPipesClient()
+        );
     this.collector = collector;
     this.prepareAdapter(extractor);
     this.numberOfEventProperties =
@@ -252,6 +255,6 @@ public IAdapterConfiguration declareConfig() {
   @Override
   public GuessSchema onSchemaRequested(IAdapterParameterExtractor extractor,
                                        IAdapterGuessSchemaContext adapterGuessSchemaContext) throws AdapterException {
-    return new OpcUaSchemaProvider().getSchema(clientProvider, extractor);
+    return new OpcUaSchemaProvider().getSchema(clientProvider, extractor, adapterGuessSchemaContext.getStreamPipesClient());
   }
 }

streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaSchemaProvider.java

@@ -18,6 +18,7 @@
 
 package org.apache.streampipes.extensions.connectors.opcua.adapter;
 
+import org.apache.streampipes.client.api.IStreamPipesClient;
 import org.apache.streampipes.commons.exceptions.connect.AdapterException;
 import org.apache.streampipes.commons.exceptions.connect.ParseException;
 import org.apache.streampipes.extensions.api.extractor.IAdapterParameterExtractor;
@@ -51,7 +52,8 @@ public class OpcUaSchemaProvider {
    * @throws ParseException
    */
   public GuessSchema getSchema(OpcUaClientProvider clientProvider,
-                               IAdapterParameterExtractor extractor)
+                               IAdapterParameterExtractor extractor,
+                               IStreamPipesClient streamPipesClient)
       throws AdapterException, ParseException {
     var builder = GuessSchemaBuilder.create();
     EventSchema eventSchema = new EventSchema();
@@ -60,7 +62,8 @@ public GuessSchema getSchema(OpcUaClientProvider clientProvider,
     List<EventProperty> allProperties = new ArrayList<>();
 
     var opcUaConfig = SpOpcUaConfigExtractor.extractAdapterConfig(
-        extractor.getStaticPropertyExtractor()
+        extractor.getStaticPropertyExtractor(),
+        streamPipesClient
     );
     try {
       var connectedClient = clientProvider.getClient(opcUaConfig);

streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/SpOpcUaConfigExtractor.java

@@ -18,6 +18,7 @@
 
 package org.apache.streampipes.extensions.connectors.opcua.config;
 
+import org.apache.streampipes.client.api.IStreamPipesClient;
 import org.apache.streampipes.extensions.api.extractor.IParameterExtractor;
 import org.apache.streampipes.extensions.api.extractor.IStaticPropertyExtractor;
 import org.apache.streampipes.extensions.connectors.opcua.config.identity.AnonymousIdentityConfig;
@@ -53,8 +54,9 @@ public class SpOpcUaConfigExtractor {
    * @param extractor extractor for user inputs
    * @return {@link OpcUaAdapterConfig}  instance based on information from {@code extractor}
    */
-  public static OpcUaAdapterConfig extractAdapterConfig(IStaticPropertyExtractor extractor) {
-    var config = extractSharedConfig(extractor, new OpcUaAdapterConfig());
+  public static OpcUaAdapterConfig extractAdapterConfig(IStaticPropertyExtractor extractor,
+                                                        IStreamPipesClient streamPipesClient) {
+    var config = extractSharedConfig(extractor, new OpcUaAdapterConfig(), streamPipesClient);
     boolean usePullMode = extractor.selectedAlternativeInternalId(ADAPTER_TYPE.name())
         .equals(PULL_MODE.name());
 
@@ -78,12 +80,14 @@ public static OpcUaAdapterConfig extractAdapterConfig(IStaticPropertyExtractor e
     return config;
   }
 
-  public static OpcUaConfig extractSinkConfig(IParameterExtractor extractor) {
-    return extractSharedConfig(extractor, new OpcUaConfig());
+  public static OpcUaConfig extractSinkConfig(IParameterExtractor extractor,
+                                              IStreamPipesClient streamPipesClient) {
+    return extractSharedConfig(extractor, new OpcUaConfig(), streamPipesClient);
   }
 
   public static <T extends OpcUaConfig> T extractSharedConfig(IParameterExtractor extractor,
-                                                              T config) {
+                                                              T config,
+                                                              IStreamPipesClient streamPipesClient) {
 
     String selectedAlternativeConnection =
         extractor.selectedAlternativeInternalId(OPC_HOST_OR_URL.name());
@@ -103,9 +107,13 @@ public static <T extends OpcUaConfig> T extractSharedConfig(IParameterExtractor
         SharedUserConfiguration.SECURITY_POLICY,
         String.class
     );
-    config.setSecurityConfig(new SecurityConfig(
-        MessageSecurityMode.valueOf(selectedSecurityMode),
-        SecurityPolicy.valueOf(selectedSecurityPolicy)));
+    config.setSecurityConfig(
+        new SecurityConfig(
+            MessageSecurityMode.valueOf(selectedSecurityMode),
+            SecurityPolicy.valueOf(selectedSecurityPolicy),
+            streamPipesClient
+        )
+    );
 
     boolean useURL = selectedAlternativeConnection.equals(OPC_URL.name());
     if (useURL) {

streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/security/CompositeCertificateValidator.java

@@ -0,0 +1,162 @@
+/*
+ * 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.streampipes.extensions.connectors.opcua.config.security;
+
+import org.apache.streampipes.client.api.IStreamPipesClient;
+import org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaUtils;
+import org.apache.streampipes.model.opcua.Certificate;
+import org.apache.streampipes.model.opcua.CertificateState;
+
+import org.eclipse.milo.opcua.stack.client.security.ClientCertificateValidator;
+import org.eclipse.milo.opcua.stack.core.StatusCodes;
+import org.eclipse.milo.opcua.stack.core.UaException;
+import org.eclipse.milo.opcua.stack.core.security.TrustListManager;
+import org.eclipse.milo.opcua.stack.core.util.validation.CertificateValidationUtil;
+import org.eclipse.milo.opcua.stack.core.util.validation.ValidationCheck;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.security.cert.PKIXCertPathBuilderResult;
+import java.security.cert.X509CRL;
+import java.security.cert.X509Certificate;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.List;
+import java.util.stream.Stream;
+
+public class CompositeCertificateValidator implements ClientCertificateValidator {
+
+  private static final Logger LOG = LoggerFactory.getLogger(CompositeCertificateValidator.class);
+
+  public static final List<Long> REJECTED_STATUS_CODES = List.of(
+      StatusCodes.Bad_CertificateChainIncomplete,
+      StatusCodes.Bad_CertificateInvalid,
+      StatusCodes.Bad_NoValidCertificates,
+      StatusCodes.Bad_CertificateUntrusted,
+      StatusCodes.Bad_CertificateUseNotAllowed,
+      StatusCodes.Bad_SecurityChecksFailed
+  );
+
+  private final TrustListManager trustListManager;
+  private final List<X509Certificate> trustedCerts;
+  private final List<ValidationCheck> validationChecks;
+  private final IStreamPipesClient streamPipesClient;
+
+  public CompositeCertificateValidator(TrustListManager trustListManager,
+                                       List<X509Certificate> trustedCerts,
+                                       List<ValidationCheck> validationChecks,
+                                       IStreamPipesClient streamPipesClient) {
+    this.trustListManager = trustListManager;
+    this.trustedCerts = trustedCerts;
+    this.validationChecks = validationChecks;
+    this.streamPipesClient = streamPipesClient;
+  }
+
+  @Override
+  public void validateCertificateChain(List<X509Certificate> certificateChain) throws UaException {
+    PKIXCertPathBuilderResult certPathResult;
+
+    try {
+      certPathResult = CertificateValidationUtil.buildTrustedCertPath(
+          certificateChain,
+          Stream.concat(trustListManager.getTrustedCertificates().stream(), trustedCerts.stream()).toList(),
+          trustListManager.getIssuerCertificates()
+      );
+    } catch (UaException e) {
+      if (isCertificateRejected(e.getStatusCode().getValue())) {
+        sendToCore(certificateChain.get(0));
+      }
+      throw e;
+    }
+
+    var crls = new ArrayList<X509CRL>();
+    crls.addAll(trustListManager.getTrustedCrls());
+    crls.addAll(trustListManager.getIssuerCrls());
+
+    CertificateValidationUtil.validateTrustedCertPath(
+        certPathResult.getCertPath(),
+        certPathResult.getTrustAnchor(),
+        crls,
+        ValidationCheck.NO_OPTIONAL_CHECKS,
+        false
+    );
+  }
+
+  @Override
+  public void validateCertificateChain(
+      List<X509Certificate> certificateChain,
+      String applicationUri,
+      String... validHostNames
+  ) throws UaException {
+
+    validateCertificateChain(certificateChain);
+
+    X509Certificate certificate = certificateChain.get(0);
+
+    try {
+      CertificateValidationUtil.checkApplicationUri(certificate, applicationUri);
+    } catch (UaException e) {
+      if (validationChecks.contains(ValidationCheck.APPLICATION_URI)) {
+        throw e;
+      } else {
+        LOG.warn(
+            "check suppressed: certificate failed application uri check: {} != {}",
+            applicationUri, CertificateValidationUtil.getSubjectAltNameUri(certificate)
+        );
+      }
+    }
+
+    try {
+      CertificateValidationUtil.checkHostnameOrIpAddress(certificate, validHostNames);
+    } catch (UaException e) {
+      if (validationChecks.contains(ValidationCheck.HOSTNAME)) {
+        throw e;
+      } else {
+        LOG.warn(
+            "check suppressed: certificate failed hostname check: {}",
+            certificate.getSubjectX500Principal().getName()
+        );
+      }
+    }
+  }
+
+  private void sendToCore(X509Certificate cert) {
+    try {
+      var certificate = new Certificate(
+          cert.getSubjectX500Principal().getName(),
+          cert.getIssuerX500Principal().getName(),
+          cert.getSerialNumber().toString(),
+          cert.getNotBefore().toString(),
+          cert.getNotAfter().toString(),
+          cert.getSigAlgName(),
+          cert.getPublicKey().getAlgorithm(),
+          Base64.getEncoder().encodeToString(cert.getEncoded()),
+          CertificateState.REJECTED
+      );
+
+      streamPipesClient.customRequest().sendPost(OpcUaUtils.getCoreCertificatePath(), certificate);
+    } catch (Exception ex) {
+      LOG.error("Failed to report rejected certificate to API", ex);
+    }
+  }
+
+  private boolean isCertificateRejected(long statusCode) {
+    return REJECTED_STATUS_CODES.contains(statusCode);
+  }
+}