Automated Certificate Management Environment (ACME) DNS Labeled With …

src/open_mpic_core/__init__.py

@@ -4,6 +4,8 @@
 from open_mpic_core.common_domain.enum.dns_record_type import DnsRecordType
 from open_mpic_core.common_domain.enum.url_scheme import UrlScheme
 
+from open_mpic_core.common_util.acme import Acme
+
 from open_mpic_core.common_domain.validation_error import MpicValidationError
 from open_mpic_core.common_domain.messages.ErrorMessages import ErrorMessages
 
@@ -15,6 +17,7 @@
     DcvDnsChangeValidationParameters,
     DcvDnsPersistentValidationParameters,
     DcvAcmeDns01ValidationParameters,
+    DcvAcmeDnsAccount01ValidationParameters,
     DcvContactPhoneTxtValidationParameters,
     DcvContactEmailCaaValidationParameters,
     DcvContactEmailTxtValidationParameters,

src/open_mpic_core/common_domain/check_parameters.py

@@ -4,7 +4,7 @@
 
 from pydantic import BaseModel, field_validator, Field, model_validator
 
-from open_mpic_core import CertificateType, DnsRecordType, DcvValidationMethod, UrlScheme
+from open_mpic_core import CertificateType, DnsRecordType, DcvValidationMethod, UrlScheme, Acme
 
 DNS_CHANGE_ALLOWED_RECORD_TYPES: Set[DnsRecordType] = {DnsRecordType.CNAME, DnsRecordType.TXT, DnsRecordType.CAA}
 IP_ADDRESS_ALLOWED_RECORD_TYPES: Set[DnsRecordType] = {DnsRecordType.A, DnsRecordType.AAAA}
@@ -151,6 +151,27 @@ class DcvAcmeDns01ValidationParameters(DcvValidationParameters):
     require_exact_case: Literal[True] = True  # ACME DNS-01 validation is always case-sensitive
 
 
+class DcvAcmeDnsAccount01ValidationParameters(DcvValidationParameters):
+    validation_method: Literal[DcvValidationMethod.DNS_ACCOUNT_01] = DcvValidationMethod.DNS_ACCOUNT_01
+    acme_account_url: str
+    key_authorization_hash: str
+    dns_record_type: Literal[DnsRecordType.TXT] = DnsRecordType.TXT
+    dns_name_prefix: str = ""
+    require_exact_case: Literal[True] = True
+
+    @field_validator("acme_account_url")
+    @classmethod
+    def validate_acme_account_url(cls, v: str) -> str:
+        if not isuri(v):
+            raise ValueError(f"acme_account_url must be a valid URI, got {v}")
+        return v
+
+    @model_validator(mode="after")
+    def set_dns_name_prefix(self) -> "DcvAcmeDnsAccount01ValidationParameters":
+        self.dns_name_prefix = Acme.dns_account_name_prefix_for(self.acme_account_url)
+        return self
+
+
 class DcvAcmeTlsAlpn01ValidationParameters(DcvValidationParameters):
     validation_method: Literal[DcvValidationMethod.ACME_TLS_ALPN_01] = DcvValidationMethod.ACME_TLS_ALPN_01
     key_authorization_hash: str
@@ -164,6 +185,7 @@ class DcvAcmeTlsAlpn01ValidationParameters(DcvValidationParameters):
         DcvDnsPersistentValidationParameters,
         DcvAcmeHttp01ValidationParameters,
         DcvAcmeDns01ValidationParameters,
+        DcvAcmeDnsAccount01ValidationParameters,
         DcvAcmeTlsAlpn01ValidationParameters,
         DcvContactEmailTxtValidationParameters,
         DcvContactEmailCaaValidationParameters,

src/open_mpic_core/common_domain/enum/dcv_validation_method.py

@@ -8,7 +8,7 @@ class DcvValidationMethod(StrEnum):
     ACME_HTTP_01 = "acme-http-01"  # CABF BRs 3.2.2.4.19 Agreed-Upon Change to Website - ACME
     ACME_DNS_01 = "acme-dns-01"  # TXT record
     ACME_TLS_ALPN_01 = "acme-tls-alpn-01"  # CABF BRs 3.2.2.4.20 TLS Using ALPN
-    DNS_ACCOUNT_01 = "dns-account-01"  # CABF BRs 3.2.2.4.21 DNS Labeled with Account ID - ACME TODO not yet implemented
+    DNS_ACCOUNT_01 = "dns-account-01"  # CABF BRs 3.2.2.4.21 DNS Labeled with Account ID - ACME
     CONTACT_EMAIL_CAA = "contact-email-caa"
     CONTACT_EMAIL_TXT = "contact-email-txt"
     CONTACT_PHONE_CAA = "contact-phone-caa"

src/open_mpic_core/common_util/acme.py

@@ -0,0 +1,20 @@
+import base64
+import hashlib
+
+
+class Acme:
+    ACME_CHALLENGE_LABEL = "_acme-challenge"
+
+    @staticmethod
+    def dns_account_label_from_url(acme_account_url: str) -> str:
+        """
+        Derive the per-account DNS label for dns-account-01 per draft-ietf-acme-dns-account-label:
+        base32(SHA-256(ACCOUNT_URL)[0:10]), lowercase, no padding.
+        """
+        digest = hashlib.sha256(acme_account_url.encode("utf-8")).digest()[:10]
+        return base64.b32encode(digest).decode("ascii").lower().rstrip("=")
+
+    @staticmethod
+    def dns_account_name_prefix_for(acme_account_url: str) -> str:
+        account_label = Acme.dns_account_label_from_url(acme_account_url)
+        return f"_{account_label}.{Acme.ACME_CHALLENGE_LABEL}"

src/open_mpic_core/mpic_dcv_checker/mpic_dcv_checker.py

@@ -590,6 +590,8 @@ def build_expected_dns_record_content(
     ) -> ExpectedDnsRecordContent:
         if validation_method == DcvValidationMethod.ACME_DNS_01:
             expected_content = ExpectedDnsRecordContent(expected_value=check_parameters.key_authorization_hash)
+        elif validation_method == DcvValidationMethod.DNS_ACCOUNT_01:
+            expected_content = ExpectedDnsRecordContent(expected_value=check_parameters.key_authorization_hash)
         elif validation_method == DcvValidationMethod.DNS_PERSISTENT:
             expected_content = ExpectedDnsRecordContent(
                 expected_value=None,  # validated via issuer_domains and account_uri

tests/unit/open_mpic_core/test_acme.py

@@ -0,0 +1,42 @@
+import pytest
+
+from open_mpic_core import Acme, DcvAcmeDnsAccount01ValidationParameters
+
+
+class TestAcme:
+    def dns_account_label_from_url__should_match_draft_test_vector(self):
+        account_url = "https://example.com/acme/acct/ExampleAccount"
+        assert Acme.dns_account_label_from_url(account_url) == "ujmmovf2vn55tgye"
+
+    @pytest.mark.parametrize(
+        "account_url",
+        [
+            "https://ca.example/acme/acct/123",
+            "http://authority.example/acme/acct/foo",
+        ],
+    )
+    def dns_account_label_from_url__should_return_lowercase_base32_without_padding(self, account_url):
+        label = Acme.dns_account_label_from_url(account_url)
+        assert label == label.lower()
+        assert "=" not in label
+        assert all(char in "abcdefghijklmnopqrstuvwxyz234567" for char in label)
+
+    def dns_account_name_prefix_for__should_build_acme_challenge_prefix(self):
+        account_url = "https://example.com/acme/acct/ExampleAccount"
+        assert Acme.dns_account_name_prefix_for(account_url) == "_ujmmovf2vn55tgye._acme-challenge"
+
+
+class TestDcvAcmeDnsAccount01ValidationParameters:
+    def dcv_acme_dns_account_01_parameters__should_derive_dns_name_prefix_from_account_url(self):
+        parameters = DcvAcmeDnsAccount01ValidationParameters(
+            acme_account_url="https://example.com/acme/acct/ExampleAccount",
+            key_authorization_hash="abc123",
+        )
+        assert parameters.dns_name_prefix == "_ujmmovf2vn55tgye._acme-challenge"
+
+    def dcv_acme_dns_account_01_parameters__should_reject_invalid_account_url(self):
+        with pytest.raises(ValueError, match="acme_account_url must be a valid URI"):
+            DcvAcmeDnsAccount01ValidationParameters(
+                acme_account_url="not-a-valid-uri",
+                key_authorization_hash="abc123",
+            )

tests/unit/open_mpic_core/test_check_request_parameters.py

@@ -7,6 +7,7 @@
     DcvDnsChangeValidationParameters,
     DcvDnsPersistentValidationParameters,
     DcvAcmeDns01ValidationParameters,
+    DcvAcmeDnsAccount01ValidationParameters,
     DcvContactPhoneTxtValidationParameters,
     DcvContactEmailCaaValidationParameters,
     DcvContactEmailTxtValidationParameters,
@@ -32,6 +33,8 @@ class TestCheckRequestDetails:
          DcvAcmeHttp01ValidationParameters),
         ('{"validation_method": "acme-dns-01", "key_authorization_hash": "test-ka"}',
          DcvAcmeDns01ValidationParameters),
+        ('{"validation_method": "dns-account-01", "acme_account_url": "https://example.com/acme/acct/ExampleAccount", "key_authorization_hash": "test-ka"}',
+         DcvAcmeDnsAccount01ValidationParameters),
         ('{"validation_method": "contact-email-txt", "challenge_value": "test-cv"}',
          DcvContactEmailTxtValidationParameters),
         ('{"validation_method": "contact-email-caa", "challenge_value": "test-cv"}',
@@ -113,6 +116,14 @@ def check_request_parameters__should_force_case_sensitivity_to_false_for_non_txt
         else:
             assert details_as_object.require_exact_case is True
 
+    @staticmethod
+    def check_request_parameters__should_disallow_setting_case_sensitivity_to_false_for_acme_dns_account_01():
+        parameters_as_json = '{"validation_method": "dns-account-01", "acme_account_url": "https://example.com/acme/acct/ExampleAccount", "key_authorization_hash": "test-kah", "require_exact_case": false}'
+        type_adapter = TypeAdapter(DcvCheckParameters)
+        with pytest.raises(Exception) as validation_error:
+            type_adapter.validate_json(parameters_as_json)
+        assert isinstance(validation_error.value, ValueError)
+
     @staticmethod
     def check_request_parameters__should_disallow_setting_case_sensitivity_to_false_for_acme_dns_01():
         parameters_as_json = '{"validation_method": "acme-dns-01", "key_authorization_hash": "test-kah", "require_exact_case": false}'

tests/unit/open_mpic_core/test_mpic_dcv_checker.py

@@ -102,6 +102,7 @@ def mpic_dcv_checker__should_be_able_to_log_at_trace_level(self):
             (DcvValidationMethod.IP_ADDRESS, DnsRecordType.AAAA),
             (DcvValidationMethod.ACME_HTTP_01, None),
             (DcvValidationMethod.ACME_DNS_01, None),
+            (DcvValidationMethod.DNS_ACCOUNT_01, None),
             (DcvValidationMethod.REVERSE_ADDRESS_LOOKUP, None),
             (DcvValidationMethod.ACME_TLS_ALPN_01, None),
         ],
@@ -160,6 +161,7 @@ async def check_dcv__should_be_case_insensitive_for_challenge_values_for_certain
         [
             (DcvValidationMethod.ACME_HTTP_01, None),
             (DcvValidationMethod.ACME_DNS_01, None),
+            (DcvValidationMethod.DNS_ACCOUNT_01, None),
         ],
     )
     async def check_dcv__should_be_case_sensitive_for_challenge_values_for_certain_validation_methods(
@@ -244,6 +246,7 @@ async def check_dcv__should_allow_issuance_only_for_well_formed_ipv4_and_ipv6_in
     @pytest.mark.parametrize("dcv_method, domain, encoded_domain", [
         (DcvValidationMethod.WEBSITE_CHANGE, "bücher.example.de", "xn--bcher-kva.example.de"),
         (DcvValidationMethod.ACME_DNS_01, "café.com", "xn--caf-dma.com"),
+        (DcvValidationMethod.DNS_ACCOUNT_01, "café.com", "xn--caf-dma.com"),
     ])
     # fmt: on
     async def check_dcv__should_handle_domains_with_non_ascii_characters(
@@ -790,7 +793,7 @@ async def contact_info_caa_lookup__should_not_pass_if_no_records_found_in_domain
         dcv_response = await self.dcv_checker.perform_general_dns_validation(dcv_request)
         assert dcv_response.check_passed is False
 
-    @pytest.mark.parametrize("dcv_method", [DcvValidationMethod.DNS_CHANGE, DcvValidationMethod.ACME_DNS_01])
+    @pytest.mark.parametrize("dcv_method", [DcvValidationMethod.DNS_CHANGE, DcvValidationMethod.ACME_DNS_01, DcvValidationMethod.DNS_ACCOUNT_01])
     async def dns_based_dcv_checks__should_not_pass_given_non_matching_dns_record(self, dcv_method, mocker):
         dcv_request = ValidCheckCreator.create_valid_dcv_check_request(dcv_method)
         test_dns_query_answer = self._create_basic_dns_response_for_mock(dcv_request, mocker)
@@ -981,7 +984,7 @@ def evaluate_persistent_dns_response__should_return_false_given_malformed_record
         result = MpicDcvChecker.evaluate_persistent_dns_response(expected_dns_record_content, [record])
         assert result is False, f"Should fail with malformed record: {record}"
 
-    @pytest.mark.parametrize("dcv_method", [DcvValidationMethod.DNS_CHANGE, DcvValidationMethod.ACME_DNS_01])
+    @pytest.mark.parametrize("dcv_method", [DcvValidationMethod.DNS_CHANGE, DcvValidationMethod.ACME_DNS_01, DcvValidationMethod.DNS_ACCOUNT_01])
     async def dns_based_dcv_checks__should_return_timestamp_and_list_of_records_seen(self, dcv_method, mocker):
         dcv_request = ValidCheckCreator.create_valid_dcv_check_request(dcv_method)
         self._mock_dns_resolve_call_getting_multiple_txt_records(dcv_request, mocker)
@@ -999,6 +1002,7 @@ async def dns_based_dcv_checks__should_return_timestamp_and_list_of_records_seen
         [
             (DcvValidationMethod.DNS_CHANGE, Rcode.NOERROR),
             (DcvValidationMethod.ACME_DNS_01, Rcode.NXDOMAIN),
+            (DcvValidationMethod.DNS_ACCOUNT_01, Rcode.NXDOMAIN),
             (DcvValidationMethod.DNS_CHANGE, Rcode.REFUSED),
         ],
     )
@@ -1015,6 +1019,8 @@ async def dns_based_dcv_checks__should_return_response_code(self, dcv_method, re
             (DcvValidationMethod.DNS_CHANGE, dns.flags.CD, False),
             (DcvValidationMethod.ACME_DNS_01, dns.flags.AD, True),
             (DcvValidationMethod.ACME_DNS_01, dns.flags.CD, False),
+            (DcvValidationMethod.DNS_ACCOUNT_01, dns.flags.AD, True),
+            (DcvValidationMethod.DNS_ACCOUNT_01, dns.flags.CD, False),
         ],
     )
     async def dns_based_dcv_checks__should_return_whether_response_has_ad_flag(
@@ -1037,7 +1043,7 @@ async def dns_based_dcv_checks__should_return_cname_chain(self, dcv_method, mock
         dcv_response = await self.dcv_checker.check_dcv(dcv_request)
         assert "sub.example.com." in dcv_response.details.cname_chain
 
-    @pytest.mark.parametrize("dcv_method", [DcvValidationMethod.DNS_CHANGE, DcvValidationMethod.ACME_DNS_01])
+    @pytest.mark.parametrize("dcv_method", [DcvValidationMethod.DNS_CHANGE, DcvValidationMethod.ACME_DNS_01, DcvValidationMethod.DNS_ACCOUNT_01])
     async def dns_based_dcv_checks__should_not_pass_with_errors_given_exception_raised(self, dcv_method, mocker):
         dcv_request = ValidCheckCreator.create_valid_dcv_check_request(dcv_method)
         no_answer_error = dns.resolver.NoAnswer()
@@ -1290,7 +1296,7 @@ def _create_basic_dns_response_for_mock(self, dcv_request: DcvCheckRequest, mock
                 record_data = {"flags": "", "tag": "contactemail", "value": check_parameters.challenge_value}
             case DcvValidationMethod.CONTACT_PHONE_CAA:
                 record_data = {"flags": "", "tag": "contactphone", "value": check_parameters.challenge_value}
-            case _:  # ACME_DNS_01
+            case DcvValidationMethod.ACME_DNS_01 | DcvValidationMethod.DNS_ACCOUNT_01:
                 record_data = {"value": check_parameters.key_authorization_hash}
         record_type = check_parameters.dns_record_type
         record_prefix = check_parameters.dns_name_prefix

tests/unit/open_mpic_core/test_mpic_dcv_request.py

@@ -87,7 +87,7 @@ def mpic_dcv_request__should_require_token_for_acme_http_01_validation(self):
             MpicDcvRequest.model_validate_json(json.dumps(request.model_dump(warnings=False)))
         assert "token" in str(validation_error.value)
 
-    @pytest.mark.parametrize("validation_method", [DcvValidationMethod.ACME_HTTP_01, DcvValidationMethod.ACME_DNS_01])
+    @pytest.mark.parametrize("validation_method", [DcvValidationMethod.ACME_HTTP_01, DcvValidationMethod.ACME_DNS_01, DcvValidationMethod.DNS_ACCOUNT_01])
     def mpic_dcv_request__should_require_key_authorization_for_acme_validations(self, validation_method):
         request = ValidMpicRequestCreator.create_valid_dcv_mpic_request(validation_method)
         # noinspection PyTypeChecker

tests/unit/test_util/valid_check_creator.py

@@ -7,6 +7,7 @@
     CaaCheckParameters,
     DcvAcmeHttp01ValidationParameters,
     DcvAcmeDns01ValidationParameters,
+    DcvAcmeDnsAccount01ValidationParameters,
     DcvAcmeTlsAlpn01ValidationParameters,
     DcvIpAddressValidationParameters,
     DcvContactEmailCaaValidationParameters,
@@ -113,6 +114,17 @@ def create_valid_acme_dns_01_check_request():
             dcv_check_parameters=DcvAcmeDns01ValidationParameters(key_authorization_hash=challenge),
         )
 
+    @staticmethod
+    def create_valid_acme_dns_account_01_check_request():
+        challenge = "challenge_111".encode().hex()
+        return DcvCheckRequest(
+            domain_or_ip_target="example.com",
+            dcv_check_parameters=DcvAcmeDnsAccount01ValidationParameters(
+                acme_account_url="https://example.com/acme/acct/ExampleAccount",
+                key_authorization_hash=challenge,
+            ),
+        )
+
     @staticmethod
     def create_valid_acme_tls_alpn_01_check_request(target="example.com"):
         challenge = "example-token.9jg46WB3rR_AHD-EBXdN7cBkH1WOu0tA3M9fm21mqTI"
@@ -148,6 +160,8 @@ def create_valid_dcv_check_request(validation_method: DcvValidationMethod, recor
                 return ValidCheckCreator.create_valid_acme_http_01_check_request()
             case DcvValidationMethod.ACME_DNS_01:
                 return ValidCheckCreator.create_valid_acme_dns_01_check_request()
+            case DcvValidationMethod.DNS_ACCOUNT_01:
+                return ValidCheckCreator.create_valid_acme_dns_account_01_check_request()
             case DcvValidationMethod.ACME_TLS_ALPN_01:
                 return ValidCheckCreator.create_valid_acme_tls_alpn_01_check_request()
             case DcvValidationMethod.IP_ADDRESS:

tests/unit/test_util/valid_mpic_request_creator.py

@@ -4,6 +4,7 @@
     DcvDnsPersistentValidationParameters,
     DcvWebsiteChangeValidationParameters,
     DcvAcmeDns01ValidationParameters,
+    DcvAcmeDnsAccount01ValidationParameters,
     DcvAcmeHttp01ValidationParameters,
     DcvContactPhoneCaaValidationParameters,
     DcvContactPhoneTxtValidationParameters,
@@ -59,6 +60,11 @@ def create_check_parameters(
                 check_parameters = DcvAcmeHttp01ValidationParameters(token="test", key_authorization="test")
             case DcvValidationMethod.ACME_DNS_01:
                 check_parameters = DcvAcmeDns01ValidationParameters(key_authorization_hash="test")
+            case DcvValidationMethod.DNS_ACCOUNT_01:
+                check_parameters = DcvAcmeDnsAccount01ValidationParameters(
+                    acme_account_url="https://example.com/acme/acct/ExampleAccount",
+                    key_authorization_hash="test",
+                )
             case DcvValidationMethod.CONTACT_PHONE_CAA:
                 check_parameters = DcvContactPhoneCaaValidationParameters(challenge_value="test")
             case DcvValidationMethod.CONTACT_PHONE_TXT: