mirror of
https://github.com/Such-Software/adaptive-multi-protocol-gateway.git
synced 2026-10-17 07:31:32 +02:00
04fcd20954
The previous fix did not work. Assigning Request.headers preserves the name in the object, and urllib then title-cases every header inside do_open, immediately before writing them to the socket. X-Request-ID still went out as X-Request-Id, which Dynadot rejects, so nothing changed. I verified that fix by inspecting Request.header_items() and it passed. A data structure holding the right string is not evidence that the right string was sent, and that is the whole of my mistake. No spelling survives .title() as X-Request-ID, so urllib cannot be made to send it. The request goes out through http.client, which writes header names verbatim. The urllib.request.Request stays as the unit handed to an injected opener so the existing tests keep inspecting what would actually be transmitted, and a 4xx is re-raised as urllib.error.HTTPError so the caller's error handling, which reads the body to surface Dynadot's own explanation, is untouched. The new test captures the headers at the socket rather than on the Request, and fails if the opener is put back to urllib.request.urlopen. That is the assertion the previous attempt should have made. Co-authored-by: jw <jw@jwmacbookm5.local>
306 lines
11 KiB
Python
306 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
from ampg.dns import ProviderDNSRecord
|
|
from ampg.dynadot_dns import DynadotDNSProvider, dynadot_signature
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, document):
|
|
self.document = document
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_arguments):
|
|
return False
|
|
|
|
def read(self):
|
|
return json.dumps(self.document).encode("utf-8")
|
|
|
|
|
|
class DynadotDNSTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.credentials = Path(self.temporary.name) / "dynadot.ini"
|
|
self.credentials.write_text(
|
|
"api_base_url=https://api-sandbox.dynadot.com\n"
|
|
"api_key=test-key\n"
|
|
"api_secret=test-secret\n"
|
|
"allowed_source_cidr=192.0.2.10/32\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def tearDown(self):
|
|
self.temporary.cleanup()
|
|
|
|
def test_signature_matches_documented_hmac_shape(self):
|
|
message = "test-key\n/restful/v2/domains/example.test/records\nrequest-id\n"
|
|
expected = base64.b64encode(
|
|
hmac.new(
|
|
b"test-secret", message.encode(), hashlib.sha256
|
|
).digest()
|
|
).decode()
|
|
self.assertEqual(
|
|
expected,
|
|
dynadot_signature(
|
|
"test-key",
|
|
"test-secret",
|
|
"/restful/v2/domains/example.test/records",
|
|
"request-id",
|
|
),
|
|
)
|
|
|
|
def test_fetch_preserves_all_provider_fields_and_secrets_stay_out_of_url(self):
|
|
requests = []
|
|
|
|
def opener(request, timeout):
|
|
requests.append(request)
|
|
return FakeResponse(
|
|
{
|
|
"code": 200,
|
|
"data": {
|
|
"glue_info": {
|
|
"glue_type": "DNS",
|
|
"ttl": 1800,
|
|
"dns_main_list": [
|
|
{
|
|
"record_type": "MX",
|
|
"value1": "mail.example.test.",
|
|
"value2": "10",
|
|
}
|
|
],
|
|
"dns_sub_list": [
|
|
{
|
|
"sub_host": "old",
|
|
"record_type": "FORWARD",
|
|
"value1": "https://example.test/",
|
|
"value2": "301",
|
|
}
|
|
],
|
|
}
|
|
},
|
|
}
|
|
)
|
|
|
|
provider = DynadotDNSProvider(
|
|
credentials=self.credentials,
|
|
opener=opener,
|
|
request_id_factory=lambda: "550e8400-e29b-41d4-a716-446655440000",
|
|
)
|
|
snapshot = provider.get_hosts("example.test")
|
|
self.assertEqual(1800, snapshot.zone_ttl)
|
|
self.assertEqual(10, snapshot.records[0].mx_pref)
|
|
self.assertEqual("301", snapshot.records[1].provider_value2)
|
|
self.assertNotIn("test-key", requests[0].full_url)
|
|
self.assertNotIn("test-secret", requests[0].full_url)
|
|
self.assertEqual("Bearer test-key", requests[0].headers["Authorization"])
|
|
|
|
def test_full_zone_write_is_canonical_and_lossless(self):
|
|
requests = []
|
|
|
|
def opener(request, timeout):
|
|
requests.append(request)
|
|
return FakeResponse({"code": 200, "data": {}})
|
|
|
|
provider = DynadotDNSProvider(
|
|
credentials=self.credentials,
|
|
opener=opener,
|
|
request_id_factory=lambda: "550e8400-e29b-41d4-a716-446655440000",
|
|
)
|
|
provider.set_hosts(
|
|
"example.test",
|
|
(
|
|
ProviderDNSRecord(
|
|
"@", "MX", "mail.example.test.", 1800, 10
|
|
),
|
|
ProviderDNSRecord(
|
|
"old",
|
|
"FORWARD",
|
|
"https://example.test/",
|
|
1800,
|
|
provider_value2="301",
|
|
),
|
|
),
|
|
mail_policy="preserve",
|
|
)
|
|
body = json.loads(requests[0].data)
|
|
self.assertFalse(body["add_dns_to_current_setting"])
|
|
self.assertEqual("10", body["dns_main_list"][0]["value2"])
|
|
self.assertEqual("301", body["dns_sub_list"][0]["value2"])
|
|
|
|
def test_http_error_body_reaches_the_operator(self):
|
|
"""A status code with no explanation is not a diagnosis.
|
|
|
|
Dynadot says why in the body. For a 400 that is the difference between
|
|
the domain not being in this account and the signature being wrong, and
|
|
the operator was handed neither.
|
|
"""
|
|
import io
|
|
import urllib.error
|
|
|
|
def opener(request, timeout=None):
|
|
raise urllib.error.HTTPError(
|
|
request.full_url,
|
|
400,
|
|
"Bad Request",
|
|
{},
|
|
io.BytesIO(b'{"error":"domain not found in this account"}'),
|
|
)
|
|
|
|
provider = DynadotDNSProvider(credentials=self.credentials, opener=opener)
|
|
with self.assertRaises(RuntimeError) as caught:
|
|
provider.get_hosts("example.test")
|
|
message = str(caught.exception)
|
|
self.assertIn("HTTP 400", message)
|
|
self.assertIn("domain not found in this account", message)
|
|
|
|
def test_an_unreadable_body_does_not_cost_the_status_code(self):
|
|
# The body is a bonus. Losing it must not lose the code as well.
|
|
import urllib.error
|
|
|
|
def opener(request, timeout=None):
|
|
error = urllib.error.HTTPError(request.full_url, 503, "x", {}, None)
|
|
|
|
def explode():
|
|
raise OSError("stream gone")
|
|
|
|
error.read = explode
|
|
raise error
|
|
|
|
provider = DynadotDNSProvider(credentials=self.credentials, opener=opener)
|
|
with self.assertRaisesRegex(RuntimeError, "HTTP 503"):
|
|
provider.get_hosts("example.test")
|
|
|
|
def test_request_id_header_keeps_its_exact_case_on_the_wire(self):
|
|
"""Dynadot matches X-Request-ID exactly, against RFC 9110.
|
|
|
|
urllib capitalises header names when they go through add_header, so
|
|
X-Request-ID left as X-request-id. Dynadot did not find it, signed with
|
|
an empty request id while we signed with the UUID, and rejected the
|
|
result as an invalid signature. Every request this provider made failed,
|
|
and the message pointed at the algorithm and the credentials.
|
|
"""
|
|
seen = {}
|
|
|
|
def opener(request, timeout=None):
|
|
seen.update(dict(request.header_items()))
|
|
return FakeResponse({"code": 200, "data": {"glue_info": {"glue_type": "DNS", "ttl": 1800, "dns_main_list": [], "dns_sub_list": []}}})
|
|
|
|
provider = DynadotDNSProvider(credentials=self.credentials, opener=opener)
|
|
provider.get_hosts("example.test")
|
|
|
|
self.assertIn("X-Request-ID", seen)
|
|
self.assertNotIn("X-request-id", seen)
|
|
# The signature must be over the id we actually sent.
|
|
self.assertIn("X-Signature", seen)
|
|
|
|
def test_the_signed_request_id_is_the_one_transmitted(self):
|
|
# Signing one value and sending another is the failure above, restated.
|
|
seen = {}
|
|
|
|
def opener(request, timeout=None):
|
|
seen.update(dict(request.header_items()))
|
|
return FakeResponse({"code": 200, "data": {"glue_info": {"glue_type": "DNS", "ttl": 1800, "dns_main_list": [], "dns_sub_list": []}}})
|
|
|
|
provider = DynadotDNSProvider(
|
|
credentials=self.credentials,
|
|
opener=opener,
|
|
request_id_factory=lambda: "fixed-request-id",
|
|
)
|
|
provider.get_hosts("example.test")
|
|
|
|
expected = dynadot_signature(
|
|
"test-key",
|
|
"test-secret",
|
|
"/restful/v2/domains/example.test/records",
|
|
"fixed-request-id",
|
|
)
|
|
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"):
|
|
provider.set_hosts(
|
|
"example.test",
|
|
(
|
|
ProviderDNSRecord("@", "A", "192.0.2.1", 300),
|
|
ProviderDNSRecord("www", "A", "192.0.2.1", 1800),
|
|
),
|
|
mail_policy="preserve",
|
|
)
|
|
self.credentials.write_text(
|
|
self.credentials.read_text().replace("192.0.2.10/32", "192.0.2.0/24")
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "IPv4 /32"):
|
|
DynadotDNSProvider(credentials=self.credentials)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|