fix(#27): Enable SSL transport options on sources

- Adds SSL transport support when configured
- Enables SSL support on AWS S3 source

README.adoc

@@ -368,6 +368,41 @@ spec:
 
 The trigger for example filters the events by its type `ce-type=dev.knative.connector.event.timer`.
 
+== Secure Knative transports
+
+Knative brokers may use TLS encrypted transport options as described in https://knative.dev/docs/eventing/features/transport-encryption/#overview
+
+This means that Event producers need to use proper SSL authentication to connect to Https Knative broker endpoints with cluster-internal CA certificates.
+
+The IntegrationSource may use a volume mount with the cluster-internal CA certificates being injected.
+
+The integration source needs to enable the SSL client via environment variables and set the path to the injected CA certs and PEM files:
+
+* CAMEL_KNATIVE_CLIENT_SSL_ENABLED=true
+* CAMEL_KNATIVE_CLIENT_SSL_KEY_PATH=/knative-custom-certs/knative-eventing-bundle.pem
+
+This enables the SSL options on the Http client that connects with the broker endpoint. The SSL client support provides these environment variables:
+
+|===
+|EnvVar |Description
+
+|CAMEL_KNATIVE_CLIENT_SSL_ENABLED
+|Enable/disable SSL options on the Http client.
+
+|CAMEL_KNATIVE_CLIENT_SSL_KEY_PATH
+|Path to the key store options configuring a list of private key and its certificate based on Privacy-enhanced Electronic Email (PEM) files.
+
+|CAMEL_KNATIVE_CLIENT_SSL_CERT_PATH
+|Optional path to the client certificate in case the CA cert is not included in the key PEM file.
+
+|CAMEL_KNATIVE_CLIENT_SSL_KEYSTORE_PATH
+|Java keystore (.jks) or (.p12) path as an alternative to PEM.
+
+|CAMEL_KNATIVE_CLIENT_SSL_KEYSTORE_PASSWORD
+|Keystore password. Value can be set via secretRef.
+
+|===
+
 == Dependencies
 
 The required Camel dependencies need to be added to the Maven POM before building and deploying.

aws-s3-source/src/main/java/dev/knative/eventing/KnativeClientOptions.java

@@ -0,0 +1,60 @@
+package dev.knative.eventing;
+
+import io.quarkus.tls.runtime.keystores.TrustAllOptions;
+import io.vertx.core.net.JksOptions;
+import io.vertx.core.net.PemKeyCertOptions;
+import io.vertx.core.net.PfxOptions;
+import io.vertx.ext.web.client.WebClientOptions;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Named;
+import org.apache.camel.CamelContext;
+import org.apache.camel.spi.PropertiesComponent;
+
+/**
+ * Knative client options able to configure secure Http transport options.
+ * SSL options are added automatically when enabled via
+ */
+@ApplicationScoped
+public class KnativeClientOptions {
+
+    private static final String PROPERTY_PREFIX = "camel.knative.client.ssl.";
+
+    @Named("knativeClientOptions")
+    public WebClientOptions knativeClientOptions(CamelContext camelContext) {
+        PropertiesComponent propertiesComponent = camelContext.getPropertiesComponent();
+
+        boolean sslEnabled = Boolean.parseBoolean(
+                propertiesComponent.resolveProperty(PROPERTY_PREFIX + "enabled").orElse("false"));
+
+        WebClientOptions options = new WebClientOptions();
+        if (sslEnabled) {
+            options.setSsl(true);
+
+            boolean verifyHostname = Boolean.parseBoolean(
+                    propertiesComponent.resolveProperty(PROPERTY_PREFIX + "verify.hostname").orElse("false"));
+            options.setVerifyHost(verifyHostname);
+
+            String keystorePath = propertiesComponent.resolveProperty(PROPERTY_PREFIX + "keystore.path").orElse("");
+            String keystorePassword = propertiesComponent.resolveProperty(PROPERTY_PREFIX + "keystore.password").orElse("");
+
+            if (!keystorePath.isEmpty()) {
+                if (keystorePath.endsWith(".p12")) {
+                    options.setKeyCertOptions(new PfxOptions().setPath(keystorePath).setPassword(keystorePassword));
+                } else {
+                    options.setKeyCertOptions(new JksOptions().setPath(keystorePath).setPassword(keystorePassword));
+                }
+            }
+
+            String keyPath = propertiesComponent.resolveProperty(PROPERTY_PREFIX + "key.path").orElse("");
+            String certPath = propertiesComponent.resolveProperty(PROPERTY_PREFIX + "cert.path").orElse(keyPath);
+
+            if (!keyPath.isEmpty()) {
+                options.setKeyCertOptions(new PemKeyCertOptions().setKeyPath(keyPath).setCertPath(certPath));
+            }
+
+            options.setTrustOptions(TrustAllOptions.INSTANCE);
+        }
+        return options;
+    }
+
+}

aws-s3-source/src/main/resources/application.properties

@@ -34,7 +34,9 @@ camel.kamelet.aws-s3-source.secretKey=
 camel.kamelet.aws-s3-source.region=
 camel.kamelet.aws-s3-source.bucketNameOrArn=
 
-
 # ConfigMap and secret based configuration
 camel.kubernetes-config.mount-path-configmaps=/etc/camel/conf.d/_configmaps/kn-source-config
 camel.kubernetes-config.mount-path-secrets=/etc/camel/conf.d/_secrets/kn-source-config
+
+# Http transport configuration (SSL enabled)
+camel.component.knative.producerFactory.clientOptions=#bean:knativeClientOptions

aws-s3-source/src/test/java/dev/knative/eventing/connector/AwsS3SourceTest.java

@@ -19,11 +19,13 @@
 import java.util.Collections;
 
 import io.quarkus.test.common.QuarkusTestResource;
+import io.quarkus.test.common.ResourceArg;
 import io.quarkus.test.junit.QuarkusTest;
 import org.citrusframework.GherkinTestActionRunner;
 import org.citrusframework.annotations.CitrusResource;
 import org.citrusframework.context.TestContext;
 import org.citrusframework.http.endpoint.builder.HttpEndpoints;
+import org.citrusframework.http.security.HttpSecureConnection;
 import org.citrusframework.http.server.HttpServer;
 import org.citrusframework.quarkus.CitrusSupport;
 import org.citrusframework.spi.BindToRegistry;
@@ -39,7 +41,9 @@
 
 @QuarkusTest
 @CitrusSupport
-@QuarkusTestResource(LocalstackTestResource.class)
+@QuarkusTestResource(value = LocalstackTestResource.class, restrictToAnnotatedClass = true, initArgs = {
+        @ResourceArg(name = "bucketNameOrArn", value = "mybucket")
+})
 public class AwsS3SourceTest {
 
     @CitrusResource

aws-s3-source/src/test/java/dev/knative/eventing/connector/LocalstackTestResource.java

@@ -35,6 +35,13 @@ public class LocalstackTestResource implements QuarkusTestResourceLifecycleManag
 
     private S3Client s3Client;
 
+    private final Map<String, String> injectedConf = new HashMap<>();
+
+    @Override
+    public void init(Map<String, String> initArgs) {
+        injectedConf.putAll(initArgs);
+    }
+
     @Override
     public Map<String, String> start() {
         localStackContainer.start();
@@ -50,17 +57,20 @@ public Map<String, String> start() {
                 .region(Region.of(localStackContainer.getRegion()))
                 .build();
 
-        s3Client.createBucket(b -> b.bucket("mybucket"));
+        String bucketNameOrArn = injectedConf.remove("bucketNameOrArn");
+        s3Client.createBucket(b -> b.bucket(bucketNameOrArn));
 
         Map<String, String> conf = new HashMap<>();
         conf.put("camel.kamelet.aws-s3-source.accessKey", localStackContainer.getAccessKey());
         conf.put("camel.kamelet.aws-s3-source.secretKey", localStackContainer.getSecretKey());
         conf.put("camel.kamelet.aws-s3-source.region", localStackContainer.getRegion());
-        conf.put("camel.kamelet.aws-s3-source.bucketNameOrArn", "mybucket");
+        conf.put("camel.kamelet.aws-s3-source.bucketNameOrArn", bucketNameOrArn);
         conf.put("camel.kamelet.aws-s3-source.uriEndpointOverride", localStackContainer.getEndpoint().toString());
         conf.put("camel.kamelet.aws-s3-source.overrideEndpoint", "true");
         conf.put("camel.kamelet.aws-s3-source.forcePathStyle", "true");
 
+        conf.putAll(injectedConf);
+
         return conf;
     }
 
@@ -80,6 +90,6 @@ public void inject(TestInjector testInjector) {
     /**
      * Annotation marks fields in test class for injection by this test resource.
      */
-    @interface Injected {
+    public @interface Injected {
     }
 }

aws-s3-source/src/test/java/dev/knative/eventing/connector/ssl/AwsS3SourceSSLTest.java

@@ -0,0 +1,116 @@
+/*
+ * Copyright the original author or authors.
+ *
+ * Licensed 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 dev.knative.eventing.connector.ssl;
+
+import java.util.Collections;
+
+import dev.knative.eventing.connector.LocalstackTestResource;
+import io.quarkus.test.common.QuarkusTestResource;
+import io.quarkus.test.common.ResourceArg;
+import io.quarkus.test.junit.QuarkusTest;
+import org.citrusframework.GherkinTestActionRunner;
+import org.citrusframework.annotations.CitrusResource;
+import org.citrusframework.context.TestContext;
+import org.citrusframework.http.endpoint.builder.HttpEndpoints;
+import org.citrusframework.http.security.HttpSecureConnection;
+import org.citrusframework.http.server.HttpServer;
+import org.citrusframework.quarkus.CitrusSupport;
+import org.citrusframework.spi.BindToRegistry;
+import org.junit.jupiter.api.Test;
+import org.springframework.http.HttpStatus;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.CompletedMultipartUpload;
+import software.amazon.awssdk.services.s3.model.CompletedPart;
+import software.amazon.awssdk.services.s3.model.CreateMultipartUploadResponse;
+
+import static org.citrusframework.http.actions.HttpActionBuilder.http;
+
+@QuarkusTest
+@CitrusSupport
+@QuarkusTestResource(value = LocalstackTestResource.class, restrictToAnnotatedClass = true, initArgs = {
+        @ResourceArg(name = "bucketNameOrArn", value = "mybucket"),
+        @ResourceArg(name = "k.sink", value = "https://localhost:8443"),
+        @ResourceArg(name = "camel.knative.client.ssl.enabled", value = "true"),
+        @ResourceArg(name = "camel.knative.client.ssl.key.path", value = "keystore/client.pem"),
+        @ResourceArg(name = "camel.knative.client.ssl.cert.path", value = "keystore/client.crt")
+        //@ResourceArg(name = "camel.knative.client.ssl.keystore.path", value = "keystore/client.jks")
+        //@ResourceArg(name = "camel.knative.client.ssl.keystore.path", value = "keystore/client.p12")
+        //@ResourceArg(name = "camel.knative.client.ssl.keystore.password", value = "secr3t")
+})
+public class AwsS3SourceSSLTest {
+
+    @CitrusResource
+    private GherkinTestActionRunner tc;
+
+    private final String s3Key = "message.txt";
+    private final String s3Data = "Hello from secured AWS S3!";
+    private final String s3BucketName = "mybucket";
+
+    @LocalstackTestResource.Injected
+    public S3Client s3Client;
+
+    @BindToRegistry
+    public HttpServer knativeBroker = HttpEndpoints.http()
+            .server()
+            .port(8080)
+            .securePort(8443)
+            .secured(HttpSecureConnection.ssl()
+                    .keyStore("classpath:keystore/server.jks", "secr3t")
+                    .trustStore("classpath:keystore/truststore.jks", "secr3t"))
+            .autoStart(true)
+            .build();
+
+    @Test
+    public void shouldProduceSecureEvents() {
+        tc.given(this::uploadS3File);
+
+        tc.when(
+            http().server(knativeBroker)
+                    .receive()
+                    .post()
+                    .message()
+                    .body(s3Data)
+                    .header("ce-id", "@matches([0-9A-Z]{15}-[0-9]{16})@")
+                    .header("ce-type", "dev.knative.connector.event.aws-s3")
+                    .header("ce-source", "dev.knative.eventing.aws-s3-source")
+                    .header("ce-subject", "aws-s3-source")
+        );
+
+        tc.then(
+            http().server(knativeBroker)
+                    .send()
+                    .response(HttpStatus.OK)
+        );
+    }
+
+    private void uploadS3File(TestContext context) {
+        CreateMultipartUploadResponse initResponse = s3Client.createMultipartUpload(b -> b.bucket(s3BucketName).key(s3Key));
+        String etag = s3Client.uploadPart(b -> b.bucket(s3BucketName)
+                        .key(s3Key)
+                        .uploadId(initResponse.uploadId())
+                        .partNumber(1),
+                RequestBody.fromString(s3Data)).eTag();
+        s3Client.completeMultipartUpload(b -> b.bucket(s3BucketName)
+                .multipartUpload(CompletedMultipartUpload.builder()
+                        .parts(Collections.singletonList(CompletedPart.builder()
+                                .partNumber(1)
+                                .eTag(etag).build())).build())
+                .key(s3Key)
+                .uploadId(initResponse.uploadId()));
+    }
+}

aws-s3-source/src/test/resources/keystore/client.crt

@@ -0,0 +1,33 @@
+-----BEGIN CERTIFICATE-----
+MIIFuTCCA6GgAwIBAgIBZTANBgkqhkiG9w0BAQsFADB+MQswCQYDVQQGEwJERTET
+MBEGA1UECAwKU29tZS1TdGF0ZTEhMB8GA1UECgwYSW50ZXJuZXQgV2lkZ2l0cyBQ
+dHkgTHRkMQ0wCwYDVQQDDARZQUtTMSgwJgYJKoZIhvcNAQkBFhl5YWtzLWRldkBn
+b29nbGVncm91cHMuY29tMB4XDTI0MDYxODA3MDM1OFoXDTI1MDYwOTA3MDM1OFow
+fjELMAkGA1UEBhMCREUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoMGElu
+dGVybmV0IFdpZGdpdHMgUHR5IEx0ZDENMAsGA1UEAwwEWUFLUzEoMCYGCSqGSIb3
+DQEJARYZeWFrcy1kZXZAZ29vZ2xlZ3JvdXBzLmNvbTCCAiIwDQYJKoZIhvcNAQEB
+BQADggIPADCCAgoCggIBANvrpZbRRbUmngCvZKKdqCsyoAtlrGtXy8zpUlyMwFcI
+RR1UZgxVNNuqjA3mQ5HdaPHYCqQdsDD8SpIreNuTjdnlmxA5vxGuW0Aq+lX1SKq+
+CjsycurX50uUPsiR/znQhkhVkbjYGOASbZ0MJ4u0jumKk/z1SSm8qpfnLsdGxEFj
+Tg+H7N7C/HJ3ZOVBbFZBLHCgIzXsVhIgz8uv/hPn/57BlvUJqoAY9z+ZCPCQOYE7
+3KNLI2VgSGbpePy0LiI4BqG14X4e7GxMa0NBkOMVWNdt/g80c2O6urrA9UEazVB9
+Ojs0FnBRKhbeF6cOoiAljuKZklq5nzslIMIlDcj9PHESwPsK3IdOR5iJAdRO2eyQ
+NGJvgDMvQuNuscV/Hpqw6LFNXZ615EBUlStJ9d1Ea8dCgBZpMQnV2bvgY6jkNazm
+StJrv9+N7hMdlRn5T/ZWzbi92YjyrOuudo5rBpQTHkzy5conLFwoHnfBpFhfB6+4
+mFUhM6RtQiq3jWjO6zylwP4xPGcDlUxb9YBSvPenXbVz3wi1p4fADkdxbAyS3xWn
+46GEKgxChvuhiwvGYExZIKDL9rjWNWaNoretJcEDQgZLm1XYjt2ie6Z3c8I6TprO
+FgfKTXlzYytUvEOqvsY+zsuCkiIDYptRhXW16LN8J5j5BcA7LFqVS5nf1zuj7kfN
+AgMBAAGjQjBAMB0GA1UdDgQWBBSGCLa6UTN3VSsBFuSC5aV0S9S/rjAfBgNVHSME
+GDAWgBSZfBQ9qjN35XGTAyXi79Wa7r0FQDANBgkqhkiG9w0BAQsFAAOCAgEAQNdT
+nBoUyL+N5/OiF7jqlkyTaNm+IIEbGX80LsjRLpmmAMbpZtnbcxXn7lF3EGEyAuUA
+9lEbVY545xymqeFw53vxZSEzboCCoxphgjTIs+/edLSSrh4SEbVvJnjlM8pQiCEh
+fZ3FsKyer8jS/mmSGr1cx4Opwcbi5UWg+lu0dsWcf1oDqS7g01EZS2AS0JtLEcmN
+fgMP0Y0YIbTLUQayk8oE6KGeG8wpB/xuPmdecRINEYpkdQKPPJ5wmdFGnTekLiOj
+obJZeBu5H1XAAfVUYNr9ShN885hmfyRIWn5tpo34S/df/hukyi83VWnUeoCLjOfx
+wsrHk0Qn+/WqXxJmzKgz5Cl7D2yVndMlGjKYlWoe/1E3/uqHWeOFzSneFm11Z+tW
+K+4etGpRFYO0ZETv5Eaf0J6Fd2Qpk2FQg7xw/5UzAV97lhDNRr6yN2GxBs5gcPcK
+tGcvay7coUkG4ZSmhUVRcqoXmNT3u1KxTMjUQDiWJpXtAGq4MHTSp37nW7fTnPKj
+PS9vNPCNoUDVj5IpWM7a7T/9mbrhsubreye7hy2qF85w/UWL8V+E797t0WQIdFRy
+eb+3EXcFenRnyVCDX81uz84Xmo7EeNawbucJKyY8Nx2XwJGzn5o5Wm7Z5xqE2rsc
+boqQ3Gqp84ATr6EnjueDKfcdfjI/XdzR8ZZQtkg=
+-----END CERTIFICATE-----