diff --git a/ampg/dynadot_dns.py b/ampg/dynadot_dns.py index 7e06e67..a9a6773 100644 --- a/ampg/dynadot_dns.py +++ b/ampg/dynadot_dns.py @@ -15,6 +15,8 @@ import json from pathlib import Path import urllib.error import urllib.parse +import http.client +import io import urllib.request import uuid from typing import Any, Callable @@ -103,6 +105,63 @@ def _record_value(record: dict[str, Any], name: str) -> str | None: return str(value) +class _Body: + """The slice of the response interface the caller uses.""" + + def __init__(self, data: bytes): + self._data = data + + def __enter__(self): + return self + + def __exit__(self, *_arguments): + return False + + def read(self) -> bytes: + return self._data + + +def _send_with_exact_header_names(request, timeout: int = 30): + """Send the request without urllib rewriting the header names. + + urllib.request title-cases every header inside do_open, so X-Request-ID goes + out as X-Request-Id. Header names are case-insensitive under RFC 9110 and + Dynadot's gateway is not: it matches X-Request-ID exactly, does not find the + rewritten form, and signs with an empty request id while we signed with the + UUID. Every request was refused as a bad signature, which is the one thing + the signature was not. + + No spelling survives .title() as X-Request-ID, so urllib cannot be persuaded + to send it and the request goes out through http.client instead. The + urllib.request.Request stays as the unit passed to an injected opener, so + tests keep inspecting exactly what would be transmitted. + """ + parsed = urllib.parse.urlsplit(request.full_url) + connection = http.client.HTTPSConnection( + parsed.hostname, parsed.port, timeout=timeout + ) + target = parsed.path + (f"?{parsed.query}" if parsed.query else "") + try: + connection.request( + request.get_method(), + target, + body=request.data, + headers=dict(request.header_items()), + ) + response = connection.getresponse() + payload = response.read() + status, reason, headers = response.status, response.reason, response.headers + finally: + connection.close() + if status >= 400: + # Raised so the caller's existing error handling, which reads the body + # to surface the provider's own explanation, keeps working unchanged. + raise urllib.error.HTTPError( + request.full_url, status, reason, headers, io.BytesIO(payload) + ) + return _Body(payload) + + class DynadotDNSProvider: name = "dynadot" backup_format = "json" @@ -127,7 +186,7 @@ class DynadotDNSProvider: raise ValueError(f"missing Dynadot api_key in {credentials}") if not self.api_secret: raise ValueError(f"missing Dynadot api_secret in {credentials}") - self._opener = opener or urllib.request.urlopen + self._opener = opener or _send_with_exact_header_names self._request_id_factory = request_id_factory or ( lambda: str(uuid.uuid4()) ) diff --git a/tests/test_dynadot_dns.py b/tests/test_dynadot_dns.py index d1960ae..44e7346 100644 --- a/tests/test_dynadot_dns.py +++ b/tests/test_dynadot_dns.py @@ -226,6 +226,63 @@ class DynadotDNSTests(unittest.TestCase): self.assertEqual("fixed-request-id", seen["X-Request-ID"]) self.assertEqual(expected, seen["X-Signature"]) + def test_the_socket_receives_the_exact_header_name(self): + """Assert on the wire, not on the Request object. + + The previous attempt at this bug checked Request.header_items() and + passed, while urllib went on to title-case every name inside do_open, so + X-Request-ID still left as X-Request-Id and nothing changed. A data + structure that holds the right string is not evidence that the right + string was sent. + """ + import http.client as http_client + from ampg import dynadot_dns + + captured = {} + + class FakeConnection: + def __init__(self, host, port=None, timeout=None): + pass + + def request(self, method, target, body=None, headers=None): + captured.update(headers or {}) + + def getresponse(self): + class Response: + status, reason, headers = 200, "OK", {} + + def read(self): + return json.dumps( + { + "code": 200, + "data": { + "glue_info": { + "glue_type": "DNS", + "ttl": 1800, + "dns_main_list": [], + "dns_sub_list": [], + } + }, + } + ).encode() + + return Response() + + def close(self): + pass + + original = http_client.HTTPSConnection + http_client.HTTPSConnection = FakeConnection + try: + provider = DynadotDNSProvider(credentials=self.credentials) + provider.get_hosts("example.test") + finally: + http_client.HTTPSConnection = original + + self.assertIn("X-Request-ID", captured) + self.assertNotIn("X-Request-Id", captured) + self.assertNotIn("X-request-id", captured) + def test_mixed_ttl_and_broad_source_allowlist_fail_closed(self): provider = DynadotDNSProvider(credentials=self.credentials) with self.assertRaisesRegex(ValueError, "one observed"):