feat(dns): allow exact protected-name record exceptions

ampg/dns_manifest.py

@@ -59,6 +59,7 @@ class DNSManifest:
     provider: str
     preserve_unmanaged: bool
     protected_names: tuple[str, ...]
+    protected_name_exceptions: tuple[tuple[str, str], ...]
     records: tuple[ManagedDNSRecord, ...]
 
 
@@ -198,6 +199,32 @@ def load_dns_manifest(path: Path) -> DNSManifest:
     ):
         raise ValueError("protected_names must be a list")
     protected_names = tuple(_record_name(item) for item in raw_protected)
+    raw_exceptions = document.get("protected_name_exceptions", [])
+    if not isinstance(raw_exceptions, list):
+        raise ValueError("protected_name_exceptions must be a list")
+    protected_name_exceptions: tuple[tuple[str, str], ...] = tuple(
+        (
+            _record_name(_required_string(item, "name")),
+            _required_string(item, "type").upper(),
+        )
+        for item in raw_exceptions
+        if isinstance(item, dict) and set(item) == {"name", "type"}
+    )
+    if len(protected_name_exceptions) != len(raw_exceptions):
+        raise ValueError(
+            "protected_name_exceptions entries require only name and type"
+        )
+    if len(set(protected_name_exceptions)) != len(protected_name_exceptions):
+        raise ValueError("protected_name_exceptions contains a duplicate")
+    for name, record_type in protected_name_exceptions:
+        if name not in protected_names:
+            raise ValueError(
+                "protected_name_exceptions must target a protected name"
+            )
+        if record_type not in _PROVIDER_RECORD_TYPES[provider]:
+            raise ValueError(
+                "protected_name_exceptions contains an unsupported type"
+            )
     raw_records = document.get("records")
     if not isinstance(raw_records, list) or not raw_records:
         raise ValueError("records must be a non-empty list")
@@ -205,10 +232,25 @@ def load_dns_manifest(path: Path) -> DNSManifest:
         raise ValueError("DNS manifest has too many managed 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:
+        managed_identity = (
+            _record_name(managed.record.name),
+            managed.record.type.upper(),
+        )
+        if (
+            managed_identity[0] in protected_names
+            and managed_identity not in protected_name_exceptions
+        ):
             raise ValueError(
                 f"managed record targets protected name {managed.record.name}"
             )
+    managed_name_types = {
+        (_record_name(managed.record.name), managed.record.type.upper())
+        for managed in records
+    }
+    if not set(protected_name_exceptions).issubset(managed_name_types):
+        raise ValueError(
+            "protected_name_exceptions must match a managed record"
+        )
     identities = [_record_identity(managed.record) for managed in records]
     if len(identities) != len(set(identities)):
         raise ValueError("DNS manifest contains a duplicate record")
@@ -219,6 +261,7 @@ def load_dns_manifest(path: Path) -> DNSManifest:
         provider=provider,
         preserve_unmanaged=True,
         protected_names=protected_names,
+        protected_name_exceptions=protected_name_exceptions,
         records=records,
     )
 
@@ -303,6 +346,10 @@ def offline_plan(manifest: DNSManifest) -> dict[str, Any]:
         "mode": "offline-plan",
         "preserve_unmanaged": True,
         "protected_names": list(manifest.protected_names),
+        "protected_name_exceptions": [
+            {"name": name, "type": record_type}
+            for name, record_type in manifest.protected_name_exceptions
+        ],
         "managed_records": [
             {
                 "name": entry.record.name,

tests/test_dns_manifest.py

@@ -108,6 +108,58 @@ class DNSManifestTests(unittest.TestCase):
         with self.assertRaisesRegex(ValueError, "protected"):
             load_dns_manifest(self.path)
 
+    def test_exact_protected_apex_txt_exception_is_allowed(self):
+        document = manifest_document()
+        document["records"][1]["name"] = "@"
+        document["records"][1]["value"] = (
+            "v=spf1 include:spf.efwd.registrar-servers.com "
+            "include:mail.suchshop.lol ~all"
+        )
+        document["protected_name_exceptions"] = [
+            {"name": "@", "type": "TXT"}
+        ]
+        self.path.write_text(json.dumps(document), encoding="utf-8")
+        manifest = load_dns_manifest(self.path)
+        self.assertEqual((("@", "TXT"),), manifest.protected_name_exceptions)
+
+        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,
+            ),
+        )
+        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.assertNotIn(
+            (
+                "@",
+                "TXT",
+                "v=spf1 include:spf.efwd.registrar-servers.com ~all",
+            ),
+            identities,
+        )
+
+    def test_protected_exception_must_match_managed_name_and_type(self):
+        document = manifest_document()
+        document["protected_name_exceptions"] = [
+            {"name": "@", "type": "TXT"}
+        ]
+        self.path.write_text(json.dumps(document), encoding="utf-8")
+        with self.assertRaisesRegex(ValueError, "match a managed record"):
+            load_dns_manifest(self.path)
+
     def test_unresolved_values_are_rejected(self):
         document = manifest_document()
         document["records"][0]["value"] = "{{ prod_ip }}"