diff --git a/ampg/dns.py b/ampg/dns.py index b6e696d..a73799e 100644 --- a/ampg/dns.py +++ b/ampg/dns.py @@ -14,7 +14,7 @@ from .addresses import effective_address_records DNS_MODES = ("static", "dynamic") -DNS_PROVIDERS = ("namecheap",) +DNS_PROVIDERS = ("namecheap", "dynadot") DNS_MAIL_POLICIES = ("preserve", "disabled") DEFAULT_TTL = 1800 NAMECHEAP_SUPPORTED_RECORD_TYPES = { @@ -91,6 +91,7 @@ class ProviderDNSRecord: value: str ttl: int = DEFAULT_TTL mx_pref: int | None = None + provider_value2: str | None = None @dataclass(frozen=True) @@ -113,6 +114,7 @@ class DNSZoneSnapshot: records: tuple[ProviderDNSRecord, ...] status: str message: str + zone_ttl: int | None = None @dataclass(frozen=True) @@ -776,7 +778,13 @@ def _default_backup_dir(config: GatewayConfig) -> Path: def _backup_domain(provider, domain: str, backup_dir: Path) -> DNSBackupResult: snapshot = provider.get_hosts(domain) - path = _write_dns_backup(provider.name, domain, snapshot.raw, backup_dir) + path = _write_dns_backup( + provider.name, + domain, + snapshot.raw, + backup_dir, + extension=getattr(provider, "backup_format", "txt"), + ) return DNSBackupResult( provider=provider.name, domain=domain, @@ -787,9 +795,16 @@ def _backup_domain(provider, domain: str, backup_dir: Path) -> DNSBackupResult: ) -def _write_dns_backup(provider: str, domain: str, raw: str, backup_dir: Path) -> Path: +def _write_dns_backup( + provider: str, + domain: str, + raw: str, + backup_dir: Path, + *, + extension: str = "xml", +) -> Path: stamp = _datetime.datetime.now(_datetime.UTC).strftime("%Y%m%d_%H%M%S") - path = backup_dir / provider / f"{domain}.{stamp}.xml" + path = backup_dir / provider / f"{domain}.{stamp}.{extension}" path.parent.mkdir(parents=True, exist_ok=True) path.write_text(raw, encoding="utf-8") return path @@ -799,9 +814,15 @@ def _records_supported_by_provider( provider: str, records: tuple[ProviderDNSRecord, ...] | list[ProviderDNSRecord], ) -> tuple[ProviderDNSRecord, ...]: - if provider != "namecheap": + if provider == "namecheap": + supported = NAMECHEAP_SUPPORTED_RECORD_TYPES + elif provider == "dynadot": + from .dynadot_dns import DYNADOT_SUPPORTED_RECORD_TYPES + + supported = DYNADOT_SUPPORTED_RECORD_TYPES + else: return tuple(records) - return tuple(record for record in records if record.type in NAMECHEAP_SUPPORTED_RECORD_TYPES) + return tuple(record for record in records if record.type in supported) def _unsupported_record_changes( @@ -809,11 +830,17 @@ def _unsupported_record_changes( provider: str, desired: tuple[ProviderDNSRecord, ...], ) -> tuple[DNSApplyChange, ...]: - if provider != "namecheap": + if provider == "namecheap": + supported = NAMECHEAP_SUPPORTED_RECORD_TYPES + elif provider == "dynadot": + from .dynadot_dns import DYNADOT_SUPPORTED_RECORD_TYPES + + supported = DYNADOT_SUPPORTED_RECORD_TYPES + else: return () changes = [] for record in desired: - if record.type in NAMECHEAP_SUPPORTED_RECORD_TYPES: + if record.type in supported: continue changes.append( DNSApplyChange( @@ -915,6 +942,7 @@ def _normalize_provider_record(record: ProviderDNSRecord) -> ProviderDNSRecord: value=record.value.strip(), ttl=int(record.ttl or DEFAULT_TTL), mx_pref=record.mx_pref, + provider_value2=record.provider_value2, ) @@ -946,11 +974,16 @@ def _url_host(value: str) -> str: def _dns_provider(provider: str, *, credentials: Path, client_ip: str | None): if provider == "namecheap": return NamecheapDNSProvider(credentials=credentials, client_ip=client_ip) + if provider == "dynadot": + from .dynadot_dns import DynadotDNSProvider + + return DynadotDNSProvider(credentials=credentials) raise ValueError(f"unsupported DNS provider {provider!r}") class NamecheapDNSProvider: name = "namecheap" + backup_format = "xml" def __init__(self, *, credentials: Path, client_ip: str | None = None): self.credentials = credentials @@ -1015,6 +1048,8 @@ class NamecheapDNSProvider: *, mail_policy: str, ) -> None: + if any(record.provider_value2 is not None for record in records): + raise ValueError("Namecheap records cannot set provider_value2") params = self._domain_params(domain) if mail_policy == "disabled": params["EmailType"] = "MX" diff --git a/ampg/dns_manifest.py b/ampg/dns_manifest.py index 8b5087d..cda05f7 100644 --- a/ampg/dns_manifest.py +++ b/ampg/dns_manifest.py @@ -21,6 +21,16 @@ from .dns import ( NamecheapDNSProvider, ProviderDNSRecord, ) +from .dynadot_dns import ( + DYNADOT_SUPPORTED_RECORD_TYPES, + DynadotDNSProvider, +) + + +_PROVIDER_RECORD_TYPES = { + "namecheap": NAMECHEAP_SUPPORTED_RECORD_TYPES, + "dynadot": DYNADOT_SUPPORTED_RECORD_TYPES, +} _HOSTNAME = re.compile( @@ -77,6 +87,7 @@ def _record_identity(record: ProviderDNSRecord) -> tuple[Any, ...]: record.value.strip(), int(record.ttl), record.mx_pref, + record.provider_value2, ) @@ -89,13 +100,14 @@ def _record_sort_key(record: ProviderDNSRecord) -> tuple[str, str, str, int]: ) -def _load_record(raw: Any) -> ManagedDNSRecord: +def _load_record(raw: Any, provider: str) -> ManagedDNSRecord: if not isinstance(raw, dict): raise ValueError("managed DNS records must be objects") name = _record_name(_required_string(raw, "name")) record_type = _required_string(raw, "type").upper() - if record_type not in NAMECHEAP_SUPPORTED_RECORD_TYPES: - raise ValueError(f"unsupported Namecheap record type: {record_type}") + supported_types = _PROVIDER_RECORD_TYPES[provider] + if record_type not in supported_types: + raise ValueError(f"unsupported {provider} record type: {record_type}") value = _required_string(raw, "value") if ( "\n" in value @@ -114,6 +126,16 @@ def _load_record(raw: Any) -> ManagedDNSRecord: raise ValueError(f"record {name}/MX requires mx_pref") elif mx_pref is not None: raise ValueError(f"record {name}/{record_type} cannot set mx_pref") + provider_value2 = raw.get("provider_value2") + if provider_value2 is not None: + if provider != "dynadot" or record_type == "MX": + raise ValueError( + f"record {name}/{record_type} cannot set provider_value2" + ) + if not isinstance(provider_value2, str) or not provider_value2: + raise ValueError( + f"record {name}/{record_type} has invalid provider_value2" + ) replace = raw.get("replace", "all") if replace not in {"all", "value_prefix"}: @@ -136,11 +158,13 @@ def _load_record(raw: Any) -> ManagedDNSRecord: ): raise ValueError(f"record {name}/{record_type} has invalid conflicts") conflict_types = tuple(item.upper() for item in raw_conflicts) - if any(item not in NAMECHEAP_SUPPORTED_RECORD_TYPES for item in conflict_types): + if any(item not in supported_types for item in conflict_types): raise ValueError(f"record {name}/{record_type} has unsupported conflicts") return ManagedDNSRecord( - record=ProviderDNSRecord(name, record_type, value, ttl, mx_pref), + record=ProviderDNSRecord( + name, record_type, value, ttl, mx_pref, provider_value2 + ), replace=replace, match_value_prefix=match_value_prefix, conflict_types=conflict_types, @@ -163,8 +187,8 @@ def load_dns_manifest(path: Path) -> DNSManifest: if not _HOSTNAME.fullmatch(zone): raise ValueError("zone must be a DNS hostname") provider = _required_string(document, "provider") - if provider != "namecheap": - raise ValueError("contract version 1 supports only Namecheap") + if provider not in _PROVIDER_RECORD_TYPES: + raise ValueError("contract version 1 has an unsupported DNS provider") if document.get("preserve_unmanaged") is not True: raise ValueError("preserve_unmanaged must be true") raw_protected = document.get("protected_names", []) @@ -179,7 +203,7 @@ def load_dns_manifest(path: Path) -> DNSManifest: raise ValueError("records must be a non-empty list") if len(raw_records) > 100: raise ValueError("DNS manifest has too many managed records") - records = tuple(_load_record(raw) for raw in raw_records) + records = tuple(_load_record(raw, provider) for raw in raw_records) for managed in records: if _record_name(managed.record.name) in protected_names: raise ValueError( @@ -241,8 +265,8 @@ def _changes( before_set = {_record_identity(record) for record in before} after_set = {_record_identity(record) for record in after} changes: list[dict[str, Any]] = [] - for identity in sorted(before_set - after_set): - name, record_type, value, ttl, mx_pref = identity + for identity in sorted(before_set - after_set, key=repr): + name, record_type, value, ttl, mx_pref, provider_value2 = identity changes.append( { "action": "remove", @@ -251,10 +275,11 @@ def _changes( "value": value, "ttl": ttl, "mx_pref": mx_pref, + "provider_value2": provider_value2, } ) - for identity in sorted(after_set - before_set): - name, record_type, value, ttl, mx_pref = identity + for identity in sorted(after_set - before_set, key=repr): + name, record_type, value, ttl, mx_pref, provider_value2 = identity changes.append( { "action": "add", @@ -263,6 +288,7 @@ def _changes( "value": value, "ttl": ttl, "mx_pref": mx_pref, + "provider_value2": provider_value2, } ) return changes @@ -273,6 +299,7 @@ def offline_plan(manifest: DNSManifest) -> dict[str, Any]: "contract_version": 1, "app_id": manifest.app_id, "zone": manifest.zone, + "provider": manifest.provider, "mode": "offline-plan", "preserve_unmanaged": True, "protected_names": list(manifest.protected_names), @@ -283,6 +310,7 @@ def offline_plan(manifest: DNSManifest) -> dict[str, Any]: "value": entry.record.value, "ttl": entry.record.ttl, "mx_pref": entry.record.mx_pref, + "provider_value2": entry.record.provider_value2, "replace": entry.replace, } for entry in manifest.records @@ -290,9 +318,15 @@ def offline_plan(manifest: DNSManifest) -> dict[str, Any]: } -def _write_backup(zone: str, raw: str, backup_dir: Path) -> Path: +def _write_backup( + provider: str, + zone: str, + raw: str, + backup_dir: Path, + extension: str, +) -> Path: stamp = datetime.datetime.now(datetime.UTC).strftime("%Y%m%d_%H%M%S_%f") - path = backup_dir / "namecheap" / f"{zone}.{stamp}.xml" + path = backup_dir / provider / f"{zone}.{stamp}.{extension}" path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) with path.open("x", encoding="utf-8") as backup: backup.write(raw) @@ -307,11 +341,21 @@ def reconcile_dns_manifest( apply: bool = False, backup_dir: Path | None = None, ) -> dict[str, Any]: + if getattr(provider, "name", None) != manifest.provider: + raise DNSManifestError("provider credentials do not match the manifest") snapshot = provider.get_hosts(manifest.zone) if snapshot.status != "ok": raise DNSManifestError( f"provider fetch failed for {manifest.zone}: {snapshot.message}" ) + if manifest.provider == "dynadot": + if snapshot.zone_ttl is None: + raise DNSManifestError("Dynadot did not return the observed zone TTL") + desired_ttls = {entry.record.ttl for entry in manifest.records} + if desired_ttls != {snapshot.zone_ttl}: + raise DNSManifestError( + "Dynadot manifest TTL must equal the observed zone-wide TTL" + ) next_records = merge_dns_manifest_records( snapshot.records, manifest.records ) @@ -320,7 +364,14 @@ def reconcile_dns_manifest( if apply and changes: if backup_dir is None or not backup_dir.is_absolute(): raise DNSManifestError("live apply requires an absolute backup_dir") - backup_path = _write_backup(manifest.zone, snapshot.raw, backup_dir) + extension = getattr(provider, "backup_format", "txt") + backup_path = _write_backup( + manifest.provider, + manifest.zone, + snapshot.raw, + backup_dir, + extension, + ) provider.set_hosts( manifest.zone, next_records, mail_policy="preserve" ) @@ -340,6 +391,7 @@ def reconcile_dns_manifest( "contract_version": 1, "app_id": manifest.app_id, "zone": manifest.zone, + "provider": manifest.provider, "mode": "apply" if apply else "connected-plan", "status": ( "applied" @@ -360,3 +412,17 @@ def namecheap_provider( return NamecheapDNSProvider( credentials=credentials, client_ip=client_ip ) + + +def provider_from_credentials( + provider: str, + credentials: Path, + client_ip: str | None, +) -> NamecheapDNSProvider | DynadotDNSProvider: + if provider == "namecheap": + return namecheap_provider(credentials, client_ip) + if provider == "dynadot": + if client_ip is not None: + raise ValueError("--client-ip is only valid for Namecheap") + return DynadotDNSProvider(credentials=credentials) + raise ValueError("unsupported DNS provider") diff --git a/ampg/dns_manifest_cli.py b/ampg/dns_manifest_cli.py index 3f1de47..f5e00dd 100644 --- a/ampg/dns_manifest_cli.py +++ b/ampg/dns_manifest_cli.py @@ -10,8 +10,8 @@ import sys from .dns_manifest import ( DNSManifestError, load_dns_manifest, - namecheap_provider, offline_plan, + provider_from_credentials, reconcile_dns_manifest, ) @@ -27,7 +27,7 @@ def _parser() -> argparse.ArgumentParser: parser.add_argument( "--credentials", type=Path, - help="Namecheap key-value credential file; required for provider access", + help="Provider key-value credential file; required for provider access", ) parser.add_argument("--client-ip") parser.add_argument( @@ -60,8 +60,8 @@ def main(arguments: list[str] | None = None) -> int: raise ValueError("--apply-zone must exactly match manifest zone") if apply and options.backup_dir is None: raise ValueError("--backup-dir is required for live apply") - provider = namecheap_provider( - options.credentials, options.client_ip + provider = provider_from_credentials( + manifest.provider, options.credentials, options.client_ip ) result = reconcile_dns_manifest( manifest, @@ -72,7 +72,7 @@ def main(arguments: list[str] | None = None) -> int: json.dump(result, sys.stdout, indent=2, sort_keys=True) sys.stdout.write("\n") return 0 - except (DNSManifestError, OSError, ValueError) as error: + except (DNSManifestError, OSError, RuntimeError, ValueError) as error: print(f"ampg-dns-manifest: {error}", file=sys.stderr) return 1 diff --git a/ampg/dynadot_dns.py b/ampg/dynadot_dns.py new file mode 100644 index 0000000..1a5ed3f --- /dev/null +++ b/ampg/dynadot_dns.py @@ -0,0 +1,316 @@ +"""Guarded Dynadot REST v2 DNS provider. + +Dynadot updates replace the full zone and use one zone-wide TTL. The provider +therefore models every documented record field and rejects lossy or mixed-TTL +writes. Credentials are sent only in signed headers, never in URLs or output. +""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import ipaddress +import json +from pathlib import Path +import urllib.error +import urllib.parse +import urllib.request +import uuid +from typing import Any, Callable + +from .dns import ( + DEFAULT_TTL, + DNSZoneSnapshot, + ProviderDNSRecord, + _read_key_value_credentials, +) + + +DYNADOT_SUPPORTED_RECORD_TYPES = { + "A", + "AAAA", + "ANAME", + "CAA", + "CNAME", + "EMAIL", + "FORWARD", + "MX", + "SRV", + "STEALTH", + "TXT", +} +DYNADOT_API_ORIGINS = { + "https://api.dynadot.com", + "https://api-sandbox.dynadot.com", +} + + +def dynadot_signature( + api_key: str, + api_secret: str, + path_and_query: str, + request_id: str, + request_body: str = "", +) -> str: + message = "\n".join( + (api_key, path_and_query, request_id or "", request_body or "") + ) + digest = hmac.new( + api_secret.encode("utf-8"), + message.encode("utf-8"), + hashlib.sha256, + ).digest() + return base64.b64encode(digest).decode("ascii") + + +def _validated_origin(value: str) -> str: + parsed = urllib.parse.urlsplit(value.rstrip("/")) + origin = f"{parsed.scheme}://{parsed.netloc}" + if ( + origin not in DYNADOT_API_ORIGINS + or parsed.path + or parsed.query + or parsed.fragment + or parsed.username + ): + raise ValueError("Dynadot api_base_url is not an approved HTTPS origin") + return origin + + +def _validated_source_cidr(value: str) -> str: + try: + network = ipaddress.ip_network(value, strict=True) + except ValueError as error: + raise ValueError("Dynadot allowed_source_cidr must be a valid CIDR") from error + if network.version != 4 or network.prefixlen != 32: + raise ValueError("Dynadot allowed_source_cidr must be an IPv4 /32") + return str(network) + + +def _required_mapping(value: Any, message: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise RuntimeError(message) + return value + + +def _record_value(record: dict[str, Any], name: str) -> str | None: + value = record.get(name) + if value is None: + return None + if not isinstance(value, (str, int)): + raise RuntimeError("Dynadot returned an unsupported DNS record shape") + return str(value) + + +class DynadotDNSProvider: + name = "dynadot" + backup_format = "json" + + def __init__( + self, + *, + credentials: Path, + opener: Callable[..., Any] | None = None, + request_id_factory: Callable[[], str] | None = None, + ): + config = _read_key_value_credentials(credentials) + self.api_base_url = _validated_origin( + config.get("api_base_url", "") + ) + self.api_key = config.get("api_key") + self.api_secret = config.get("api_secret") + self.allowed_source_cidr = _validated_source_cidr( + config.get("allowed_source_cidr", "") + ) + if not self.api_key: + 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._request_id_factory = request_id_factory or ( + lambda: str(uuid.uuid4()) + ) + + def get_hosts(self, domain: str) -> DNSZoneSnapshot: + response = self._request("GET", self._records_path(domain)) + data = _required_mapping( + response.get("data"), "Dynadot DNS response omitted data" + ) + glue = _required_mapping( + data.get("glue_info"), "Dynadot DNS response omitted glue_info" + ) + glue_type = str(glue.get("glue_type", "")).upper() + if glue_type and glue_type != "DNS": + raise RuntimeError("Dynadot zone is not using Dynadot DNS") + ttl = glue.get("ttl", DEFAULT_TTL) + if not isinstance(ttl, int) or ttl < 60 or ttl > 86400: + raise RuntimeError("Dynadot returned an invalid zone TTL") + records: list[ProviderDNSRecord] = [] + for field, name_field in ( + ("dns_main_list", None), + ("dns_sub_list", "sub_host"), + ): + raw_records = glue.get(field, []) + if not isinstance(raw_records, list): + raise RuntimeError("Dynadot returned an unsupported DNS record list") + for raw_record in raw_records: + item = _required_mapping( + raw_record, "Dynadot returned an unsupported DNS record" + ) + record_type = str(item.get("record_type", "")).upper() + if record_type not in DYNADOT_SUPPORTED_RECORD_TYPES: + raise RuntimeError( + "Dynadot returned an unsupported DNS record type" + ) + name = "@" if name_field is None else _record_value(item, name_field) + value1 = _record_value(item, "value1") + value2 = _record_value(item, "value2") + if not name or value1 is None: + raise RuntimeError("Dynadot returned an incomplete DNS record") + mx_pref = None + provider_value2 = value2 + if record_type == "MX": + try: + mx_pref = int(value2) if value2 is not None else None + except ValueError as error: + raise RuntimeError( + "Dynadot returned an invalid MX preference" + ) from error + if mx_pref is None or not 0 <= mx_pref <= 65535: + raise RuntimeError("Dynadot returned an invalid MX preference") + provider_value2 = None + records.append( + ProviderDNSRecord( + name=name, + type=record_type, + value=value1, + ttl=ttl, + mx_pref=mx_pref, + provider_value2=provider_value2, + ) + ) + raw = json.dumps(response, indent=2, sort_keys=True) + "\n" + return DNSZoneSnapshot( + self.name, + domain, + raw, + tuple(records), + "ok", + "zone fetched from Dynadot", + ttl, + ) + + def set_hosts( + self, + domain: str, + records: tuple[ProviderDNSRecord, ...], + *, + mail_policy: str, + ) -> None: + if mail_policy != "preserve": + raise ValueError("Dynadot full-zone updates require mail_policy=preserve") + if not records: + raise ValueError("refusing to replace a Dynadot zone with no records") + ttls = {int(record.ttl) for record in records} + if len(ttls) != 1: + raise ValueError("Dynadot full-zone updates require one observed zone TTL") + ttl = next(iter(ttls)) + if ttl < 60 or ttl > 86400: + raise ValueError("Dynadot zone TTL is invalid") + main: list[dict[str, str]] = [] + sub: list[dict[str, str]] = [] + for record in records: + record_type = record.type.upper() + if record_type not in DYNADOT_SUPPORTED_RECORD_TYPES: + raise ValueError("Dynadot record type is unsupported") + item = { + "record_type": record_type, + "value1": record.value, + } + value2 = ( + str(record.mx_pref) + if record_type == "MX" + else record.provider_value2 + ) + if record_type == "MX" and record.mx_pref is None: + raise ValueError("Dynadot MX records require mx_pref") + if value2 is not None: + item["value2"] = value2 + normalized_name = record.name.strip().rstrip(".").lower() or "@" + if normalized_name == "@": + main.append(item) + else: + item["sub_host"] = normalized_name + sub.append(item) + if len(main) > 20 or len(sub) > 100: + raise ValueError("Dynadot zone exceeds documented record limits") + body = { + "dns_main_list": main, + "dns_sub_list": sub, + "ttl": ttl, + "add_dns_to_current_setting": False, + } + self._request("POST", self._records_path(domain), body) + + @staticmethod + def _records_path(domain: str) -> str: + normalized = domain.strip().rstrip(".").lower() + if not normalized or "/" in normalized: + raise ValueError("invalid Dynadot domain") + return "/restful/v2/domains/" + urllib.parse.quote(normalized) + "/records" + + def _request( + self, + method: str, + path: str, + body: dict[str, Any] | None = None, + ) -> dict[str, Any]: + request_body = ( + json.dumps(body, sort_keys=True, separators=(",", ":")) + if body is not None + else "" + ) + request_id = self._request_id_factory() + headers = { + "Accept": "application/json", + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json", + "X-Request-ID": request_id, + "X-Signature": dynadot_signature( + self.api_key, + self.api_secret, + path, + request_id, + request_body, + ), + } + request = urllib.request.Request( + self.api_base_url + path, + data=request_body.encode("utf-8") if body is not None else None, + headers=headers, + method=method, + ) + try: + with self._opener(request, timeout=30) as response: + raw = response.read().decode("utf-8") + except urllib.error.HTTPError as error: + raise RuntimeError( + f"Dynadot API rejected {method} {path} with HTTP {error.code}" + ) from error + except urllib.error.URLError as error: + raise RuntimeError( + f"Dynadot API connectivity failed for {method} {path}" + ) from error + try: + document = json.loads(raw) + except json.JSONDecodeError as error: + raise RuntimeError("Dynadot API returned non-JSON data") from error + if not isinstance(document, dict): + raise RuntimeError("Dynadot API returned an unsupported response") + code = document.get("code") + if code is not None and str(code) not in {"200", "201", "0"}: + raise RuntimeError(f"Dynadot API rejected {method} {path}") + if document.get("success") is False: + raise RuntimeError(f"Dynadot API rejected {method} {path}") + return document diff --git a/docs/domain-onboarding.md b/docs/domain-onboarding.md index 3a8e6c9..8d571a6 100644 --- a/docs/domain-onboarding.md +++ b/docs/domain-onboarding.md @@ -1,6 +1,6 @@ # Domain Onboarding -> Status: draft | Updated 2026-07-07 | Applies to: clearnet DNS setup +> Status: reviewed adapter contract | Updated 2026-07-31 | Applies to: clearnet DNS setup AMPG treats clearnet naming as an operator choice. A site can use a normal owned domain, a community free subdomain, a Dynamic DNS name, or no clearnet name at all when only @@ -42,6 +42,15 @@ whole zone before applying. The documented Namecheap API record types do not inc unsupported record types and relies on TXT discovery hints unless another provider path is configured. +Dynadot REST v2 is also supported. Its DNS update replaces the complete zone and +uses one zone-wide TTL, so AMPG preserves every documented `value1`, `value2`, and +`sub_host` field and refuses a write if the merged records have mixed TTLs. The +Dynadot credential input must contain `api_base_url`, `api_key`, `api_secret`, and +an operator-observed IPv4 `/32` `allowed_source_cidr`. Keys and signatures travel +only in request headers. Because Dynadot labels REST v2 beta, production use also +requires a pinned adapter commit, a sandbox proof, a connected production plan, +and a private raw backup before the exact-zone apply. + Application and service onboarding can use a secretless JSON manifest without creating a synthetic `gateway.toml`: @@ -59,6 +68,11 @@ ampg-dns-manifest \ --apply-zone example.com ``` +For Dynadot, omit `--client-ip` and provide the ephemeral four-field Dynadot +credential file. A connected plan first observes the existing zone-wide TTL. +The secretless manifest must be rendered with that exact TTL before apply; +AMPG fails closed instead of changing the TTL on unmanaged records. + The generic writer is also read-only by default. The exact `--apply-zone` sentinel and an absolute backup directory are mandatory for a write. `protected_names` can forbid all changes at sensitive names such as `@`. diff --git a/tests/test_dns_manifest.py b/tests/test_dns_manifest.py index 8aa74df..cce78c7 100644 --- a/tests/test_dns_manifest.py +++ b/tests/test_dns_manifest.py @@ -44,6 +44,7 @@ def manifest_document(): class FakeProvider: name = "namecheap" + backup_format = "xml" def __init__(self, records): self.records = tuple(records) @@ -134,6 +135,62 @@ class DNSManifestTests(unittest.TestCase): self.assertEqual("verified", second["status"]) self.assertEqual(1, provider.writes) + def test_dynadot_requires_observed_zone_ttl(self): + document = manifest_document() + document["provider"] = "dynadot" + document["records"][0]["conflict_types"] = ["CNAME", "ANAME"] + self.path.write_text(json.dumps(document), encoding="utf-8") + manifest = load_dns_manifest(self.path) + provider = FakeProvider([]) + provider.name = "dynadot" + with self.assertRaisesRegex(Exception, "observed zone TTL"): + reconcile_dns_manifest(manifest, provider) + + def test_provider_credentials_must_match_manifest(self): + manifest = load_dns_manifest(self.path) + provider = FakeProvider([]) + provider.name = "dynadot" + with self.assertRaisesRegex(Exception, "do not match"): + reconcile_dns_manifest(manifest, provider) + + def test_dynadot_apply_uses_json_backup_and_is_idempotent(self): + document = manifest_document() + document["provider"] = "dynadot" + document["records"][0]["conflict_types"] = ["CNAME", "ANAME"] + self.path.write_text(json.dumps(document), encoding="utf-8") + manifest = load_dns_manifest(self.path) + + class FakeDynadot(FakeProvider): + name = "dynadot" + backup_format = "json" + + def get_hosts(self, zone): + return DNSZoneSnapshot( + self.name, + zone, + '{"zone":"snapshot"}\n', + self.records, + "ok", + "ok", + 300, + ) + + provider = FakeDynadot( + [ProviderDNSRecord("@", "A", "153.75.248.86", 300)] + ) + backup_dir = Path(self.temporary.name) / "backups" + result = reconcile_dns_manifest( + manifest, provider, apply=True, backup_dir=backup_dir + ) + self.assertEqual("applied", result["status"]) + self.assertTrue(result["backup_path"].endswith(".json")) + self.assertEqual(1, provider.writes) + second = reconcile_dns_manifest( + manifest, provider, apply=True, backup_dir=backup_dir + ) + self.assertEqual("verified", second["status"]) + self.assertEqual(1, provider.writes) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_dynadot_dns.py b/tests/test_dynadot_dns.py new file mode 100644 index 0000000..1765acb --- /dev/null +++ b/tests/test_dynadot_dns.py @@ -0,0 +1,158 @@ +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_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()