feat: Implement mTLS event stream authentication support (#1402)

## Summary
This PR implements comprehensive mTLS (mutual TLS) authentication
support for Event Streams, enabling secure client certificate-based
authentication with the Gateway service.

https://issues.redhat.com/browse/AAP-46060

### Key Features
- **New mTLS Credential Type**: EventStreamCredentialType.MTLS for
secure client authentication
- **Certificate Synchronization**: Automatic sync with Gateway API
during Event Stream lifecycle
- **Certificate Validation**: PEM format validation with expiration
checking and RFC 2253 subject validation
- **Lifecycle Management**: Automatic certificate create/update/delete
operations

### Core Components
- `SyncCertificates` service for Gateway certificate management
- Enhanced `EdaCredential` model with mTLS-specific validation
- `EventStream` views with automatic certificate synchronization
- New API exceptions for Gateway and credential error handling

### Security Enhancements
- Certificate format validation using cryptography library
- Subject DN validation with proper X.509 attribute checking
- Secure certificate storage and transmission to Gateway
- Comprehensive error handling for authentication failures

## Test plan
- [x] Unit tests for certificate validation logic
- [x] Integration tests for EventStream mTLS workflows
- [x] Service-level tests for Gateway synchronization
- [x] Edge case testing for certificate lifecycle management
- [x] All existing tests continue to pass
- [x] Linting and code quality checks pass

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-authored-by: Claude <[email protected]>

Author: hsong-rh

src/aap_eda/api/event_stream_authentication.py

@@ -32,6 +32,7 @@
 from rest_framework.exceptions import AuthenticationFailed
 
 from aap_eda.core.enums import SignatureEncodingType
+from aap_eda.core.utils.credentials import validate_x509_subject_match
 
 logger = logging.getLogger(__name__)
 DEFAULT_TIMEOUT = 30
@@ -110,11 +111,29 @@ class MTLSAuthentication(EventStreamAuthentication):
 
     def authenticate(self, _body=None):
         """Handle mTLS authentication."""
-        if self.subject and self.subject != self.value:
-            message = f"Subject Name mismatch : {self.value}"
+        if self.subject and not self.validate_subject(
+            self.subject, self.value
+        ):
+            message = f"Subject: {self.value} does not match {self.subject}"
             logger.warning(message)
             raise AuthenticationFailed(message)
 
+    def validate_subject(self, expected: str, actual: str) -> bool:
+        """Validate that actual subject matches expected subject pattern.
+
+        Uses shared X.509 standard-compliant DN parsing for attribute-level
+        matching. Supports wildcards and is order-independent per X.509
+        standards.
+
+        Args:
+            expected: Official subject pattern (may contain * wildcards)
+            actual: Input subject from user
+
+        Returns:
+            bool: True if actual matches expected pattern, False otherwise
+        """
+        return validate_x509_subject_match(expected, actual)
+
 
 @dataclass
 class BasicAuthentication(EventStreamAuthentication):

src/aap_eda/api/exceptions.py

@@ -35,6 +35,8 @@
     "Conflict",
     "Unprocessable",
     "PermissionDenied",
+    "GatewayAPIError",
+    "MissingCredentialsError",
     "api_fallback_handler",
 )
 
@@ -125,3 +127,20 @@ class ExternalSMSError(APIException):
     default_detail = (
         "External SMS Error: not able to fetch secrets from external SMS"
     )
+
+
+class GatewayAPIError(APIException):
+    status_code = status.HTTP_502_BAD_GATEWAY
+    default_code = "gateway_api_error"
+    default_detail = _(
+        "Gateway API Error: Unable to communicate with the Gateway service"
+    )
+
+
+class MissingCredentialsError(APIException):
+    status_code = status.HTTP_400_BAD_REQUEST
+    default_code = "missing_credentials"
+    default_detail = _(
+        "Missing Credentials: Required credentials are not available "
+        "for Gateway operations"
+    )

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,13 @@
 
 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 as CoreGatewayAPIError,
+    MissingCredentials as CoreMissingCredentials,
+)
 from aap_eda.core.utils import logging_utils
+from aap_eda.services.sync_certs import SyncCertificates
 
 logger = logging.getLogger(__name__)
 
@@ -99,7 +104,13 @@ def retrieve(self, request, *args, **kwargs):
         responses={
             status.HTTP_204_NO_CONTENT: OpenApiResponse(
                 None, description="Delete successful."
-            )
+            ),
+            status.HTTP_400_BAD_REQUEST: OpenApiResponse(
+                description="Missing credentials for certificate deletion."
+            ),
+            status.HTTP_502_BAD_GATEWAY: OpenApiResponse(
+                description="Gateway API error during certificate deletion."
+            ),
         },
     )
     def destroy(self, request, *args, **kwargs):
@@ -110,6 +121,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(
@@ -160,7 +172,10 @@ def list(self, request, *args, **kwargs):
                 description="Return the new event stream.",
             ),
             status.HTTP_400_BAD_REQUEST: OpenApiResponse(
-                description="Invalid data to create event stream."
+                description="Invalid data or missing credentials."
+            ),
+            status.HTTP_502_BAD_GATEWAY: OpenApiResponse(
+                description="Gateway API error during certificate sync."
             ),
         },
     )
@@ -182,6 +197,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(
@@ -206,12 +222,17 @@ def create(self, request, *args, **kwargs):
                 description="Update successful, return the new event stream.",
             ),
             status.HTTP_400_BAD_REQUEST: OpenApiResponse(
-                description="Unable to update event stream."
+                description="Update failed or missing credentials."
+            ),
+            status.HTTP_502_BAD_GATEWAY: OpenApiResponse(
+                description="Gateway API error during certificate sync."
             ),
         },
     )
     def partial_update(self, request, *args, **kwargs):
         event_stream = self.get_object()
+        new_eda_credential_id = request.data.get("eda_credential_id")
+
         old_data = model_to_dict(event_stream)
         context = {"request": request}
         serializer = serializers.EventStreamInSerializer(
@@ -233,6 +254,13 @@ def partial_update(self, request, *args, **kwargs):
             setattr(event_stream, key, value)
 
         with transaction.atomic():
+            # Check if we need to destroy old certificates before saving
+            if (
+                new_eda_credential_id
+                and event_stream.eda_credential.id != new_eda_credential_id
+            ):
+                self._sync_certificates(event_stream, "destroy")
+
             event_stream.save()
             check_related_permissions(
                 request.user,
@@ -241,6 +269,9 @@ def partial_update(self, request, *args, **kwargs):
                 model_to_dict(event_stream),
             )
 
+            if new_eda_credential_id:
+                self._sync_certificates(event_stream, "update")
+
         logger.info(
             logging_utils.generate_simple_audit_log(
                 "Update",
@@ -307,3 +338,35 @@ def activations(self, request, id):
             )
         )
         return self.get_paginated_response(serializer.data)
+
+    def _sync_certificates(
+        self,
+        event_stream: models.EventStream,
+        action: str,
+    ) -> None:
+        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 CoreGatewayAPIError as ex:
+                logger.error("Could not %s certificates: %s", action, str(ex))
+                raise api_exc.GatewayAPIError(
+                    detail=f"Gateway API error during certificate {action}: "
+                    f"{str(ex)}"
+                )
+            except CoreMissingCredentials as ex:
+                logger.error(
+                    "Missing credentials for certificate %s: %s",
+                    action,
+                    str(ex),
+                )
+                raise api_exc.MissingCredentialsError(
+                    detail=f"Missing credentials for certificate {action}: "
+                    f"{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,55 @@
     "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. It can contains regular "
+                "expression to match indivisual attributes in the subject "
+                "name. E.g., CN=[agent1,agent2].example.com,ST=[NJ|NY]"
+            ),
+        },
+        {
+            "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 +2094,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."
+        ),
+    },
 ]