diff --git a/pyproject.toml b/pyproject.toml index 83040a1..5322872 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,11 +24,11 @@ classifiers = [ ] dependencies = [ "pyyaml==6.0.2", - "requests==2.32.4", + "requests==2.33.0", "dnspython==2.7.0", "pydantic==2.11.7", - "aiohttp==3.13.3", - "cryptography==46.0.5", + "aiohttp==3.13.4", + "cryptography==46.0.6", "uritools==6.0.1", ] @@ -60,7 +60,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.8.0" +spec_version = "3.9.0" spec_repository = "https://github.com/open-mpic/open-mpic-specification" [tool.hatch.envs.default] diff --git a/src/open_mpic_core/__about__.py b/src/open_mpic_core/__about__.py index 76d9e7e..f0a00b3 100644 --- a/src/open_mpic_core/__about__.py +++ b/src/open_mpic_core/__about__.py @@ -1 +1 @@ -__version__ = "6.3.4" +__version__ = "6.4.0" diff --git a/src/open_mpic_core/common_domain/check_parameters.py b/src/open_mpic_core/common_domain/check_parameters.py index c33524e..224e0f1 100644 --- a/src/open_mpic_core/common_domain/check_parameters.py +++ b/src/open_mpic_core/common_domain/check_parameters.py @@ -2,7 +2,7 @@ from typing import Literal, Union, Any, Set, Annotated from uritools import isuri -from pydantic import BaseModel, field_validator, Field +from pydantic import BaseModel, field_validator, Field, model_validator from open_mpic_core import CertificateType, DnsRecordType, DcvValidationMethod, UrlScheme @@ -18,6 +18,7 @@ class CaaCheckParameters(BaseModel): class DcvValidationParameters(BaseModel, ABC): validation_method: DcvValidationMethod + require_exact_case: bool # DNS records have 5 fields: name, ttl, class, type, rdata (which can be multipart itself) # A or AAAA: name=domain_name type=A (ip address) # CNAME: name=domain_name_x type=CNAME @@ -31,6 +32,7 @@ class DcvWebsiteChangeValidationParameters(DcvValidationParameters): url_scheme: UrlScheme = UrlScheme.HTTP http_headers: dict[str, Any] | None = None match_regex: str | None = None + require_exact_case: bool = True # TODO add optional flag to iterate up through the domain hierarchy @@ -42,7 +44,8 @@ class DcvGeneralDnsValidationParameters(DcvValidationParameters, ABC): class DcvDnsChangeValidationParameters(DcvGeneralDnsValidationParameters): validation_method: Literal[DcvValidationMethod.DNS_CHANGE] = DcvValidationMethod.DNS_CHANGE - require_exact_match: bool = False + require_exact_match: bool = False # if False, looks for a matching substring (rather than entire string) + require_exact_case: bool = True # noinspection PyNestedDecorators @field_validator("dns_record_type") @@ -52,6 +55,12 @@ def validate_record_type(cls, v: DnsRecordType) -> DnsRecordType: raise ValueError(f"Record type must be one of {DNS_CHANGE_ALLOWED_RECORD_TYPES}, got {v}") return v + @model_validator(mode="after") + def validate_require_exact_case(self) -> 'DcvDnsChangeValidationParameters': + if self.dns_record_type is not DnsRecordType.TXT: + self.require_exact_case = False # case-sensitivity only applies to TXT records; force to False for others + return self + class DcvDnsPersistentValidationParameters(DcvValidationParameters): validation_method: Literal[DcvValidationMethod.DNS_PERSISTENT] = DcvValidationMethod.DNS_PERSISTENT @@ -59,6 +68,7 @@ class DcvDnsPersistentValidationParameters(DcvValidationParameters): 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 + require_exact_case: Literal[False] = False @field_validator("expected_account_uri") @classmethod @@ -83,27 +93,32 @@ class DcvContactEmailTxtValidationParameters(DcvGeneralDnsValidationParameters): validation_method: Literal[DcvValidationMethod.CONTACT_EMAIL_TXT] = DcvValidationMethod.CONTACT_EMAIL_TXT dns_record_type: Literal[DnsRecordType.TXT] = DnsRecordType.TXT dns_name_prefix: Literal["_validation-contactemail"] = "_validation-contactemail" + require_exact_case: Literal[False] = False class DcvContactEmailCaaValidationParameters(DcvGeneralDnsValidationParameters): validation_method: Literal[DcvValidationMethod.CONTACT_EMAIL_CAA] = DcvValidationMethod.CONTACT_EMAIL_CAA dns_record_type: Literal[DnsRecordType.CAA] = DnsRecordType.CAA + require_exact_case: Literal[False] = False class DcvContactPhoneTxtValidationParameters(DcvGeneralDnsValidationParameters): validation_method: Literal[DcvValidationMethod.CONTACT_PHONE_TXT] = DcvValidationMethod.CONTACT_PHONE_TXT dns_record_type: Literal[DnsRecordType.TXT] = DnsRecordType.TXT dns_name_prefix: Literal["_validation-contactphone"] = "_validation-contactphone" + require_exact_case: Literal[False] = False class DcvContactPhoneCaaValidationParameters(DcvGeneralDnsValidationParameters): validation_method: Literal[DcvValidationMethod.CONTACT_PHONE_CAA] = DcvValidationMethod.CONTACT_PHONE_CAA dns_record_type: Literal[DnsRecordType.CAA] = DnsRecordType.CAA + require_exact_case: Literal[False] = False class DcvIpAddressValidationParameters(DcvGeneralDnsValidationParameters): validation_method: Literal[DcvValidationMethod.IP_ADDRESS] = DcvValidationMethod.IP_ADDRESS dns_record_type: DnsRecordType + require_exact_case: Literal[False] = False # noinspection PyNestedDecorators @field_validator("dns_record_type") @@ -117,6 +132,7 @@ def validate_record_type(cls, v: DnsRecordType) -> DnsRecordType: class DcvReverseAddressLookupValidationParameters(DcvGeneralDnsValidationParameters): validation_method: Literal[DcvValidationMethod.REVERSE_ADDRESS_LOOKUP] = DcvValidationMethod.REVERSE_ADDRESS_LOOKUP dns_record_type: Literal[DnsRecordType.PTR] = DnsRecordType.PTR + require_exact_case: Literal[False] = False class DcvAcmeHttp01ValidationParameters(DcvValidationParameters): @@ -124,6 +140,7 @@ class DcvAcmeHttp01ValidationParameters(DcvValidationParameters): token: str key_authorization: str http_headers: dict[str, Any] | None = None + require_exact_case: Literal[True] = True # ACME HTTP-01 validation is always case-sensitive, per RFC8555 class DcvAcmeDns01ValidationParameters(DcvValidationParameters): @@ -131,11 +148,13 @@ class DcvAcmeDns01ValidationParameters(DcvValidationParameters): key_authorization_hash: str dns_record_type: Literal[DnsRecordType.TXT] = DnsRecordType.TXT dns_name_prefix: Literal["_acme-challenge"] = "_acme-challenge" + require_exact_case: Literal[True] = True # ACME DNS-01 validation is always case-sensitive class DcvAcmeTlsAlpn01ValidationParameters(DcvValidationParameters): validation_method: Literal[DcvValidationMethod.ACME_TLS_ALPN_01] = DcvValidationMethod.ACME_TLS_ALPN_01 key_authorization_hash: str + require_exact_case: Literal[False] = False DcvCheckParameters = Annotated[ diff --git a/src/open_mpic_core/mpic_dcv_checker/mpic_dcv_checker.py b/src/open_mpic_core/mpic_dcv_checker/mpic_dcv_checker.py index 7639c17..db1949a 100644 --- a/src/open_mpic_core/mpic_dcv_checker/mpic_dcv_checker.py +++ b/src/open_mpic_core/mpic_dcv_checker/mpic_dcv_checker.py @@ -129,6 +129,8 @@ async def perform_general_dns_validation(self, request: DcvCheckRequest) -> DcvC validation_method, check_parameters ) + require_exact_case = check_parameters.require_exact_case + if validation_method == DcvValidationMethod.DNS_CHANGE: # DNS_CHANGE may allow for non-exact match exact_match = check_parameters.require_exact_match @@ -141,7 +143,13 @@ async def perform_general_dns_validation(self, request: DcvCheckRequest) -> DcvC ): lookup = await self.perform_dns_resolution(name_to_resolve, validation_method, dns_record_type) MpicDcvChecker.evaluate_dns_lookup_response( - dcv_check_response, lookup, validation_method, dns_record_type, expected_dns_record_content, exact_match + dcv_check_response, + lookup, + validation_method, + dns_record_type, + expected_dns_record_content, + exact_match, + require_exact_case, ) except dns.exception.DNSException as e: log_msg = f"DNS lookup error for {name_to_resolve}: {str(e)}. Trace ID: {request.trace_identifier}" @@ -289,11 +297,18 @@ async def evaluate_http_lookup_response( # ACME requires an exact match dcv_check_response.check_passed = challenge_value == result else: - # Case-insensitive substring check for WEBSITE_CHANGE; also checks regex if provided - dcv_check_response.check_passed = challenge_value.lower() in result.lower() + # For WEBSITE_CHANGE: substring check; case-sensitive by default, unless require_exact_case=False + require_exact_case = dcv_check_request.dcv_check_parameters.require_exact_case + if require_exact_case: + dcv_check_response.check_passed = challenge_value in result + else: + dcv_check_response.check_passed = challenge_value.lower() in result.lower() if match_regex is not None and len(match_regex) > 0: match = re.search(match_regex, result) - dcv_check_response.check_passed = challenge_value.lower() in result.lower() and match is not None + challenge_value_found = ( + challenge_value in result if require_exact_case else challenge_value.lower() in result.lower() + ) + dcv_check_response.check_passed = challenge_value_found and match is not None dcv_check_response.details.response_status_code = http_response.status dcv_check_response.details.response_url = target_url dcv_check_response.details.response_history = response_history @@ -324,6 +339,7 @@ def evaluate_dns_lookup_response( dns_record_type: DnsRecordType, expected_dns_record_content: ExpectedDnsRecordContent | None, exact_match: bool = True, + require_exact_case: bool = False, ) -> None: if dns_response is None: dcv_check_response.check_passed = False @@ -374,10 +390,10 @@ def evaluate_dns_lookup_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 + if require_exact_case: + expected_dns_value = expected_dns_record_content.expected_value else: - expected_dns_value = expected_dns_record_content.expected_value.lower() # all others case-insensitive + expected_dns_value = expected_dns_record_content.expected_value.lower() # 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. diff --git a/tests/unit/open_mpic_core/test_check_request_parameters.py b/tests/unit/open_mpic_core/test_check_request_parameters.py index 5a5ffa6..77cbf87 100644 --- a/tests/unit/open_mpic_core/test_check_request_parameters.py +++ b/tests/unit/open_mpic_core/test_check_request_parameters.py @@ -13,6 +13,7 @@ DcvContactPhoneCaaValidationParameters, DcvIpAddressValidationParameters, DcvCheckParameters, + DnsRecordType ) @@ -94,6 +95,40 @@ def check_request_parameters__should_accept_valid_uri_format_for_expected_accoun assert isinstance(details_as_object, DcvDnsPersistentValidationParameters) assert details_as_object.expected_account_uri == account_uri + # fmt: off + @pytest.mark.parametrize("record_type, is_always_case_insensitive", [ + (DnsRecordType.CNAME, True), (DnsRecordType.TXT, False), (DnsRecordType.CAA, True) + ]) + # fmt: on + def check_request_parameters__should_force_case_sensitivity_to_false_for_non_txt_dns_records( + self, record_type, is_always_case_insensitive + ): + # notice require_exact_case is true in the serialized JSON; it should be forced to False for non-TXT records + parameters_as_json = f'{{"validation_method": "dns-change", "dns_record_type": "{record_type}", "challenge_value": "test-cv", "require_exact_case": true}}' + type_adapter = TypeAdapter(DcvCheckParameters) + details_as_object: DcvCheckParameters = type_adapter.validate_json(parameters_as_json) + assert isinstance(details_as_object, DcvDnsChangeValidationParameters) + if is_always_case_insensitive: + assert details_as_object.require_exact_case is False # should be forced to False for non-TXT records + 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_01(): + parameters_as_json = '{"validation_method": "acme-dns-01", "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_http_01(): + parameters_as_json = '{"validation_method": "acme-http-01", "token:" "test", "key_authorization": "test-ka", "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) + if __name__ == "__main__": pytest.main() diff --git a/tests/unit/open_mpic_core/test_mpic_dcv_checker.py b/tests/unit/open_mpic_core/test_mpic_dcv_checker.py index f382782..e2dcf68 100644 --- a/tests/unit/open_mpic_core/test_mpic_dcv_checker.py +++ b/tests/unit/open_mpic_core/test_mpic_dcv_checker.py @@ -119,30 +119,23 @@ async def check_dcv__should_perform_appropriate_check_and_allow_issuance_given_t assert dcv_response.check_passed is True @pytest.mark.parametrize( - "dcv_method, record_type, is_case_insensitive", + "dcv_method, record_type", [ - (DcvValidationMethod.WEBSITE_CHANGE, None, True), - (DcvValidationMethod.DNS_CHANGE, DnsRecordType.TXT, True), - (DcvValidationMethod.DNS_CHANGE, DnsRecordType.CNAME, True), - (DcvValidationMethod.DNS_CHANGE, DnsRecordType.CAA, True), - # (DcvValidationMethod.DNS_PERSISTENT, None, True), # Skipped: no challenge_value - (DcvValidationMethod.CONTACT_EMAIL_TXT, None, True), - (DcvValidationMethod.CONTACT_EMAIL_CAA, None, True), - (DcvValidationMethod.CONTACT_PHONE_TXT, None, True), - (DcvValidationMethod.CONTACT_PHONE_CAA, None, True), - # (DcvValidationMethod.IP_ADDRESS, DnsRecordType.A, False), # A records should not have letters anyway - (DcvValidationMethod.IP_ADDRESS, DnsRecordType.AAAA, True), - (DcvValidationMethod.ACME_HTTP_01, None, False), - (DcvValidationMethod.ACME_DNS_01, None, False), - (DcvValidationMethod.REVERSE_ADDRESS_LOOKUP, None, True), + (DcvValidationMethod.DNS_CHANGE, DnsRecordType.CNAME), + (DcvValidationMethod.DNS_CHANGE, DnsRecordType.CAA), # TODO do we want to move this to flagged casing? + # (DcvValidationMethod.DNS_PERSISTENT, None), # Skipped: no challenge_value + (DcvValidationMethod.CONTACT_EMAIL_TXT, None), + (DcvValidationMethod.CONTACT_EMAIL_CAA, None), + (DcvValidationMethod.CONTACT_PHONE_TXT, None), + (DcvValidationMethod.CONTACT_PHONE_CAA, None), + # (DcvValidationMethod.IP_ADDRESS, DnsRecordType.A), # A records should not have letters anyway + (DcvValidationMethod.IP_ADDRESS, DnsRecordType.AAAA), + (DcvValidationMethod.REVERSE_ADDRESS_LOOKUP, None), ], ) - async def check_dcv__should_be_case_insensitive_for_challenge_values_for_all_validation_methods_except_acme( - self, dcv_method, record_type, is_case_insensitive, mocker + async def check_dcv__should_be_case_insensitive_for_challenge_values_for_certain_validation_methods( + self, dcv_method, record_type, mocker ): - if dcv_method == DcvValidationMethod.DNS_PERSISTENT: - pytest.skip("DNS_PERSISTENT does not use challenge_value for case sensitivity test") - dcv_request = ValidCheckCreator.create_valid_dcv_check_request(dcv_method, record_type) if dcv_method in (DcvValidationMethod.CONTACT_PHONE_TXT, DcvValidationMethod.CONTACT_PHONE_CAA): # technically this should be case-insensitive, but also it would usually have digits... @@ -151,7 +144,29 @@ async def check_dcv__should_be_case_insensitive_for_challenge_values_for_all_val dcv_request.dcv_check_parameters.challenge_value = "2001:0DB8:85A3:0000:0000:8A2E:03C0:7B34" # set up mocks prior which will return the original challenge value in the dcv_request - if dcv_method in (DcvValidationMethod.WEBSITE_CHANGE, DcvValidationMethod.ACME_HTTP_01): + self._mock_request_specific_dns_resolve_call(dcv_request, mocker) + + # set up the challenge value casing to be different from the original + dcv_request.dcv_check_parameters.challenge_value = TestMpicDcvChecker.shuffle_case( + dcv_request.dcv_check_parameters.challenge_value + ) + + dcv_response = await self.dcv_checker.check_dcv(dcv_request) + assert dcv_response.check_passed is True + + @pytest.mark.parametrize( + "dcv_method, record_type", + [ + (DcvValidationMethod.ACME_HTTP_01, None), + (DcvValidationMethod.ACME_DNS_01, None), + ], + ) + async def check_dcv__should_be_case_sensitive_for_challenge_values_for_certain_validation_methods( + self, dcv_method, record_type, mocker + ): + dcv_request = ValidCheckCreator.create_valid_dcv_check_request(dcv_method, record_type) + + if dcv_method is DcvValidationMethod.ACME_HTTP_01: self._mock_request_specific_http_response(dcv_request, mocker) else: self._mock_request_specific_dns_resolve_call(dcv_request, mocker) @@ -161,17 +176,45 @@ async def check_dcv__should_be_case_insensitive_for_challenge_values_for_all_val dcv_request.dcv_check_parameters.key_authorization = TestMpicDcvChecker.shuffle_case( dcv_request.dcv_check_parameters.key_authorization ) - elif dcv_method == DcvValidationMethod.ACME_DNS_01: + else: dcv_request.dcv_check_parameters.key_authorization_hash = TestMpicDcvChecker.shuffle_case( dcv_request.dcv_check_parameters.key_authorization_hash ) - else: - dcv_request.dcv_check_parameters.challenge_value = TestMpicDcvChecker.shuffle_case( - dcv_request.dcv_check_parameters.challenge_value + + dcv_response = await self.dcv_checker.check_dcv(dcv_request) + assert dcv_response.check_passed is False # should fail because casing was shuffled + + @pytest.mark.parametrize( + "dcv_method, record_type, allow_case_insensitive", + [ + (DcvValidationMethod.WEBSITE_CHANGE, None, False), + (DcvValidationMethod.WEBSITE_CHANGE, None, True), + (DcvValidationMethod.DNS_CHANGE, DnsRecordType.TXT, False), + (DcvValidationMethod.DNS_CHANGE, DnsRecordType.TXT, True), + ], + ) + async def check_dcv__should_be_case_sensitive_unless_flagged_otherwise_for_challenge_values_for_certain_validation_methods( + self, dcv_method, record_type, allow_case_insensitive, mocker + ): + if allow_case_insensitive: + dcv_request = ValidCheckCreator.create_valid_dcv_check_request( + dcv_method, record_type, require_exact_case=False ) + else: + dcv_request = ValidCheckCreator.create_valid_dcv_check_request(dcv_method, record_type) + + if dcv_method is DcvValidationMethod.WEBSITE_CHANGE: + self._mock_request_specific_http_response(dcv_request, mocker) + else: + self._mock_request_specific_dns_resolve_call(dcv_request, mocker) + dcv_request.dcv_check_parameters.challenge_value = TestMpicDcvChecker.shuffle_case( + dcv_request.dcv_check_parameters.challenge_value + ) + + # pass or fail depending on case sensitivity flag dcv_response = await self.dcv_checker.check_dcv(dcv_request) - assert dcv_response.check_passed is is_case_insensitive + assert dcv_response.check_passed is allow_case_insensitive # fmt: off @pytest.mark.parametrize("record_type, target_record_data, mock_record_data, should_allow_issuance", [ @@ -842,11 +885,14 @@ def evaluate_persistent_dns_response__should_return_true_given_any_record_in_the result = MpicDcvChecker.evaluate_persistent_dns_response(expected_dns_record_content, records) assert result is True, "Should pass if any record is valid" - @pytest.mark.parametrize("account_uri", [ - "acct:foo123@example.com", - "https://example.com/acct/123", - "thiscanbeanything:bar@baz.com", - ]) + @pytest.mark.parametrize( + "account_uri", + [ + "acct:foo123@example.com", + "https://example.com/acct/123", + "thiscanbeanything:bar@baz.com", + ], + ) def evaluate_persistent_dns_response__should_accept_any_spec_compliant_account_uri(self, account_uri): issuer_domain_names = ["ca.example.com"] expected_account_uri = account_uri diff --git a/tests/unit/test_util/valid_check_creator.py b/tests/unit/test_util/valid_check_creator.py index 226f4aa..df62e97 100644 --- a/tests/unit/test_util/valid_check_creator.py +++ b/tests/unit/test_util/valid_check_creator.py @@ -34,22 +34,24 @@ def create_valid_caa_check_request() -> CaaCheckRequest: ) @staticmethod - def create_valid_http_check_request() -> DcvCheckRequest: + def create_valid_http_check_request(require_exact_case=True) -> DcvCheckRequest: return DcvCheckRequest( domain_or_ip_target="example.com", dcv_check_parameters=DcvWebsiteChangeValidationParameters( - http_token_path="token111_ca1.txt", challenge_value="challenge_111", url_scheme=UrlScheme.HTTP + http_token_path="token111_ca1.txt", challenge_value="challenge_111", url_scheme=UrlScheme.HTTP, + require_exact_case=require_exact_case ), ) @staticmethod - def create_valid_dns_check_request(record_type=DnsRecordType.TXT) -> DcvCheckRequest: + def create_valid_dns_check_request(record_type=DnsRecordType.TXT, require_exact_case=True) -> DcvCheckRequest: check_request = DcvCheckRequest( domain_or_ip_target="example.com", dcv_check_parameters=DcvDnsChangeValidationParameters( dns_name_prefix="_dnsauth", dns_record_type=record_type, challenge_value=f"{record_type}_challenge_111.ca1.com.", + require_exact_case=require_exact_case, ), ) return check_request @@ -129,14 +131,17 @@ def create_valid_reverse_address_lookup_check_request() -> DcvCheckRequest: ) @staticmethod - def create_valid_dcv_check_request(validation_method: DcvValidationMethod, record_type=None): + def create_valid_dcv_check_request(validation_method: DcvValidationMethod, record_type=None, require_exact_case=True) -> DcvCheckRequest: match validation_method: case DcvValidationMethod.WEBSITE_CHANGE: - return ValidCheckCreator.create_valid_http_check_request() + return ValidCheckCreator.create_valid_http_check_request(require_exact_case=require_exact_case) case DcvValidationMethod.DNS_CHANGE: if record_type is None: record_type = DnsRecordType.TXT - return ValidCheckCreator.create_valid_dns_check_request(record_type) + if record_type is DnsRecordType.TXT: # accept a flag for TXT; for the rest, don't... + return ValidCheckCreator.create_valid_dns_check_request(record_type, require_exact_case=require_exact_case) + else: + return ValidCheckCreator.create_valid_dns_check_request(record_type) case DcvValidationMethod.DNS_PERSISTENT: return ValidCheckCreator.create_valid_dns_persistent_check_request() case DcvValidationMethod.ACME_HTTP_01: