fix(api): harden Azure certificate serializer and complete schema/tests

This commit is contained in:
Lydia Vilchez authored and GitHub committed 2026-08-31 18:22:57 +02:00
1 parent fcf06e148c
commit f67cf7c3e5
5 files changed
+213 -27

No files matched your search

+24 -16
View File
@@ -21316,8 +21316,9 @@ components:
in Azure AD.
certificate_content:
type: string
description: Base64-encoded PEM certificate and private key
content for certificate-based authentication.
description: Base64-encoded PEM or PKCS#12/PFX bundle containing the certificate
and its matching private key for certificate-based authentication.
maxLength: 68266
tenant_id:
type: string
description: The Azure tenant ID, representing the directory
@@ -21368,8 +21369,9 @@ components:
where the application is registered.
certificate_content:
type: string
description: The certificate content in base64 format for
certificate-based authentication.
description: Base64-encoded PEM or PKCS#12/PFX bundle containing the certificate
and its matching private key for certificate-based authentication.
maxLength: 68266
required:
- client_id
- tenant_id
@@ -23465,8 +23467,9 @@ components:
in Azure AD.
certificate_content:
type: string
description: Base64-encoded PEM certificate and private key content
for certificate-based authentication.
description: Base64-encoded PEM or PKCS#12/PFX bundle containing the certificate
and its matching private key for certificate-based authentication.
maxLength: 68266
tenant_id:
type: string
description: The Azure tenant ID, representing the directory where
@@ -23517,8 +23520,9 @@ components:
the application is registered.
certificate_content:
type: string
description: The certificate content in base64 format for certificate-based
authentication.
description: Base64-encoded PEM or PKCS#12/PFX bundle containing the certificate
and its matching private key for certificate-based authentication.
maxLength: 68266
required:
- client_id
- tenant_id
@@ -23930,8 +23934,9 @@ components:
in Azure AD.
certificate_content:
type: string
description: Base64-encoded PEM certificate and private key
content for certificate-based authentication.
description: Base64-encoded PEM or PKCS#12/PFX bundle containing the certificate
and its matching private key for certificate-based authentication.
maxLength: 68266
tenant_id:
type: string
description: The Azure tenant ID, representing the directory
@@ -23982,8 +23987,9 @@ components:
where the application is registered.
certificate_content:
type: string
description: The certificate content in base64 format for
certificate-based authentication.
description: Base64-encoded PEM or PKCS#12/PFX bundle containing the certificate
and its matching private key for certificate-based authentication.
maxLength: 68266
required:
- client_id
- tenant_id
@@ -24417,8 +24423,9 @@ components:
in Azure AD.
certificate_content:
type: string
description: Base64-encoded PEM certificate and private key content
for certificate-based authentication.
description: Base64-encoded PEM or PKCS#12/PFX bundle containing the certificate
and its matching private key for certificate-based authentication.
maxLength: 68266
tenant_id:
type: string
description: The Azure tenant ID, representing the directory where
@@ -24469,8 +24476,9 @@ components:
the application is registered.
certificate_content:
type: string
description: The certificate content in base64 format for certificate-based
authentication.
description: Base64-encoded PEM or PKCS#12/PFX bundle containing the certificate
and its matching private key for certificate-based authentication.
maxLength: 68266
required:
- client_id
- tenant_id
@@ -333,6 +333,98 @@ class TestAzureProviderSecret:
assert "client_secret" in serializer.errors
assert "certificate_content" in serializer.errors
def test_rejects_encrypted_pem_certificate_content(self):
# `load_pem_private_key(password=None)` raises TypeError for
# encrypted keys — the narrowed `except (binascii.Error, TypeError,
# ValueError)` in the serializer must catch it and surface the
# typed `azure-certificate-content` code rather than a 500.
import base64
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
certificate = self._self_signed_certificate()
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"),
)
bundle = (
certificate.public_bytes(serialization.Encoding.PEM) + encrypted_key_pem
)
serializer = AzureProviderSecret(
data={
**self.BASE,
"certificate_content": base64.b64encode(bundle).decode("ascii"),
}
)
assert not serializer.is_valid()
assert "certificate_content" in serializer.errors
assert (
serializer.errors["certificate_content"][0].code
== "azure-certificate-content"
)
def test_rejects_oversized_certificate_content(self):
# Payloads larger than the base64 cap must be rejected at the DRF
# field layer before base64 decoding or bundle parsing runs, so a
# multi-MB blob cannot exhaust API-worker memory.
from api.v1.serializers import _MAX_CERTIFICATE_CONTENT_LENGTH
serializer = AzureProviderSecret(
data={
**self.BASE,
"certificate_content": "A" * (_MAX_CERTIFICATE_CONTENT_LENGTH + 1),
}
)
assert not serializer.is_valid()
assert "certificate_content" in serializer.errors
def test_mutex_errors_carry_stable_codes(self):
# JSON:API clients key on `code`; without it they cannot tell the
# mutex ("both provided") apart from the required-material error
# ("neither provided") without string-matching the message.
both = AzureProviderSecret(
data={
**self.BASE,
"client_secret": "fake-client-secret",
"certificate_content": self.certificate_bundle(),
}
)
assert not both.is_valid()
assert both.errors["non_field_errors"][0].code == "azure-credential-mutex"
neither = AzureProviderSecret(data=self.BASE)
assert not neither.is_valid()
assert neither.errors["non_field_errors"][0].code == "azure-credential-required"
@staticmethod
def _self_signed_certificate():
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")])
return (
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())
)
class TestOracleCloudProviderSecret:
def valid_secret(self, **overrides):
+64 -1
View File
@@ -42,7 +42,7 @@ def test_certificate_bundle_rejects_key_only_pkcs12():
encryption_algorithm=serialization.NoEncryption(),
)
with pytest.raises(ValueError, match="certificate and its private key"):
with pytest.raises(ValueError, match="does not contain a certificate"):
validate_certificate_bundle(key_only_pkcs12)
@@ -63,6 +63,69 @@ def test_certificate_bundle_rejects_mismatched_pem_key():
validate_certificate_bundle(mismatched_bundle)
def test_certificate_bundle_rejects_encrypted_pem_key():
# `cryptography.load_pem_private_key(..., password=None)` raises
# TypeError for encrypted keys; the API relies on that specific type
# to route to `azure-certificate-content`, not a generic 500.
certificate, _ = _certificate_and_key()
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"),
)
encrypted_bundle = (
certificate.public_bytes(serialization.Encoding.PEM) + encrypted_key_pem
)
with pytest.raises(TypeError):
validate_certificate_bundle(encrypted_bundle)
def test_certificate_bundle_normalizes_multi_key_bundle_when_second_key_matches():
# 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 validator must find the key that actually pairs with the leaf
# instead of stopping at the first `-----BEGIN PRIVATE KEY-----`.
leaf_cert, leaf_key = _certificate_and_key()
_, unrelated_key = _certificate_and_key()
leaf_cert_pem = leaf_cert.public_bytes(serialization.Encoding.PEM)
unrelated_key_pem = unrelated_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption(),
)
leaf_key_pem = leaf_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption(),
)
bundle = leaf_cert_pem + unrelated_key_pem + leaf_key_pem
normalized = validate_certificate_bundle(bundle)
# The leaf still leads (azure-identity's thumbprint invariant) and the
# matching key is the one paired in the normalized output.
assert normalized.startswith(leaf_cert_pem)
assert leaf_key_pem in normalized
def test_certificate_bundle_rejects_oversized_payload():
# Legitimate PEM/PFX bundles are well under 10 KiB. Reject anything
# above the 50 KiB cap before base64 decoding + PKCS#12/PEM parsing
# allocate the doubled memory a multi-MB payload would need.
from prowler.providers.azure.lib.certificate import (
_MAX_CERTIFICATE_BUNDLE_BYTES,
)
oversized = b"\x00" * (_MAX_CERTIFICATE_BUNDLE_BYTES + 1)
with pytest.raises(ValueError, match="maximum bundle size"):
validate_certificate_bundle(oversized)
def test_lighthouse_base_url_rejects_http_scheme():
with pytest.raises(ValidationError, match="HTTPS"):
validate_lighthouse_openai_compatible_base_url(
@@ -107,7 +107,7 @@ from rest_framework_json_api import serializers
},
"certificate_content": {
"type": "string",
"description": "Base64-encoded PEM certificate and private key content for certificate-based authentication.",
"description": "Base64-encoded PEM or PKCS#12/PFX bundle containing the certificate and its matching private key for certificate-based authentication.",
},
"tenant_id": {
"type": "string",
@@ -169,7 +169,7 @@ from rest_framework_json_api import serializers
},
"certificate_content": {
"type": "string",
"description": "The certificate content in base64 format for certificate-based authentication.",
"description": "Base64-encoded PEM or PKCS#12/PFX bundle containing the certificate and its matching private key for certificate-based authentication.",
},
},
"required": [
+31 -8
View File
@@ -1,4 +1,5 @@
import base64
import binascii
import json
import logging
from datetime import UTC, datetime, timedelta
@@ -1788,20 +1789,31 @@ class AwsProviderSecret(serializers.Serializer):
resource_name = "provider-secrets"
# Base64 cap that matches the SDK's 50 KiB `_MAX_CERTIFICATE_BUNDLE_BYTES`
# limit on the decoded bundle. Rejects oversized payloads at the request
# layer so DRF never allocates the doubled memory that base64 decoding plus
# PKCS#12/PEM parsing would need for a multi-MB blob.
_MAX_CERTIFICATE_CONTENT_LENGTH = 68266
class AzureProviderSecret(serializers.Serializer):
client_id = serializers.CharField()
client_secret = serializers.CharField(required=False)
tenant_id = serializers.CharField()
certificate_content = serializers.CharField(required=False)
certificate_content = serializers.CharField(
required=False, max_length=_MAX_CERTIFICATE_CONTENT_LENGTH
)
def validate(self, attrs):
if attrs.get("client_secret") and attrs.get("certificate_content"):
raise serializers.ValidationError(
"You cannot provide both client_secret and certificate_content."
"You cannot provide both client_secret and certificate_content.",
code="azure-credential-mutex",
)
if not attrs.get("client_secret") and not attrs.get("certificate_content"):
raise serializers.ValidationError(
"You must provide either client_secret or certificate_content."
"You must provide either client_secret or certificate_content.",
code="azure-credential-required",
)
return super().validate(attrs)
@@ -1811,7 +1823,12 @@ class AzureProviderSecret(serializers.Serializer):
try:
certificate_data = base64.b64decode(certificate_content, validate=True)
validate_certificate_bundle(certificate_data)
except Exception as e:
# `binascii.Error` (bad base64), `TypeError` (encrypted PEM key)
# and `ValueError` (mismatched cert/key, oversized bundle,
# malformed bytes) are the failure modes `validate_certificate_bundle`
# and `base64.b64decode` surface. Anything else is a real bug
# and should propagate.
except (binascii.Error, TypeError, ValueError) as e:
logger.error(
f"{e.__class__.__name__}[{e.__traceback__.tb_lineno}]: {e}"
)
@@ -1836,16 +1853,20 @@ class M365ProviderSecret(serializers.Serializer):
tenant_id = serializers.CharField()
user = serializers.EmailField(required=False)
password = serializers.CharField(required=False)
certificate_content = serializers.CharField(required=False)
certificate_content = serializers.CharField(
required=False, max_length=_MAX_CERTIFICATE_CONTENT_LENGTH
)
def validate(self, attrs):
if attrs.get("client_secret") and attrs.get("certificate_content"):
raise serializers.ValidationError(
"You cannot provide both client_secret and certificate_content."
"You cannot provide both client_secret and certificate_content.",
code="m365-credential-mutex",
)
if not attrs.get("client_secret") and not attrs.get("certificate_content"):
raise serializers.ValidationError(
"You must provide either client_secret or certificate_content."
"You must provide either client_secret or certificate_content.",
code="m365-credential-required",
)
return super().validate(attrs)
@@ -1854,7 +1875,9 @@ class M365ProviderSecret(serializers.Serializer):
if certificate_content:
try:
base64.b64decode(certificate_content, validate=True)
except Exception as e:
# `base64.b64decode(validate=True)` raises `binascii.Error`; the
# legacy alias `ValueError` is preserved for older builds.
except (binascii.Error, ValueError) as e:
# DRF field validators are already keyed to `certificate_content`,
# so raising a dict here would double-nest the JSON:API pointer.
raise serializers.ValidationError(