fix(azure): cap certificate bundle at 50 KiB before parsing

Legitimate PEM/PKCS#12 bundles are well under 10 KiB. A multi-MB
payload would waste memory on base64 decoding plus PKCS#12/PEM parsing
before the validator rejects it, so the size check runs before any
parsing. Prevents memory-exhaustion attacks from callers that hand
untrusted bytes to `validate_certificate_bundle`.

.gitignore

@@ -175,3 +175,4 @@ docker-compose.override.yml
 docker-compose-dev.override.yml
 # Local Pi runtime state
 .atl/
+.gga

prowler/changelog.d/azure-certificate-authentication.added.md

@@ -0,0 +1 @@
+Certificate-based service principal authentication for the Azure provider, alongside the existing client-secret flow

prowler/providers/azure/exceptions/exceptions.py

@@ -102,6 +102,14 @@ class AzureBaseException(ProwlerException):
             "message": "The provided provider_id does not match with the available subscriptions",
             "remediation": "Check the provider_id and ensure it is a valid subscription for the given credentials.",
         },
+        (2024, "AzureNotValidCertificateContentError"): {
+            "message": "The provided certificate content is not valid",
+            "remediation": "Check that the certificate content is a valid base64-encoded PEM or PFX bound to the app registration's keyCredentials.",
+        },
+        (2025, "AzureNotValidCertificatePathError"): {
+            "message": "The provided certificate path is not valid",
+            "remediation": "Check that the certificate file exists, is readable, and matches an entry in the app registration's keyCredentials.",
+        },
     }
 
     def __init__(self, code, file=None, original_exception=None, message=None):
@@ -291,3 +299,17 @@ class AzureInvalidProviderIdError(AzureBaseException):
         super().__init__(
             2023, file=file, original_exception=original_exception, message=message
         )
+
+
+class AzureNotValidCertificateContentError(AzureCredentialsError):
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            2024, file=file, original_exception=original_exception, message=message
+        )
+
+
+class AzureNotValidCertificatePathError(AzureCredentialsError):
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            2025, file=file, original_exception=original_exception, message=message
+        )

prowler/providers/azure/lib/arguments/arguments.py

@@ -29,6 +29,21 @@ def init_parser(self):
         action="store_true",
         help="Use managed identity authentication to log in against Azure ",
     )
+    azure_auth_modes_group.add_argument(
+        "--certificate-auth",
+        action="store_true",
+        help="Use certificate authentication to log in against Azure",
+    )
+    # Modifier of --certificate-auth, not a separate mode. The pairing is
+    # enforced in `validate_arguments` so a stray combination like
+    # `--browser-auth --certificate-path X` fails fast instead of dropping
+    # the certificate silently.
+    azure_parser.add_argument(
+        "--certificate-path",
+        nargs="?",
+        default=None,
+        help="Path to the certificate file to be used with --certificate-auth option",
+    )
     # Subscriptions
     azure_subscriptions_subparser = azure_parser.add_argument_group("Subscriptions")
     azure_subscriptions_subparser.add_argument(

prowler/providers/azure/lib/certificate.py

@@ -0,0 +1,147 @@
+import re
+from typing import Optional
+
+from cryptography import x509
+from cryptography.exceptions import UnsupportedAlgorithm
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.serialization import pkcs12
+
+# Reject bundles larger than this before parsing: legitimate PEM and PFX
+# bundles are under ~10 KiB, and a multi-MB payload would waste memory on
+# the doubling that base64 decoding plus PKCS#12/PEM parsing perform.
+_MAX_CERTIFICATE_BUNDLE_BYTES = 50 * 1024
+
+_CERTIFICATE_BLOCK_RE = re.compile(
+    rb"-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----",
+    re.DOTALL,
+)
+
+# Covers PKCS#8 (encrypted or not), legacy RSA/EC/DSA and OpenSSH PEM labels.
+# The actual decoding is delegated to `load_pem_private_key` below.
+_PRIVATE_KEY_BLOCK_RE = re.compile(
+    rb"-----BEGIN (?:ENCRYPTED |OPENSSH |RSA |EC |DSA )?PRIVATE KEY-----"
+    rb".*?"
+    rb"-----END (?:ENCRYPTED |OPENSSH |RSA |EC |DSA )?PRIVATE KEY-----",
+    re.DOTALL,
+)
+
+
+def validate_certificate_bundle(certificate_data: bytes) -> bytes:
+    """Validate the bundle and return a normalized copy safe for azure-identity.
+
+    Accepts either a PKCS#12/PFX blob (encrypted or unencrypted with a null
+    password) or a concatenated PEM bundle. Raises ``ValueError`` when the
+    payload exceeds the maximum bundle size, is missing a certificate,
+    missing a private key, or contains a pair whose public keys do not
+    match. Raises ``TypeError`` for password-protected PEM private keys,
+    which cryptography surfaces from
+    ``load_pem_private_key(..., password=None)``.
+
+    The normalized bytes always place the leaf certificate before the private
+    key so ``azure.identity.CertificateCredential`` — which uses the first
+    ``BEGIN CERTIFICATE`` block to compute the credential thumbprint — never
+    picks an intermediate CA over the matching leaf. PKCS#12 blobs are
+    returned as-is.
+    """
+    if len(certificate_data) > _MAX_CERTIFICATE_BUNDLE_BYTES:
+        raise ValueError(
+            f"the payload exceeds the maximum bundle size of "
+            f"{_MAX_CERTIFICATE_BUNDLE_BYTES} bytes"
+        )
+    try:
+        private_key, certificate, additional_certs = pkcs12.load_key_and_certificates(
+            certificate_data, None, default_backend()
+        )
+    except (ValueError, UnsupportedAlgorithm) as error:
+        # `load_key_and_certificates` also raises `ValueError` when the
+        # PKCS#12 archive is password-protected (message text: "Invalid
+        # password or PKCS12 data"). Fall through to the PEM parser only
+        # when the payload smells like PEM; otherwise raise a specific
+        # error so the caller does not see the misleading "missing
+        # certificate or key" message from `_normalize_pem_bundle`.
+        if b"-----BEGIN" not in certificate_data:
+            raise ValueError(
+                "the payload is not a valid PEM bundle nor an unencrypted "
+                "PKCS#12 archive; password-protected PKCS#12 archives are "
+                "not supported"
+            ) from error
+        return _normalize_pem_bundle(certificate_data)
+
+    if private_key is None:
+        raise ValueError("the PKCS#12 archive does not contain a private key")
+    if certificate is None and not additional_certs:
+        raise ValueError("the PKCS#12 archive does not contain a certificate")
+
+    encoding = serialization.Encoding.DER
+    public_format = serialization.PublicFormat.SubjectPublicKeyInfo
+    key_public_bytes = private_key.public_key().public_bytes(encoding, public_format)
+    if (
+        certificate is not None
+        and certificate.public_key().public_bytes(encoding, public_format)
+        == key_public_bytes
+    ):
+        # PKCS#12 blobs are consumed directly by azure-identity; no reordering.
+        return certificate_data
+
+    # Some `openssl pkcs12 -export -certfile` workflows write the leaf
+    # certificate into the additional-certs bag instead of the primary
+    # slot. azure-identity reads the primary certificate for the
+    # thumbprint, so accepting the archive as-is would authenticate
+    # against the wrong thumbprint. Detect that case and raise an
+    # actionable error rather than the opaque "does not match" message.
+    for candidate in additional_certs or ():
+        if (
+            candidate.public_key().public_bytes(encoding, public_format)
+            == key_public_bytes
+        ):
+            raise ValueError(
+                "the PKCS#12 archive has the certificate matching the "
+                "private key in the additional-certs bag; re-export the "
+                "archive with the leaf certificate as the primary entry"
+            )
+    raise ValueError("the certificate does not match the private key")
+
+
+def _normalize_pem_bundle(certificate_data: bytes) -> bytes:
+    """Validate a PEM bundle and return it with the matching leaf first."""
+    certificate_blocks = _CERTIFICATE_BLOCK_RE.findall(certificate_data)
+    private_key_matches = list(_PRIVATE_KEY_BLOCK_RE.finditer(certificate_data))
+
+    if not certificate_blocks or not private_key_matches:
+        raise ValueError("the payload must contain a certificate and its private key")
+
+    # A bundle may carry more than one private key block (e.g. legacy tools
+    # export both an RSA and a PKCS#8 copy). Try each in order; preserve the
+    # first parse error so that a single encrypted key still surfaces the
+    # TypeError callers rely on to route to the typed certificate errors.
+    first_key_error: Optional[Exception] = None
+    for key_match in private_key_matches:
+        try:
+            private_key = serialization.load_pem_private_key(
+                key_match.group(), password=None, backend=default_backend()
+            )
+        except (ValueError, TypeError) as error:
+            if first_key_error is None:
+                first_key_error = error
+            continue
+        key_public_bytes = private_key.public_key().public_bytes(
+            serialization.Encoding.DER,
+            serialization.PublicFormat.SubjectPublicKeyInfo,
+        )
+        for pem_block in certificate_blocks:
+            candidate = x509.load_pem_x509_certificate(pem_block, default_backend())
+            candidate_public_bytes = candidate.public_key().public_bytes(
+                serialization.Encoding.DER,
+                serialization.PublicFormat.SubjectPublicKeyInfo,
+            )
+            if candidate_public_bytes == key_public_bytes:
+                # azure-identity's CertificateCredential uses the first BEGIN
+                # CERTIFICATE block to compute the credential thumbprint. Put
+                # the matching leaf first so authentication uses the correct
+                # certificate regardless of the bundle's original ordering.
+                return pem_block + b"\n" + key_match.group() + b"\n"
+
+    if first_key_error is not None:
+        raise first_key_error
+    raise ValueError("the certificate does not match the private key")

prowler/providers/azure/models.py

@@ -11,6 +11,7 @@ class AzureIdentityInfo(BaseModel):
     identity_type: str = ""
     tenant_ids: list[str] = []
     tenant_domain: str = "Unknown tenant domain (missing AAD permissions)"
+    certificate_thumbprint: str = ""
     subscriptions: dict = {}
     locations: dict = {}
 

prowler/providers/common/provider.py

@@ -404,6 +404,8 @@ class Provider(ABC):
                         sp_env_auth=arguments.sp_env_auth,
                         browser_auth=arguments.browser_auth,
                         managed_identity_auth=arguments.managed_identity_auth,
+                        certificate_auth=arguments.certificate_auth,
+                        certificate_path=arguments.certificate_path,
                         tenant_id=arguments.tenant_id,
                         region=arguments.azure_region,
                         subscription_ids=arguments.subscription_id,

tests/lib/cli/parser_test.py

@@ -12,6 +12,7 @@ from prowler.providers.aws.lib.arguments.arguments import (
     validate_role_session_name,
 )
 from prowler.providers.azure.lib.arguments.arguments import validate_azure_region
+from prowler.providers.common.provider import Provider
 
 prowler_command = "prowler"
 
@@ -154,6 +155,53 @@ class Test_Parser:
         assert not parsed.managed_identity_auth
         assert not parsed.shodan
 
+    def test_azure_certificate_auth_arguments(self):
+        certificate_path = "/secure/path/prowler-cert.pem"
+
+        parsed = self.parser.parse(
+            [
+                prowler_command,
+                "azure",
+                "--certificate-auth",
+                "--certificate-path",
+                certificate_path,
+            ]
+        )
+
+        assert parsed.certificate_auth
+        assert parsed.certificate_path == certificate_path
+
+    def test_azure_certificate_auth_arguments_are_forwarded(self):
+        certificate_path = "/secure/path/prowler-cert.pem"
+        parsed = self.parser.parse(
+            [
+                prowler_command,
+                "azure",
+                "--certificate-auth",
+                "--certificate-path",
+                certificate_path,
+            ]
+        )
+        captured = {}
+
+        class AzureProviderStub:
+            def __init__(self, **kwargs):
+                captured.update(kwargs)
+
+        with (
+            patch.object(Provider, "_global", None),
+            patch.object(Provider, "get_class", return_value=AzureProviderStub),
+            patch.object(Provider, "is_builtin", return_value=True),
+            patch(
+                "prowler.providers.common.provider.load_and_validate_config_file",
+                return_value={},
+            ),
+        ):
+            Provider.init_global_provider(parsed)
+
+        assert captured["certificate_auth"] is True
+        assert captured["certificate_path"] == certificate_path
+
     def test_default_parser_no_arguments_gcp(self):
         provider = "gcp"
         command = [prowler_command, provider]