test(dns): cover GoDaddy relative-name derivation, RRset merge and request handling

Twelve tests, no network and no database, in the existing pure-logic style. The merge helpers are covered directly (additive add, idempotent republish, tombstone filtering, remove-one-of-many, remove-the-last-value signalling DELETE), but helper math alone would stay green if the write path stopped using it, so add_txt_record and remove_txt_record are also driven against a recording stub: the assertions pin that a sibling value survives a publish, that an already-published value issues no write, that an unreadable read raises instead of replacing the set, that removing the last value emits DELETE and never an empty PUT, and that no call is ever aimed at a zone-wide path. _request is exercised through a fake response for the cases that only appear against the real API: an empty 204 body must not raise, a 3xx must not read as success (redirects are not followed), a transport failure mid-read must not be mistaken for an empty body, and each error status must produce a message naming what the operator has to fix. Also covers the auth header in both forms, that the sanitizer strips credentials from composed error text, that the module does not log at all, the credential-field schema against the upsert validator's own key and length rules, and the two-key encryption round trip. Verified by mutation: nine deliberate breakages of the provider - single-value PUT, empty PUT instead of DELETE, coercing an unreadable read to empty, treating 3xx as success, following redirects, swallowing transport errors, dropping the dot-segment guard, lowering the TTL below the API floor, and removing sanitization - are each caught by at least one test.
2026-09-12 05:48:58 +00:00 · 2026-08-07 09:01:23 +03:00
parent 8e534ef170
commit 0a0226c758
1 changed files with 399 additions and 4 deletions
@@ -2,7 +2,8 @@

 Covers the TXT-value math (RFC 8555 §8.4 — raw SHA-256 digest, base64url, NOT hex),
 the _acme-challenge record-name derivation (wildcard stripping), credential encryption
-round-trip + tamper handling, and the DNS provider registry/allow-list.
+round-trip + tamper handling, the DNS provider registry/allow-list, and (v1.10.0) the
+GoDaddy provider's zone-relative name derivation and additive RRset merge math.
 """
 import base64
 import hashlib
@@ -53,9 +54,9 @@ def test_decrypt_invalid_token_returns_none():

 def test_provider_registry_and_allow_list():
    names = {p["name"] for p in list_providers()}
-    assert {"manual", "cloudflare"} <= names
-    assert is_supported("manual") and is_supported("cloudflare")
-    assert not is_supported("route53")  # not in MVP allow-list
+    assert {"manual", "cloudflare", "godaddy"} <= names
+    assert is_supported("manual") and is_supported("cloudflare") and is_supported("godaddy")
+    assert not is_supported("route53")  # not in the allow-list

    assert get_provider("manual").automated is False
    cf = get_provider("cloudflare", {"api_token": "x"})
@@ -91,6 +92,400 @@ def test_cloudflare_token_sanitize():
    assert p._raw_token == '"my-token_123"'


+# --- v1.10.0: GoDaddy provider (pure logic only — no network, no DB) ---
+
+
+def test_godaddy_credential_fields_schema():
+    # Re-assert DnsCredentialsUpsert's validator rules directly against the declared schema, so the
+    # UI can never render a field whose submission the API would reject with a 422.
+    import re
+    from services.dns_providers.godaddy import GoDaddyDNSProvider
+
+    fields = GoDaddyDNSProvider.credential_fields
+    assert [f["key"] for f in fields] == ["api_key", "api_secret"]
+    for f in fields:
+        assert re.match(r"^[a-zA-Z0-9_]{1,50}$", f["key"])         # DnsCredentialsUpsert key regex
+        assert f["type"] == "password"                              # renders Input.Password, not Input
+        assert isinstance(f["max_length"], int) and 0 < f["max_length"] <= 4000  # validator value cap
+        assert f["help"] and isinstance(f["help"], str)             # shown in the Form.Item `extra` slot
+    # api_secret is optional on purpose: leaving it blank is how a Personal Access Token is used
+    # (Bearer), which is the migration path off the sso-key scheme GoDaddy is retiring.
+    assert fields[0]["required"] is True and fields[1]["required"] is False
+    # Must not reuse Cloudflare's field name: the register modal's credential Form.Items are named
+    # cred_<key> in a SHARED form and are not cleared when the provider dropdown changes.
+    assert "api_token" not in {f["key"] for f in fields}
+
+
+def test_godaddy_provider_is_automated():
+    p = get_provider("godaddy", {"api_key": "k", "api_secret": "s"})
+    assert p.automated is True      # else the orchestrator takes the manual-confirm branch
+    assert p.name == "godaddy" and 1 <= len(p.name) <= 50  # dns_provider Field(min_length=1, max_length=50)
+    assert p.label == "GoDaddy"
+
+
+def test_godaddy_missing_credentials_returns_not_ok():
+    # verify_credentials must RETURN {"ok": False}, never raise: the router turns any non-
+    # DnsProviderError into the information-free generic 422 and the user never sees the reason.
+    import asyncio
+
+    for creds in ({}, {"api_secret": "s"}):  # blank UI fields arrive as MISSING keys, not ""
+        r = asyncio.run(get_provider("godaddy", creds).verify_credentials())
+        assert r["ok"] is False and r["detail"]
+    # Short-circuits before any request, so this touches no network.
+
+
+def test_godaddy_auth_header_formats_and_secret_never_leaks():
+    from services.dns_providers.godaddy import GoDaddyDNSProvider, _scrub
+
+    sentinel = "SENTINEL-SECRET-DO-NOT-LEAK"
+    p = GoDaddyDNSProvider({"api_key": "KEY123", "api_secret": sentinel})
+    # Literal prefix, one space, a single colon — no base64, no quoting.
+    assert p._auth_header() == f"sso-key KEY123:{sentinel}"
+    # No secret -> Personal Access Token. This one branch is the whole sso-key-sunset migration.
+    assert GoDaddyDNSProvider({"api_key": "PAT"})._auth_header() == "Bearer PAT"
+    # _scrub removes credential substrings from anything bound for a log or an order event.
+    assert sentinel not in _scrub(f"boom {sentinel} boom", "KEY123", sentinel)
+    assert "KEY123" not in _scrub("boom KEY123", "KEY123", sentinel)
+    assert _scrub("x" * 500, "KEY123") == "x" * 300  # bounded, so a huge body can't flood an event
+
+    # The channel that actually persists text: _http_error composes the message an order event and
+    # letsencrypt_orders.error_detail will carry, so it must scrub its own inputs — a caller that
+    # forgets to pre-scrub must not be able to leak. (Regression guard: scrubbing used to live at
+    # the single call site in _request instead of here.)
+    exc = p._http_error(403, f"DENIED_{sentinel}", f"token {sentinel} rejected", None)
+    assert sentinel not in str(exc) and "***" in str(exc)
+
+    # This module must not log at all — logging is the one channel _scrub cannot reach, since the
+    # arguments would be formatted by the logging framework rather than passed through it.
+    import inspect
+    import re as _re
+    from services.dns_providers import godaddy as gd_mod
+
+    assert not _re.search(r"\blogger\.\w+\(", inspect.getsource(gd_mod)), \
+        "godaddy.py must not log; surface everything through DnsProviderError so it is scrubbed"
+
+
+def test_godaddy_relative_record_name():
+    # GoDaddy names are RELATIVE to the zone with no trailing dot; the apex is the literal "@".
+    from services.dns_providers.godaddy import _relative_name
+
+    assert _relative_name("_acme-challenge.example.com", "example.com") == "_acme-challenge"
+    assert _relative_name("_acme-challenge.foo.bar.example.com", "example.com") == "_acme-challenge.foo.bar"
+    assert _relative_name("example.com", "example.com") == "@"          # never "" — see _rrset_path
+    assert _relative_name("_acme-challenge.example.com.", "example.com") == "_acme-challenge"
+    assert _relative_name("_ACME-Challenge.Example.COM", "example.com") == "_acme-challenge"
+    # Apex and wildcard produce the SAME relative name — which is exactly why the merge below
+    # has to be additive.
+    apex = ACMEService._challenge_dns_name("example.com")
+    wild = ACMEService._challenge_dns_name("*.example.com")
+    assert _relative_name(apex, "example.com") == _relative_name(wild, "example.com") == "_acme-challenge"
+
+
+def test_godaddy_rrset_merge_is_additive():
+    # THE critical test: GoDaddy's PUT REPLACES an entire RRset, so the merge math is the only thing
+    # keeping a wildcard+apex certificate's two coexisting TXT values alive.
+    from services.dns_providers.godaddy import _live_values, _merge_add, _merge_remove
+
+    def vals(body):
+        return sorted(r["data"] for r in body)
+
+    assert vals(_merge_add([{"data": "valueA", "ttl": 600}], "valueB")) == ["valueA", "valueB"]
+    assert _merge_add([{"data": "valueA"}], "valueA") is None      # idempotent; ACME retries land here
+    # Total, not an all()-over-a-computed-list (which passes vacuously on an empty result): the
+    # first publish at a fresh name must emit exactly one element, carrying the 600s TTL floor.
+    assert _merge_add([], "v") == [{"data": "v", "ttl": 600}]      # below 600 GoDaddy answers 422
+    # Tombstone rows ({"data": ""}) must never be echoed back — GoDaddy answers 422 INVALID_BODY.
+    assert _live_values([{"data": ""}, {"data": "x"}, {}]) == ["x"]
+    assert vals(_merge_add([{"data": ""}, {"data": "valueA"}], "valueB")) == ["valueA", "valueB"]
+
+    assert vals(_merge_remove([{"data": "valueA"}, {"data": "valueB"}], "valueB")) == ["valueA"]
+    assert _merge_remove([{"data": "valueA"}], "valueZ") is None    # already gone — tolerate
+    assert _merge_remove([], "valueZ") is None
+    # [] means "use DELETE": PUT with an empty array is rejected (422, "Records must be specified").
+    assert _merge_remove([{"data": "valueA"}], "valueA") == []
+    assert _merge_remove([{"data": ""}, {"data": "valueA"}], "valueA") == []
+
+
+def test_godaddy_never_builds_a_zone_wide_txt_path():
+    # A 3-segment path (.../records/TXT) is the endpoint that wipes EVERY TXT in the zone — SPF,
+    # DKIM, DMARC, domain verifications. An empty relative name must never be able to produce it.
+    from services.dns_providers.godaddy import _rrset_path
+
+    p = _rrset_path("example.com", "_acme-challenge")
+    assert p == "/domains/example.com/records/TXT/_acme-challenge"
+    assert p.count("/") == 5 and not p.endswith("/TXT")
+    assert _rrset_path("example.com", "@").endswith("/%40")  # apex percent-encoded for proxy safety
+    # "." and ".." survive quote() and are then normalized away by yarl when the URL is built, so
+    # ".../records/TXT/.." would resolve to the whole-zone endpoint. They must be refused too.
+    for bad in [("example.com", ""), ("", "_acme-challenge"), ("example.com", "."),
+                ("example.com", ".."), ("example.com", "...")]:
+        raised = False
+        try:
+            _rrset_path(*bad)
+        except DnsProviderError:
+            raised = True
+        assert raised, f"_rrset_path{bad} must refuse to build a zone-wide TXT path"
+    # And the only way to reach those inputs — a malformed domain — really does produce them.
+    from services.dns_providers.godaddy import _relative_name as _rel
+    assert _rel("..example.com", "example.com") == "."
+
+
+def test_godaddy_credential_encryption_roundtrip():
+    # The two-field credential dict rides the same Fernet blob as Cloudflare's single token.
+    reset_fernet_for_tests()
+    creds = {"api_key": "gd-key-plaintext", "api_secret": "gd-secret-plaintext"}
+    token = encrypt_dns_credentials(creds)
+    assert "gd-key-plaintext" not in token and "gd-secret-plaintext" not in token  # ciphertext
+    assert decrypt_dns_credentials(token) == creds
+    # This sorted key list is exactly what GET /dns-credentials exposes as credential_fields_present
+    # — names only, never values.
+    assert sorted(decrypt_dns_credentials(token).keys()) == ["api_key", "api_secret"]
+
+
+_GD_NS = "/domains/example.com/records/NS"
+_GD_TXT = "/domains/example.com/records/TXT/_acme-challenge"
+
+
+def _gd_provider(responses):
+    """A GoDaddy provider whose _request is replaced by a recorder.
+
+    The pure-merge tests above prove the MATH; this proves the WRITE PATH actually uses it. Without
+    it, replacing the merge with a single-value PUT — the mutation that silently destroys the
+    sibling value of every wildcard+apex certificate — leaves the whole suite green.
+
+    `responses` maps (method, path) -> value to return, or an Exception to raise. Unmapped calls
+    return None, which is how the zone suffix-walk's failed probes are modelled.
+    """
+    import types
+    from services.dns_providers.godaddy import GoDaddyDNSProvider
+
+    calls = []
+
+    async def _fake_request(self, session, method, path, **kwargs):
+        calls.append((method, path, kwargs.get("json")))
+        result = responses.get((method, path))
+        if isinstance(result, Exception):
+            raise result
+        return result
+
+    p = GoDaddyDNSProvider({"api_key": "k", "api_secret": "s"})
+    p._request = types.MethodType(_fake_request, p)
+    return p, calls
+
+
+def _assert_never_zone_wide(calls):
+    # A write to .../records or .../records/TXT replaces every TXT (or every record) in the zone.
+    for method, path, _json in calls:
+        if method in ("PUT", "DELETE"):
+            assert not path.endswith("/records"), f"zone-wide write: {method} {path}"
+            assert not path.endswith("/records/TXT"), f"type-wide write: {method} {path}"
+
+
+def test_godaddy_add_write_path_merges_siblings():
+    import asyncio
+
+    # An existing sibling value at the same name — the apex half of an apex+wildcard certificate.
+    p, calls = _gd_provider({
+        ("GET", _GD_NS): [{"data": "ns1.domaincontrol.com"}],
+        ("GET", _GD_TXT): [{"data": "valueA", "ttl": 600}],
+    })
+    asyncio.run(p.add_txt_record("_acme-challenge.example.com", "valueB"))
+
+    writes = [c for c in calls if c[0] in ("PUT", "PATCH", "DELETE")]
+    assert len(writes) == 1 and writes[0][0] == "PUT" and writes[0][1] == _GD_TXT
+    # BOTH values must be in the body: GoDaddy's PUT replaces the whole RRset.
+    assert sorted(r["data"] for r in writes[0][2]) == ["valueA", "valueB"]
+    _assert_never_zone_wide(calls)
+
+
+def test_godaddy_add_write_path_is_idempotent_and_fails_closed():
+    import asyncio
+
+    # Already published -> no write at all (this is where an ACME retry cycle lands).
+    p, calls = _gd_provider({
+        ("GET", _GD_NS): [{"data": "ns1.domaincontrol.com"}],
+        ("GET", _GD_TXT): [{"data": "valueB", "ttl": 600}],
+    })
+    asyncio.run(p.add_txt_record("_acme-challenge.example.com", "valueB"))
+    assert [c for c in calls if c[0] != "GET"] == []
+
+    # Unreadable RRset read (2xx whose body did not parse as a list) must FAIL, never be treated as
+    # an empty RRset — the PUT that follows would replace the sibling values with only ours.
+    p, calls = _gd_provider({
+        ("GET", _GD_NS): [{"data": "ns1.domaincontrol.com"}],
+        ("GET", _GD_TXT): None,
+    })
+    raised = False
+    try:
+        asyncio.run(p.add_txt_record("_acme-challenge.example.com", "valueB"))
+    except DnsProviderError:
+        raised = True
+    assert raised, "an unreadable RRset read must not be coerced into an empty RRset"
+    assert [c for c in calls if c[0] != "GET"] == []
+
+
+def test_godaddy_remove_write_path_uses_delete_for_the_last_value():
+    import asyncio
+
+    # Two values -> PUT back the survivor only.
+    p, calls = _gd_provider({
+        ("GET", _GD_NS): [{"data": "ns1.domaincontrol.com"}],
+        ("GET", _GD_TXT): [{"data": "valueA"}, {"data": "valueB"}],
+    })
+    asyncio.run(p.remove_txt_record("_acme-challenge.example.com", "valueB"))
+    writes = [c for c in calls if c[0] != "GET"]
+    assert len(writes) == 1 and writes[0][0] == "PUT"
+    assert [r["data"] for r in writes[0][2]] == ["valueA"]
+
+    # Last value -> DELETE. `PUT []` is rejected by GoDaddy (422 INVALID_BODY), so an empty PUT
+    # body would make every cleanup fail forever.
+    p, calls = _gd_provider({
+        ("GET", _GD_NS): [{"data": "ns1.domaincontrol.com"}],
+        ("GET", _GD_TXT): [{"data": "valueA"}],
+    })
+    asyncio.run(p.remove_txt_record("_acme-challenge.example.com", "valueA"))
+    writes = [c for c in calls if c[0] != "GET"]
+    assert len(writes) == 1 and writes[0] == ("DELETE", _GD_TXT, None)
+    assert not any(c[0] == "PUT" and c[2] == [] for c in calls)
+
+    # Value already gone -> no write, no error.
+    p, calls = _gd_provider({
+        ("GET", _GD_NS): [{"data": "ns1.domaincontrol.com"}],
+        ("GET", _GD_TXT): [{"data": "valueA"}],
+    })
+    asyncio.run(p.remove_txt_record("_acme-challenge.example.com", "valueZ"))
+    assert [c for c in calls if c[0] != "GET"] == []
+    _assert_never_zone_wide(calls)
+
+
+class _FakeGDResponse:
+    """Minimal stand-in for aiohttp's ClientResponse: status, headers, and json()."""
+
+    _NO_BODY = object()
+
+    def __init__(self, status, body=_NO_BODY, headers=None):
+        self.status = status
+        self._body = body
+        self.headers = headers or {}
+
+    async def json(self, content_type=None):
+        if self._body is _FakeGDResponse._NO_BODY:
+            raise ValueError("no body to decode")  # what an empty 204 does
+        return self._body
+
+
+class _FakeGDSession:
+    def __init__(self, response):
+        self._response = response
+        self.calls = []
+
+    def request(self, method, url, **kwargs):
+        self.calls.append((method, url, kwargs))
+        response = self._response
+
+        class _Ctx:
+            async def __aenter__(self_inner):
+                return response
+
+            async def __aexit__(self_inner, *exc):
+                return False
+
+        return _Ctx()
+
+
+def test_godaddy_request_status_handling():
+    import asyncio
+    from services.dns_providers.godaddy import GoDaddyDNSProvider
+
+    p = GoDaddyDNSProvider({"api_key": "KEY123", "api_secret": "SEC456"})
+
+    def call(response):
+        session = _FakeGDSession(response)
+        try:
+            return asyncio.run(p._request(session, "PUT", "/domains/example.com/records/TXT/x",
+                                          json=[{"data": "v", "ttl": 600}])), None, session
+        except DnsProviderError as exc:
+            return None, str(exc), session
+
+    # 204 with an EMPTY body is the normal answer to every GoDaddy write — it must not raise.
+    body, err, session = call(_FakeGDResponse(204))
+    assert body is None and err is None
+    # Redirects are deliberately not followed (aiohttp would forward the Authorization header), so
+    # a 3xx is a FAILED call. Treating it as success would report a redirected write as a no-op.
+    _kw = session.calls[0][2]
+    assert _kw["allow_redirects"] is False
+    assert _kw["headers"]["Authorization"] == "sso-key KEY123:SEC456"
+    assert _kw["headers"]["Accept"] == "application/json"
+    for status in (301, 302, 307):
+        body, err, _ = call(_FakeGDResponse(status))
+        assert body is None and err and str(status) in err, f"HTTP {status} must not read as success"
+
+    # 200 with a list is passed through verbatim.
+    body, err, _ = call(_FakeGDResponse(200, [{"data": "v"}]))
+    assert err is None and body == [{"data": "v"}]
+
+    # Error mapping: each message must name what the operator has to fix.
+    _, err, _ = call(_FakeGDResponse(401, {"code": "UNABLE_TO_AUTHENTICATE", "message": "nope"}))
+    assert "PRODUCTION" in err and "UNABLE_TO_AUTHENTICATE" in err
+    _, err, _ = call(_FakeGDResponse(403, {"code": "ACCESS_DENIED", "message": "not allowed"}))
+    assert "domains.dns:update" in err
+    # 429: Retry-After wins; the legacy body field is the fallback; absent both -> 60s default.
+    _, err, _ = call(_FakeGDResponse(429, None, {"Retry-After": "17"}))
+    assert "~17s" in err
+    _, err, _ = call(_FakeGDResponse(429, {"retryAfterSec": 42}))
+    assert "~42s" in err
+    _, err, _ = call(_FakeGDResponse(429, {"Retry-After": "not-a-number"}))
+    assert "~60s" in err
+    # A non-dict error body must not crash the error mapper.
+    _, err, _ = call(_FakeGDResponse(500, "<html>gateway</html>"))
+    assert "500" in err
+
+    # A transport failure MID-READ must not be mistaken for "empty body". Only a decode error may
+    # be swallowed: a caller reading an RRset would otherwise see None and could take it for an
+    # empty record set, and the full-RRset PUT that follows would destroy the sibling values.
+    import aiohttp
+
+    class _TruncatedResponse(_FakeGDResponse):
+        async def json(self, content_type=None):
+            raise aiohttp.ClientPayloadError("connection closed mid-body")
+
+    body, err, _ = call(_TruncatedResponse(200))
+    assert body is None and err and "GoDaddy" in err
+
+
+def test_godaddy_zone_resolution_walks_suffixes_and_caches():
+    import asyncio
+    from services.dns_providers.godaddy import _GoDaddyHTTPError
+
+    # The deepest candidate is not a zone (404 = "not this zone"); the walk must continue to the
+    # registrable domain and then reuse it, so the second challenge at the same name costs no probe.
+    p, calls = _gd_provider({
+        ("GET", "/domains/_acme-challenge.example.com/records/NS"):
+            _GoDaddyHTTPError("nope", status=404, code="UNKNOWN_DOMAIN"),
+        ("GET", _GD_NS): [{"data": "ns1.domaincontrol.com"}],
+        ("GET", _GD_TXT): [],
+    })
+    asyncio.run(p.add_txt_record("_acme-challenge.example.com", "valueA"))
+    asyncio.run(p.add_txt_record("_acme-challenge.example.com", "valueB"))
+    probes = [c for c in calls if c[1].endswith("/records/NS")]
+    assert len(probes) == 2, "the resolved zone must be cached for the life of the provider"
+    # Relative name derived from the RESOLVED zone, never from the deepest candidate.
+    assert all(c[1] == _GD_TXT for c in calls if "/records/TXT/" in c[1])
+
+    # A credential/eligibility failure during the walk must surface, not be swallowed as
+    # "no managed domain" — otherwise the operator chases a DNS problem that is really a bad key.
+    p, calls = _gd_provider({
+        ("GET", "/domains/_acme-challenge.example.com/records/NS"):
+            _GoDaddyHTTPError("denied", status=403, code="ACCESS_DENIED"),
+    })
+    raised = ""
+    try:
+        asyncio.run(p.add_txt_record("_acme-challenge.example.com", "v"))
+    except DnsProviderError as exc:
+        raised = str(exc)
+    assert "denied" in raised and "No managed GoDaddy domain" not in raised
+
+
 def test_b64url_decode_padding_roundtrip():
    # Issue #35 v1.8.2: _b64url_decode must round-trip for EVERY length, including base64url strings
    # whose length is a multiple of 4 (the case the old padding formula '=' * (4 - len%4) over-padded).