feat: add support for mTLS EventStreams

src/aap_eda/api/views/eda_credential.py

@@ -204,6 +204,7 @@ def partial_update(self, request, pk):
             setattr(eda_credential, key, value)
 
         with transaction.atomic():
+            eda_credential._request = request
             eda_credential.save()
             check_related_permissions(
                 request.user,

src/aap_eda/api/views/event_stream.py

@@ -30,8 +30,10 @@
 
 from aap_eda.api import exceptions as api_exc, filters, serializers
 from aap_eda.core import models
-from aap_eda.core.enums import ResourceType
+from aap_eda.core.enums import EventStreamAuthType, ResourceType
+from aap_eda.core.exceptions import GatewayAPIError, MissingCredentials
 from aap_eda.core.utils import logging_utils
+from aap_eda.services.sync_certs import SyncCertificates
 
 logger = logging.getLogger(__name__)
 
@@ -110,6 +112,7 @@ def destroy(self, request, *args, **kwargs):
                 f"Event stream '{event_stream.name}' is being referenced by "
                 f"{ref_count} activation(s) and cannot be deleted"
             )
+        self._sync_certificates(event_stream, "destroy")
         self.perform_destroy(event_stream)
 
         logger.info(
@@ -182,6 +185,7 @@ def create(self, request, *args, **kwargs):
             RoleDefinition.objects.give_creator_permissions(
                 request.user, serializer.instance
             )
+            self._sync_certificates(response, "create")
 
         logger.info(
             logging_utils.generate_simple_audit_log(
@@ -307,3 +311,21 @@ def activations(self, request, id):
             )
         )
         return self.get_paginated_response(serializer.data)
+
+    def _sync_certificates(
+        self,
+        event_stream: models.EventStream,
+        action: str,
+    ):
+        if (
+            event_stream.eda_credential.credential_type.kind
+            == EventStreamAuthType.MTLS
+        ):
+            try:
+                obj = SyncCertificates(event_stream.eda_credential.id)
+                if action == "destroy":
+                    obj.delete(event_stream.id)
+                else:
+                    obj.update()
+            except (GatewayAPIError, MissingCredentials) as ex:
+                logger.error("Could not %s certificates %s", action, str(ex))

src/aap_eda/core/exceptions.py

@@ -55,3 +55,11 @@ class CredentialPluginError(Exception):
 
 class UnknownPluginTypeError(Exception):
     pass
+
+
+class GatewayAPIError(Exception):
+    pass
+
+
+class MissingCredentials(Exception):
+    pass

src/aap_eda/core/management/commands/create_initial_data.py

@@ -48,7 +48,7 @@
 AUTH_TYPE_LABEL = "Event Stream Authentication Type"
 SIGNATURE_ENCODING_LABEL = "Signature Encoding"
 HTTP_HEADER_LABEL = "HTTP Header Key"
-DEPRECATED_CREDENTIAL_KINDS = ["mtls"]
+DEPRECATED_CREDENTIAL_KINDS = [""]
 LABEL_PATH_TO_AUTH = "Path to Auth"
 LABEL_CLIENT_CERTIFICATE = "Client Certificate"
 LABEL_CLIENT_SECRET = "Client Secret"
@@ -1762,6 +1762,53 @@
     "required": ["app_or_client_id", "install_id", "private_rsa_key"],
 }
 
+EVENT_STREAM_MTLS_INPUTS = {
+    "fields": [
+        {
+            "id": "auth_type",
+            "label": AUTH_TYPE_LABEL,
+            "type": "string",
+            "default": "mtls",
+            "hidden": True,
+        },
+        {
+            "id": "certificate",
+            "label": "Certificate",
+            "type": "string",
+            "multiline": True,
+            "format": "pem_certificate",
+            "help_text": (
+                "The Certificate collection in PEM format. You can have "
+                "multiple certificates in this field separated by "
+                "-----BEGIN CERTIFICATE----- "
+                "and ending in -----END CERTIFICATE-----"
+                "If a certificate is provided it will be transferred "
+                "to the Gateway, otherwise its assumed that the Gateway "
+                "already has the CA certificates in place to validate "
+                "the incoming client certificate."
+            ),
+        },
+        {
+            "id": "subject",
+            "label": "Certificate Subject",
+            "type": "string",
+            "help_text": (
+                "The Subject from Certificate compliant with RFC 2253."
+                "This is optional and can be used to check the subject "
+                "defined in the certificate."
+            ),
+        },
+        {
+            "id": "http_header_key",
+            "label": HTTP_HEADER_LABEL,
+            "type": "string",
+            "default": "Subject",
+            "hidden": True,
+        },
+    ],
+    "required": ["auth_type", "http_header_key"],
+}
+
 CREDENTIAL_TYPES = [
     {
         "name": enums.DefaultCredentialType.SOURCE_CONTROL,
@@ -2045,6 +2092,21 @@
         "injectors": {},
         "managed": True,
     },
+    {
+        "name": enums.EventStreamCredentialType.MTLS,
+        "namespace": "event_stream",
+        "kind": "mtls",
+        "inputs": EVENT_STREAM_MTLS_INPUTS,
+        "injectors": {},
+        "managed": True,
+        "description": (
+            "Credential for Event Streams that use mutual TLS. "
+            "If CA Certificates are defined in the UI it will "
+            "be transferred to the Gateway proxy for validation "
+            "of incoming requests. We can optionally validate the "
+            "Subject defined in the inbound Certificate."
+        ),
+    },
 ]
 
 

src/aap_eda/services/sync_certs.py

@@ -0,0 +1,205 @@
+#  Copyright 2025 Red Hat, Inc.
+#
+#  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.
+"""Synchronize Certificates with Gateway."""
+import hashlib
+import logging
+from typing import Optional
+from urllib.parse import urljoin
+
+import requests
+import yaml
+from ansible_base.resource_registry import resource_server
+from django.conf import settings
+from django.db.models.signals import post_save
+from django.dispatch import receiver
+from rest_framework import status
+
+from aap_eda.core import enums, models
+from aap_eda.core.exceptions import GatewayAPIError, MissingCredentials
+
+LOGGER = logging.getLogger(__name__)
+SLUG = "api/gateway/v1/ca_certificates/"
+DEFAULT_TIMEOUT = 30
+SERVICE_TOKEN_HEADER = "X-ANSIBLE-SERVICE-AUTH"
+
+
+class SyncCertificates:
+    """This class synchronizes the certificates with Gateway."""
+
+    def __init__(self, eda_credential_id: int):
+        self.eda_credential_id = eda_credential_id
+        self.gateway_url = settings.RESOURCE_SERVER["URL"]
+        self.gateway_ssl_verify = settings.RESOURCE_SERVER.get(
+            "VALIDATE_HTTPS", True
+        )
+
+        self.eda_credential = models.EdaCredential.objects.get(
+            id=self.eda_credential_id
+        )
+
+    def update(self):
+        """Handle creating and updating the certificate in Gateway."""
+        inputs = yaml.safe_load(self.eda_credential.inputs.get_secret_value())
+        existing_object = self._fetch_from_gateway()
+
+        # If the user had a certificate and then they deleted it
+        # remove it from Gateway
+        if existing_object and not inputs["certificate"]:
+            return self.delete()
+
+        # If the user has not provided any certificate nothing to do
+        if not inputs["certificate"]:
+            return
+
+        sha256 = hashlib.sha256(
+            inputs["certificate"].encode("utf-8")
+        ).hexdigest()
+
+        if existing_object.get("sha256", "") != sha256:
+            data = {
+                "name": self.eda_credential.name,
+                "pem_data": inputs["certificate"],
+                "sha256": sha256,
+                "eda_credential_id": self._get_remote_id(),
+            }
+            headers = self._prep_headers()
+            if existing_object:
+                slug = f"{SLUG}/{existing_object['id']}/"
+                url = urljoin(self.gateway_url, slug)
+                response = requests.patch(
+                    url,
+                    json=data,
+                    headers=headers,
+                    verify=self.gateway_ssl_verify,
+                    timeout=DEFAULT_TIMEOUT,
+                )
+            else:
+                url = urljoin(self.gateway_url, SLUG)
+                response = requests.post(
+                    url,
+                    json=data,
+                    headers=headers,
+                    verify=self.gateway_ssl_verify,
+                    timeout=DEFAULT_TIMEOUT,
+                )
+
+            if response.status_code in [
+                status.HTTP_200_OK,
+                status.HTTP_201_CREATED,
+            ]:
+                LOGGER.debug("Certificate updated")
+            elif response.status_code == status.HTTP_400_BAD_REQUEST:
+                LOGGER.error("Update failed")
+            else:
+                LOGGER.error("Couldn't update certificate")
+
+        else:
+            LOGGER.debug("No changes detected")
+
+    def delete(self, event_stream_id: Optional[int]):
+        """Delete the Certificate from Gateway."""
+        existing_object = self._fetch_from_gateway()
+        if not existing_object:
+            return
+
+        objects = models.EventStream.objects.filter(
+            eda_credential_id=self.eda_credential
+        )
+        if not event_stream_id:
+            self._delete_from_gateway(existing_object)
+        elif len(objects) == 1 and event_stream_id == objects[0].id:
+            self._delete_from_gateway(existing_object)
+
+    def _delete_from_gateway(self, existing_object: dict):
+        slug = f"{SLUG}/{existing_object['id']}/"
+        url = urljoin(self.gateway_url, slug)
+        headers = self._prep_headers()
+        response = requests.delete(
+            url,
+            headers=headers,
+            verify=self.gateway_ssl_verify,
+            timeout=DEFAULT_TIMEOUT,
+        )
+        if response.status_code == status.HTTP_200_OK:
+            LOGGER.debug("Certificate object deleted")
+        if response.status_code == status.HTTP_404_NOT_FOUND:
+            LOGGER.warning("Certificate object missing during delete")
+        else:
+            LOGGER.error(
+                "Could not delete certificate object in gateway. "
+                f"Error code: {response.status_code}"
+            )
+            LOGGER.error(f"Error message: {response.text}")
+            raise GatewayAPIError
+
+    def _fetch_from_gateway(self):
+        slug = f"{SLUG}/?eda_credential_id={self._get_remote_id()}"
+        url = urljoin(self.gateway_url, slug)
+        headers = self._prep_headers()
+        response = requests.get(
+            url,
+            headers=headers,
+            verify=self.gateway_ssl_verify,
+            timeout=DEFAULT_TIMEOUT,
+        )
+
+        if response.status_code == status.HTTP_200_OK:
+            LOGGER.debug("Certificate object exists in gateway")
+            data = response.json()
+            if data["count"] > 0:
+                return data["results"][0]
+            else:
+                return {}
+        if response.status_code == status.HTTP_404_NOT_FOUND:
+            LOGGER.debug("Certificate object does not exist in gateway")
+            return {}
+
+        LOGGER.error(
+            "Error fetching certificate object. "
+            f"Error code: {response.status_code}"
+        )
+        LOGGER.error(f"Error message: {response.text}")
+        raise GatewayAPIError
+
+    def _get_remote_id(self) -> str:
+        return f"eda_{self.eda_credential_id}"
+
+    def _prep_headers(self) -> dict:
+        token = resource_server.get_service_token()
+        if token:
+            return {SERVICE_TOKEN_HEADER: token}
+
+        LOGGER.error("Cannot connect to gateway service token")
+        raise MissingCredentials
+
+
+@receiver(post_save, sender=models.EdaCredential)
+def gw_handler(sender, instance, **kwargs):
+    """Handle updates to EdaCredential object and force a certificate sync."""
+    if (
+        instance.credential_type is not None
+        and instance.credential_type.name
+        == enums.EventStreamCredentialType.MTLS
+        and hasattr(instance, "_request")
+    ):
+        try:
+            objects = models.EventStream.objects.filter(
+                eda_credential_id=instance.id
+            )
+            if len(objects) > 0:
+                SyncCertificates(instance.id).update()
+        except (GatewayAPIError, MissingCredentials) as ex:
+            LOGGER.error(
+                "Couldn't trigger gateway certificate updates %s", str(ex)
+            )