Fetch client cert (#216)

Author: gerbrand-bosch

app/services/client_oauth.py

@@ -3,7 +3,7 @@
 import hmac
 import logging
 import ssl
-from typing import Any, Dict
+from typing import Any, Dict, List
 
 import jwt
 from cryptography import x509
@@ -13,7 +13,6 @@
 
 from app.config import ConfigClientOAuth
 from app.models.ura import UraNumber
-from app.utils.certificates.utils import enforce_cert_newlines
 
 SSL_CLIENT_CERT_HEADER_NAME = "x-forwarded-tls-client-cert"  # "x-proxy-ssl_client_cert"
 
@@ -34,14 +33,22 @@ def _create_ssl_context(self) -> ssl.SSLContext:
         """
         Create an SSL context for mTLS connections to the JWKS endpoint.
         """
-        if self.config.mtls_cert is None or self.config.mtls_key is None or self.config.verify_ca is None:
-            raise ValueError("mTLS certificate and key must be provided for Client OAuth2")
+        if (
+            self.config.mtls_cert is None
+            or self.config.mtls_key is None
+            or self.config.verify_ca is None
+        ):
+            raise ValueError(
+                "mTLS certificate and key must be provided for Client OAuth2"
+            )
 
         context = ssl.create_default_context()
         if isinstance(self.config.verify_ca, bool) and self.config.verify_ca is True:
             context.verify_mode = ssl.CERT_REQUIRED
 
-        context.load_cert_chain(certfile=self.config.mtls_cert, keyfile=self.config.mtls_key)
+        context.load_cert_chain(
+            certfile=self.config.mtls_cert, keyfile=self.config.mtls_key
+        )
         if isinstance(self.config.verify_ca, str):
             context.load_verify_locations(cafile=self.config.verify_ca)
 
@@ -65,7 +72,9 @@ def verify(self, request: Request) -> Dict[str, Any]:
         """
         auth_header = request.headers.get("authorization")
         if not auth_header or not auth_header.lower().startswith("bearer "):
-            raise HTTPException(status_code=401, detail="Missing or invalid Authorization header")
+            raise HTTPException(
+                status_code=401, detail="Missing or invalid Authorization header"
+            )
 
         token = auth_header[7:]  # Remove "Bearer "
         claims = self._verify_token(token)
@@ -77,7 +86,9 @@ def _verify_token(self, token: str) -> Dict[str, Any]:
         """
         Verify JWT token and return claims.
         """
-        jwk_client = PyJWKClient(self.config.jwks_url, cache_keys=True, ssl_context=self._ssl_context)
+        jwk_client = PyJWKClient(
+            self.config.jwks_url, cache_keys=True, ssl_context=self._ssl_context
+        )
         signing_key = jwk_client.get_signing_key_from_jwt(token).key
 
         try:
@@ -105,16 +116,16 @@ def _verify_mtls(request: Request, claims: Dict[str, Any]) -> None:
         Verify mTLS binding by checking cnf.x5t#S256 against presented client certificate thumbprint.
         """
 
-        # Extract presented client certificate thumbprint from request
-        cert_pem = request.headers.get(SSL_CLIENT_CERT_HEADER_NAME)
-        if not cert_pem:
+        certs = ClientOAuthService.get_pem_from_request(request)
+        if not certs:
             logger.error("Client certificate not presented or verification failed")
-            raise HTTPException(status_code=401, detail="Client certificate not presented or verification failed")
-
-        cert_pem = enforce_cert_newlines(cert_pem)
+            raise HTTPException(
+                status_code=401,
+                detail="Client certificate not presented or verification failed",
+            )
 
-        # Calculate thumbprint of presented certificate
-        presented_cert = x509.load_pem_x509_certificate(cert_pem.encode())
+        # Calculate thumbprint of presented client certificate, ignoring the chain
+        presented_cert = x509.load_pem_x509_certificate(certs[0].encode())
         cert_der = presented_cert.public_bytes(serialization.Encoding.DER)
         sha256_hash = hashlib.sha256(cert_der).digest()
         request_cert_fp = base64.urlsafe_b64encode(sha256_hash).rstrip(b"=").decode()
@@ -135,3 +146,38 @@ def _verify_mtls(request: Request, claims: Dict[str, Any]) -> None:
         if not hmac.compare_digest(fp, request_cert_fp):
             logger.debug("mTLS binding failed")
             raise HTTPException(status_code=401, detail="Invalid token")
+
+    @staticmethod
+    def get_pem_from_request(request: Request) -> List[str]:
+        """
+        Extracts and returns the PEM-encoded client certificate from the request headers.
+        """
+        if (
+            SSL_CLIENT_CERT_HEADER_NAME not in request.headers
+            or not request.headers.get(SSL_CLIENT_CERT_HEADER_NAME)
+        ):
+            logger.debug("Client certificate not found or verification failed.")
+            return []
+
+        certs = request.headers.get(SSL_CLIENT_CERT_HEADER_NAME, "").split(",")
+        return [
+            ClientOAuthService.fixup_cert_headers_and_footers(cert) for cert in certs
+        ]
+
+    @staticmethod
+    def fixup_cert_headers_and_footers(cert: str) -> str:
+        # Add PEM headers/footers if missing
+        if not cert.startswith("-----BEGIN CERTIFICATE-----"):
+            cert = (
+                "-----BEGIN CERTIFICATE-----\n" + cert + "\n-----END CERTIFICATE-----"
+            )
+
+        # If we are by any chance missing newlines after/before the headers, add them
+        if not cert.startswith("-----BEGIN CERTIFICATE-----\n"):
+            cert = cert.replace(
+                "-----BEGIN CERTIFICATE-----", "-----BEGIN CERTIFICATE-----\n"
+            )
+            cert = cert.replace(
+                "-----END CERTIFICATE-----", "\n-----END CERTIFICATE-----"
+            )
+        return cert