feat: Upload certiticates to Gateway, mTLS support

https://issues.redhat.com/browse/AAP-29924
EDA will manage certificates so that it can be used
by the Secret Discovery Service in Envoy.

Author: mkanoor

src/aap_eda/api/views/event_stream.py

@@ -15,7 +15,6 @@
 import logging
 from urllib.parse import urljoin
 
-import yaml
 from ansible_base.rbac.api.related import check_related_permissions
 from ansible_base.rbac.models import RoleDefinition
 from django.conf import settings
@@ -34,7 +33,9 @@
 from aap_eda.api import exceptions as api_exc, filters, serializers
 from aap_eda.core import models
 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__)
 
@@ -115,6 +116,8 @@ 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(
@@ -187,11 +190,11 @@ def create(self, request, *args, **kwargs):
             RoleDefinition.objects.give_creator_permissions(
                 request.user, serializer.instance
             )
-            inputs = yaml.safe_load(
-                response.eda_credential.inputs.get_secret_value()
-            )
             sub_path = f"{EVENT_STREAM_EXTERNAL_PATH}/{response.uuid}/post/"
-            if inputs["auth_type"] == EventStreamAuthType.MTLS:
+            if (
+                response.eda_credential.credential_type.kind
+                == EventStreamAuthType.MTLS_V2.value
+            ):
                 response.url = urljoin(
                     settings.EVENT_STREAM_MTLS_BASE_URL, sub_path
                 )
@@ -200,6 +203,7 @@ def create(self, request, *args, **kwargs):
                     settings.EVENT_STREAM_BASE_URL, sub_path
                 )
             response.save(update_fields=["url"])
+            self._sync_certificates(response, "create")
 
         logger.info(
             logging_utils.generate_simple_audit_log(
@@ -325,3 +329,19 @@ 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_V2.value
+        ):
+            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/api/views/external_event_stream.py

@@ -148,7 +148,7 @@ def _handle_auth(self, request, inputs):
                     secret=inputs["secret"].encode("utf-8"),
                 )
                 obj.authenticate(request.body)
-            elif inputs["auth_type"] == EventStreamAuthType.MTLS:
+            elif inputs["auth_type"] == EventStreamAuthType.MTLS_V2:
                 obj = MTLSAuthentication(
                     subject=inputs.get("subject", ""),
                     value=request.headers[inputs["http_header_key"]],
@@ -199,7 +199,7 @@ def _handle_auth(self, request, inputs):
                 )
                 obj.authenticate(request.body)
             else:
-                message = "Unknown auth type"
+                message = f"Unknown auth type {inputs['auth_type']}"
                 logger.error(message)
                 raise ParseError(message)
         except AuthenticationFailed as err:

src/aap_eda/core/enums.py

@@ -148,7 +148,7 @@ class EventStreamAuthType(DjangoStrEnum):
     OAUTH2 = "oauth2"
     OAUTH2JWT = "oauth2-jwt"
     ECDSA = "ecdsa"
-    MTLS = "mtls"
+    MTLS_V2 = "mtls_v2"
 
 
 class SignatureEncodingType(DjangoStrEnum):
@@ -165,7 +165,7 @@ class EventStreamCredentialType(DjangoStrEnum):
     OAUTH2 = "OAuth2 Event Stream"
     OAUTH2_JWT = "OAuth2 JWT Event Stream"
     ECDSA = "ECDSA Event Stream"
-    MTLS = "mTLS Event Stream"
+    MTLS_V2 = "mTLS Event Stream"
 
 
 class CustomEventStreamCredentialType(DjangoStrEnum):

src/aap_eda/core/exceptions.py

@@ -31,3 +31,11 @@ class PGNotifyError(Exception):
 
 class ParseError(Exception):
     pass
+
+
+class MissingCredentials(Exception):
+    pass
+
+
+class GatewayAPIError(Exception):
+    pass

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

@@ -834,6 +834,48 @@
     "required": ["auth_type", "username", "password", "http_header_key"],
 }
 
+EVENT_STREAM_MTLS_V2_INPUTS = {
+    "fields": [
+        {
+            "id": "auth_type",
+            "label": AUTH_TYPE_LABEL,
+            "type": "string",
+            "default": "mtls_v2",
+            "hidden": True,
+        },
+        {
+            "id": "certificate",
+            "label": "Certificate",
+            "type": "string",
+            "multiline": True,
+            "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-----"
+            ),
+        },
+        {
+            "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", "certificate", "http_header_key"],
+}
+
 CREDENTIAL_TYPES = [
     {
         "name": enums.DefaultCredentialType.SOURCE_CONTROL,
@@ -957,6 +999,21 @@
             "the signature."
         ),
     },
+    {
+        "name": enums.EventStreamCredentialType.MTLS_V2,
+        "namespace": "event_stream",
+        "kind": "mtls_v2",
+        "inputs": EVENT_STREAM_MTLS_V2_INPUTS,
+        "injectors": {},
+        "managed": True,
+        "description": (
+            "Credential for Event Streams that use mutual TLS. "
+            "The Certificates can be defined in the UI and it "
+            "be transferred to the Gateway proxy for validation "
+            "of incoming requests. We can optionally validate the "
+            "Subject defined in the inbound Certificate."
+        ),
+    },
     {
         "name": enums.CustomEventStreamCredentialType.GITLAB,
         "namespace": "event_stream",

src/aap_eda/core/validators.py

@@ -278,7 +278,7 @@ def valid_hash_format(fmt: str):
 def _validate_event_stream_settings(auth_type: str):
     """Check event stream settings."""
     if (
-        auth_type == enums.EventStreamCredentialType.MTLS
+        auth_type == enums.EventStreamCredentialType.MTLS_V2
         and not settings.EVENT_STREAM_MTLS_BASE_URL
     ):
         raise serializers.ValidationError(
@@ -290,7 +290,7 @@ def _validate_event_stream_settings(auth_type: str):
         )
 
     if (
-        auth_type != enums.EventStreamCredentialType.MTLS
+        auth_type != enums.EventStreamCredentialType.MTLS_V2
         and not settings.EVENT_STREAM_BASE_URL
     ):
         raise serializers.ValidationError(

src/aap_eda/services/sync_certs.py

@@ -0,0 +1,189 @@
+#  Copyright 2024 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 base64
+import hashlib
+import logging
+from urllib.parse import urljoin
+
+import requests
+import yaml
+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
+
+
+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.GATEWAY_URL
+        self.gateway_user = settings.GATEWAY_USER
+        self.gateway_password = settings.GATEWAY_PASSWORD
+        self.gateway_ssl_verify = settings.GATEWAY_SSL_VERIFY
+        self.gateway_token = settings.GATEWAY_TOKEN
+        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())
+        sha256 = hashlib.sha256(
+            inputs["certificate"].encode("utf-8")
+        ).hexdigest()
+        existing_object = self._fetch_from_gateway()
+        LOGGER.info(f"Existing object is {existing_object}")
+
+        if existing_object.get("sha256", "") != sha256:
+            data = {
+                "name": self.eda_credential.name,
+                "pem_data": inputs["certificate"],
+                "sha256": sha256,
+                "remote_id": self.eda_credential_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: int):
+        """Delete the Certificate from Gateway if no other EventStream
+        is using it.
+        Parameters
+        ----------
+        event_stream_id: int
+            The event_stream associated with this credential, if this is
+            the only event stream using this certificate we can safely
+            delete it
+        """
+        existing_object = self._fetch_from_gateway()
+        if not existing_object:
+            return
+
+        objects = models.EventStream.objects.filter(
+            eda_credential_id=self.eda_credential
+        )
+        if len(objects) == 1 and event_stream_id == objects[0].id:
+            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("Couldn't delete certificate object in gateway")
+                raise GatewayAPIError
+
+    def _fetch_from_gateway(self):
+        slug = f"{SLUG}/?remote_id={self.eda_credential_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")
+        raise GatewayAPIError
+
+    def _prep_headers(self) -> dict:
+        if self.gateway_token:
+            return {"Authorization": f"Bearer {self.gateway_token}"}
+
+        if self.gateway_user and self.gateway_password:
+            user_pass = f"{self.gateway_user}:{self.gateway_password}"
+            auth_value = (
+                f"Basic {base64.b64encode(user_pass.encode()).decode()}"
+            )
+            return {"Authorization": auth_value}
+
+        LOGGER.error("Cannot connect to gateway missing Credentials")
+        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.name
+        == enums.EventStreamCredentialType.MTLS_V2
+    ):
+        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)
+            )