Merge branch 'main' into cryptokeys

Author: rwxd

powerdns_api_proxy/models.py

@@ -1,7 +1,7 @@
 from functools import lru_cache
 from typing import TypedDict
 
-from pydantic import BaseModel, field_validator
+from pydantic import BaseModel, field_validator, model_validator
 
 from powerdns_api_proxy.logging import logger
 from powerdns_api_proxy.utils import (
@@ -26,8 +26,8 @@ class ProxyConfigZone(BaseModel):
     `admin` enabled creating and deleting the zone.
     `subzones` sets the same permissions on all subzones.
     `all_records` will be set to `True` if no `records` are defined.
-    `read_only` will be set to `True` if `global_read_only` is `True`.
     `cryptokeys` enables management of DNSSEC.
+    `read_only` controls write permissions for this specific zone.
     """
 
     name: str
@@ -55,7 +55,7 @@ def __init__(self, **data):
 class ProxyConfigEnvironment(BaseModel):
     name: str
     token_sha512: str
-    zones: list[ProxyConfigZone]
+    zones: list[ProxyConfigZone] = []
     global_read_only: bool = False
     global_search: bool = False
     global_cryptokeys: bool = False
@@ -77,17 +77,20 @@ def validate_token(cls, token_sha512):
             raise ValueError("A SHA512 hash must be 128 digits long")
         return token_sha512
 
+    @model_validator(mode="after")
+    def validate_zones_or_global_read_only(self):
+        if not self.zones and not self.global_read_only:
+            raise ValueError(
+                "Either 'zones' must be non-empty or 'global_read_only' must be True"
+            )
+        return self
+
     def __init__(self, **data):
         super().__init__(**data)
-        if self.global_read_only:
-            logger.debug(
-                "Setting all subzones to read_only, because global_read_only is true"
-            )
-            for zone in self.zones:
-                zone.read_only = True
 
-                # populate zones lookup
-                self._zones_lookup[zone.name] = zone
+        # populate zones lookup
+        for zone in self.zones:
+            self._zones_lookup[zone.name] = zone
 
     def __hash__(self):
         return hash(

powerdns_api_proxy/proxy.py

@@ -3,12 +3,9 @@
 from http import HTTPStatus
 from typing import Literal
 
-import sentry_sdk
 from fastapi import APIRouter, Depends, FastAPI, Header, Request, Response
 from fastapi.responses import HTMLResponse, JSONResponse
 from prometheus_fastapi_instrumentator import Instrumentator, metrics
-from sentry_sdk.integrations.aiohttp import AioHttpIntegration
-from sentry_sdk.integrations.fastapi import FastApiIntegration
 from starlette.exceptions import HTTPException as StarletteHTTPException
 
 from powerdns_api_proxy.config import (
@@ -40,6 +37,10 @@
 from powerdns_api_proxy.pdns import PDNSConnector, handle_pdns_response
 
 if os.getenv("SENTRY_DSN"):
+    import sentry_sdk
+    from sentry_sdk.integrations.aiohttp import AioHttpIntegration
+    from sentry_sdk.integrations.fastapi import FastApiIntegration
+
     sentry_sdk.init(
         traces_sample_rate=float(os.getenv("SENTRY_TRACES_SAMPLE_RATE") or 0.1),
         environment=os.getenv("ENVIRONMENT") or "DEV",

tests/unit/config_test.py

@@ -616,3 +616,79 @@ def test_tsigkeys_allowed_globally():
     environment = deepcopy(dummy_proxy_environment)
     environment.global_tsigkeys = True
     assert check_pdns_tsigkeys_allowed(environment) is True
+
+
+def test_global_read_only_without_zones():
+    """Test that global_read_only=True allows empty zones list"""
+    env = ProxyConfigEnvironment(
+        name="Test Global Read Only",
+        token_sha512=dummy_proxy_environment_token_sha512,
+        global_read_only=True,
+    )
+    assert env.global_read_only is True
+    assert env.zones == []
+
+
+def test_environment_with_neither_zones_nor_global_read_only_fails():
+    """Test that providing neither zones nor global_read_only fails validation"""
+    with pytest.raises(ValueError) as err:
+        ProxyConfigEnvironment(
+            name="test", token_sha512=dummy_proxy_environment_token_sha512
+        )
+    assert "Either 'zones' must be non-empty or 'global_read_only' must be True" in str(
+        err.value
+    )
+
+
+def test_environment_with_empty_zones_and_no_global_read_only_fails():
+    """Test that explicitly providing empty zones without global_read_only fails"""
+    with pytest.raises(ValueError) as err:
+        ProxyConfigEnvironment(
+            name="test", token_sha512=dummy_proxy_environment_token_sha512, zones=[]
+        )
+    assert "Either 'zones' must be non-empty or 'global_read_only' must be True" in str(
+        err.value
+    )
+
+
+def test_proxy_config_with_global_read_only_environment():
+    """Test that ProxyConfig works with global_read_only environment without zones"""
+    config = ProxyConfig(
+        pdns_api_url="https://powerdns-api.example.com",
+        pdns_api_token="blablub",
+        environments=[
+            ProxyConfigEnvironment(
+                name="foo",
+                token_sha512=dummy_proxy_environment_token_sha512,
+                global_read_only=True,
+            )
+        ],
+    )
+    assert config.environments[0].global_read_only is True
+    assert config.environments[0].zones == []
+
+
+def test_global_read_only_with_explicit_zones_keeps_zone_permissions():
+    """Test that global_read_only=True doesn't force explicit zones to be read_only"""
+    # Create a zone that should remain writable
+    writable_zone = ProxyConfigZone(name="example.com", read_only=False)
+    readonly_zone = ProxyConfigZone(name="readonly.com", read_only=True)
+
+    env = ProxyConfigEnvironment(
+        name="Test Global Read Only with Zones",
+        token_sha512=dummy_proxy_environment_token_sha512,
+        zones=[writable_zone, readonly_zone],
+        global_read_only=True,
+    )
+
+    # global_read_only should be True
+    assert env.global_read_only is True
+
+    # But explicit zones should keep their original read_only settings
+    assert env.zones[0].read_only is False  # writable_zone should remain writable
+    assert env.zones[1].read_only is True  # readonly_zone should remain read_only
+
+    # Should have access to zones via lookup
+    assert len(env._zones_lookup) == 2
+    assert "example.com" in env._zones_lookup
+    assert "readonly.com" in env._zones_lookup