DNS Persistent DCV validation method

pyproject.toml

@@ -27,7 +27,7 @@ dependencies = [
   "requests==2.32.4",
   "dnspython==2.7.0",
   "pydantic==2.11.7",
-  "aiohttp==3.12.14",
+  "aiohttp==3.13.3",
   "black==25.1.0",
   "cryptography==45.0.4",
 ]
@@ -57,7 +57,7 @@ build.targets.wheel.packages = ["src/open_mpic_core"]
 "./tests/unit/test_util" = "open_mpic_core_test/test_util"  # include tests in the wheel to facilitate integration testing in wrapper projects
 
 [tool.api]
-spec_version = "3.7.0"
+spec_version = "3.8.0"
 spec_repository = "https://github.com/open-mpic/open-mpic-specification"
 
 [tool.hatch.envs.default]
@@ -99,8 +99,6 @@ markers = [
 addopts = [
     "--import-mode=prepend",  # explicit default, as the tests rely on it for proper import resolution
 ]
-spec_header_format = "Spec for {test_case} ({path}):"
-spec_test_format = "{result} {docstring_summary}"  # defaults to {name} if docstring is not present in test
 asyncio_mode = "auto"  # defaults to "strict"
 asyncio_default_fixture_loop_scope = "function"
 

src/open_mpic_core/__about__.py

@@ -1 +1 @@
-__version__ = "6.2.0"
+__version__ = "6.3.0"

src/open_mpic_core/__init__.py

@@ -13,6 +13,7 @@
     DcvAcmeHttp01ValidationParameters,
     DcvWebsiteChangeValidationParameters,
     DcvDnsChangeValidationParameters,
+    DcvDnsPersistentValidationParameters,
     DcvAcmeDns01ValidationParameters,
     DcvContactPhoneTxtValidationParameters,
     DcvContactEmailCaaValidationParameters,

src/open_mpic_core/common_domain/check_parameters.py

@@ -54,6 +54,14 @@ def validate_record_type(cls, v: DnsRecordType) -> DnsRecordType:
         return v
 
 
+class DcvDnsPersistentValidationParameters(DcvValidationParameters):
+    validation_method: Literal[DcvValidationMethod.DNS_PERSISTENT] = DcvValidationMethod.DNS_PERSISTENT
+    dns_record_type: Literal[DnsRecordType.TXT] = DnsRecordType.TXT
+    dns_name_prefix: Literal["_validation-persist"] = "_validation-persist"
+    issuer_domain_names: list[str]  # Disclosed issuer domain names from CA's CP/CPS
+    expected_account_uri: str  # The specific account URI to validate
+
+
 class DcvContactEmailTxtValidationParameters(DcvGeneralDnsValidationParameters):
     validation_method: Literal[DcvValidationMethod.CONTACT_EMAIL_TXT] = DcvValidationMethod.CONTACT_EMAIL_TXT
     dns_record_type: Literal[DnsRecordType.TXT] = DnsRecordType.TXT
@@ -117,6 +125,7 @@ class DcvAcmeTlsAlpn01ValidationParameters(DcvValidationParameters):
     Union[
         DcvWebsiteChangeValidationParameters,
         DcvDnsChangeValidationParameters,
+        DcvDnsPersistentValidationParameters,
         DcvAcmeHttp01ValidationParameters,
         DcvAcmeDns01ValidationParameters,
         DcvAcmeTlsAlpn01ValidationParameters,

src/open_mpic_core/common_domain/check_response_details.py

@@ -27,12 +27,14 @@ class DcvHttpCheckResponseDetails(BaseModel):
 class DcvDnsCheckResponseDetails(BaseModel):
     validation_method: Literal[
         DcvValidationMethod.DNS_CHANGE,
+        DcvValidationMethod.DNS_PERSISTENT,
         DcvValidationMethod.IP_ADDRESS,
         DcvValidationMethod.CONTACT_EMAIL_CAA,
         DcvValidationMethod.CONTACT_EMAIL_TXT,
         DcvValidationMethod.CONTACT_PHONE_CAA,
         DcvValidationMethod.CONTACT_PHONE_TXT,
         DcvValidationMethod.ACME_DNS_01,
+        DcvValidationMethod.DNS_ACCOUNT_01,
         DcvValidationMethod.REVERSE_ADDRESS_LOOKUP,
     ]
     records_seen: list[str] | None = None  # list of records found in DNS query; not base64 encoded
@@ -41,6 +43,7 @@ class DcvDnsCheckResponseDetails(BaseModel):
     found_at: str | None = None  # domain where DNS record was found
     cname_chain: list[str] | None = None # List of CNAMEs followed to obtain the final result.
 
+
 class DcvTlsAlpnCheckResponseDetails(BaseModel):
     validation_method: Literal[DcvValidationMethod.ACME_TLS_ALPN_01]
     common_name: str | None = None  # common name seen in certificate.
@@ -56,8 +59,10 @@ def build_response_details(validation_method: DcvValidationMethod) -> DcvCheckRe
         types = {
             DcvValidationMethod.WEBSITE_CHANGE: DcvHttpCheckResponseDetails,
             DcvValidationMethod.DNS_CHANGE: DcvDnsCheckResponseDetails,
+            DcvValidationMethod.DNS_PERSISTENT: DcvDnsCheckResponseDetails,
             DcvValidationMethod.ACME_HTTP_01: DcvHttpCheckResponseDetails,
             DcvValidationMethod.ACME_DNS_01: DcvDnsCheckResponseDetails,
+            DcvValidationMethod.DNS_ACCOUNT_01: DcvDnsCheckResponseDetails,
             DcvValidationMethod.ACME_TLS_ALPN_01: DcvTlsAlpnCheckResponseDetails,
             DcvValidationMethod.CONTACT_PHONE_TXT: DcvDnsCheckResponseDetails,
             DcvValidationMethod.CONTACT_PHONE_CAA: DcvDnsCheckResponseDetails,

src/open_mpic_core/common_domain/enum/dcv_validation_method.py

@@ -2,11 +2,13 @@
 
 
 class DcvValidationMethod(StrEnum):
-    WEBSITE_CHANGE = 'website-change'
+    WEBSITE_CHANGE = 'website-change'  # CABF BRs 3.2.2.4.18 Agreed-Upon Change to Website v2
     DNS_CHANGE = 'dns-change'  # CNAME, TXT, or CAA record
-    ACME_HTTP_01 = 'acme-http-01'
+    DNS_PERSISTENT = 'dns-persistent'  # CABF BRs 3.2.2.4.22 DNS TXT Record with Persistent Value
+    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'
+    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
     CONTACT_EMAIL_CAA = 'contact-email-caa'
     CONTACT_EMAIL_TXT = 'contact-email-txt'
     CONTACT_PHONE_CAA = 'contact-phone-caa'

src/open_mpic_core/mpic_dcv_checker/mpic_dcv_checker.py

@@ -25,6 +25,15 @@
 logger = get_logger(__name__)
 
 
+class ExpectedDnsRecordContent:
+    def __init__(self, expected_value=None, possible_values=None, expected_parameters=None):
+        self.expected_value = expected_value
+        self.possible_values = None
+        if self.expected_value is None:
+            self.possible_values = possible_values
+        self.expected_parameters = expected_parameters
+
+
 # noinspection PyUnusedLocal
 class MpicDcvChecker:
     WELL_KNOWN_PKI_PATH = ".well-known/pki-validation"
@@ -92,11 +101,10 @@ async def check_dcv(self, dcv_request: DcvCheckRequest) -> DcvCheckResponse:
                 result = await self.perform_http_based_validation(dcv_request)
             case DcvValidationMethod.ACME_TLS_ALPN_01:
                 result = await self.acme_tls_alpn_validator.perform_tls_alpn_validation(dcv_request)
-            case _:  # ACME_DNS_01 | DNS_CHANGE | IP_LOOKUP | CONTACT_EMAIL | CONTACT_PHONE | REVERSE_ADDRESS_LOOKUP
+            case _:  # all DNS based methods
                 result = await self.perform_general_dns_validation(dcv_request)
 
         # noinspection PyUnresolvedReferences
-
         self.logger.trace(
             "Completed DCV for %s with method %s. Trace ID: %s",
             dcv_request.domain_or_ip_target,
@@ -117,12 +125,11 @@ async def perform_general_dns_validation(self, request: DcvCheckRequest) -> DcvC
         else:
             name_to_resolve = request.domain_or_ip_target
 
-        if validation_method == DcvValidationMethod.ACME_DNS_01:
-            expected_dns_record_content = check_parameters.key_authorization_hash
-        else:
-            expected_dns_record_content = check_parameters.challenge_value
+        expected_dns_record_content = MpicDcvChecker.build_expected_dns_record_content(
+            validation_method, check_parameters
+        )
 
-        if validation_method == DcvValidationMethod.DNS_CHANGE:
+        if validation_method == DcvValidationMethod.DNS_CHANGE:  # DNS_CHANGE may allow for non-exact match
             exact_match = check_parameters.require_exact_match
 
         dcv_check_response = DcvUtils.create_empty_check_response(validation_method)
@@ -170,7 +177,8 @@ async def perform_dns_resolution(self, name_to_resolve, validation_method, dns_r
                 except (dns.resolver.NoAnswer, dns.resolver.NXDOMAIN):
                     domain = domain.parent()
         else:
-            lookup = await self.resolver.resolve(qname=name_to_resolve, rdtype=dns_rdata_type)
+            domain = dns.name.from_text(name_to_resolve)  # to ensure trailing dot is added
+            lookup = await self.resolver.resolve(qname=domain, rdtype=dns_rdata_type)
         return lookup
 
     @staticmethod
@@ -193,14 +201,14 @@ async def perform_http_based_validation(self, request: DcvCheckRequest) -> DcvCh
             expected_response_content = request.dcv_check_parameters.challenge_value
             url_scheme = request.dcv_check_parameters.url_scheme
             token_path = request.dcv_check_parameters.http_token_path
-            token_url = f"{url_scheme}://{formatted_host}/{MpicDcvChecker.WELL_KNOWN_PKI_PATH}/{token_path}"  # noqa E501 (http)
+            token_url = (
+                f"{url_scheme}://{formatted_host}/{MpicDcvChecker.WELL_KNOWN_PKI_PATH}/{token_path}"  # noqa E501 (http)
+            )
             dcv_check_response = DcvUtils.create_empty_check_response(DcvValidationMethod.WEBSITE_CHANGE)
         else:
             expected_response_content = request.dcv_check_parameters.key_authorization
             token = request.dcv_check_parameters.token
-            token_url = (
-                f"http://{formatted_host}/{MpicDcvChecker.WELL_KNOWN_ACME_PATH}/{token}"  # noqa E501 (http)
-            )
+            token_url = f"http://{formatted_host}/{MpicDcvChecker.WELL_KNOWN_ACME_PATH}/{token}"  # noqa E501 (http)
             dcv_check_response = DcvUtils.create_empty_check_response(DcvValidationMethod.ACME_HTTP_01)
         try:
             async with self.get_async_http_client() as async_http_client:
@@ -314,9 +322,9 @@ def evaluate_dns_lookup_response(
         dns_response: dns.resolver.Answer,
         validation_method: DcvValidationMethod,
         dns_record_type: DnsRecordType,
-        expected_dns_record_content: str,
+        expected_dns_record_content: ExpectedDnsRecordContent | None,
         exact_match: bool = True,
-    ):
+    ) -> None:
         if dns_response is None:
             dcv_check_response.check_passed = False
             dcv_check_response.check_completed = True
@@ -356,24 +364,31 @@ def evaluate_dns_lookup_response(
         dcv_check_response.details.cname_chain = cname_chain_str
         dcv_check_response.details.found_at = dns_response.qname.to_text(omit_final_dot=True)
 
-        # Case-insensitive comparison for all validation methods except ACME and IP Address
-        if validation_method not in (DcvValidationMethod.ACME_DNS_01, DcvValidationMethod.IP_ADDRESS):
-            expected_dns_record_content = expected_dns_record_content.lower()
-            records_as_strings = [record.lower() for record in records_as_strings]
-
-        # exact_match=True requires at least one record matches and will fail even if whitespace is different.
-        # exact_match=False simply runs a contains check.
-        if exact_match:
-            if validation_method == DcvValidationMethod.IP_ADDRESS:
-                dcv_check_response.check_passed = MpicDcvChecker.is_expected_ip_address_in_response(
-                    expected_dns_record_content, records_as_strings
-                )
-            else:
-                dcv_check_response.check_passed = expected_dns_record_content in records_as_strings
-        else:
-            dcv_check_response.check_passed = any(
-                expected_dns_record_content in record for record in records_as_strings
+        # handle "special logic" validation methods first
+        if validation_method == DcvValidationMethod.IP_ADDRESS:
+            dcv_check_response.check_passed = MpicDcvChecker.is_expected_ip_address_in_response(
+                expected_dns_record_content.expected_value, records_as_strings
+            )
+        elif validation_method == DcvValidationMethod.DNS_PERSISTENT:
+            dcv_check_response.check_passed = MpicDcvChecker.evaluate_persistent_dns_response(
+                expected_dns_record_content, records_as_strings
             )
+        else:
+            if validation_method == DcvValidationMethod.ACME_DNS_01:
+                expected_dns_value = expected_dns_record_content.expected_value  # case-sensitive per ACME spec
+            else:
+                expected_dns_value = expected_dns_record_content.expected_value.lower()  # all others case-insensitive
+                records_as_strings = [record.lower() for record in records_as_strings]
+
+            # exact_match=True requires at least one record matches and will fail even if whitespace is different.
+            # exact_match=False simply runs a contains check.
+            if exact_match:
+                dcv_check_response.check_passed = expected_dns_value in records_as_strings
+            else:
+                dcv_check_response.check_passed = any(
+                    expected_dns_value in record for record in records_as_strings
+                )
+
         dcv_check_response.check_completed = True
 
     @staticmethod
@@ -397,6 +412,86 @@ def is_expected_ip_address_in_response(ip_address_as_string: str, records_as_str
                     continue
         return ip_address_found
 
+    @staticmethod
+    def evaluate_persistent_dns_response(
+        expected_dns_record_content: ExpectedDnsRecordContent,
+        records_as_strings: list[str],
+    ) -> bool:
+        """
+        Evaluate DNS TXT records for persistent validation per CA/Browser Forum requirements.
+        Expected format follows RFC 8659 CAA issue-value syntax:
+        "issuer-domain-name; accounturi=<uri>; persistUntil=<timestamp>"
+        The persistUntil parameter is optional.
+        """
+        found_valid_record = False
+        accepted_domain_names = [domain.lower() for domain in expected_dns_record_content.possible_values]
+        expected_account_uri = expected_dns_record_content.expected_parameters['accounturi'].lower()
+
+        for txt_record in records_as_strings:
+            # Split on semicolon (parameter delimiter) and strip whitespace from each part
+            parts = [part.strip() for part in txt_record.rstrip().split(";")]
+            if len(parts) < 2:
+                continue  # Need at least issuer-domain-name and one parameter
+
+            issuer_domain_name = parts[0].lower()
+            param_list = parts[1:]
+
+            # First check issuer-domain-name matches one of the expected values
+            if issuer_domain_name not in accepted_domain_names:
+                continue
+
+            # Look for required accounturi parameter
+            valid_account_uri = False
+            within_allowed_time = True  # Assume valid unless proven otherwise
+
+            if not (len(param_list) == 1 and param_list[0].strip() == ""):  # if actual parameters follow the semicolon
+                for parameter in param_list:
+                    name_and_value = parameter.split("=", 1)
+                    if len(name_and_value) != 2:
+                        break  # malformed parameter; skip to next record
+                    param_name = name_and_value[0].strip().lower()
+                    param_value = name_and_value[1].strip()
+
+                    if param_name == "accounturi":
+                        if param_value.lower() == expected_account_uri:
+                            valid_account_uri = True
+                        else:
+                            break  # accounturi does not match; skip to next record
+                    elif param_name == "persistuntil":
+                        try:
+                            persist_until_in_seconds = int(param_value)  # seconds since epoch
+                            current_seconds = int(time.time())
+                            if persist_until_in_seconds < current_seconds:
+                                within_allowed_time = False
+                                break
+                        except (ValueError, OSError):
+                            within_allowed_time = False
+                            break  # Invalid timestamp format
+                        # Additional parameters are ignored per CA/Browser Forum spec
+
+            # Record is valid if issuer matches, account URI matches, and not expired
+            if valid_account_uri and within_allowed_time:
+                found_valid_record = True
+
+        return found_valid_record
+
+    @staticmethod
+    def build_expected_dns_record_content(
+        validation_method: DcvValidationMethod,
+        check_parameters,
+    ) -> ExpectedDnsRecordContent:
+        if validation_method == DcvValidationMethod.ACME_DNS_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
+                possible_values=check_parameters.issuer_domain_names,
+                expected_parameters={"accounturi": check_parameters.expected_account_uri},
+            )
+        else:
+            expected_content = ExpectedDnsRecordContent(expected_value=check_parameters.challenge_value)
+        return expected_content
+
     # noinspection PyUnresolvedReferences
     @staticmethod
     def extract_value_from_record(record: dns.rdata.Rdata) -> str:

tests/unit/open_mpic_core/test_check_request_parameters.py

@@ -5,6 +5,7 @@
     DcvAcmeHttp01ValidationParameters,
     DcvWebsiteChangeValidationParameters,
     DcvDnsChangeValidationParameters,
+    DcvDnsPersistentValidationParameters,
     DcvAcmeDns01ValidationParameters,
     DcvContactPhoneTxtValidationParameters,
     DcvContactEmailCaaValidationParameters,
@@ -24,6 +25,8 @@ class TestCheckRequestDetails:
          DcvDnsChangeValidationParameters),
         ('{"validation_method": "dns-change", "dns_record_type": "CNAME", "challenge_value": "test-cv"}',
          DcvDnsChangeValidationParameters),
+        ('{"validation_method": "dns-persistent", "issuer_domain_names": ["authority.example"], "expected_account_uri": "https://authority.example/acct/123"}',
+         DcvDnsPersistentValidationParameters),
         ('{"validation_method": "acme-http-01", "token": "test-t", "key_authorization": "test-ka"}',
          DcvAcmeHttp01ValidationParameters),
         ('{"validation_method": "acme-dns-01", "key_authorization_hash": "test-ka"}',
@@ -57,6 +60,10 @@ def check_request_parameters__should_automatically_deserialize_into_correct_obje
          "should fail validation when DNS record type is invalid for Contact Phone"),
         ('{"validation_method": "ip-address", "dns_record_type": "TXT", "challenge_value": "test-cv"}',
          "should fail validation when DNS record type is invalid like TXT for IP Address"),
+        ('{"validation_method": "dns-persistent", "expected_account_uri": "https://authority.example/acct/123"}',
+         "should fail validation when required issuer_domain_names is missing for DNS Persistent"),
+        ('{"validation_method": "dns-persistent", "issuer_domain_names": ["authority.example"]}',
+         "should fail validation when required expected_account_uri is missing for DNS Persistent"),
     ])
     # fmt: on
     def check_request_parameters__should_fail_validation_when_serialized_object_is_malformed(

tests/unit/open_mpic_core/test_mpic_caa_request.py

@@ -15,15 +15,15 @@ class TestMpicCaaRequest:
 
     def model_validate_json__should_return_caa_mpic_request_given_valid_caa_json(self):
         request = ValidMpicRequestCreator.create_valid_caa_mpic_request()
-        mpic_request = MpicCaaRequest.model_validate_json(json.dumps(request.model_dump()))
+        mpic_request = MpicCaaRequest.model_validate_json(json.dumps(request.model_dump(warnings=False)))
         assert mpic_request.domain_or_ip_target == request.domain_or_ip_target
 
     def mpic_caa_request__should_require_domain_or_ip_target(self):
         request = ValidMpicRequestCreator.create_valid_caa_mpic_request()
         # noinspection PyTypeChecker
         request.domain_or_ip_target = None
         with pytest.raises(pydantic.ValidationError) as validation_error:
-            MpicCaaRequest.model_validate_json(json.dumps(request.model_dump()))
+            MpicCaaRequest.model_validate_json(json.dumps(request.model_dump(warnings=False)))
         assert "domain_or_ip_target" in str(validation_error.value)
 
     @pytest.mark.parametrize("certificate_type", ["invalid"])