test(azure): cover certificate authentication flows

This commit is contained in:
Hugo P.Brito committed 2026-08-22 20:19:18 +01:00
1 parent 2b34abe30c
commit bb2a0031d0
3 files changed
+460 -8

No files matched your search

@@ -1,14 +1,68 @@
import socket
from datetime import UTC, datetime, timedelta
import pytest
from api.validators import (
resolve_lighthouse_openai_compatible_host,
validate_certificate_bundle,
validate_lighthouse_openai_compatible_base_url,
)
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.serialization import pkcs12
from cryptography.x509.oid import NameOID
from django.core.exceptions import ValidationError
from django.test import override_settings
def _certificate_and_key():
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(UTC))
.not_valid_after(datetime.now(UTC) + timedelta(days=1))
.sign(private_key, hashes.SHA256())
)
return certificate, private_key
def test_certificate_bundle_rejects_key_only_pkcs12():
_, private_key = _certificate_and_key()
key_only_pkcs12 = pkcs12.serialize_key_and_certificates(
name=b"prowler",
key=private_key,
cert=None,
cas=None,
encryption_algorithm=serialization.NoEncryption(),
)
with pytest.raises(ValueError, match="certificate and its private key"):
validate_certificate_bundle(key_only_pkcs12)
def test_certificate_bundle_rejects_mismatched_pem_key():
certificate, _ = _certificate_and_key()
different_private_key = rsa.generate_private_key(
public_exponent=65537, key_size=2048
)
mismatched_bundle = certificate.public_bytes(
serialization.Encoding.PEM
) + different_private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption(),
)
with pytest.raises(ValueError, match="does not match"):
validate_certificate_bundle(mismatched_bundle)
def test_lighthouse_base_url_rejects_http_scheme():
with pytest.raises(ValidationError, match="HTTPS"):
validate_lighthouse_openai_compatible_base_url(
+399 -1
View File
@@ -4,7 +4,7 @@ from uuid import uuid4
import pytest
from azure.core.credentials import AccessToken
from azure.core.exceptions import HttpResponseError
from azure.core.exceptions import ClientAuthenticationError, HttpResponseError
from azure.identity import DefaultAzureCredential
from mock import MagicMock
@@ -23,6 +23,10 @@ from prowler.providers.azure.exceptions.exceptions import (
AzureNoAuthenticationMethodError,
AzureNotValidCertificateContentError,
AzureNotValidCertificatePathError,
AzureNotValidClientIdError,
AzureNotValidClientSecretError,
AzureNotValidTenantIdError,
AzureSetUpSessionError,
AzureTenantIDNoBrowserAuthError,
)
from prowler.providers.azure.models import AzureIdentityInfo, AzureRegionConfig
@@ -1325,6 +1329,21 @@ class TestAzureProviderCertificateAuth:
certificate_path=None,
)
def test_validate_arguments_rejects_client_id_without_credential_material(self):
with pytest.raises(AzureConfigCredentialsError, match="must provide"):
AzureProvider.validate_arguments(
az_cli_auth=False,
sp_env_auth=False,
browser_auth=False,
managed_identity_auth=False,
certificate_auth=False,
tenant_id=self._TENANT_ID,
client_id=self._CLIENT_ID,
client_secret=None,
certificate_content=None,
certificate_path=None,
)
def test_check_certificate_creds_env_vars_missing_content(self, monkeypatch):
monkeypatch.setenv("AZURE_CLIENT_ID", self._CLIENT_ID)
monkeypatch.setenv("AZURE_TENANT_ID", self._TENANT_ID)
@@ -1383,6 +1402,42 @@ class TestAzureProviderCertificateAuth:
region_config=self._region_config(),
)
def test_validate_static_credentials_rejects_key_only_pkcs12(self):
import base64
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.serialization import pkcs12
_, private_key = self._certificate_and_key()
key_only_pkcs12 = pkcs12.serialize_key_and_certificates(
name=b"prowler",
key=private_key,
cert=None,
cas=None,
encryption_algorithm=serialization.NoEncryption(),
)
with pytest.raises(AzureNotValidCertificateContentError):
AzureProvider.validate_static_credentials(
tenant_id=self._TENANT_ID,
client_id=self._CLIENT_ID,
client_secret=None,
certificate_content=base64.b64encode(key_only_pkcs12).decode("ascii"),
certificate_path=None,
region_config=self._region_config(),
)
def test_validate_static_credentials_rejects_missing_credential_material(self):
with pytest.raises(AzureNotValidClientSecretError, match="must provide"):
AzureProvider.validate_static_credentials(
tenant_id=self._TENANT_ID,
client_id=self._CLIENT_ID,
client_secret=None,
certificate_content=None,
certificate_path=None,
region_config=self._region_config(),
)
def test_validate_static_credentials_rejects_mismatched_certificate_and_key(self):
import base64
from datetime import datetime, timedelta, timezone
@@ -1600,6 +1655,43 @@ class TestAzureProviderCertificateAuth:
# The dict content is passed through base64.b64decode into bytes.
assert isinstance(kwargs["certificate_data"], (bytes, bytearray))
def test_setup_session_static_credentials_certificate_path_reads_bundle(
self, tmp_path
):
certificate_path = tmp_path / "prowler-cert.pem"
certificate_path.write_bytes(b"certificate bundle")
expected_credentials = MagicMock()
with patch(
"prowler.providers.azure.azure_provider._build_certificate_credential",
return_value=expected_credentials,
) as build_certificate_credential:
credentials = AzureProvider.setup_session(
az_cli_auth=False,
sp_env_auth=False,
browser_auth=False,
managed_identity_auth=False,
certificate_auth=False,
certificate_path=None,
tenant_id=self._TENANT_ID,
azure_credentials={
"tenant_id": self._TENANT_ID,
"client_id": self._CLIENT_ID,
"client_secret": None,
"certificate_content": None,
"certificate_path": str(certificate_path),
},
region_config=self._region_config(),
)
assert credentials is expected_credentials
build_certificate_credential.assert_called_once_with(
tenant_id=self._TENANT_ID,
client_id=self._CLIENT_ID,
certificate_data=b"certificate bundle",
authority=None,
)
def test_setup_session_env_cert_auth_uses_certificate_credential(self, monkeypatch):
monkeypatch.setenv("AZURE_CLIENT_ID", self._CLIENT_ID)
monkeypatch.setenv("AZURE_TENANT_ID", self._TENANT_ID)
@@ -1623,6 +1715,293 @@ class TestAzureProviderCertificateAuth:
assert kwargs["tenant_id"] == self._TENANT_ID
assert kwargs["client_id"] == self._CLIENT_ID
def test_setup_session_env_cert_auth_reads_certificate_path(
self, monkeypatch, tmp_path
):
certificate_path = tmp_path / "prowler-cert.pem"
certificate_path.write_bytes(b"environment certificate bundle")
monkeypatch.setenv("AZURE_CLIENT_ID", self._CLIENT_ID)
monkeypatch.setenv("AZURE_TENANT_ID", self._TENANT_ID)
expected_credentials = MagicMock()
with patch(
"prowler.providers.azure.azure_provider._build_certificate_credential",
return_value=expected_credentials,
) as build_certificate_credential:
credentials = AzureProvider.setup_session(
az_cli_auth=False,
sp_env_auth=False,
browser_auth=False,
managed_identity_auth=False,
certificate_auth=True,
certificate_path=str(certificate_path),
tenant_id=None,
azure_credentials=None,
region_config=self._region_config(),
)
assert credentials is expected_credentials
build_certificate_credential.assert_called_once_with(
tenant_id=self._TENANT_ID,
client_id=self._CLIENT_ID,
certificate_data=b"environment certificate bundle",
authority=None,
)
def test_setup_session_env_cert_auth_preserves_missing_variable_error(self):
expected_error = AzureEnvironmentVariableError(
file="azure_provider.py",
message="Missing certificate credentials.",
)
with (
patch.object(
AzureProvider,
"check_certificate_creds_env_vars",
side_effect=expected_error,
),
pytest.raises(AzureEnvironmentVariableError) as exception,
):
AzureProvider.setup_session(
az_cli_auth=False,
sp_env_auth=False,
browser_auth=False,
managed_identity_auth=False,
certificate_auth=True,
certificate_path=None,
tenant_id=None,
azure_credentials=None,
region_config=self._region_config(),
)
assert exception.value is expected_error
def test_setup_session_env_cert_auth_translates_authentication_error(
self, monkeypatch
):
monkeypatch.setenv("AZURE_CLIENT_ID", self._CLIENT_ID)
monkeypatch.setenv("AZURE_TENANT_ID", self._TENANT_ID)
monkeypatch.setenv("AZURE_CERTIFICATE_CONTENT", self._CERT_CONTENT_B64)
with (
patch(
"prowler.providers.azure.azure_provider._build_certificate_credential",
side_effect=ClientAuthenticationError("invalid certificate"),
),
pytest.raises(AzureSetUpSessionError, match="client authentication"),
):
AzureProvider.setup_session(
az_cli_auth=False,
sp_env_auth=False,
browser_auth=False,
managed_identity_auth=False,
certificate_auth=True,
certificate_path=None,
tenant_id=None,
azure_credentials=None,
region_config=self._region_config(),
)
@pytest.mark.parametrize(
("tenant_id", "client_id", "error_description", "expected_error"),
[
(
"invalid-tenant",
_CLIENT_ID,
"Tenant 'invalid-tenant' was not found",
AzureNotValidTenantIdError,
),
(
_TENANT_ID,
"invalid-client",
"Application with identifier 'invalid-client' was not found",
AzureNotValidClientIdError,
),
(
_TENANT_ID,
_CLIENT_ID,
"Invalid client secret provided",
AzureNotValidClientSecretError,
),
],
)
def test_verify_client_translates_token_endpoint_errors(
self, tenant_id, client_id, error_description, expected_error
):
with (
patch("prowler.providers.azure.azure_provider.requests.post") as post,
pytest.raises(expected_error),
):
post.return_value.json.return_value = {
"error_codes": [700016],
"error_description": error_description,
}
AzureProvider.verify_client(
tenant_id,
client_id,
"invalid-secret",
self._region_config(),
)
def test_verify_client_certificate_content_acquires_graph_token(self):
import base64
with patch(
"prowler.providers.azure.azure_provider.CertificateCredential"
) as certificate_credential:
AzureProvider.verify_client(
self._TENANT_ID,
self._CLIENT_ID,
client_secret=None,
region_config=self._region_config(),
certificate_content=self._CERT_CONTENT_B64,
)
certificate_credential.assert_called_once_with(
client_id=self._CLIENT_ID,
tenant_id=self._TENANT_ID,
certificate_data=base64.b64decode(self._CERT_CONTENT_B64),
authority=None,
)
certificate_credential.return_value.get_token.assert_called_once_with(
"https://graph.microsoft.com/.default"
)
def test_verify_client_certificate_path_reads_bundle(self, tmp_path):
certificate_path = tmp_path / "prowler-cert.pem"
certificate_path.write_bytes(b"certificate bundle")
with patch(
"prowler.providers.azure.azure_provider.CertificateCredential"
) as certificate_credential:
AzureProvider.verify_client(
self._TENANT_ID,
self._CLIENT_ID,
client_secret=None,
region_config=self._region_config(),
certificate_path=str(certificate_path),
)
assert (
certificate_credential.call_args.kwargs["certificate_data"]
== b"certificate bundle"
)
certificate_credential.return_value.get_token.assert_called_once_with(
"https://graph.microsoft.com/.default"
)
@pytest.mark.parametrize("credential_source", ["content", "path"])
def test_verify_client_translates_certificate_authentication_error(
self, credential_source, tmp_path
):
certificate_content = None
certificate_path = None
expected_error = AzureNotValidCertificateContentError
if credential_source == "content":
certificate_content = self._CERT_CONTENT_B64
else:
path = tmp_path / "prowler-cert.pem"
path.write_bytes(b"certificate bundle")
certificate_path = str(path)
expected_error = AzureNotValidCertificatePathError
with (
patch(
"prowler.providers.azure.azure_provider.CertificateCredential",
side_effect=ClientAuthenticationError("invalid certificate"),
),
pytest.raises(expected_error),
):
AzureProvider.verify_client(
self._TENANT_ID,
self._CLIENT_ID,
client_secret=None,
region_config=self._region_config(),
certificate_content=certificate_content,
certificate_path=certificate_path,
)
def test_verify_client_without_credential_material_returns(self):
assert (
AzureProvider.verify_client(
self._TENANT_ID,
self._CLIENT_ID,
client_secret=None,
region_config=self._region_config(),
)
is None
)
def test_certificate_identity_and_printed_credentials_include_thumbprint(self):
from cryptography.hazmat.primitives import hashes, serialization
from prowler.providers.azure.azure_provider import (
_build_certificate_credential,
)
certificate, private_key = self._certificate_and_key()
bundle = certificate.public_bytes(
serialization.Encoding.PEM
) + private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption(),
)
expected_thumbprint = certificate.fingerprint(hashes.SHA1()).hex().upper()
credentials = _build_certificate_credential(
tenant_id=self._TENANT_ID,
client_id=self._CLIENT_ID,
certificate_data=bundle,
authority=None,
)
with patch.object(AzureProvider, "__init__", return_value=None):
provider = AzureProvider()
provider._session = credentials
provider._region_config = self._region_config()
graph_client = MagicMock()
graph_client.domains.get = AsyncMock(return_value=MagicMock(value=[]))
subscription_client = MagicMock()
subscription_client.subscriptions.list.return_value = [
MagicMock(display_name="Subscription", subscription_id="subscription-id")
]
subscription_client.tenants.list.return_value = [
MagicMock(tenant_id=self._TENANT_ID)
]
with (
patch(
"prowler.providers.azure.azure_provider.GraphServiceClient",
return_value=graph_client,
),
patch(
"prowler.providers.azure.azure_provider.SubscriptionClient",
return_value=subscription_client,
),
):
identity = provider.setup_identity(
az_cli_auth=False,
sp_env_auth=False,
browser_auth=False,
managed_identity_auth=False,
certificate_auth=False,
subscription_ids=[],
client_id=self._CLIENT_ID,
)
assert identity.identity_id == self._CLIENT_ID
assert identity.identity_type == "Service Principal with Certificate"
assert identity.certificate_thumbprint == expected_thumbprint
provider._identity = identity
provider._resource_groups = {}
with patch("prowler.providers.azure.azure_provider.print_boxes") as print_boxes:
provider.print_credentials()
report_lines = print_boxes.call_args.args[0]
assert any(expected_thumbprint in line for line in report_lines)
def test_setup_session_attaches_computed_thumbprint(self):
# Regression guard for the thumbprint handoff added in PROWLER-2378.
# The earlier `test_setup_session_static_credentials_cert_content_...`
@@ -1813,6 +2192,25 @@ class TestAzureProviderCertificateThumbprint:
assert _compute_certificate_thumbprint(b"definitely not a cert") is None
def test_returns_none_for_key_only_pkcs12(self):
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.serialization import pkcs12
from prowler.providers.azure.azure_provider import (
_compute_certificate_thumbprint,
)
_, _, key, _ = self._self_signed_pem_and_thumbprint()
key_only_pkcs12 = pkcs12.serialize_key_and_certificates(
name=b"prowler",
key=key,
cert=None,
cas=None,
encryption_algorithm=serialization.NoEncryption(),
)
assert _compute_certificate_thumbprint(key_only_pkcs12) is None
def serialization_no_encryption():
# Kept out of the test method so the PKCS#12 test reads cleanly. Wraps
@@ -69,7 +69,7 @@ describe("AzureCertificateCredentialsForm", () => {
href: "https://docs.prowler.com/user-guide/providers/azure/authentication#certificate-authentication",
},
{
element: screen.getByRole("link", { name: "New App Registration" }),
element: screen.getByRole("link", { name: "Open Azure" }),
href: "https://portal.azure.com/#view/Microsoft_AAD_RegisteredApps/CreateApplicationBlade",
},
{
@@ -103,22 +103,22 @@ describe("AzureCertificateCredentialsForm", () => {
const steps = screen.getAllByRole("listitem");
expect(steps).toHaveLength(6);
expect(steps[0]).toHaveTextContent(
"1. Create an App Registration. Copy its Directory (tenant) ID and Application (client) ID.",
"Register an Azure application, then copy its Directory (tenant) ID and Application (client) ID.",
);
expect(steps[1]).toHaveTextContent(
"2. Generate the certificate bundle and upload the public prowler-cert.cer file under Certificates & secrets. The private bundle is filled below and is submitted to Prowler only when you connect.",
"Generate the certificate. The private bundle is filled below and submitted to Prowler only when you connect.",
);
expect(steps[2]).toHaveTextContent(
"3. Add Microsoft Graph application permissions: AuditLog.Read.All, Directory.Read.All (or Domain.Read.All), and Policy.Read.All. Then grant admin consent.",
"Upload prowler-cert.cer under Certificates & secrets. Add Microsoft Graph application permissions: AuditLog.Read.All, Directory.Read.All (or Domain.Read.All), and Policy.Read.All. Then grant admin consent.",
);
expect(steps[3]).toHaveTextContent(
"4. Copy the Service Principal Object ID from Enterprise applications.",
"Copy the Service Principal Object ID from Enterprise applications.",
);
expect(steps[4]).toHaveTextContent(
"5. Deploy subscription RBAC with that Service Principal Object ID. The template creates ProwlerRole and assigns Reader and ProwlerRole; it does not create Entra resources.",
"Deploy subscription RBAC with the Service Principal Object ID. The template creates ProwlerRole and assigns Reader and ProwlerRole; it does not create Entra resources.",
);
expect(steps[5]).toHaveTextContent(
"6. Return to Prowler and connect. Paste the Tenant ID and Application Client ID below; the generated certificate bundle is already filled.",
"Return to Prowler and complete the fields below. The generated certificate bundle is already filled.",
);
expect(
screen.getByRole("textbox", {