feat(dnssec): support cryptokeys endpoint

This allows for optional management of DNSSEC.

Signed-off-by: Georg Pfuetzenreuter <mail@georg-pfuetzenreuter.net>

Author: tacerus

README.md

@@ -232,6 +232,23 @@ environments:
       global_tsigkeys: true
 ```
 
+#### CryptoKeys (DNSSEC)
+
+Global or zone-specific CryptoKeys access can be enabled.
+
+This allows for reading and writing of DNSSEC key material.
+
+```yaml
+...
+environments:
+    - name: "Test1"
+      global_cryptokeys: true
+    - name: example.com
+      zones:
+        - name: "example.com"
+          cryptokeys: true
+```
+
 ### Metrics of the proxy
 
 The proxy exposes metrics on the `/metrics` endpoint.

powerdns_api_proxy/config.py

@@ -190,6 +190,20 @@ def check_acme_record_allowed(zone: ProxyConfigZone, rrset: RRSET) -> bool:
     return False
 
 
+def check_pdns_cryptokeys_allowed(
+    environment: ProxyConfigEnvironment, zone: str
+) -> bool:
+    if environment.global_cryptokeys:
+        return True
+
+    try:
+        return environment.get_zone_if_allowed(zone).cryptokeys
+    except ZoneNotAllowedException:
+        pass
+
+    return False
+
+
 def check_pdns_tsigkeys_allowed(environment: ProxyConfigEnvironment) -> bool:
     if environment.global_tsigkeys:
         return True

powerdns_api_proxy/models.py

@@ -27,6 +27,7 @@ class ProxyConfigZone(BaseModel):
     `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.
     """
 
     name: str
@@ -39,6 +40,7 @@ class ProxyConfigZone(BaseModel):
     subzones: bool = False
     all_records: bool = False
     read_only: bool = False
+    cryptokeys: bool = False
 
     def __init__(self, **data):
         super().__init__(**data)
@@ -56,6 +58,7 @@ class ProxyConfigEnvironment(BaseModel):
     zones: list[ProxyConfigZone]
     global_read_only: bool = False
     global_search: bool = False
+    global_cryptokeys: bool = False
     global_tsigkeys: bool = False
     _zones_lookup: dict[str, ProxyConfigZone] = {}
     metrics_proxy: bool = False

powerdns_api_proxy/proxy.py

@@ -13,6 +13,7 @@
 
 from powerdns_api_proxy.config import (
     check_pdns_search_allowed,
+    check_pdns_cryptokeys_allowed,
     check_pdns_tsigkeys_allowed,
     check_pdns_zone_admin,
     check_pdns_zone_allowed,
@@ -479,6 +480,115 @@ async def search_data(
     return JSONResponse(content=pdns_response.data, status_code=status_code)
 
 
+@router_pdns.get("/servers/{server_id}/zones/{zone_id}/cryptokeys")
+async def list_cryptokeys(server_id: str, zone_id: str, X_API_Key: str = Header()):
+    """
+    Get all CryptoKeys for a zone, except the private key.
+
+    <https://doc.powerdns.com/authoritative/http-api/cryptokey.html#get--servers-server_id-zones-zone_id-cryptokeys>
+    """
+    environment = get_environment_for_token(config, X_API_Key)
+    if not check_pdns_cryptokeys_allowed(environment, zone_id):
+        logger.info(f"CryptoKeys not allowed for environment {environment.name}")
+        raise ZoneNotAllowedException()
+    resp = await pdns.get(f"/api/v1/servers/{server_id}/zones/{zone_id}/cryptokeys")
+    pdns_response = await handle_pdns_response(resp)
+    status_code = pdns_response.raise_for_error()
+    return JSONResponse(content=pdns_response.data, status_code=status_code)
+
+
+@router_pdns.post("/servers/{server_id}/zones/{zone_id}/cryptokeys")
+async def create_cryptokey(
+    request: Request, server_id: str, zone_id: str, X_API_Key: str = Header()
+):
+    """
+    Creates a Cryptokey.
+
+    This method adds a new key to a zone.
+
+    <https://doc.powerdns.com/authoritative/http-api/cryptokey.html#post--servers-server_id-zones-zone_id-cryptokeys>
+    """
+    environment = get_environment_for_token(config, X_API_Key)
+    if not check_pdns_cryptokeys_allowed(environment, zone_id):
+        logger.info(f"CryptoKeys not allowed for environment {environment.name}")
+        raise ZoneNotAllowedException()
+    resp = await pdns.post(
+        f"/api/v1/servers/{server_id}/zones/{zone_id}/cryptokeys",
+        payload=await request.json(),
+    )
+    pdns_response = await handle_pdns_response(resp)
+    status_code = pdns_response.raise_for_error()
+    return JSONResponse(content=pdns_response.data, status_code=status_code)
+
+
+@router_pdns.get("/servers/{server_id}/zones/{zone_id}/cryptokeys/{cryptokey_id}")
+async def fetch_cryptokey(
+    server_id: str, zone_id: str, cryptokey_id: str, X_API_Key: str = Header()
+):
+    """
+    Returns all data about the CryptoKey, including the private key.
+
+    <https://doc.powerdns.com/authoritative/http-api/cryptokey.html#get--servers-server_id-zones-zone_id-cryptokeys-cryptokey_id>
+    """
+    environment = get_environment_for_token(config, X_API_Key)
+    if not check_pdns_cryptokeys_allowed(environment, zone_id):
+        logger.info(f"CryptoKeys not allowed for environment {environment.name}")
+        raise ZoneNotAllowedException()
+    resp = await pdns.get(
+        f"/api/v1/servers/{server_id}/zones/{zone_id}/cryptokeys/{cryptokey_id}"
+    )
+    pdns_response = await handle_pdns_response(resp)
+    status_code = pdns_response.raise_for_error()
+    return JSONResponse(content=pdns_response.data, status_code=status_code)
+
+
+@router_pdns.put("/servers/{server_id}/zones/{zone_id}/cryptokeys/{cryptokey_id}")
+async def update_cryptokey(
+    request: Request,
+    server_id: str,
+    zone_id: str,
+    cryptokey_id: str,
+    X_API_Key: str = Header(),
+):
+    """
+    This method (de)activates a key from zone_name specified by cryptokey_id.
+
+    <https://doc.powerdns.com/authoritative/http-api/cryptokey.html#put--servers-server_id-zones-zone_id-cryptokeys-cryptokey_id>
+    """
+    environment = get_environment_for_token(config, X_API_Key)
+    if not check_pdns_cryptokeys_allowed(environment, zone_id):
+        logger.info(f"CryptoKeys not allowed for environment {environment.name}")
+        raise ZoneNotAllowedException()
+    resp = await pdns.put(
+        f"/api/v1/servers/{server_id}/zones/{zone_id}/cryptokeys/{cryptokey_id}",
+        payload=await request.json(),
+    )
+    pdns_response = await handle_pdns_response(resp)
+    status_code = pdns_response.raise_for_error()
+    return JSONResponse(content=pdns_response.data, status_code=status_code)
+
+
+@router_pdns.delete("/servers/{server_id}/zones/{zone_id}/cryptokeys/{cryptokey_id}")
+async def delete_cryptokey(
+    server_id: str, zone_id: str, cryptokey_id: str, X_API_Key: str = Header()
+):
+    """
+    This method deletes a key specified by cryptokey_id.
+
+    <https://doc.powerdns.com/authoritative/http-api/cryptokey.html#delete--servers-server_id-zones-zone_id-cryptokeys-cryptokey_id>
+    """
+    environment = get_environment_for_token(config, X_API_Key)
+    if not check_pdns_cryptokeys_allowed(environment, zone_id):
+        logger.info(f"CryptoKeys not allowed for environment {environment.name}")
+        raise ZoneNotAllowedException()
+    resp = await pdns.delete(
+        f"/api/v1/servers/{server_id}/zones/{zone_id}/cryptokeys/{cryptokey_id}"
+    )
+    pdns_response = await handle_pdns_response(resp)
+    status_code = pdns_response.raise_for_error()
+    return JSONResponse(content=pdns_response.data, status_code=status_code)
+
+
 @router_pdns.get("/servers/{server_id}/tsigkeys")
 async def list_tsigkeys(server_id: str, X_API_Key: str = Header()):
     """

tests/unit/config_test.py

@@ -7,6 +7,7 @@
 from powerdns_api_proxy.config import (
     check_acme_record_allowed,
     check_pdns_search_allowed,
+    check_pdns_cryptokeys_allowed,
     check_pdns_tsigkeys_allowed,
     check_pdns_zone_admin,
     check_pdns_zone_allowed,
@@ -225,7 +226,7 @@ def test_check_pdns_zone_allowed_allowed_without_trailing_point():
 
 
 def test_check_pdns_zone_allowed_allowed_without_trailing_point_point_last_item():
-    env = dummy_proxy_environment
+    env = deepcopy(dummy_proxy_environment)
     env.zones[0].name = "blablub.example.com+"
     zone = "blablub.example.com"
     assert not check_pdns_zone_allowed(env, zone)
@@ -585,6 +586,26 @@ def test_search_allowed_globally():
     assert check_pdns_search_allowed(environment, "test", "all") is True
 
 
+def test_cryptokeys_not_allowed():
+    environment = dummy_proxy_environment
+    assert check_pdns_cryptokeys_allowed(environment, "test") is False
+    assert check_pdns_cryptokeys_allowed(environment, "test.example.com.") is False
+
+
+def test_cryptokeys_allowed_zone_only():
+    environment = deepcopy(dummy_proxy_environment)
+    environment.zones[0].cryptokeys = True
+    assert check_pdns_cryptokeys_allowed(environment, "test") is False
+    assert check_pdns_cryptokeys_allowed(environment, "test.example.com.") is True
+
+
+def test_cryptokeys_allowed_global():
+    environment = deepcopy(dummy_proxy_environment)
+    environment.global_cryptokeys = True
+    assert check_pdns_cryptokeys_allowed(environment, "test") is True
+    assert check_pdns_cryptokeys_allowed(environment, "test.example.com.") is True
+
+
 def test_tsigkeys_not_allowed():
     environment = deepcopy(dummy_proxy_environment)
     environment.global_tsigkeys = False