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

- Adds Http client SSL transport options
- Uses options as a Camel bean and configures via environment variables when SSL is enabled
- Enables SSL support on all sources

README.adoc

@@ -368,6 +368,57 @@ 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. Default value is `false`.
+
+|CAMEL_KNATIVE_CLIENT_SSL_VERIFY_HOSTNAME
+|Enable/disable hostname verification. Default value is `true`.
+
+|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_KEY_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 using PEM files.
+
+|CAMEL_KNATIVE_CLIENT_SSL_KEYSTORE_PASSWORD
+|Keystore password. Value can be set via secretKeyRef.
+
+|CAMEL_KNATIVE_CLIENT_SSL_TRUST_CERT_PATH
+|Path to the trust certificate provided as a PEM file.
+
+|CAMEL_KNATIVE_CLIENT_SSL_TRUSTSTORE_PATH
+|Java truststore (.jks) or (.p12) path as an alternative to using PEM files.
+
+|CAMEL_KNATIVE_CLIENT_SSL_TRUSTSTORE_PASSWORD
+|Truststore password. Value can be set via secretKeyRef.
+
+|===
+
+As you can see the SSL client support provides multiple ways to configure keystore and truststore options.
+It is recommended to set keystore/truststore passwords vie secretKeyRef on the IntegrationSource spec.
+When no truststore configuration is given the SSL client support defaults to using trust all options.
+
 == Dependencies
 
 The required Camel dependencies need to be added to the Maven POM before building and deploying.

aws-ddb-streams-source/pom.xml

@@ -15,6 +15,12 @@
   <name>Knative Connectors :: AWS DDB Streams Source</name>
 
   <dependencies>
+    <dependency>
+      <groupId>dev.knative.eventing</groupId>
+      <artifactId>connector-utils</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
     <dependency>
       <groupId>org.apache.camel.quarkus</groupId>
       <artifactId>camel-quarkus-core</artifactId>

aws-ddb-streams-source/src/main/java/dev/knative/eventing/SourceOptions.java

@@ -0,0 +1,31 @@
+/*
+ * 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;
+
+import dev.knative.eventing.source.KnativeHttpClientOptions;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Named;
+import org.apache.camel.CamelContext;
+
+@ApplicationScoped
+public class SourceOptions {
+
+    @Named("knativeHttpClientOptions")
+    public KnativeHttpClientOptions knativeHttpClientOptions(CamelContext camelContext) {
+        return new KnativeHttpClientOptions(camelContext);
+    }
+}

aws-ddb-streams-source/src/main/resources/application.properties

@@ -37,3 +37,6 @@ camel.kamelet.aws-ddb-streams-source.table=
 # 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:knativeHttpClientOptions

aws-s3-source/pom.xml

@@ -15,6 +15,12 @@
   <name>Knative Connectors :: AWS S3 Source</name>
 
   <dependencies>
+    <dependency>
+      <groupId>dev.knative.eventing</groupId>
+      <artifactId>connector-utils</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
     <dependency>
       <groupId>org.apache.camel.quarkus</groupId>
       <artifactId>camel-quarkus-core</artifactId>

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

@@ -0,0 +1,31 @@
+/*
+ * 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;
+
+import dev.knative.eventing.source.KnativeHttpClientOptions;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Named;
+import org.apache.camel.CamelContext;
+
+@ApplicationScoped
+public class SourceOptions {
+
+    @Named("knativeHttpClientOptions")
+    public KnativeHttpClientOptions knativeHttpClientOptions(CamelContext camelContext) {
+        return new KnativeHttpClientOptions(camelContext);
+    }
+}

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

@@ -38,3 +38,6 @@ 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:knativeHttpClientOptions

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.verify.hostname", value = "false"),
+        @ResourceArg(name = "camel.knative.client.ssl.key.path", value = "keystore/client.pem"),
+        @ResourceArg(name = "camel.knative.client.ssl.key.cert.path", value = "keystore/client.crt"),
+        @ResourceArg(name = "camel.knative.client.ssl.truststore.path", value = "keystore/truststore.jks"),
+        @ResourceArg(name = "camel.knative.client.ssl.truststore.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()));
+    }
+}