Repository: getmoto/moto
Pull request #7950 (2024-08-08 11:27:19)

ACM-PCA: Fix x.509 certificate generation
This PR fixes ACM-PCA so that it generates valid set of certificates. The certificate parameters (x.509 extensions, etc.) now resemble AWS.

- `CreateCertificateAuthority` now obeys certificate subject attributes provided by the client (only country, state, organisation, organisational unit and common name)
- `GetCertificateAuthorityCsr` now produces a CSR with proper CA attributes and extensions
- `IssueCertificate` gains limited support for [certificate templates](https://docs.aws.amazon.com/privateca/latest/userguide/UsingTemplates.html) (only [RootCACertificate/V1](https://docs.aws.amazon.com/privateca/latest/userguide/UsingTemplates.html#RootCACertificate-V1) and [EndEntityCertificate/V1](https://docs.aws.amazon.com/privateca/latest/userguide/UsingTemplates.html#EndEntityCertificate-V1)). Accordingly, appropriate X.509 extensions are added to the issued certificate.
- `GetCertificateAuthorityCertificate` now raises if the CA is not in correct state (as opposed to returning empty values earlier)

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/getmoto/moto/pull/7950?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 94.36%. Comparing base [(`9289332`)](https://app.codecov.io/gh/getmoto/moto/commit/928933254546b5f143b16fa8bb774f0baf0beae7?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) to head [(`a7f004f`)](https://app.codecov.io/gh/getmoto/moto/commit/a7f004fa3480f08196808d1eb3d0d8b1c07a319e?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).

<details><summary>Additional details and impacted files</summary>


```diff
@@           Coverage Diff           @@
##           master    #7950   +/-   ##
=======================================
  Coverage   94.36%   94.36%           
=======================================
  Files        1120     1120           
  Lines       95980    96011   +31     
=======================================
+ Hits        90567    90597   +30     
- Misses       5413     5414    +1     
```

| [Flag](https://app.codecov.io/gh/getmoto/moto/pull/7950/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | Coverage Δ | |
|---|---|---|
| [servertests](https://app.codecov.io/gh/getmoto/moto/pull/7950/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `28.95% <0.00%> (-0.01%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/getmoto/moto/pull/7950/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `94.33% <100.00%> (+<0.01%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto#carryforward-flags-in-the-pull-request-comment) to find out more.

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/getmoto/moto/pull/7950?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).


Patch:
diff --git a/moto/acmpca/exceptions.py b/moto/acmpca/exceptions.py
index 9a584d67c..b17c9b245 100644
--- a/moto/acmpca/exceptions.py
+++ b/moto/acmpca/exceptions.py
@@ -16,3 +16,13 @@ class InvalidS3ObjectAclInCrlConfiguration(JsonRESTError):
             "InvalidS3ObjectAclInCrlConfiguration",
             f"Invalid value for parameter RevocationConfiguration.CrlConfiguration.S3ObjectAcl, value: {value}, valid values: ['PUBLIC_READ', 'BUCKET_OWNER_FULL_CONTROL']",
         )
+
+
+class InvalidStateException(JsonRESTError):
+    code = 400
+
+    def __init__(self, arn: str):
+        super().__init__(
+            "InvalidStateException",
+            f"The certificate authority {arn} is not in the correct state to have a certificate signing request.",
+        )
diff --git a/moto/acmpca/models.py b/moto/acmpca/models.py
index 950c73adb..02e0413f8 100644
--- a/moto/acmpca/models.py
+++ b/moto/acmpca/models.py
@@ -4,11 +4,10 @@ import base64
 import datetime
 from typing import Any, Dict, List, Optional, Tuple
 
-import cryptography.hazmat.primitives.asymmetric.rsa
-import cryptography.x509
+from cryptography import x509
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import hashes, serialization
-from cryptography.x509 import Certificate, NameOID, load_pem_x509_certificate
+from cryptography.hazmat.primitives.asymmetric import rsa
 
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
@@ -17,7 +16,11 @@ from moto.moto_api._internal import mock_random
 from moto.utilities.tagging_service import TaggingService
 from moto.utilities.utils import get_partition
 
-from .exceptions import InvalidS3ObjectAclInCrlConfiguration, ResourceNotFoundException
+from .exceptions import (
+    InvalidS3ObjectAclInCrlConfiguration,
+    InvalidStateException,
+    ResourceNotFoundException,
+)
 
 
 class CertificateAuthority(BaseModel):
@@ -46,84 +49,193 @@ class CertificateAuthority(BaseModel):
         self.usage_mode = "SHORT_LIVED_CERTIFICATE"
         self.security_standard = security_standard or "FIPS_140_2_LEVEL_3_OR_HIGHER"
 
-        self.common_name = self.certificate_authority_configuration.get(
-            "Subject", {}
-        ).get("CommonName", "Moto.org")
-        self.key = cryptography.hazmat.primitives.asymmetric.rsa.generate_private_key(
-            public_exponent=65537, key_size=2048
-        )
+        self.key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
         self.password = str(mock_random.uuid4()).encode("utf-8")
         self.private_bytes = self.key.private_bytes(
             encoding=serialization.Encoding.PEM,
             format=serialization.PrivateFormat.TraditionalOpenSSL,
             encryption_algorithm=serialization.BestAvailableEncryption(self.password),
         )
-        self.certificate: Optional[Certificate] = None
+        self.certificate: Optional[x509.Certificate] = None
         self.certificate_chain: Optional[bytes] = None
-        self.csr = self.generate_csr(self.common_name)
-
         self.issued_certificates: Dict[str, bytes] = dict()
 
-    def generate_cert(self, common_name: str, subject: cryptography.x509.Name) -> bytes:
-        issuer = cryptography.x509.Name(
-            [  # C = US, O = Amazon, OU = Server CA 1B, CN = Amazon
-                cryptography.x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
-                cryptography.x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Amazon"),
-                cryptography.x509.NameAttribute(
-                    NameOID.ORGANIZATIONAL_UNIT_NAME, "Server CA 1B"
-                ),
-                cryptography.x509.NameAttribute(NameOID.COMMON_NAME, common_name),
-            ]
-        )
-        cert = (
-            cryptography.x509.CertificateBuilder()
+        subject = self.certificate_authority_configuration.get("Subject", {})
+        name_attributes = []
+        if "Country" in subject:
+            name_attributes.append(
+                x509.NameAttribute(x509.NameOID.COUNTRY_NAME, subject["Country"])
+            )
+        if "State" in subject:
+            name_attributes.append(
+                x509.NameAttribute(
+                    x509.NameOID.STATE_OR_PROVINCE_NAME, subject["State"]
+                )
+            )
+        if "Organization" in subject:
+            name_attributes.append(
+                x509.NameAttribute(
+                    x509.NameOID.ORGANIZATION_NAME, subject["Organization"]
+                )
+            )
+        if "OrganizationalUnit" in subject:
+            name_attributes.append(
+                x509.NameAttribute(
+                    x509.NameOID.ORGANIZATIONAL_UNIT_NAME, subject["OrganizationalUnit"]
+                )
+            )
+        if "CommonName" in subject:
+            name_attributes.append(
+                x509.NameAttribute(x509.NameOID.COMMON_NAME, subject["CommonName"])
+            )
+        self.issuer = x509.Name(name_attributes)
+        self.csr = self._ca_csr(self.issuer)
+
+    def generate_cert(
+        self,
+        subject: x509.Name,
+        public_key: rsa.RSAPublicKey,
+        extensions: List[Tuple[x509.ExtensionType, bool]],
+    ) -> bytes:
+        builder = (
+            x509.CertificateBuilder()
             .subject_name(subject)
-            .issuer_name(issuer)
-            .public_key(self.key.public_key())
-            .serial_number(cryptography.x509.random_serial_number())
+            .issuer_name(self.issuer)
+            .public_key(public_key)
+            .serial_number(x509.random_serial_number())
             .not_valid_before(utcnow())
             .not_valid_after(utcnow() + datetime.timedelta(days=365))
-            .sign(self.key, hashes.SHA512(), default_backend())
         )
 
+        for extension, critical in extensions:
+            builder = builder.add_extension(extension, critical)
+
+        cert = builder.sign(self.key, hashes.SHA512(), default_backend())
+
         return cert.public_bytes(serialization.Encoding.PEM)
 
-    def generate_csr(self, common_name: str) -> bytes:
+    def _ca_csr(self, issuer: x509.Name) -> bytes:
         csr = (
-            cryptography.x509.CertificateSigningRequestBuilder()
-            .subject_name(
-                cryptography.x509.Name(
-                    [
-                        cryptography.x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
-                        cryptography.x509.NameAttribute(
-                            NameOID.STATE_OR_PROVINCE_NAME, "California"
-                        ),
-                        cryptography.x509.NameAttribute(
-                            NameOID.LOCALITY_NAME, "San Francisco"
-                        ),
-                        cryptography.x509.NameAttribute(
-                            NameOID.ORGANIZATION_NAME, "My Company"
-                        ),
-                        cryptography.x509.NameAttribute(
-                            NameOID.COMMON_NAME, common_name
-                        ),
-                    ]
-                )
+            x509.CertificateSigningRequestBuilder()
+            .subject_name(issuer)
+            .add_extension(
+                x509.BasicConstraints(ca=True, path_length=None),
+                critical=True,
             )
             .sign(self.key, hashes.SHA256())
         )
         return csr.public_bytes(serialization.Encoding.PEM)
 
-    def issue_certificate(self, csr_bytes: bytes) -> str:
-        cert = cryptography.x509.load_pem_x509_csr(base64.b64decode(csr_bytes))
+    def issue_certificate(self, csr_bytes: bytes, template_arn: Optional[str]) -> str:
+        csr = x509.load_pem_x509_csr(base64.b64decode(csr_bytes))
+        extensions = self._x509_extensions(csr, template_arn)
         new_cert = self.generate_cert(
-            common_name=self.common_name, subject=cert.subject
+            subject=csr.subject,
+            public_key=csr.public_key(),  # type: ignore[arg-type]
+            extensions=extensions,
         )
+
         cert_id = str(mock_random.uuid4()).replace("-", "")
         cert_arn = f"arn:{get_partition(self.region_name)}:acm-pca:{self.region_name}:{self.account_id}:certificate-authority/{self.id}/certificate/{cert_id}"
         self.issued_certificates[cert_arn] = new_cert
         return cert_arn
 
+    def _x509_extensions(
+        self, csr: x509.CertificateSigningRequest, template_arn: Optional[str]
+    ) -> List[Tuple[x509.ExtensionType, bool]]:
+        """
+        Uses a PCA certificate template ARN to return a list of X.509 extensions.
+        These extensions are part of the constructed certificate.
+
+        See https://docs.aws.amazon.com/privateca/latest/userguide/UsingTemplates.html
+        """
+        extensions = []
+
+        if template_arn == "arn:aws:acm-pca:::template/RootCACertificate/V1":
+            extensions.extend(
+                [
+                    (
+                        x509.BasicConstraints(ca=True, path_length=None),
+                        True,
+                    ),
+                    (
+                        x509.KeyUsage(
+                            crl_sign=True,
+                            key_cert_sign=True,
+                            digital_signature=True,
+                            content_commitment=False,
+                            key_encipherment=False,
+                            data_encipherment=False,
+                            key_agreement=False,
+                            encipher_only=False,
+                            decipher_only=False,
+                        ),
+                        True,
+                    ),
+                    (
+                        x509.SubjectKeyIdentifier.from_public_key(csr.public_key()),
+                        False,
+                    ),
+                ]
+            )
+
+        elif template_arn in (
+            "arn:aws:acm-pca:::template/EndEntityCertificate/V1",
+            None,
+        ):
+            extensions.extend(
+                [
+                    (
+                        x509.BasicConstraints(ca=False, path_length=None),
+                        True,
+                    ),
+                    (
+                        x509.AuthorityKeyIdentifier.from_issuer_public_key(
+                            self.key.public_key()
+                        ),
+                        False,
+                    ),
+                    (
+                        x509.SubjectKeyIdentifier.from_public_key(csr.public_key()),
+                        False,
+                    ),
+                    (
+                        x509.KeyUsage(
+                            crl_sign=False,
+                            key_cert_sign=False,
+                            digital_signature=True,
+                            content_commitment=False,
+                            key_encipherment=True,
+                            data_encipherment=False,
+                            key_agreement=False,
+                            encipher_only=False,
+                            decipher_only=False,
+                        ),
+                        True,
+                    ),
+                    (
+                        x509.ExtendedKeyUsage(
+                            [
+                                x509.ExtendedKeyUsageOID.SERVER_AUTH,
+                                x509.ExtendedKeyUsageOID.CLIENT_AUTH,
+                            ]
+                        ),
+                        False,
+                    ),
+                ]
+            )
+
+        cn = csr.subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME)
+        if cn:
+            extensions.append(
+                (
+                    x509.SubjectAlternativeName([x509.DNSName(cn[0].value)]),  # type: ignore[arg-type]
+                    False,
+                ),
+            )
+
+        return extensions
+
     def get_certificate(self, certificate_arn: str) -> bytes:
         return self.issued_certificates[certificate_arn]
 
@@ -171,7 +283,7 @@ class CertificateAuthority(BaseModel):
     def import_certificate_authority_certificate(
         self, certificate: bytes, certificate_chain: Optional[bytes]
     ) -> None:
-        self.certificate = load_pem_x509_certificate(certificate)
+        self.certificate = x509.load_pem_x509_certificate(certificate)
         self.certificate_chain = certificate_chain
         self.status = "ACTIVE"
         self.updated_at = unix_time()
@@ -243,6 +355,8 @@ class ACMPCABackend(BaseBackend):
         self, certificate_authority_arn: str
     ) -> Tuple[bytes, Optional[bytes]]:
         ca = self.describe_certificate_authority(certificate_authority_arn)
+        if ca.status != "ACTIVE":
+            raise InvalidStateException(certificate_authority_arn)
         return ca.certificate_bytes, ca.certificate_chain
 
     def get_certificate_authority_csr(self, certificate_authority_arn: str) -> bytes:
@@ -273,13 +387,15 @@ class ACMPCABackend(BaseBackend):
         ca = self.describe_certificate_authority(certificate_authority_arn)
         ca.status = "DELETED"
 
-    def issue_certificate(self, certificate_authority_arn: str, csr: bytes) -> str:
+    def issue_certificate(
+        self, certificate_authority_arn: str, csr: bytes, template_arn: Optional[str]
+    ) -> str:
         """
-        The following parameters are not yet implemented: ApiPassthrough, SigningAlgorithm, TemplateArn, Validity, ValidityNotBefore, IdempotencyToken
+        The following parameters are not yet implemented: ApiPassthrough, SigningAlgorithm, Validity, ValidityNotBefore, IdempotencyToken
         Some fields of the resulting certificate will have default values, instead of using the CSR
         """
         ca = self.describe_certificate_authority(certificate_authority_arn)
-        certificate_arn = ca.issue_certificate(csr)
+        certificate_arn = ca.issue_certificate(csr, template_arn)
         return certificate_arn
 
     def get_certificate(
diff --git a/moto/acmpca/responses.py b/moto/acmpca/responses.py
index d7f3c0f5b..899f23127 100644
--- a/moto/acmpca/responses.py
+++ b/moto/acmpca/responses.py
@@ -100,10 +100,12 @@ class ACMPCAResponse(BaseResponse):
     def issue_certificate(self) -> str:
         params = json.loads(self.body)
         certificate_authority_arn = params.get("CertificateAuthorityArn")
+        template_arn = params.get("TemplateArn")
         csr = params.get("Csr").encode("utf-8")
         certificate_arn = self.acmpca_backend.issue_certificate(
             certificate_authority_arn=certificate_authority_arn,
             csr=csr,
+            template_arn=template_arn,
         )
         return json.dumps(dict(CertificateArn=certificate_arn))
 
diff --git a/moto/kms/models.py b/moto/kms/models.py
index 43d78940f..aed7cec72 100644
--- a/moto/kms/models.py
+++ b/moto/kms/models.py
@@ -1,5 +1,6 @@
 import json
 import os
+import typing
 from collections import defaultdict
 from copy import copy
 from datetime import datetime, timedelta
@@ -78,6 +79,15 @@ class Key(CloudFormationModel):
         self.description = description or ""
         self.enabled = True
         self.multi_region = multi_region
+        if self.multi_region:
+            self.multi_region_configuration: typing.Dict[str, Any] = {
+                "MultiRegionKeyType": "PRIMARY",
+                "PrimaryKey": {
+                    "Arn": f"arn:{get_partition(region)}:kms:{region}:{account_id}:key/{self.id}",
+                    "Region": self.region,
+                },
+                "ReplicaKeys": [],
+            }
         self.key_rotation_status = False
         self.deletion_date: Optional[datetime] = None
         self.key_material = generate_master_key()
@@ -88,7 +98,6 @@ class Key(CloudFormationModel):
         self.arn = (
             f"arn:{get_partition(region)}:kms:{region}:{account_id}:key/{self.id}"
         )
-
         self.grants: Dict[str, Grant] = dict()
 
     def add_grant(
@@ -206,6 +215,10 @@ class Key(CloudFormationModel):
                 "SigningAlgorithms": self.signing_algorithms,
             }
         }
+        if key_dict["KeyMetadata"]["MultiRegion"]:
+            key_dict["KeyMetadata"]["MultiRegionConfiguration"] = (
+                self.multi_region_configuration
+            )
         if self.key_state == "PendingDeletion":
             key_dict["KeyMetadata"]["DeletionDate"] = unix_time(self.deletion_date)
         return key_dict
@@ -326,8 +339,24 @@ class KmsBackend(BaseBackend):
         replica_key = copy(self.keys[key_id])
         replica_key.region = replica_region
         replica_key.arn = replica_key.arn.replace(self.region_name, replica_region)
+
+        if replica_key.multi_region:
+            existing_replica = any(
+                replica["Region"] == replica_region
+                for replica in replica_key.multi_region_configuration["ReplicaKeys"]
+            )
+
+            if not existing_replica:
+                replica_payload = {"Arn": replica_key.arn, "Region": replica_region}
+                replica_key.multi_region_configuration["ReplicaKeys"].append(
+                    replica_payload
+                )
+
         to_region_backend = kms_backends[self.account_id][replica_region]
         to_region_backend.keys[replica_key.id] = replica_key
+
+        self.multi_region_configuration = copy(replica_key.multi_region_configuration)
+
         return replica_key
 
     def update_key_description(self, key_id: str, description: str) -> None:
@@ -348,6 +377,16 @@ class KmsBackend(BaseBackend):
         key_id = self.get_key_id(key_id)
         if r"alias/" in str(key_id).lower():
             key_id = self.get_key_id_from_alias(key_id)  # type: ignore[assignment]
+
+        key = self.keys[self.get_key_id(key_id)]
+
+        if key.multi_region:
+            if key.arn == key.multi_region_configuration["PrimaryKey"]["Arn"]:
+                return self.keys[self.get_key_id(key_id)]
+            else:
+                key.multi_region_configuration["MultiRegionKeyType"] = "REPLICA"
+                return key
+
         return self.keys[self.get_key_id(key_id)]
 
     def list_keys(self) -> Iterable[Key]:


Test patch:
diff --git a/tests/test_acmpca/test_acmpca.py b/tests/test_acmpca/test_acmpca.py
index bbcc4c071..a9c592cd5 100644
--- a/tests/test_acmpca/test_acmpca.py
+++ b/tests/test_acmpca/test_acmpca.py
@@ -7,7 +7,7 @@ import pytest
 from botocore.exceptions import ClientError
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import hashes, serialization
-from cryptography.x509 import NameOID
+from cryptography.x509 import DNSName, NameOID
 
 from moto import mock_aws
 from moto.core import DEFAULT_ACCOUNT_ID
@@ -121,10 +121,15 @@ def test_get_certificate_authority_certificate():
         IdempotencyToken="terraform-20221125230308947400000001",
     )["CertificateAuthorityArn"]
 
-    resp = client.get_certificate_authority_certificate(CertificateAuthorityArn=ca_arn)
-
-    # Certificate is empty for now,  until we call import_certificate_authority_certificate
-    assert resp["Certificate"] == ""
+    # CA is not usable yet
+    with pytest.raises(ClientError) as exc:
+        client.get_certificate_authority_certificate(CertificateAuthorityArn=ca_arn)
+    err = exc.value.response["Error"]
+    assert err["Code"] == "InvalidStateException"
+    assert (
+        err["Message"]
+        == f"The certificate authority {ca_arn} is not in the correct state to have a certificate signing request."
+    )
 
 
 @mock_aws
@@ -134,15 +139,52 @@ def test_get_certificate_authority_csr():
         CertificateAuthorityConfiguration={
             "KeyAlgorithm": "RSA_4096",
             "SigningAlgorithm": "SHA512WITHRSA",
-            "Subject": {"CommonName": "yscb41lw.test"},
+            "Subject": {
+                "CommonName": "evilcorp.com",
+                "Country": "US",
+                "State": "MA",
+                "Organization": "Evil Corp",
+                "OrganizationalUnit": "Finance",
+            },
         },
-        CertificateAuthorityType="SUBORDINATE",
+        CertificateAuthorityType="ROOT",
         IdempotencyToken="terraform-20221125230308947400000001",
     )["CertificateAuthorityArn"]
 
     resp = client.get_certificate_authority_csr(CertificateAuthorityArn=ca_arn)
 
     assert "Csr" in resp
+    csr = cryptography.x509.load_pem_x509_csr(resp["Csr"].encode("utf-8"))
+    assert (
+        csr.subject.get_attributes_for_oid(cryptography.x509.NameOID.COMMON_NAME)[
+            0
+        ].value
+        == "evilcorp.com"
+    )
+    assert (
+        csr.subject.get_attributes_for_oid(cryptography.x509.NameOID.COUNTRY_NAME)[
+            0
+        ].value
+        == "US"
+    )
+    assert (
+        csr.subject.get_attributes_for_oid(
+            cryptography.x509.NameOID.STATE_OR_PROVINCE_NAME
+        )[0].value
+        == "MA"
+    )
+    assert (
+        csr.subject.get_attributes_for_oid(cryptography.x509.NameOID.ORGANIZATION_NAME)[
+            0
+        ].value
+        == "Evil Corp"
+    )
+    assert (
+        csr.subject.get_attributes_for_oid(
+            cryptography.x509.NameOID.ORGANIZATIONAL_UNIT_NAME
+        )[0].value
+        == "Finance"
+    )
 
 
 @mock_aws
@@ -364,6 +406,67 @@ def test_import_certificate_authority_certificate():
     assert "-----BEGIN CERTIFICATE-----" in resp["Certificate"]
 
 
+@mock_aws
+def test_end_entity_certificate_issuance():
+    client = boto3.client("acm-pca", region_name="us-east-2")
+
+    # Create a CA
+    ca_arn = client.create_certificate_authority(
+        CertificateAuthorityConfiguration={
+            "KeyAlgorithm": "RSA_4096",
+            "SigningAlgorithm": "SHA512WITHRSA",
+            "Subject": {"CommonName": "evilcorp.com"},
+        },
+        CertificateAuthorityType="ROOT",
+    )["CertificateAuthorityArn"]
+
+    # Create CA certificate
+    csr = client.get_certificate_authority_csr(CertificateAuthorityArn=ca_arn)["Csr"]
+    certificate_arn = client.issue_certificate(
+        CertificateAuthorityArn=ca_arn,
+        Csr=csr,
+        SigningAlgorithm="SHA256WITHRSA",
+        TemplateArn="arn:aws:acm-pca:::template/RootCACertificate/V1",
+        Validity={"Type": "YEARS", "Value": 10},
+    )["CertificateArn"]
+
+    resp = client.get_certificate(
+        CertificateAuthorityArn=ca_arn, CertificateArn=certificate_arn
+    )
+    ca_cert = resp["Certificate"]
+
+    client.import_certificate_authority_certificate(
+        CertificateAuthorityArn=ca_arn,
+        Certificate=ca_cert,
+    )
+
+    # Create a end-entity certificate and sign it by the CA
+    private_key = cryptography.hazmat.primitives.asymmetric.rsa.generate_private_key(
+        public_exponent=65537, key_size=2048
+    )
+    ee_csr = create_csr(private_key, "US", "WA", "BezosCorp", "bezoscorp.com")
+    ee_cert_arn = client.issue_certificate(
+        CertificateAuthorityArn=ca_arn,
+        Csr=ee_csr,
+        SigningAlgorithm="SHA256WITHRSA",
+        Validity={"Type": "YEARS", "Value": 1},
+    )["CertificateArn"]
+    ee_cert = client.get_certificate(
+        CertificateAuthorityArn=ca_arn, CertificateArn=ee_cert_arn
+    )["Certificate"]
+
+    # Ensure that this is a valid single-hierarchy CA and all certificates are valid
+    store = cryptography.x509.verification.Store(
+        [cryptography.x509.load_pem_x509_certificate(ca_cert.encode("utf-8"))]
+    )
+    builder = cryptography.x509.verification.PolicyBuilder().store(store)
+    verifier = builder.build_server_verifier(DNSName("bezoscorp.com"))
+    chain = verifier.verify(
+        cryptography.x509.load_pem_x509_certificate(ee_cert.encode("utf-8")), []
+    )
+    assert len(chain) == 2
+
+
 @mock_aws
 def test_tag_certificate_authority():
     client = boto3.client("acm-pca", region_name="eu-west-1")
@@ -410,6 +513,36 @@ def test_untag_certificate_authority():
     assert resp["Tags"] == [{"Key": "t2", "Value": "v2"}]
 
 
+def create_csr(private_key, country, state, org, cn):
+    csr = (
+        cryptography.x509.CertificateSigningRequestBuilder()
+        .subject_name(
+            cryptography.x509.Name(
+                [
+                    cryptography.x509.NameAttribute(
+                        cryptography.x509.NameOID.COUNTRY_NAME, country
+                    ),
+                    cryptography.x509.NameAttribute(
+                        cryptography.x509.NameOID.STATE_OR_PROVINCE_NAME, state
+                    ),
+                    cryptography.x509.NameAttribute(
+                        cryptography.x509.NameOID.ORGANIZATION_NAME, org
+                    ),
+                    cryptography.x509.NameAttribute(
+                        cryptography.x509.NameOID.COMMON_NAME, cn
+                    ),
+                ]
+            )
+        )
+        .add_extension(
+            cryptography.x509.BasicConstraints(ca=False, path_length=None),
+            critical=True,
+        )
+        .sign(private_key, hashes.SHA256())
+    )
+    return csr.public_bytes(serialization.Encoding.PEM)
+
+
 def create_cert():
     serial_number = cryptography.x509.random_serial_number()
     subject = cryptography.x509.Name(
diff --git a/tests/test_kms/test_kms_boto3.py b/tests/test_kms/test_kms_boto3.py
index 9e4c69179..455ef368b 100644
--- a/tests/test_kms/test_kms_boto3.py
+++ b/tests/test_kms/test_kms_boto3.py
@@ -134,6 +134,86 @@ def test_create_multi_region_key():
     assert key["KeyMetadata"]["MultiRegion"] is True
 
 
+@mock_aws
+def test_create_multi_region_configuration_key():
+    region_to_replicate_from = "us-east-1"
+    region_to_replicate_to = "us-west-1"
+    region_to_replicate_to_second_region = "us-west-2"
+
+    from_region_client = boto3.client("kms", region_name=region_to_replicate_from)
+    to_region_client_us_west_1 = boto3.client("kms", region_name=region_to_replicate_to)
+    to_region_client_us_west_2 = boto3.client(
+        "kms", region_name=region_to_replicate_to_second_region
+    )
+
+    response = from_region_client.create_key(
+        Policy="my policy",
+        Description="my key",
+        KeyUsage="ENCRYPT_DECRYPT",
+        MultiRegion=True,
+        Tags=[{"TagKey": "project", "TagValue": "moto"}],
+    )
+    key_id = response["KeyMetadata"]["KeyId"]
+
+    first_replica_response = from_region_client.replicate_key(
+        KeyId=key_id, ReplicaRegion=region_to_replicate_to
+    )
+
+    second_replica_response = from_region_client.replicate_key(
+        KeyId=key_id, ReplicaRegion=region_to_replicate_to_second_region
+    )
+
+    third_replica_response = from_region_client.replicate_key(
+        KeyId=key_id, ReplicaRegion=region_to_replicate_to_second_region
+    )
+
+    replica_key_id = first_replica_response["ReplicaKeyMetadata"]["KeyId"]
+    replica_key_second_id = second_replica_response["ReplicaKeyMetadata"]["KeyId"]
+    replica_key_third_id = third_replica_response["ReplicaKeyMetadata"]["KeyId"]
+
+    from_region_describe_key = from_region_client.describe_key(KeyId=key_id)
+    to_region_describe_key = to_region_client_us_west_1.describe_key(
+        KeyId=replica_key_id
+    )
+    to_region_describe_key_second_replica = to_region_client_us_west_2.describe_key(
+        KeyId=replica_key_second_id
+    )
+    to_region_describe_key_third_replica = to_region_client_us_west_2.describe_key(
+        KeyId=replica_key_third_id
+    )
+
+    assert from_region_describe_key["KeyMetadata"]["MultiRegion"] is True
+    assert to_region_describe_key["KeyMetadata"]["MultiRegion"] is True
+    assert to_region_describe_key_second_replica["KeyMetadata"]["MultiRegion"] is True
+    assert (
+        from_region_describe_key["KeyMetadata"]["MultiRegionConfiguration"][
+            "MultiRegionKeyType"
+        ]
+        == "PRIMARY"
+    )
+    assert (
+        to_region_describe_key["KeyMetadata"]["MultiRegionConfiguration"][
+            "MultiRegionKeyType"
+        ]
+        == "REPLICA"
+    )
+    assert (
+        to_region_describe_key_second_replica["KeyMetadata"][
+            "MultiRegionConfiguration"
+        ]["MultiRegionKeyType"]
+        == "REPLICA"
+    )
+    # This test below tests the expected failure of making a replica in an already existing region.
+    assert (
+        len(
+            to_region_describe_key_third_replica["KeyMetadata"][
+                "MultiRegionConfiguration"
+            ]["ReplicaKeys"]
+        )
+        == 2
+    )
+
+
 @mock_aws
 def test_non_multi_region_keys_should_not_have_multi_region_properties():
     conn = boto3.client("kms", region_name="us-east-1")