diff --git a/prowler/providers/azure/azure_provider.py b/prowler/providers/azure/azure_provider.py index 294b0da661..cc28dc78bc 100644 --- a/prowler/providers/azure/azure_provider.py +++ b/prowler/providers/azure/azure_provider.py @@ -213,10 +213,10 @@ class AzureProvider(Provider): fixer_config(self): Returns the fixer configuration. output_options(self, options: tuple): Sets the output options for the Azure provider. mutelist(self) -> AzureMutelist: Returns the mutelist object associated with the Azure provider. - validate_arguments(cls, az_cli_auth, sp_env_auth, browser_auth, managed_identity_auth, tenant_id): Validates the authentication arguments for the Azure provider. + validate_arguments(az_cli_auth, sp_env_auth, browser_auth, managed_identity_auth, tenant_id, client_id, client_secret, *, certificate_auth, certificate_content, certificate_path): Validates the authentication arguments for the Azure provider. setup_region_config(cls, region): Sets up the region configuration for the Azure provider. print_credentials(self): Prints the Azure credentials information. - setup_session(cls, az_cli_auth, sp_env_auth, browser_auth, managed_identity_auth, tenant_id, region_config): Set up the Azure session with the specified authentication method. + setup_session(az_cli_auth, sp_env_auth, browser_auth, managed_identity_auth, tenant_id, azure_credentials, region_config, *, certificate_auth, certificate_path): Set up the Azure session with the specified authentication method. """ _type: str = "azure" @@ -371,12 +371,12 @@ class AzureProvider(Provider): sp_env_auth, browser_auth, managed_identity_auth, - certificate_auth, tenant_id, client_id, client_secret, - certificate_content, - certificate_path, + certificate_auth=certificate_auth, + certificate_content=certificate_content, + certificate_path=certificate_path, ) logger.info("Checking if region is different than default one") @@ -404,11 +404,11 @@ class AzureProvider(Provider): sp_env_auth, browser_auth, managed_identity_auth, - certificate_auth, - certificate_path, tenant_id, azure_credentials, self._region_config, + certificate_auth=certificate_auth, + certificate_path=certificate_path, ) # Set up the identity @@ -505,12 +505,16 @@ class AzureProvider(Provider): sp_env_auth: bool, browser_auth: bool, managed_identity_auth: bool, - certificate_auth: bool, tenant_id: str, client_id: str, client_secret: str, - certificate_content: str, - certificate_path: str, + # Certificate kwargs are keyword-only so they cannot displace + # `tenant_id`/`client_id`/`client_secret` for callers passing + # positionally the way the pre-cert-auth signature accepted. + *, + certificate_auth: bool = False, + certificate_content: str = None, + certificate_path: str = None, ): """ Validates the authentication arguments for the Azure provider. @@ -687,11 +691,15 @@ class AzureProvider(Provider): sp_env_auth: bool, browser_auth: bool, managed_identity_auth: bool, - certificate_auth: bool, - certificate_path: str, tenant_id: str, azure_credentials: dict, region_config: AzureRegionConfig, + # Certificate kwargs are keyword-only so callers that pass + # `tenant_id`/`azure_credentials`/`region_config` positionally the + # way the pre-cert-auth signature accepted keep binding those slots. + *, + certificate_auth: bool = False, + certificate_path: str = None, ): """Returns the Azure credentials object. @@ -989,12 +997,12 @@ class AzureProvider(Provider): sp_env_auth, browser_auth, managed_identity_auth, - certificate_auth, tenant_id, client_id, client_secret, - certificate_content, - certificate_path, + certificate_auth=certificate_auth, + certificate_content=certificate_content, + certificate_path=certificate_path, ) region_config = AzureProvider.setup_region_config(region) @@ -1020,11 +1028,11 @@ class AzureProvider(Provider): sp_env_auth, browser_auth, managed_identity_auth, - certificate_auth, - certificate_path, tenant_id, azure_credentials, region_config, + certificate_auth=certificate_auth, + certificate_path=certificate_path, ) # Create a SubscriptionClient subscription_client = SubscriptionClient( @@ -1217,26 +1225,24 @@ class AzureProvider(Provider): def check_certificate_creds_env_vars( check_certificate_content: bool, tenant_id: Optional[str] = None, - client_id: Optional[str] = None, ): """ Checks the presence of required environment variables for certificate-based service principal authentication against Azure. Environment variables are only required when the caller has not already - supplied the value explicitly: an explicit ``tenant_id`` or ``client_id`` - substitutes for AZURE_TENANT_ID / AZURE_CLIENT_ID respectively. + supplied the value explicitly: an explicit ``tenant_id`` substitutes + for AZURE_TENANT_ID. There is no CLI/API path that overrides + AZURE_CLIENT_ID on the env-var flow, so it must always be set. Raises: AzureEnvironmentVariableError: If any required environment variable is missing and no explicit value was supplied. """ logger.info("Azure provider: checking certificate environment variables ...") - env_vars: list[str] = [] + env_vars: list[str] = ["AZURE_CLIENT_ID"] if not tenant_id: env_vars.append("AZURE_TENANT_ID") - if not client_id: - env_vars.append("AZURE_CLIENT_ID") if check_certificate_content: env_vars.append("AZURE_CERTIFICATE_CONTENT") for env_var in env_vars: @@ -1714,6 +1720,7 @@ class AzureProvider(Provider): client_id, client_secret, region_config: AzureRegionConfig = None, + *, certificate_content: str = None, certificate_path: str = None, ) -> None: @@ -1811,6 +1818,7 @@ class AzureProvider(Provider): # here is to prove the private key matches an active `keyCredentials` # entry on the app registration. credential = None + transport = None try: if certificate_content: certificate_data = base64.b64decode(certificate_content, validate=True) @@ -1825,16 +1833,20 @@ class AzureProvider(Provider): # running `get_token`, so timeouts and non-timeout failures # leaked a worker per validation under Entra ID degradation. # `retry_total=0` stops Azure Core from replaying each request - # and multiplying the effective deadline. + # and multiplying the effective deadline. The transport is + # bound to a local so `finally` can close it even when + # `CertificateCredential.__init__` raises before the + # credential is assigned. + transport = RequestsTransport( + connection_timeout=_TOKEN_ACQUISITION_TIMEOUT_SECONDS, + read_timeout=_TOKEN_ACQUISITION_TIMEOUT_SECONDS, + ) credential = CertificateCredential( client_id=client_id, tenant_id=tenant_id, certificate_data=validate_certificate_bundle(certificate_data), authority=region_config.authority, - transport=RequestsTransport( - connection_timeout=_TOKEN_ACQUISITION_TIMEOUT_SECONDS, - read_timeout=_TOKEN_ACQUISITION_TIMEOUT_SECONDS, - ), + transport=transport, retry_total=0, ) credential.get_token(region_config.graph_scope) @@ -1893,9 +1905,12 @@ class AzureProvider(Provider): original_exception=error, ) finally: - # Always release the transient credential (and its transport): - # a `close()` failure must be logged and swallowed so it cannot - # replace the primary typed exception on its way up. + # Always release the transient credential and its transport: + # cleanup failures are logged and swallowed so they cannot + # replace the primary typed exception on its way up. The + # transport is closed independently to cover the case where + # `CertificateCredential.__init__` raised before `credential` + # took ownership of it. if credential is not None: try: credential.close() @@ -1904,3 +1919,11 @@ class AzureProvider(Provider): f"{cleanup_error.__class__.__name__}: failed to close " f"CertificateCredential during verify_client cleanup" ) + elif transport is not None: + try: + transport.close() + except Exception as cleanup_error: + logger.warning( + f"{cleanup_error.__class__.__name__}: failed to close " + f"RequestsTransport during verify_client cleanup" + ) diff --git a/prowler/providers/azure/lib/certificate.py b/prowler/providers/azure/lib/certificate.py index c76cf1aa2f..ace1dca877 100644 --- a/prowler/providers/azure/lib/certificate.py +++ b/prowler/providers/azure/lib/certificate.py @@ -1,4 +1,5 @@ import re +from typing import Optional from cryptography import x509 from cryptography.exceptions import UnsupportedAlgorithm @@ -63,30 +64,42 @@ def validate_certificate_bundle(certificate_data: bytes) -> bytes: 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_match = _PRIVATE_KEY_BLOCK_RE.search(certificate_data) + private_key_matches = list(_PRIVATE_KEY_BLOCK_RE.finditer(certificate_data)) - if not certificate_blocks or not private_key_match: + if not certificate_blocks or not private_key_matches: raise ValueError("the payload must contain a certificate and its private key") - private_key = serialization.load_pem_private_key( - private_key_match.group(), password=None, backend=default_backend() - ) - 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( + # 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, ) - 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" + private_key_match.group() + b"\n" + 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") diff --git a/tests/providers/azure/azure_provider_test.py b/tests/providers/azure/azure_provider_test.py index bf1237c8e5..eb58b91f12 100644 --- a/tests/providers/azure/azure_provider_test.py +++ b/tests/providers/azure/azure_provider_test.py @@ -3004,6 +3004,142 @@ class TestAzureProviderSignatureCompatibility: parameters[name].kind is inspect.Parameter.KEYWORD_ONLY ), f"validate_static_credentials: {name} must be keyword-only" + def test_validate_arguments_positional_call_still_binds_tenant_and_client(self): + import inspect + + # Master signature: `..., managed_identity_auth, tenant_id, client_id, + # client_secret`. Any external caller passing everything positionally + # must keep binding those three slots to the right names. + signature = inspect.signature(AzureProvider.validate_arguments) + bound = signature.bind( + False, # az_cli_auth + False, # sp_env_auth + False, # browser_auth + False, # managed_identity_auth + self._TENANT_ID, # tenant_id + self._CLIENT_ID, # client_id + "some-secret", # client_secret + ) + + assert bound.arguments["tenant_id"] == self._TENANT_ID + assert bound.arguments["client_id"] == self._CLIENT_ID + assert bound.arguments["client_secret"] == "some-secret" + assert "certificate_auth" not in bound.arguments + assert "certificate_content" not in bound.arguments + assert "certificate_path" not in bound.arguments + + self._assert_certificate_kwargs_are_keyword_only( + AzureProvider.validate_arguments + ) + + def test_setup_session_positional_call_still_binds_tenant_and_credentials(self): + import inspect + + # Master signature: `..., managed_identity_auth, tenant_id, + # azure_credentials, region_config`. Callers passing the pre-cert + # positional shape must not rebind those slots. + signature = inspect.signature(AzureProvider.setup_session) + region_config = AzureProvider.setup_region_config("AzureCloud") + bound = signature.bind( + False, # az_cli_auth + False, # sp_env_auth + False, # browser_auth + False, # managed_identity_auth + self._TENANT_ID, # tenant_id + {"tenant_id": self._TENANT_ID}, # azure_credentials + region_config, # region_config + ) + + assert bound.arguments["tenant_id"] == self._TENANT_ID + assert bound.arguments["azure_credentials"] == {"tenant_id": self._TENANT_ID} + assert bound.arguments["region_config"] is region_config + assert "certificate_auth" not in bound.arguments + assert "certificate_path" not in bound.arguments + + # Only the two certificate parameters are keyword-only in + # `setup_session` — `certificate_content` is not part of its shape. + parameters = inspect.signature(AzureProvider.setup_session).parameters + for name in ("certificate_auth", "certificate_path"): + assert ( + parameters[name].kind is inspect.Parameter.KEYWORD_ONLY + ), f"setup_session: {name} must be keyword-only" + + +class TestValidateCertificateBundleMultiKey: + """A PEM bundle may legitimately carry more than one private key block + (e.g. legacy tools that export both RSA and PKCS#8 encodings). The + normalizer must locate the key that actually pairs with the leaf + certificate, not stop at the first `-----BEGIN PRIVATE KEY-----`.""" + + def _self_signed(self): + from datetime import datetime, timedelta, timezone + + from cryptography import x509 + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import rsa + from cryptography.x509.oid import NameOID + + private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Prowler")]) + certificate = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(subject) + .public_key(private_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(datetime.now(timezone.utc)) + .not_valid_after(datetime.now(timezone.utc) + timedelta(days=1)) + .sign(private_key, hashes.SHA256()) + ) + cert_pem = certificate.public_bytes(serialization.Encoding.PEM) + key_pem = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + return cert_pem, key_pem + + def test_normalizes_when_matching_key_is_second_in_bundle(self): + from prowler.providers.azure.lib.certificate import ( + validate_certificate_bundle, + ) + + _, unrelated_key_pem = self._self_signed() + leaf_cert_pem, leaf_key_pem = self._self_signed() + + # First private key belongs to an unrelated pair; the leaf's key is + # the second block. `.search` used to stop at the first match and + # falsely reject the bundle. + bundle = leaf_cert_pem + unrelated_key_pem + leaf_key_pem + + normalized = validate_certificate_bundle(bundle) + + assert normalized.startswith(leaf_cert_pem) + assert leaf_key_pem in normalized + + def test_encrypted_single_key_still_raises_type_error(self): + # Guardrail for the multi-key iteration: an encrypted single key + # must still surface `TypeError`, which the SDK caller relies on to + # route to `AzureNotValidCertificateContentError`. + from cryptography.hazmat.primitives import serialization + from cryptography.hazmat.primitives.asymmetric import rsa + + from prowler.providers.azure.lib.certificate import ( + validate_certificate_bundle, + ) + + cert_pem, _ = self._self_signed() + encrypted_key_pem = rsa.generate_private_key( + public_exponent=65537, key_size=2048 + ).private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.BestAvailableEncryption(b"prowler"), + ) + + with pytest.raises(TypeError): + validate_certificate_bundle(cert_pem + encrypted_key_pem) + def serialization_no_encryption(): # Kept out of the test method so the PKCS#12 test reads cleanly. Wraps