added a flag allow_lookup_failure to CAA check parameters

pyproject.toml

@@ -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.5.0"
+spec_version = "3.6.0"
 spec_repository = "https://github.com/open-mpic/open-mpic-specification"
 
 [tool.hatch.envs.default]

src/open_mpic_core/__about__.py

@@ -1 +1 @@
-__version__ = "6.0.0"
+__version__ = "6.1.0"

src/open_mpic_core/common_domain/check_parameters.py

@@ -13,7 +13,7 @@
 class CaaCheckParameters(BaseModel):
     certificate_type: CertificateType = CertificateType.TLS_SERVER
     caa_domains: list[str] | None = None
-    # contact_info_query: bool | False = False  # to better accommodate email/phone based DCV using contact info in CAA
+    allow_lookup_failure: bool = False  # Baseline Requirements have a carve-out for CAA lookup failure; use carefully!
 
 
 class DcvValidationParameters(BaseModel, ABC):

src/open_mpic_core/mpic_caa_checker/mpic_caa_checker.py

@@ -58,11 +58,7 @@ async def find_caa_records_and_domain(self, caa_request) -> tuple[RRset, Name]:
                 break
             except (dns.resolver.NoAnswer, dns.resolver.NXDOMAIN):
                 domain = domain.parent()
-            except Exception as e:
-                self.logger.error(
-                    f"Exception during CAA lookup for {caa_request.domain_or_ip_target}: {e}. Trace ID: {caa_request.trace_identifier}"
-                )
-                raise MpicCaaLookupException(f"{e}") from e
+            # will raise other exceptions that we want to catch in the calling function
 
         return rrset, domain
 
@@ -84,7 +80,8 @@ async def check_caa(self, caa_request: CaaCheckRequest) -> CaaCheckResponse:
         if caa_request.domain_or_ip_target.startswith("*."):
             is_wc_domain = True
 
-        caa_lookup_error = False
+        error_encountered = False
+        caa_lookup_error = None
         caa_found = False
         domain = None
         rrset = None
@@ -105,15 +102,22 @@ async def check_caa(self, caa_request: CaaCheckRequest) -> CaaCheckResponse:
             async with self.logger.trace_timing(f"CAA lookup for target {caa_request.domain_or_ip_target}"):
                 rrset, domain = await self.find_caa_records_and_domain(caa_request)
             caa_found = rrset is not None
-        except (MpicCaaLookupException, ValueError) as e:
-            caa_lookup_error = True
+        except Exception as e:
+            error_encountered = True
+            caa_lookup_error = e
             error_message = f"Error during CAA lookup for {caa_request.domain_or_ip_target}: {e}. Trace ID: {caa_request.trace_identifier}"
+            self.logger.error(error_message)
             caa_check_response.errors = [MpicValidationError.create(ErrorMessages.CAA_LOOKUP_ERROR, error_message)]
             caa_check_response.details.found_at = None
             caa_check_response.details.records_seen = None
 
-        if caa_lookup_error:
-            pass
+        if error_encountered:  # if there was an error during lookup
+            # check if allow_lookup_failure is set to True, and allow issuance depending on error
+            if isinstance(caa_lookup_error, (dns.resolver.LifetimeTimeout, dns.resolver.NoNameservers)):
+                if caa_request.caa_check_parameters and caa_request.caa_check_parameters.allow_lookup_failure:
+                    # if the error was from the lookup process itself (e.g. timeout), allow issuance
+                    caa_check_response.check_completed = True
+                    caa_check_response.check_passed = True
         elif not caa_found:  # if domain has no CAA records: valid for issuance
             caa_check_response.check_completed = True
             caa_check_response.check_passed = True

tests/unit/open_mpic_core/test_mpic_caa_checker.py

@@ -191,6 +191,18 @@ async def check_caa__should_allow_issuance_given_matching_caa_record_found_in_pa
         caa_response = await caa_checker.check_caa(caa_request)
         assert caa_response.check_passed is True
 
+    async def check_caa__should_allow_issuance_relying_on_default_caa_domains(self, mocker):
+        caa_checker = TestMpicCaaChecker.create_configured_caa_checker()
+        resolver = caa_checker.resolver
+        record_name, expected_domain = "example.com", "example.com."
+        test_dns_query_answer = MockDnsObjectCreator.create_caa_query_answer(record_name, 0, "issue", "ca2.net", mocker)
+        self.patch_resolver_to_expect_domain(
+            mocker, resolver, expected_domain, test_dns_query_answer, dns.resolver.NoAnswer
+        )
+        caa_request = CaaCheckRequest(domain_or_ip_target="example.com")
+        caa_response = await caa_checker.check_caa(caa_request)
+        assert caa_response.check_passed is True
+
     async def check_caa__should_disallow_issuance_given_non_matching_caa_record_found(self, mocker):
         caa_checker = TestMpicCaaChecker.create_configured_caa_checker()
         resolver = caa_checker.resolver
@@ -220,18 +232,7 @@ async def check_caa__should_disallow_smime_issuance_given_non_matching_caa_issue
         caa_response = await caa_checker.check_caa(caa_request)
         assert caa_response.check_passed is False
 
-    async def check_caa__should_allow_issuance_relying_on_default_caa_domains(self, mocker):
-        caa_checker = TestMpicCaaChecker.create_configured_caa_checker()
-        resolver = caa_checker.resolver
-        record_name, expected_domain = "example.com", "example.com."
-        test_dns_query_answer = MockDnsObjectCreator.create_caa_query_answer(record_name, 0, "issue", "ca2.net", mocker)
-        self.patch_resolver_to_expect_domain(
-            mocker, resolver, expected_domain, test_dns_query_answer, dns.resolver.NoAnswer
-        )
-        caa_request = CaaCheckRequest(domain_or_ip_target="example.com")
-        caa_response = await caa_checker.check_caa(caa_request)
-        assert caa_response.check_passed is True
-
+    # noinspection PyUnresolvedReferences
     @pytest.mark.parametrize("should_complete_check", [True, False])
     async def check_caa__should_set_check_completed_true_if_no_errors_encountered_and_false_otherwise(
         self, should_complete_check, mocker
@@ -248,6 +249,34 @@ async def check_caa__should_set_check_completed_true_if_no_errors_encountered_an
         assert caa_result.check_passed is should_complete_check
         assert caa_result.check_completed is should_complete_check
 
+    # fmt: off
+    @pytest.mark.parametrize("error_type, allow_failure", [
+        (dns.resolver.NoNameservers, True),
+        (dns.resolver.LifetimeTimeout, True),
+        (dns.resolver.YXDOMAIN, False),
+    ])
+    async def check_caa__should_allow_issuance_on_certain_lookup_failures_when_lookup_failure_is_explicitly_allowed(
+        self, error_type, allow_failure, mocker
+    ):
+        caa_checker = TestMpicCaaChecker.create_configured_caa_checker()
+        resolver = caa_checker.resolver
+        if error_type == dns.resolver.NoNameservers:
+            resolver_error = dns.resolver.NoNameservers(
+                request=dns.message.make_query("example.com", "CAA", "IN"), errors=[("192.0.0.1", True, 53, "SERVFAIL")]
+            )
+        elif error_type == dns.resolver.LifetimeTimeout:
+            resolver_error = dns.resolver.LifetimeTimeout(
+                timeout=10, errors=[("192.0.0.1", True, 53, "The DNS operation timed out after 10.000 seconds")]
+            )
+        else:
+            resolver_error = dns.resolver.YXDOMAIN()
+        self.patch_resolver_with_answer_or_exception(mocker, resolver, resolver_error)
+        caa_request = self.create_caa_check_request("example.com", ["ca111.com"])
+        caa_request.caa_check_parameters.allow_lookup_failure = True
+        caa_response = await caa_checker.check_caa(caa_request)
+        assert caa_response.check_passed is allow_failure
+        assert caa_response.check_completed is allow_failure
+
     async def check_caa__should_include_timestamp_in_nanos_in_result(self, mocker):
         caa_checker = TestMpicCaaChecker.create_configured_caa_checker()
         resolver = caa_checker.resolver
@@ -256,25 +285,36 @@ async def check_caa__should_include_timestamp_in_nanos_in_result(self, mocker):
         caa_response = await caa_checker.check_caa(caa_request)
         assert caa_response.timestamp_ns is not None
 
-    async def check_caa__should_return_failure_response_with_errors_given_error_in_dns_lookup(self, mocker):
+    async def check_caa__should_return_failure_response_given_error_in_dns_lookup(self, mocker):
         caa_checker = TestMpicCaaChecker.create_configured_caa_checker()
         resolver = caa_checker.resolver
         dns_lookup_error = dns.resolver.NoNameservers(
             request=dns.message.make_query("example.com", "CAA", "IN"),
-            errors=[
-                ("192.0.2.1", True, 53, dns.exception.Timeout("Timeout resolving example.com"))
-            ],  # List of (server, error) tuples
+            errors=[("192.0.2.1", True, 53, "SERVFAIL")],  # List of (server, error) tuples
         )
         self.patch_resolver_with_answer_or_exception(mocker, resolver, dns_lookup_error)
         caa_request = self.create_caa_check_request("example.com", ["ca111.com"])
         caa_response = await caa_checker.check_caa(caa_request)
         check_response_details = CaaCheckResponseDetails(caa_record_present=None)  # if error, don't know this detail
-        caa_response.timestamp_ns = None  # ignore timestamp for comparison
         assert caa_response.check_passed is False
         assert caa_response.check_completed is False
         assert caa_response.details == check_response_details
+
+    @pytest.mark.parametrize("allow_lookup_failure", [True, False])
+    async def check_caa__should_return_errors_in_response_given_error_in_dns_lookup(self, allow_lookup_failure, mocker):
+        caa_checker = TestMpicCaaChecker.create_configured_caa_checker()
+        resolver = caa_checker.resolver
+        dns_lookup_error = dns.resolver.NoNameservers(
+            request=dns.message.make_query("example.com", "CAA", "IN"),
+            errors=[("192.0.2.1", True, 53, "SERVFAIL")],  # List of (server, error) tuples
+        )
+        self.patch_resolver_with_answer_or_exception(mocker, resolver, dns_lookup_error)
+        caa_request = self.create_caa_check_request("example.com", ["ca111.com"])
+        caa_request.caa_check_parameters.allow_lookup_failure = allow_lookup_failure
+        caa_response = await caa_checker.check_caa(caa_request)
+        assert len(caa_response.errors) == 1
+        assert "SERVFAIL" in caa_response.errors[0].error_message
         assert caa_response.errors[0].error_type == ErrorMessages.CAA_LOOKUP_ERROR.key
-        assert "Timeout resolving example.com" in caa_response.errors[0].error_message
 
     @pytest.mark.parametrize("caa_answer_value, check_passed", [("ca1allowed.org", True), ("ca1notallowed.org", False)])
     async def check_caa__should_return_rrset_and_domain_given_domain_with_caa_record_on_success_or_failure(