feat(aws): add acmpca service and pqc key algorithm check (#11318)

Co-authored-by: Hugo P.Brito <hugopbrit@gmail.com>
2026-07-23 12:31:54 +00:00 · 2026-06-22 18:13:17 +02:00
parent 10d9fc35e6
commit bdd44a0dce
12 changed files with 427 additions and 0 deletions
@@ -439,6 +439,11 @@ mainConfig:
    elbv2_min_azs: 2

    # AWS Post-Quantum TLS Configuration
+    # aws.acmpca_certificate_authority_pqc_key_algorithm
+    acmpca_pqc_key_algorithms:
+      - "ML_DSA_44"
+      - "ML_DSA_65"
+      - "ML_DSA_87"
    # aws.cloudfront_distributions_pqc_tls_enabled
    cloudfront_pqc_min_protocol_versions:
      - "TLSv1.3_2025"
@@ -13,11 +13,13 @@ Additionally, you can input a custom configuration file using the `--config-file
 ## AWS

 ### Configurable Checks
+
 The following list includes all the AWS checks with configurable variables that can be changed in the configuration yaml file:

 | Check Name                                                    | Value                                            | Type            |
 |---------------------------------------------------------------|--------------------------------------------------|-----------------|
 | `acm_certificates_expiration_check`                           | `days_to_expire_threshold`                       | Integer         |
+| `acmpca_certificate_authority_pqc_key_algorithm`              | `acmpca_pqc_key_algorithms`                      | List of Strings |
 | `appstream_fleet_maximum_session_duration`                    | `max_session_duration_seconds`                   | Integer         |
 | `appstream_fleet_session_disconnect_timeout`                  | `max_disconnect_timeout_in_seconds`              | Integer         |
 | `appstream_fleet_session_idle_disconnect_timeout`             | `max_idle_disconnect_timeout_in_seconds`         | Integer         |
@@ -47,6 +47,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - `cloudfront_distributions_pqc_tls_enabled` check for AWS provider to verify CloudFront distributions enforce a post-quantum TLS 1.3 security policy [(#11317)](https://github.com/prowler-cloud/prowler/pull/11317)
 - `apigateway_domain_name_pqc_tls_enabled` check for AWS provider to verify API Gateway custom domain names use a post-quantum TLS security policy [(#11316)](https://github.com/prowler-cloud/prowler/pull/11316)
 - `transfer_server_pqc_ssh_kex_enabled` check for AWS provider to verify Transfer Family servers use a post-quantum hybrid SSH key exchange security policy [(#11315)](https://github.com/prowler-cloud/prowler/pull/11315)
+- `acmpca_certificate_authority_pqc_key_algorithm` check and new `acmpca` service for AWS provider to verify AWS Private CA certificate authorities use a post-quantum (ML-DSA) key algorithm [(#11318)](https://github.com/prowler-cloud/prowler/pull/11318)

 ### 🔄 Changed

@@ -381,6 +381,12 @@ aws:
  elbv2_min_azs: 2

  # AWS Post-Quantum TLS Configuration
+  # aws.acmpca_certificate_authority_pqc_key_algorithm
+  # Allowed post-quantum key algorithms for AWS Private CA certificate authorities
+  acmpca_pqc_key_algorithms:
+    - "ML_DSA_44"
+    - "ML_DSA_65"
+    - "ML_DSA_87"
  # aws.cloudfront_distributions_pqc_tls_enabled
  # Allowed CloudFront MinimumProtocolVersion values that enable post-quantum hybrid key exchange
  cloudfront_pqc_min_protocol_versions:
@@ -0,0 +1,41 @@
+{
+  "Provider": "aws",
+  "CheckID": "acmpca_certificate_authority_pqc_key_algorithm",
+  "CheckTitle": "AWS Private CA certificate authorities use a post-quantum (ML-DSA) key algorithm",
+  "CheckType": [
+    "Software and Configuration Checks/AWS Security Best Practices"
+  ],
+  "ServiceName": "acmpca",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "low",
+  "ResourceType": "AwsAcmPcaCertificateAuthority",
+  "ResourceGroup": "security",
+  "Description": "**AWS Private Certificate Authorities (Private CAs)** are assessed for use of a **post-quantum digital signature key algorithm** (`ML_DSA_44`, `ML_DSA_65`, `ML_DSA_87`). CAs that still issue certificates with RSA or ECC algorithms produce signatures vulnerable to forgery once a cryptographically relevant quantum computer is available.",
+  "Risk": "RSA and ECC signatures can be broken by Shor's algorithm on a sufficiently large quantum computer. A compromised CA private key would let an attacker issue arbitrary certificates trusted across the PKI, undermining identity and code-signing controls. Migrating CAs to **ML-DSA** (NIST FIPS 204) provides quantum-resistant signatures so issued certificates retain integrity in the post-quantum era.",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://docs.aws.amazon.com/privateca/latest/userguide/PcaTerms.html",
+    "https://aws.amazon.com/about-aws/whats-new/2025/11/aws-private-ca-post-quantum-digital-certificates/",
+    "https://aws.amazon.com/blogs/security/post-quantum-ml-dsa-code-signing-with-aws-private-ca-and-aws-kms/",
+    "https://csrc.nist.gov/pubs/fips/204/final"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "aws acm-pca create-certificate-authority --certificate-authority-configuration '{\"KeyAlgorithm\":\"ML_DSA_65\",\"SigningAlgorithm\":\"ML_DSA_65\",\"Subject\":{...}}' --certificate-authority-type SUBORDINATE",
+      "NativeIaC": "```yaml\nResources:\n  <example_resource_name>:\n    Type: AWS::ACMPCA::CertificateAuthority\n    Properties:\n      Type: SUBORDINATE\n      KeyAlgorithm: ML_DSA_65        # FIX: post-quantum signature algorithm\n      SigningAlgorithm: ML_DSA_65\n      Subject:\n        CommonName: example-pqc-ca\n```",
+      "Other": "Existing CAs cannot have their key algorithm changed; create a new CA with KeyAlgorithm = ML_DSA_44 / ML_DSA_65 / ML_DSA_87, re-issue certificates from it, and decommission the legacy CA once dependent workloads have rotated.",
+      "Terraform": "```hcl\nresource \"aws_acmpca_certificate_authority\" \"<example_resource_name>\" {\n  type = \"SUBORDINATE\"\n  certificate_authority_configuration {\n    key_algorithm     = \"ML_DSA_65\" # FIX: post-quantum signature algorithm\n    signing_algorithm = \"ML_DSA_65\"\n    subject {\n      common_name = \"example-pqc-ca\"\n    }\n  }\n}\n```"
+    },
+    "Recommendation": {
+      "Text": "Create new Private CAs with a **post-quantum key algorithm** (`ML_DSA_44`, `ML_DSA_65`, or `ML_DSA_87`) and migrate workloads off legacy RSA/ECC CAs. Plan crypto-agility for your PKI so that quantum-resistant trust anchors can be rolled out before threat actors gain access to a cryptographically relevant quantum computer.",
+      "Url": "https://hub.prowler.com/check/acmpca_certificate_authority_pqc_key_algorithm"
+    }
+  },
+  "Categories": [
+    "encryption"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -0,0 +1,49 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.acmpca.acmpca_client import acmpca_client
+
+PQC_PCA_KEY_ALGORITHMS_DEFAULT = [
+    "ML_DSA_44",
+    "ML_DSA_65",
+    "ML_DSA_87",
+]
+
+
+class acmpca_certificate_authority_pqc_key_algorithm(Check):
+    """Verify that every AWS Private CA uses a post-quantum key algorithm.
+
+    A Private CA PASSES when its ``KeyAlgorithm`` belongs to the configured
+    allowlist of post-quantum signature algorithms (ML-DSA family).
+    Deleted CAs are skipped.
+    """
+
+    def execute(self) -> list[Check_Report_AWS]:
+        """Execute the check against AWS Private CAs.
+
+        Returns:
+            A list of reports with each non-deleted CA PQC key algorithm status.
+        """
+
+        findings = []
+        pqc_algorithms = acmpca_client.audit_config.get(
+            "acmpca_pqc_key_algorithms", PQC_PCA_KEY_ALGORITHMS_DEFAULT
+        )
+        for ca in acmpca_client.certificate_authorities.values():
+            if ca.status == "DELETED":
+                continue
+            report = Check_Report_AWS(metadata=self.metadata(), resource=ca)
+            algorithm = ca.key_algorithm or "<none>"
+            if ca.key_algorithm in pqc_algorithms:
+                report.status = "PASS"
+                report.status_extended = (
+                    f"AWS Private CA {ca.id} uses post-quantum key algorithm "
+                    f"{algorithm}."
+                )
+            else:
+                report.status = "FAIL"
+                report.status_extended = (
+                    f"AWS Private CA {ca.id} uses key algorithm {algorithm}, "
+                    "which is not post-quantum (ML-DSA)."
+                )
+            findings.append(report)
+
+        return findings
@@ -0,0 +1,4 @@
+from prowler.providers.aws.services.acmpca.acmpca_service import ACMPCA
+from prowler.providers.common.provider import Provider
+
+acmpca_client = ACMPCA(Provider.get_global_provider())
@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+from typing import Any, Dict, List
+
+from pydantic.v1 import BaseModel, Field
+
+from prowler.lib.logger import logger
+from prowler.lib.scan_filters.scan_filters import is_resource_filtered
+from prowler.providers.aws.aws_provider import AwsProvider
+from prowler.providers.aws.lib.service.service import AWSService
+
+
+class ACMPCA(AWSService):
+    """AWS Private CA service class to list certificate authorities."""
+
+    def __init__(self, provider: AwsProvider) -> None:
+        """Initialize the AWS Private CA service.
+
+        Args:
+            provider: AWS provider instance with session and audit context.
+        """
+
+        # The boto3 client identifier for AWS Private CA is "acm-pca"
+        super().__init__("acm-pca", provider)
+        self.certificate_authorities: dict[str, CertificateAuthority] = {}
+        self.__threading_call__(self._list_certificate_authorities)
+
+    def _list_certificate_authorities(self, regional_client: Any) -> None:
+        """List AWS Private CAs and their tags in a region.
+
+        Args:
+            regional_client: Regional AWS Private CA client.
+        """
+
+        logger.info("ACM PCA - Listing Certificate Authorities...")
+        try:
+            paginator = regional_client.get_paginator("list_certificate_authorities")
+            for page in paginator.paginate():
+                for ca in page.get("CertificateAuthorities", []):
+                    arn = ca.get("Arn", "")
+                    if not arn:
+                        continue
+                    if self.audit_resources and not is_resource_filtered(
+                        arn, self.audit_resources
+                    ):
+                        continue
+                    config = ca.get("CertificateAuthorityConfiguration", {})
+                    tags = []
+                    try:
+                        tags = regional_client.list_tags(
+                            CertificateAuthorityArn=arn
+                        ).get("Tags", [])
+                    except Exception as error:
+                        logger.error(
+                            f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                        )
+                    self.certificate_authorities[arn] = CertificateAuthority(
+                        arn=arn,
+                        id=arn.split("/")[-1],
+                        region=regional_client.region,
+                        status=ca.get("Status", ""),
+                        type=ca.get("Type", ""),
+                        usage_mode=ca.get("UsageMode", ""),
+                        key_algorithm=config.get("KeyAlgorithm", ""),
+                        signing_algorithm=config.get("SigningAlgorithm", ""),
+                        tags=tags,
+                    )
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+
+class CertificateAuthority(BaseModel):
+    """AWS Private Certificate Authority metadata.
+
+    Attributes:
+        arn: Certificate authority ARN.
+        id: Certificate authority identifier.
+        region: AWS region where the certificate authority exists.
+        status: Certificate authority lifecycle status.
+        type: Certificate authority type.
+        usage_mode: Certificate authority usage mode.
+        key_algorithm: Key algorithm configured for the certificate authority.
+        signing_algorithm: Signing algorithm configured for the certificate authority.
+        tags: Tags attached to the certificate authority.
+    """
+
+    arn: str
+    id: str
+    region: str
+    status: str = ""
+    type: str = ""
+    usage_mode: str = ""
+    key_algorithm: str = ""
+    signing_algorithm: str = ""
+    tags: List[Dict[str, str]] = Field(default_factory=list)
@@ -0,0 +1,157 @@
+from unittest import mock
+
+from prowler.providers.aws.services.acmpca.acmpca_service import CertificateAuthority
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+CA_ID = "12345678-1234-1234-1234-123456789012"
+CA_ARN = f"arn:aws:acm-pca:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:certificate-authority/{CA_ID}"
+
+
+def _build_client(certificate_authorities, audit_config=None):
+    acmpca_client = mock.MagicMock()
+    acmpca_client.certificate_authorities = certificate_authorities
+    acmpca_client.audit_config = audit_config or {}
+    return acmpca_client
+
+
+def _ca(key_algorithm: str, status: str = "ACTIVE"):
+    return CertificateAuthority(
+        arn=CA_ARN,
+        id=CA_ID,
+        region=AWS_REGION_US_EAST_1,
+        status=status,
+        type="SUBORDINATE",
+        usage_mode="GENERAL_PURPOSE",
+        key_algorithm=key_algorithm,
+        signing_algorithm="ML_DSA_65" if "ML_DSA" in key_algorithm else "SHA256WITHRSA",
+    )
+
+
+class Test_acmpca_certificate_authority_pqc_key_algorithm:
+    def test_no_cas(self):
+        acmpca_client = _build_client({})
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.acmpca.acmpca_certificate_authority_pqc_key_algorithm.acmpca_certificate_authority_pqc_key_algorithm.acmpca_client",
+                new=acmpca_client,
+            ),
+        ):
+            from prowler.providers.aws.services.acmpca.acmpca_certificate_authority_pqc_key_algorithm.acmpca_certificate_authority_pqc_key_algorithm import (
+                acmpca_certificate_authority_pqc_key_algorithm,
+            )
+
+            check = acmpca_certificate_authority_pqc_key_algorithm()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_ml_dsa_65(self):
+        acmpca_client = _build_client({CA_ARN: _ca("ML_DSA_65")})
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.acmpca.acmpca_certificate_authority_pqc_key_algorithm.acmpca_certificate_authority_pqc_key_algorithm.acmpca_client",
+                new=acmpca_client,
+            ),
+        ):
+            from prowler.providers.aws.services.acmpca.acmpca_certificate_authority_pqc_key_algorithm.acmpca_certificate_authority_pqc_key_algorithm import (
+                acmpca_certificate_authority_pqc_key_algorithm,
+            )
+
+            check = acmpca_certificate_authority_pqc_key_algorithm()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert "ML_DSA_65" in result[0].status_extended
+            assert result[0].resource_id == CA_ID
+            assert result[0].resource_arn == CA_ARN
+
+    def test_rsa_2048_fails(self):
+        acmpca_client = _build_client({CA_ARN: _ca("RSA_2048")})
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.acmpca.acmpca_certificate_authority_pqc_key_algorithm.acmpca_certificate_authority_pqc_key_algorithm.acmpca_client",
+                new=acmpca_client,
+            ),
+        ):
+            from prowler.providers.aws.services.acmpca.acmpca_certificate_authority_pqc_key_algorithm.acmpca_certificate_authority_pqc_key_algorithm import (
+                acmpca_certificate_authority_pqc_key_algorithm,
+            )
+
+            check = acmpca_certificate_authority_pqc_key_algorithm()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert "RSA_2048" in result[0].status_extended
+
+    def test_deleted_ca_skipped(self):
+        acmpca_client = _build_client({CA_ARN: _ca("RSA_2048", status="DELETED")})
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.acmpca.acmpca_certificate_authority_pqc_key_algorithm.acmpca_certificate_authority_pqc_key_algorithm.acmpca_client",
+                new=acmpca_client,
+            ),
+        ):
+            from prowler.providers.aws.services.acmpca.acmpca_certificate_authority_pqc_key_algorithm.acmpca_certificate_authority_pqc_key_algorithm import (
+                acmpca_certificate_authority_pqc_key_algorithm,
+            )
+
+            check = acmpca_certificate_authority_pqc_key_algorithm()
+            result = check.execute()
+
+            assert len(result) == 0
+
+    def test_configurable_allowlist(self):
+        acmpca_client = _build_client(
+            {CA_ARN: _ca("RSA_2048")},
+            audit_config={"acmpca_pqc_key_algorithms": ["ML_DSA_65", "RSA_2048"]},
+        )
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.acmpca.acmpca_certificate_authority_pqc_key_algorithm.acmpca_certificate_authority_pqc_key_algorithm.acmpca_client",
+                new=acmpca_client,
+            ),
+        ):
+            from prowler.providers.aws.services.acmpca.acmpca_certificate_authority_pqc_key_algorithm.acmpca_certificate_authority_pqc_key_algorithm import (
+                acmpca_certificate_authority_pqc_key_algorithm,
+            )
+
+            check = acmpca_certificate_authority_pqc_key_algorithm()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
@@ -0,0 +1,65 @@
+from unittest.mock import patch
+
+import botocore
+from moto import mock_aws
+
+from prowler.providers.aws.services.acmpca.acmpca_service import (
+    ACMPCA,
+    CertificateAuthority,
+)
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+CA_ID = "12345678-1234-1234-1234-123456789012"
+CA_ARN = f"arn:aws:acm-pca:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:certificate-authority/{CA_ID}"
+
+make_api_call = botocore.client.BaseClient._make_api_call
+
+
+def mock_make_api_call(self, operation_name, kwarg):
+    if operation_name == "ListCertificateAuthorities":
+        return {
+            "CertificateAuthorities": [
+                {
+                    "Arn": CA_ARN,
+                    "Status": "ACTIVE",
+                    "Type": "SUBORDINATE",
+                    "UsageMode": "GENERAL_PURPOSE",
+                    "CertificateAuthorityConfiguration": {
+                        "KeyAlgorithm": "ML_DSA_65",
+                        "SigningAlgorithm": "ML_DSA_65",
+                    },
+                }
+            ]
+        }
+    if operation_name == "ListTags":
+        assert kwarg["CertificateAuthorityArn"] == CA_ARN
+        return {"Tags": [{"Key": "Environment", "Value": "test"}]}
+    return make_api_call(self, operation_name, kwarg)
+
+
+class Test_ACMPCA_Service:
+    @mock_aws
+    def test_service(self):
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+        acmpca = ACMPCA(aws_provider)
+        assert acmpca.service == "acm-pca"
+
+    @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
+    @mock_aws
+    def test_list_certificate_authorities(self):
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+        acmpca = ACMPCA(aws_provider)
+        assert len(acmpca.certificate_authorities) == 1
+        ca = acmpca.certificate_authorities[CA_ARN]
+        assert isinstance(ca, CertificateAuthority)
+        assert ca.id == CA_ID
+        assert ca.region == AWS_REGION_US_EAST_1
+        assert ca.status == "ACTIVE"
+        assert ca.type == "SUBORDINATE"
+        assert ca.key_algorithm == "ML_DSA_65"
+        assert ca.signing_algorithm == "ML_DSA_65"
+        assert ca.tags == [{"Key": "Environment", "Value": "test"}]