mirror of
https://github.com/Such-Software/adaptive-multi-protocol-gateway.git
synced 2026-10-17 15:41:41 +02:00
jw
b5df743fbb
A 400 from Dynadot arrived as "HTTP 400" and nothing else. Dynadot explains the refusal in the response body and we were discarding it, so the operator was left with a number that does not distinguish "the domain is not in this account" from "the signature is wrong" from "the caller address is not allowlisted". All three are 400 and all three have different fixes. The body is bounded to 400 characters and quoted as remote text. No credential can appear in it: the key travels in an Authorization header and the signature is computed rather than echoed. Reading the body is best-effort. A stream that cannot be read must not cost the status code as well, so the failure to read is swallowed and the code is still raised, which is what the second test pins.
202 lines
6.8 KiB
Python
202 lines
6.8 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_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()
|