feat: add allowed_record_types and enforce append_only as superset check

README.md

@@ -140,6 +140,58 @@ environments:
             acme: true
 ```
 
+##### Allowed record types
+
+Under a `zone` the option `allowed_record_types` can be set to a list of DNS record types.
+
+Only the listed types may be written. Any attempt to write a different type is rejected with `HTTP 403`.
+An empty list (the default) means all record types are permitted.
+
+```yaml
+...
+environments:
+    - name: "Test1"
+      zones:
+        - name: "example.com"
+          allowed_record_types:
+            - "TXT"
+            - "AAAA"
+```
+
+##### Append only
+
+Under a `zone` the option `append_only: true` can be set.
+
+This prevents any write from removing existing records:
+
+- `DELETE` changesets are rejected with `HTTP 403`.
+- `REPLACE` changesets are verified against the live PowerDNS state. The incoming records must contain **all records that are currently in PowerDNS** for that name and type. Any missing record results in `HTTP 403`.
+
+```yaml
+...
+environments:
+    - name: "Test1"
+      zones:
+        - name: "example.com"
+          append_only: true
+```
+
+`allowed_record_types` and `append_only` can be combined. The example below creates a token that can read all zones and may only ever add TXT records — never update or remove any:
+
+```yaml
+...
+environments:
+  - name: "append-txt-only"
+    token_sha512: "..."
+    global_read_only: true
+    zones:
+      - name: ".*"
+        regex: true
+        allowed_record_types:
+          - "TXT"
+        append_only: true
+```
+
 ##### Admin
 
 Under a `zone` `admin` rights can be defined.

powerdns_api_proxy/config.py

@@ -152,6 +152,16 @@ def check_rrset_allowed(zone: ProxyConfigZone, rrset: RRSET) -> bool:
     if zone.read_only:
         return False
 
+    if zone.append_only and rrset.get("changetype") == "DELETE":
+        logger.debug(f"RRSET {rrset['name']} not allowed: DELETE changetype blocked by append_only")
+        return False
+
+    if zone.allowed_record_types and rrset.get("type") not in zone.allowed_record_types:
+        logger.debug(
+            f"RRSET {rrset['name']} not allowed: type {rrset.get('type')} not in allowed_record_types {zone.allowed_record_types}"
+        )
+        return False
+
     if zone.all_records:
         return True
 
@@ -173,6 +183,41 @@ def check_rrset_allowed(zone: ProxyConfigZone, rrset: RRSET) -> bool:
     return False
 
 
+def check_append_only_records_intact(
+    existing_rrsets: list[dict], incoming_rrset: RRSET
+) -> bool:
+    """
+    For append_only zones, verify that a REPLACE changeset does not remove any
+    existing records from the RRset.
+
+    Finds the existing RRset matching name+type and checks that every existing
+    record content is still present in the incoming records list.
+
+    Returns True if the incoming records are a superset of existing records
+    (i.e. nothing is removed), False otherwise.
+    """
+    name = incoming_rrset["name"]
+    rtype = incoming_rrset["type"]
+
+    existing = next(
+        (r for r in existing_rrsets if r["name"] == name and r["type"] == rtype),
+        None,
+    )
+    if existing is None:
+        return True  # no existing RRset for this name+type, nothing can be lost
+
+    existing_contents = {rec["content"] for rec in existing.get("records", [])}
+    incoming_contents = {rec["content"] for rec in incoming_rrset.get("records", [])}
+
+    missing = existing_contents - incoming_contents
+    if missing:
+        logger.debug(
+            f"RRSET {name} append_only violation: existing records would be removed: {missing}"
+        )
+        return False
+    return True
+
+
 def check_acme_record_allowed(zone: ProxyConfigZone, rrset: RRSET) -> bool:
     if zone.all_records:
         logger.debug("ACME challenge allowed, because all records are allowed")
@@ -210,14 +255,31 @@ def check_pdns_tsigkeys_allowed(environment: ProxyConfigEnvironment) -> bool:
     return False
 
 
-def ensure_rrsets_request_allowed(zone: ProxyConfigZone, request: RRSETRequest) -> bool:
-    """Raises HTTPException if RRSET is not allowed"""
+def ensure_rrsets_request_allowed(
+    zone: ProxyConfigZone,
+    request: RRSETRequest,
+    existing_rrsets: list[dict] = [],
+) -> bool:
+    """Raises HTTPException if RRSET is not allowed.
+
+    `existing_rrsets` should be passed for append_only zones so that REPLACE
+    changesets can be verified not to remove any currently live records.
+    """
     if zone.read_only:
         logger.info("RRSET update not allowed with read only token")
         raise HTTPException(403, "RRSET update not allowed with read only token")
     for rrset in request["rrsets"]:
         if not check_rrset_allowed(zone, rrset):
             logger.info(f"RRSET {rrset['name']} not allowed in zone {zone.name}")
             raise HTTPException(403, f"RRSET {rrset['name']} not allowed")
+        if zone.append_only and rrset.get("changetype") == "REPLACE":
+            if not check_append_only_records_intact(existing_rrsets, rrset):
+                logger.info(
+                    f"RRSET {rrset['name']} append_only violation in zone {zone.name}"
+                )
+                raise HTTPException(
+                    403,
+                    f"RRSET {rrset['name']} append_only violation: existing records would be removed",
+                )
         logger.info(f"RRSET {rrset['name']} allowed")
     return True

powerdns_api_proxy/models.py

@@ -28,6 +28,11 @@ class ProxyConfigZone(BaseModel):
     `all_records` will be set to `True` if no `records` are defined.
     `cryptokeys` enables management of DNSSEC.
     `read_only` controls write permissions for this specific zone.
+    `allowed_record_types` restricts writes to the given DNS record types (e.g. ["TXT"]).
+      An empty list means all record types are allowed.
+    `append_only` blocks DELETE changesets and ensures REPLACE changesets never
+      remove existing records — the incoming records must be a superset of the
+      records currently in PowerDNS.
     """
 
     name: str
@@ -41,6 +46,8 @@ class ProxyConfigZone(BaseModel):
     all_records: bool = False
     read_only: bool = False
     cryptokeys: bool = False
+    allowed_record_types: list[str] = []
+    append_only: bool = False
 
     def __init__(self, **data):
         super().__init__(**data)

powerdns_api_proxy/proxy.py

@@ -3,7 +3,7 @@
 from http import HTTPStatus
 from typing import Literal
 
-from fastapi import APIRouter, Depends, FastAPI, Header, Request, Response
+from fastapi import APIRouter, Depends, FastAPI, Header, HTTPException, Request, Response
 from fastapi.responses import HTMLResponse, JSONResponse
 from prometheus_fastapi_instrumentator import Instrumentator, metrics
 from starlette.exceptions import HTTPException as StarletteHTTPException
@@ -361,10 +361,24 @@ async def update_zone_rrset(
         logger.info(f"Zone {zone_id} not allowed for environment {environment.name}")
         raise ZoneNotAllowedException()
     zone = environment.get_zone_if_allowed(zone_id)
-    ensure_rrsets_request_allowed(zone, await request.json())
+    payload = await request.json()
+
+    existing_rrsets = []
+    if zone.append_only:
+        zone_resp = await pdns.get(f"/api/v1/servers/{server_id}/zones/{zone_id}")
+        zone_pdns_response = await handle_pdns_response(zone_resp)
+        zone_pdns_response.raise_for_error()
+        existing_rrsets = (
+            zone_pdns_response.data.get("rrsets", [])
+            if isinstance(zone_pdns_response.data, dict)
+            else []
+        )
+
+    ensure_rrsets_request_allowed(zone, payload, existing_rrsets)
+
     resp = await pdns.patch(
         f"/api/v1/servers/{server_id}/zones/{zone_id}",
-        payload=await request.json(),
+        payload=payload,
     )
     pdns_response = await handle_pdns_response(resp)
     status_code = pdns_response.raise_for_error()