diff --git a/ampg/dns_manifest.py b/ampg/dns_manifest.py new file mode 100644 index 0000000..8b5087d --- /dev/null +++ b/ampg/dns_manifest.py @@ -0,0 +1,362 @@ +"""Generic, preserve-unmanaged DNS manifest reconciliation. + +The existing AMPG DNS provider already performs safe Namecheap getHosts and +setHosts calls. This module adds a provider-neutral manifest projection for +application onboarding without coupling it to a gateway.toml transport plan. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import datetime +import json +import os +from pathlib import Path +import re +from typing import Any, Iterable + +from .dns import ( + DEFAULT_TTL, + NAMECHEAP_SUPPORTED_RECORD_TYPES, + NamecheapDNSProvider, + ProviderDNSRecord, +) + + +_HOSTNAME = re.compile( + r"(?=^.{1,253}$)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+" + r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$" +) +_RELATIVE_NAME = re.compile( + r"@|(?:[a-z0-9_*](?:[a-z0-9_*-]{0,61}[a-z0-9_*])?\.)*" + r"[a-z0-9_*](?:[a-z0-9_*-]{0,61}[a-z0-9_*])?" +) + + +@dataclass(frozen=True) +class ManagedDNSRecord: + record: ProviderDNSRecord + replace: str + match_value_prefix: str | None + conflict_types: tuple[str, ...] + + +@dataclass(frozen=True) +class DNSManifest: + contract_version: int + app_id: str + zone: str + provider: str + preserve_unmanaged: bool + protected_names: tuple[str, ...] + records: tuple[ManagedDNSRecord, ...] + + +class DNSManifestError(RuntimeError): + """Safe-to-display manifest or reconciliation failure.""" + + +def _required_string(document: dict[str, Any], name: str) -> str: + value = document.get(name) + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{name} must be a non-empty string") + return value.strip() + + +def _record_name(value: str) -> str: + normalized = value.strip().rstrip(".").lower() or "@" + if len(normalized) > 253 or not _RELATIVE_NAME.fullmatch(normalized): + raise ValueError(f"invalid relative DNS name: {value}") + return normalized + + +def _record_identity(record: ProviderDNSRecord) -> tuple[Any, ...]: + return ( + _record_name(record.name), + record.type.upper(), + record.value.strip(), + int(record.ttl), + record.mx_pref, + ) + + +def _record_sort_key(record: ProviderDNSRecord) -> tuple[str, str, str, int]: + return ( + _record_name(record.name), + record.type.upper(), + record.value.strip(), + int(record.mx_pref or 0), + ) + + +def _load_record(raw: Any) -> 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}") + value = _required_string(raw, "value") + if ( + "\n" in value + or "\r" in value + or "{{" in value + or "${" in value + or (value.startswith("<") and value.endswith(">")) + ): + raise ValueError(f"record {name}/{record_type} has an unresolved value") + ttl = raw.get("ttl", DEFAULT_TTL) + if not isinstance(ttl, int) or ttl < 60 or ttl > 86400: + raise ValueError(f"record {name}/{record_type} has an invalid TTL") + mx_pref = raw.get("mx_pref") + if record_type == "MX": + if not isinstance(mx_pref, int) or not 0 <= mx_pref <= 65535: + 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") + + replace = raw.get("replace", "all") + if replace not in {"all", "value_prefix"}: + raise ValueError(f"record {name}/{record_type} has invalid replace mode") + match_value_prefix = raw.get("match_value_prefix") + if replace == "value_prefix": + if not isinstance(match_value_prefix, str) or not match_value_prefix: + raise ValueError( + f"record {name}/{record_type} requires match_value_prefix" + ) + elif match_value_prefix is not None: + raise ValueError( + f"record {name}/{record_type} cannot set match_value_prefix" + ) + + raw_conflicts = raw.get("conflict_types", []) + if ( + not isinstance(raw_conflicts, list) + or not all(isinstance(item, str) for item in raw_conflicts) + ): + 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): + raise ValueError(f"record {name}/{record_type} has unsupported conflicts") + + return ManagedDNSRecord( + record=ProviderDNSRecord(name, record_type, value, ttl, mx_pref), + replace=replace, + match_value_prefix=match_value_prefix, + conflict_types=conflict_types, + ) + + +def load_dns_manifest(path: Path) -> DNSManifest: + if path.is_symlink() or not path.is_file(): + raise ValueError("DNS manifest must be a regular, non-symlink file") + try: + document = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError as error: + raise ValueError("DNS manifest is not valid JSON") from error + if not isinstance(document, dict) or document.get("contract_version") != 1: + raise ValueError("unsupported DNS manifest contract") + app_id = _required_string(document, "app_id") + if not re.fullmatch(r"[a-z][a-z0-9_]{1,63}", app_id): + raise ValueError("app_id is invalid") + zone = _required_string(document, "zone").lower().rstrip(".") + 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 document.get("preserve_unmanaged") is not True: + raise ValueError("preserve_unmanaged must be true") + raw_protected = document.get("protected_names", []) + if ( + not isinstance(raw_protected, list) + or not all(isinstance(item, str) for item in raw_protected) + ): + raise ValueError("protected_names must be a list") + protected_names = tuple(_record_name(item) for item in raw_protected) + raw_records = document.get("records") + if not isinstance(raw_records, list) or not raw_records: + 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) + for managed in records: + if _record_name(managed.record.name) in protected_names: + raise ValueError( + f"managed record targets protected name {managed.record.name}" + ) + identities = [_record_identity(managed.record) for managed in records] + if len(identities) != len(set(identities)): + raise ValueError("DNS manifest contains a duplicate record") + return DNSManifest( + contract_version=1, + app_id=app_id, + zone=zone, + provider=provider, + preserve_unmanaged=True, + protected_names=protected_names, + records=records, + ) + + +def _is_replaced( + existing: ProviderDNSRecord, managed: ManagedDNSRecord +) -> bool: + existing_name = _record_name(existing.name) + desired_name = _record_name(managed.record.name) + existing_type = existing.type.upper() + desired_type = managed.record.type.upper() + if existing_name != desired_name: + return False + if existing_type in managed.conflict_types: + return True + if existing_type != desired_type: + return False + if managed.replace == "all": + return True + assert managed.match_value_prefix is not None + return existing.value.lower().startswith( + managed.match_value_prefix.lower() + ) + + +def merge_dns_manifest_records( + existing: Iterable[ProviderDNSRecord], + managed_records: Iterable[ManagedDNSRecord], +) -> tuple[ProviderDNSRecord, ...]: + managed = tuple(managed_records) + retained = [ + record + for record in existing + if not any(_is_replaced(record, desired) for desired in managed) + ] + desired = [entry.record for entry in managed] + return tuple(sorted(retained + desired, key=_record_sort_key)) + + +def _changes( + before: Iterable[ProviderDNSRecord], + after: Iterable[ProviderDNSRecord], +) -> list[dict[str, Any]]: + 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 + changes.append( + { + "action": "remove", + "name": name, + "type": record_type, + "value": value, + "ttl": ttl, + "mx_pref": mx_pref, + } + ) + for identity in sorted(after_set - before_set): + name, record_type, value, ttl, mx_pref = identity + changes.append( + { + "action": "add", + "name": name, + "type": record_type, + "value": value, + "ttl": ttl, + "mx_pref": mx_pref, + } + ) + return changes + + +def offline_plan(manifest: DNSManifest) -> dict[str, Any]: + return { + "contract_version": 1, + "app_id": manifest.app_id, + "zone": manifest.zone, + "mode": "offline-plan", + "preserve_unmanaged": True, + "protected_names": list(manifest.protected_names), + "managed_records": [ + { + "name": entry.record.name, + "type": entry.record.type, + "value": entry.record.value, + "ttl": entry.record.ttl, + "mx_pref": entry.record.mx_pref, + "replace": entry.replace, + } + for entry in manifest.records + ], + } + + +def _write_backup(zone: str, raw: str, backup_dir: Path) -> Path: + stamp = datetime.datetime.now(datetime.UTC).strftime("%Y%m%d_%H%M%S_%f") + path = backup_dir / "namecheap" / f"{zone}.{stamp}.xml" + path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + with path.open("x", encoding="utf-8") as backup: + backup.write(raw) + os.chmod(path, 0o600) + return path + + +def reconcile_dns_manifest( + manifest: DNSManifest, + provider: Any, + *, + apply: bool = False, + backup_dir: Path | None = None, +) -> dict[str, Any]: + snapshot = provider.get_hosts(manifest.zone) + if snapshot.status != "ok": + raise DNSManifestError( + f"provider fetch failed for {manifest.zone}: {snapshot.message}" + ) + next_records = merge_dns_manifest_records( + snapshot.records, manifest.records + ) + changes = _changes(snapshot.records, next_records) + backup_path = None + 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) + provider.set_hosts( + manifest.zone, next_records, mail_policy="preserve" + ) + verified = provider.get_hosts(manifest.zone) + if verified.status != "ok": + raise DNSManifestError("post-apply provider verification failed") + projected = merge_dns_manifest_records( + verified.records, manifest.records + ) + if {_record_identity(record) for record in projected} != { + _record_identity(record) for record in verified.records + }: + raise DNSManifestError( + "post-apply zone does not match the managed projection" + ) + return { + "contract_version": 1, + "app_id": manifest.app_id, + "zone": manifest.zone, + "mode": "apply" if apply else "connected-plan", + "status": ( + "applied" + if apply and changes + else "planned" + if changes + else "verified" + ), + "preserve_unmanaged": True, + "changes": changes, + "backup_path": str(backup_path) if backup_path else None, + } + + +def namecheap_provider( + credentials: Path, client_ip: str | None +) -> NamecheapDNSProvider: + return NamecheapDNSProvider( + credentials=credentials, client_ip=client_ip + ) diff --git a/ampg/dns_manifest_cli.py b/ampg/dns_manifest_cli.py new file mode 100644 index 0000000..3f1de47 --- /dev/null +++ b/ampg/dns_manifest_cli.py @@ -0,0 +1,81 @@ +"""CLI for generic preserve-unmanaged DNS manifest reconciliation.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +import sys + +from .dns_manifest import ( + DNSManifestError, + load_dns_manifest, + namecheap_provider, + offline_plan, + reconcile_dns_manifest, +) + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="ampg-dns-manifest") + parser.add_argument("--manifest", required=True, type=Path) + parser.add_argument( + "--offline", + action="store_true", + help="validate and render the manifest without provider access", + ) + parser.add_argument( + "--credentials", + type=Path, + help="Namecheap key-value credential file; required for provider access", + ) + parser.add_argument("--client-ip") + parser.add_argument( + "--backup-dir", + type=Path, + help="absolute backup root; required for live apply", + ) + parser.add_argument( + "--apply-zone", + help="exact zone sentinel required for writes; omit for read-only plan", + ) + return parser + + +def main(arguments: list[str] | None = None) -> int: + options = _parser().parse_args(arguments) + try: + manifest = load_dns_manifest(options.manifest) + if options.offline: + if options.apply_zone or options.credentials: + raise ValueError( + "--offline cannot be combined with credentials or apply" + ) + result = offline_plan(manifest) + else: + if options.credentials is None: + raise ValueError("--credentials is required") + apply = options.apply_zone is not None + if apply and options.apply_zone != manifest.zone: + 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 + ) + result = reconcile_dns_manifest( + manifest, + provider, + apply=apply, + backup_dir=options.backup_dir, + ) + json.dump(result, sys.stdout, indent=2, sort_keys=True) + sys.stdout.write("\n") + return 0 + except (DNSManifestError, OSError, ValueError) as error: + print(f"ampg-dns-manifest: {error}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/docs/domain-onboarding.md b/docs/domain-onboarding.md index 0808308..3a8e6c9 100644 --- a/docs/domain-onboarding.md +++ b/docs/domain-onboarding.md @@ -42,6 +42,31 @@ 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. +Application and service onboarding can use a secretless JSON manifest without +creating a synthetic `gateway.toml`: + +```sh +ampg-dns-manifest --manifest ~/Build/example/dns-manifest.json --offline +ampg-dns-manifest \ + --manifest ~/Build/example/dns-manifest.json \ + --credentials /tmp/ephemeral-namecheap.ini \ + --client-ip 203.0.113.10 +ampg-dns-manifest \ + --manifest ~/Build/example/dns-manifest.json \ + --credentials /tmp/ephemeral-namecheap.ini \ + --client-ip 203.0.113.10 \ + --backup-dir ~/Build/example/dns-backups \ + --apply-zone example.com +``` + +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 `@`. +Each managed record explicitly chooses whether to replace all values at its +name/type or only values with a known prefix; unrelated TXT values survive. +Credential files must be short-lived inputs created by the approved secret +handoff and must never be committed or placed in Seafile. + `dns plan --free-domain-hints` prints `AMPG_FREE_DOMAIN_HINT` rows for optional third-party services that may provide free subdomains for personal sites, hobby apps, or open-source projects. AMPG does not register these names automatically because each diff --git a/pyproject.toml b/pyproject.toml index 4410970..4ab1c4a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,6 +7,7 @@ license = "Apache-2.0" [project.scripts] ampg = "ampg.cli:main" +ampg-dns-manifest = "ampg.dns_manifest_cli:main" [tool.pytest.ini_options] testpaths = ["tests"] diff --git a/tests/test_dns_manifest.py b/tests/test_dns_manifest.py new file mode 100644 index 0000000..8aa74df --- /dev/null +++ b/tests/test_dns_manifest.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +import json +from pathlib import Path +import tempfile +import unittest + +from ampg.dns import DNSZoneSnapshot, ProviderDNSRecord +from ampg.dns_manifest import ( + load_dns_manifest, + merge_dns_manifest_records, + reconcile_dns_manifest, +) + + +def manifest_document(): + return { + "contract_version": 1, + "app_id": "bloomword", + "zone": "bloomword.earth", + "provider": "namecheap", + "preserve_unmanaged": True, + "protected_names": ["@"], + "records": [ + { + "name": "shop", + "type": "A", + "value": "94.72.115.61", + "ttl": 300, + "replace": "all", + "conflict_types": ["CNAME", "ALIAS"], + }, + { + "name": "shop", + "type": "TXT", + "value": "v=spf1 include:mail.baby ~all", + "ttl": 300, + "replace": "value_prefix", + "match_value_prefix": "v=spf1", + }, + ], + } + + +class FakeProvider: + name = "namecheap" + + def __init__(self, records): + self.records = tuple(records) + self.writes = 0 + + def get_hosts(self, zone): + return DNSZoneSnapshot( + "namecheap", zone, "", self.records, "ok", "ok" + ) + + def set_hosts(self, zone, records, *, mail_policy): + self.writes += 1 + self.records = tuple(records) + + +class DNSManifestTests(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.path = Path(self.temporary.name) / "dns.json" + self.path.write_text( + json.dumps(manifest_document()), encoding="utf-8" + ) + + def tearDown(self): + self.temporary.cleanup() + + def test_merge_preserves_apex_mail_and_unrelated_shop_txt(self): + manifest = load_dns_manifest(self.path) + existing = ( + ProviderDNSRecord("@", "A", "153.75.248.86", 1800), + ProviderDNSRecord( + "@", "MX", "eforward1.registrar-servers.com.", 1800, 10 + ), + ProviderDNSRecord( + "@", + "TXT", + "v=spf1 include:spf.efwd.registrar-servers.com ~all", + 1800, + ), + ProviderDNSRecord("shop", "CNAME", "parking.example.", 1800), + ProviderDNSRecord("shop", "TXT", "verification=keep", 1800), + ProviderDNSRecord("shop", "TXT", "v=spf1 -all", 1800), + ) + merged = merge_dns_manifest_records(existing, manifest.records) + identities = { + (record.name, record.type, record.value) for record in merged + } + self.assertIn(("@", "A", "153.75.248.86"), identities) + self.assertIn( + ("@", "MX", "eforward1.registrar-servers.com."), identities + ) + self.assertIn(("shop", "TXT", "verification=keep"), identities) + self.assertNotIn(("shop", "CNAME", "parking.example."), identities) + self.assertNotIn(("shop", "TXT", "v=spf1 -all"), identities) + self.assertIn(("shop", "A", "94.72.115.61"), identities) + + def test_protected_apex_cannot_be_managed(self): + document = manifest_document() + document["records"][0]["name"] = "@" + self.path.write_text(json.dumps(document), encoding="utf-8") + with self.assertRaisesRegex(ValueError, "protected"): + load_dns_manifest(self.path) + + def test_unresolved_values_are_rejected(self): + document = manifest_document() + document["records"][0]["value"] = "{{ prod_ip }}" + self.path.write_text(json.dumps(document), encoding="utf-8") + with self.assertRaisesRegex(ValueError, "unresolved"): + load_dns_manifest(self.path) + + def test_connected_plan_is_read_only_and_apply_is_idempotent(self): + manifest = load_dns_manifest(self.path) + provider = FakeProvider( + [ProviderDNSRecord("@", "A", "153.75.248.86", 1800)] + ) + planned = reconcile_dns_manifest(manifest, provider) + self.assertEqual("planned", planned["status"]) + self.assertEqual(0, provider.writes) + backup_dir = Path(self.temporary.name) / "backups" + applied = reconcile_dns_manifest( + manifest, provider, apply=True, backup_dir=backup_dir + ) + self.assertEqual("applied", applied["status"]) + 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()