mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-10-09 21:14:22 +00:00
fix(azure): validate certificate key pairs
- Reject malformed, key-only, and mismatched certificate bundles - Support matching PEM and unencrypted PKCS#12 credentials - Cover SDK and API validation paths
This commit is contained in:
8 files changed
+420
-20
No files matched your search
@@ -1 +1 @@
|
||||
Accept an optional `certificate_content` field on the Azure provider secret and enforce mutual exclusion with `client_secret` in the serializer, matching the M365 provider secret shape
|
||||
`certificate_content` support for Azure provider secrets, with mutual exclusion against `client_secret` and certificate/private-key bundle validation
|
||||
@@ -208,9 +208,37 @@ class TestAzureProviderSecret:
|
||||
"client_id": "87654321-4321-4321-4321-210987654321",
|
||||
"tenant_id": "12345678-1234-1234-1234-123456789012",
|
||||
}
|
||||
# Valid base64 of a tiny DER-shaped payload; the serializer only checks
|
||||
# that the string decodes as base64, not that it parses as a real cert.
|
||||
CERT_CONTENT_B64 = "MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA"
|
||||
|
||||
@staticmethod
|
||||
def certificate_bundle():
|
||||
import base64
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
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(UTC))
|
||||
.not_valid_after(datetime.now(UTC) + timedelta(days=1))
|
||||
.sign(private_key, hashes.SHA256())
|
||||
)
|
||||
bundle = certificate.public_bytes(
|
||||
serialization.Encoding.PEM
|
||||
) + private_key.private_bytes(
|
||||
encoding=serialization.Encoding.PEM,
|
||||
format=serialization.PrivateFormat.PKCS8,
|
||||
encryption_algorithm=serialization.NoEncryption(),
|
||||
)
|
||||
return base64.b64encode(bundle).decode("ascii")
|
||||
|
||||
def test_accepts_client_secret_only(self):
|
||||
# Backwards-compatibility guard: rows saved by the previous serializer
|
||||
@@ -223,11 +251,12 @@ class TestAzureProviderSecret:
|
||||
assert "certificate_content" not in serializer.validated_data
|
||||
|
||||
def test_accepts_certificate_content_only(self):
|
||||
certificate_content = self.certificate_bundle()
|
||||
serializer = AzureProviderSecret(
|
||||
data={**self.BASE, "certificate_content": self.CERT_CONTENT_B64}
|
||||
data={**self.BASE, "certificate_content": certificate_content}
|
||||
)
|
||||
assert serializer.is_valid(), serializer.errors
|
||||
assert serializer.validated_data["certificate_content"] == self.CERT_CONTENT_B64
|
||||
assert serializer.validated_data["certificate_content"] == certificate_content
|
||||
assert "client_secret" not in serializer.validated_data
|
||||
|
||||
def test_rejects_both_client_secret_and_certificate_content(self):
|
||||
@@ -238,7 +267,7 @@ class TestAzureProviderSecret:
|
||||
data={
|
||||
**self.BASE,
|
||||
"client_secret": "fake-client-secret",
|
||||
"certificate_content": self.CERT_CONTENT_B64,
|
||||
"certificate_content": self.certificate_bundle(),
|
||||
}
|
||||
)
|
||||
assert not serializer.is_valid()
|
||||
@@ -259,6 +288,41 @@ class TestAzureProviderSecret:
|
||||
assert not serializer.is_valid()
|
||||
assert "certificate_content" in serializer.errors
|
||||
|
||||
def test_rejects_invalid_tenant_and_client_ids(self):
|
||||
serializer = AzureProviderSecret(
|
||||
data={
|
||||
"tenant_id": "not-a-uuid",
|
||||
"client_id": "also-not-a-uuid",
|
||||
"client_secret": "fake-client-secret",
|
||||
}
|
||||
)
|
||||
|
||||
assert not serializer.is_valid()
|
||||
assert "tenant_id" in serializer.errors
|
||||
assert "client_id" in serializer.errors
|
||||
|
||||
def test_rejects_key_only_certificate_content(self):
|
||||
import base64
|
||||
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import rsa
|
||||
|
||||
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
||||
key_only_pem = private_key.private_bytes(
|
||||
encoding=serialization.Encoding.PEM,
|
||||
format=serialization.PrivateFormat.PKCS8,
|
||||
encryption_algorithm=serialization.NoEncryption(),
|
||||
)
|
||||
serializer = AzureProviderSecret(
|
||||
data={
|
||||
**self.BASE,
|
||||
"certificate_content": base64.b64encode(key_only_pem).decode("ascii"),
|
||||
}
|
||||
)
|
||||
|
||||
assert not serializer.is_valid()
|
||||
assert "certificate_content" in serializer.errors
|
||||
|
||||
def test_rejects_empty_strings_for_both(self):
|
||||
# DRF's CharField rejects "" at field-level before `validate()` runs.
|
||||
# The errors surface per-field rather than as non_field_errors, but
|
||||
|
||||
@@ -2,6 +2,7 @@ import base64
|
||||
import json
|
||||
import logging
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from uuid import UUID
|
||||
|
||||
import yaml
|
||||
from api.celery_utils import decode_celery_field
|
||||
@@ -60,7 +61,10 @@ from api.v1.serializer_utils.lighthouse import (
|
||||
)
|
||||
from api.v1.serializer_utils.processors import ProcessorConfigField
|
||||
from api.v1.serializer_utils.providers import ProviderSecretField
|
||||
from api.validators import validate_lighthouse_openai_compatible_base_url
|
||||
from api.validators import (
|
||||
validate_certificate_bundle,
|
||||
validate_lighthouse_openai_compatible_base_url,
|
||||
)
|
||||
from config.custom_logging import BackendLogger
|
||||
from django.conf import settings
|
||||
from django.contrib.auth import authenticate
|
||||
@@ -1791,6 +1795,24 @@ class AzureProviderSecret(serializers.Serializer):
|
||||
tenant_id = serializers.CharField()
|
||||
certificate_content = serializers.CharField(required=False)
|
||||
|
||||
def validate_client_id(self, client_id):
|
||||
try:
|
||||
UUID(client_id)
|
||||
except (TypeError, ValueError) as error:
|
||||
raise serializers.ValidationError(
|
||||
"Client ID must be a valid UUID."
|
||||
) from error
|
||||
return client_id
|
||||
|
||||
def validate_tenant_id(self, tenant_id):
|
||||
try:
|
||||
UUID(tenant_id)
|
||||
except (TypeError, ValueError) as error:
|
||||
raise serializers.ValidationError(
|
||||
"Tenant ID must be a valid UUID."
|
||||
) from error
|
||||
return tenant_id
|
||||
|
||||
def validate(self, attrs):
|
||||
if attrs.get("client_secret") and attrs.get("certificate_content"):
|
||||
raise serializers.ValidationError(
|
||||
@@ -1803,10 +1825,11 @@ class AzureProviderSecret(serializers.Serializer):
|
||||
return super().validate(attrs)
|
||||
|
||||
def validate_certificate_content(self, certificate_content):
|
||||
"""Validate that Azure certificate content is valid base64 encoded data."""
|
||||
"""Validate the Azure certificate and matching private-key bundle."""
|
||||
if certificate_content:
|
||||
try:
|
||||
base64.b64decode(certificate_content, validate=True)
|
||||
certificate_data = base64.b64decode(certificate_content, validate=True)
|
||||
validate_certificate_bundle(certificate_data)
|
||||
except Exception as e:
|
||||
# Field validators are invoked per-field; DRF already knows
|
||||
# this error belongs to `certificate_content` and will nest
|
||||
@@ -1814,9 +1837,9 @@ class AzureProviderSecret(serializers.Serializer):
|
||||
# double-nest the JSON:API pointer as
|
||||
# `/certificate_content/certificate_content`.
|
||||
raise serializers.ValidationError(
|
||||
f"The provided certificate content is not valid base64 encoded data: {str(e)}",
|
||||
"Certificate content must be valid base64 containing an X.509 certificate and its matching private key.",
|
||||
code="azure-certificate-content",
|
||||
)
|
||||
) from e
|
||||
return certificate_content
|
||||
|
||||
class Meta:
|
||||
|
||||
@@ -1,8 +1,13 @@
|
||||
import ipaddress
|
||||
import re
|
||||
import socket
|
||||
import string
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from cryptography import x509
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
from cryptography.hazmat.primitives.serialization import pkcs12
|
||||
from django.conf import settings
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.utils.translation import gettext as _
|
||||
@@ -20,6 +25,45 @@ LIGHTHOUSE_BLOCKED_METADATA_HOSTS = frozenset(
|
||||
)
|
||||
|
||||
|
||||
def validate_certificate_bundle(certificate_data: bytes) -> None:
|
||||
"""Validate that certificate data contains a matching certificate and key."""
|
||||
try:
|
||||
private_key, certificate, _ = pkcs12.load_key_and_certificates(
|
||||
certificate_data, None, default_backend()
|
||||
)
|
||||
except ValueError:
|
||||
certificate_match = re.search(
|
||||
rb"-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----",
|
||||
certificate_data,
|
||||
re.DOTALL,
|
||||
)
|
||||
private_key_match = re.search(
|
||||
rb"-----BEGIN (?:RSA |EC )?PRIVATE KEY-----.*?-----END (?:RSA |EC )?PRIVATE KEY-----",
|
||||
certificate_data,
|
||||
re.DOTALL,
|
||||
)
|
||||
if not certificate_match or not private_key_match:
|
||||
raise ValueError(
|
||||
"the payload must contain a certificate and its private key"
|
||||
)
|
||||
certificate = x509.load_pem_x509_certificate(
|
||||
certificate_match.group(), default_backend()
|
||||
)
|
||||
private_key = serialization.load_pem_private_key(
|
||||
private_key_match.group(), password=None, backend=default_backend()
|
||||
)
|
||||
|
||||
if certificate is None or private_key is None:
|
||||
raise ValueError("the payload must contain a certificate and its private key")
|
||||
|
||||
encoding = serialization.Encoding.DER
|
||||
public_format = serialization.PublicFormat.SubjectPublicKeyInfo
|
||||
if certificate.public_key().public_bytes(
|
||||
encoding, public_format
|
||||
) != private_key.public_key().public_bytes(encoding, public_format):
|
||||
raise ValueError("the certificate does not match the private key")
|
||||
|
||||
|
||||
def _normalize_hostname(hostname: str) -> str:
|
||||
return hostname.rstrip(".").lower()
|
||||
|
||||
|
||||
@@ -1 +1 @@
|
||||
Add certificate-based service principal authentication to the Azure provider. `AzureProvider` now accepts a `keyCredentials` certificate — as base64 content or file path — and authenticates via `azure.identity.CertificateCredential`, mirroring the existing M365 provider flow
|
||||
Certificate-based Service Principal authentication for Azure using a base64-encoded or file-based certificate/private-key bundle
|
||||
@@ -63,6 +63,7 @@ from prowler.providers.azure.exceptions.exceptions import (
|
||||
AzureTenantIDNoBrowserAuthError,
|
||||
)
|
||||
from prowler.providers.azure.lib.arguments.arguments import validate_azure_region
|
||||
from prowler.providers.azure.lib.certificate import validate_certificate_bundle
|
||||
from prowler.providers.azure.lib.mutelist.mutelist import AzureMutelist
|
||||
from prowler.providers.azure.lib.regions.regions import get_regions_config
|
||||
from prowler.providers.azure.models import AzureIdentityInfo, AzureRegionConfig
|
||||
@@ -217,8 +218,8 @@ class AzureProvider(Provider):
|
||||
client_id (str): The Azure client ID.
|
||||
client_secret (str): The Azure client secret.
|
||||
certificate_auth (bool): Flag indicating whether to use certificate authentication with environment variables (AZURE_TENANT_ID, AZURE_CLIENT_ID, AZURE_CERTIFICATE_CONTENT).
|
||||
certificate_content (str): Base64-encoded certificate content bound to the app registration's `keyCredentials`.
|
||||
certificate_path (str): Path to a certificate file bound to the app registration's `keyCredentials`.
|
||||
certificate_content (str): Base64-encoded certificate and private key bundle matching the App Registration certificate.
|
||||
certificate_path (str): Path to a certificate and private key bundle matching the App Registration certificate.
|
||||
resource_groups (list): List of resource group names.
|
||||
|
||||
Returns:
|
||||
@@ -1460,7 +1461,7 @@ class AzureProvider(Provider):
|
||||
AzureNotValidClientIdError: If the provided Azure Client ID is not valid.
|
||||
AzureNotValidClientSecretError: If the provided Azure Client Secret is not valid.
|
||||
AzureNotValidCertificateContentError: If the provided base64 certificate content is not valid.
|
||||
AzureNotValidCertificatePathError: If the provided certificate path cannot be read.
|
||||
AzureNotValidCertificatePathError: If the provided certificate path cannot be read or does not contain a valid certificate/private-key bundle.
|
||||
AzureClientIdAndClientSecretNotBelongingToTenantIdError: If the provided Azure Client ID and Client Secret do not belong to the specified Tenant ID.
|
||||
AzureTenantIdAndClientSecretNotBelongingToClientIdError: If the provided Azure Tenant ID and Client Secret do not belong to the specified Client ID.
|
||||
AzureTenantIdAndClientIdNotBelongingToClientSecretError: If the provided Azure Tenant ID and Client ID do not belong to the specified Client Secret.
|
||||
@@ -1507,21 +1508,22 @@ class AzureProvider(Provider):
|
||||
# azure-identity: `CertificateCredential` raises an opaque
|
||||
# exception several call frames deeper if this fails, which
|
||||
# makes for a bad UX in the API/UI.
|
||||
base64.b64decode(certificate_content, validate=True)
|
||||
certificate_data = base64.b64decode(certificate_content, validate=True)
|
||||
validate_certificate_bundle(certificate_data)
|
||||
except Exception as e:
|
||||
raise AzureNotValidCertificateContentError(
|
||||
file=os.path.basename(__file__),
|
||||
message=f"The provided certificate content is not valid base64 encoded data: {str(e)}",
|
||||
message=f"The provided certificate content is not a valid base64-encoded certificate/private-key bundle: {str(e)}",
|
||||
)
|
||||
|
||||
if certificate_path:
|
||||
try:
|
||||
with open(certificate_path, "rb") as cert_file:
|
||||
cert_file.read()
|
||||
validate_certificate_bundle(cert_file.read())
|
||||
except Exception as e:
|
||||
raise AzureNotValidCertificatePathError(
|
||||
file=os.path.basename(__file__),
|
||||
message=f"The provided certificate path is not valid: {str(e)}",
|
||||
message=f"The provided certificate path does not contain a valid certificate/private-key bundle: {str(e)}",
|
||||
)
|
||||
|
||||
if region_config is None:
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
import re
|
||||
|
||||
from cryptography import x509
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
from cryptography.hazmat.primitives.serialization import pkcs12
|
||||
|
||||
|
||||
def validate_certificate_bundle(certificate_data: bytes) -> None:
|
||||
"""Validate that certificate data contains a matching certificate and key."""
|
||||
try:
|
||||
private_key, certificate, _ = pkcs12.load_key_and_certificates(
|
||||
certificate_data, None, default_backend()
|
||||
)
|
||||
except ValueError:
|
||||
certificate_match = re.search(
|
||||
rb"-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----",
|
||||
certificate_data,
|
||||
re.DOTALL,
|
||||
)
|
||||
private_key_match = re.search(
|
||||
rb"-----BEGIN (?:RSA |EC )?PRIVATE KEY-----.*?-----END (?:RSA |EC )?PRIVATE KEY-----",
|
||||
certificate_data,
|
||||
re.DOTALL,
|
||||
)
|
||||
if not certificate_match or not private_key_match:
|
||||
raise ValueError(
|
||||
"the payload must contain a certificate and its private key"
|
||||
)
|
||||
certificate = x509.load_pem_x509_certificate(
|
||||
certificate_match.group(), default_backend()
|
||||
)
|
||||
private_key = serialization.load_pem_private_key(
|
||||
private_key_match.group(), password=None, backend=default_backend()
|
||||
)
|
||||
|
||||
if certificate is None or private_key is None:
|
||||
raise ValueError("the payload must contain a certificate and its private key")
|
||||
|
||||
encoding = serialization.Encoding.DER
|
||||
public_format = serialization.PublicFormat.SubjectPublicKeyInfo
|
||||
if certificate.public_key().public_bytes(
|
||||
encoding, public_format
|
||||
) != private_key.public_key().public_bytes(encoding, public_format):
|
||||
raise ValueError("the certificate does not match the private key")
|
||||
@@ -1255,6 +1255,29 @@ class TestAzureProviderCertificateAuth:
|
||||
def _region_config(self):
|
||||
return AzureProvider.setup_region_config("AzureCloud")
|
||||
|
||||
@staticmethod
|
||||
def _certificate_and_key():
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
from cryptography import x509
|
||||
from cryptography.hazmat.primitives import hashes
|
||||
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(UTC))
|
||||
.not_valid_after(datetime.now(UTC) + timedelta(days=1))
|
||||
.sign(private_key, hashes.SHA256())
|
||||
)
|
||||
return certificate, private_key
|
||||
|
||||
def test_validate_arguments_rejects_client_secret_and_cert_together(self):
|
||||
with pytest.raises(AzureConfigCredentialsError) as exception:
|
||||
AzureProvider.validate_arguments(
|
||||
@@ -1334,6 +1357,77 @@ class TestAzureProviderCertificateAuth:
|
||||
region_config=self._region_config(),
|
||||
)
|
||||
|
||||
def test_validate_static_credentials_rejects_key_only_pem(self):
|
||||
import base64
|
||||
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import rsa
|
||||
|
||||
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
||||
key_only_pem = private_key.private_bytes(
|
||||
encoding=serialization.Encoding.PEM,
|
||||
format=serialization.PrivateFormat.PKCS8,
|
||||
encryption_algorithm=serialization.NoEncryption(),
|
||||
)
|
||||
|
||||
with (
|
||||
patch.object(AzureProvider, "verify_client"),
|
||||
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_pem).decode("ascii"),
|
||||
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
|
||||
|
||||
from cryptography import x509
|
||||
from cryptography.hazmat.primitives import hashes, serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import rsa
|
||||
from cryptography.x509.oid import NameOID
|
||||
|
||||
certificate_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
||||
different_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(certificate_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(certificate_key, hashes.SHA256())
|
||||
)
|
||||
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 (
|
||||
patch.object(AzureProvider, "verify_client"),
|
||||
pytest.raises(AzureNotValidCertificateContentError),
|
||||
):
|
||||
AzureProvider.validate_static_credentials(
|
||||
tenant_id=self._TENANT_ID,
|
||||
client_id=self._CLIENT_ID,
|
||||
client_secret=None,
|
||||
certificate_content=base64.b64encode(mismatched_bundle).decode("ascii"),
|
||||
certificate_path=None,
|
||||
region_config=self._region_config(),
|
||||
)
|
||||
|
||||
def test_validate_static_credentials_rejects_missing_cert_file(self):
|
||||
with pytest.raises(AzureNotValidCertificatePathError):
|
||||
AzureProvider.validate_static_credentials(
|
||||
@@ -1345,6 +1439,134 @@ class TestAzureProviderCertificateAuth:
|
||||
region_config=self._region_config(),
|
||||
)
|
||||
|
||||
def test_validate_static_credentials_rejects_invalid_cert_file(self, tmp_path):
|
||||
certificate_path = tmp_path / "invalid-certificate.pem"
|
||||
certificate_path.write_bytes(b"not a certificate bundle")
|
||||
|
||||
with (
|
||||
patch.object(AzureProvider, "verify_client"),
|
||||
pytest.raises(AzureNotValidCertificatePathError),
|
||||
):
|
||||
AzureProvider.validate_static_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(),
|
||||
)
|
||||
|
||||
def test_validate_static_credentials_rejects_key_only_cert_file(self, tmp_path):
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
|
||||
_, private_key = self._certificate_and_key()
|
||||
certificate_path = tmp_path / "key-only.pem"
|
||||
certificate_path.write_bytes(
|
||||
private_key.private_bytes(
|
||||
encoding=serialization.Encoding.PEM,
|
||||
format=serialization.PrivateFormat.PKCS8,
|
||||
encryption_algorithm=serialization.NoEncryption(),
|
||||
)
|
||||
)
|
||||
|
||||
with (
|
||||
patch.object(AzureProvider, "verify_client"),
|
||||
pytest.raises(AzureNotValidCertificatePathError),
|
||||
):
|
||||
AzureProvider.validate_static_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(),
|
||||
)
|
||||
|
||||
def test_validate_static_credentials_rejects_mismatched_cert_file(self, tmp_path):
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import rsa
|
||||
|
||||
certificate, _ = self._certificate_and_key()
|
||||
different_private_key = rsa.generate_private_key(
|
||||
public_exponent=65537, key_size=2048
|
||||
)
|
||||
certificate_path = tmp_path / "mismatched-certificate.pem"
|
||||
certificate_path.write_bytes(
|
||||
certificate.public_bytes(serialization.Encoding.PEM)
|
||||
+ different_private_key.private_bytes(
|
||||
encoding=serialization.Encoding.PEM,
|
||||
format=serialization.PrivateFormat.PKCS8,
|
||||
encryption_algorithm=serialization.NoEncryption(),
|
||||
)
|
||||
)
|
||||
|
||||
with (
|
||||
patch.object(AzureProvider, "verify_client"),
|
||||
pytest.raises(AzureNotValidCertificatePathError),
|
||||
):
|
||||
AzureProvider.validate_static_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(),
|
||||
)
|
||||
|
||||
def test_validate_static_credentials_accepts_valid_pem_cert_file(self, tmp_path):
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
|
||||
certificate, private_key = self._certificate_and_key()
|
||||
certificate_path = tmp_path / "certificate-bundle.pem"
|
||||
certificate_path.write_bytes(
|
||||
certificate.public_bytes(serialization.Encoding.PEM)
|
||||
+ private_key.private_bytes(
|
||||
encoding=serialization.Encoding.PEM,
|
||||
format=serialization.PrivateFormat.PKCS8,
|
||||
encryption_algorithm=serialization.NoEncryption(),
|
||||
)
|
||||
)
|
||||
|
||||
with patch.object(AzureProvider, "verify_client"):
|
||||
credentials = AzureProvider.validate_static_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["certificate_path"] == str(certificate_path)
|
||||
|
||||
def test_validate_static_credentials_accepts_valid_pkcs12_cert_file(self, tmp_path):
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
from cryptography.hazmat.primitives.serialization import pkcs12
|
||||
|
||||
certificate, private_key = self._certificate_and_key()
|
||||
certificate_path = tmp_path / "certificate-bundle.pfx"
|
||||
certificate_path.write_bytes(
|
||||
pkcs12.serialize_key_and_certificates(
|
||||
name=b"prowler",
|
||||
key=private_key,
|
||||
cert=certificate,
|
||||
cas=None,
|
||||
encryption_algorithm=serialization.NoEncryption(),
|
||||
)
|
||||
)
|
||||
|
||||
with patch.object(AzureProvider, "verify_client"):
|
||||
credentials = AzureProvider.validate_static_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["certificate_path"] == str(certificate_path)
|
||||
|
||||
def test_setup_session_static_credentials_cert_content_uses_certificate_credential(
|
||||
self,
|
||||
):
|
||||
|
||||
Reference in new issue
Block a user