feat(tsigkeys): implement tsigkeys API proxying

Author: quantum5

README.md

@@ -218,6 +218,21 @@ environments:
       global_search: true
 ```
 
+#### Global TSIGKeys
+
+Global TSIGKeys access can be defined under an `environment`.
+
+For this the `Environment` must have the option `global_tsigkeys: true`.
+
+This allows the token to read and modify all TSIGKeys in the PowerDNS.
+
+```yaml
+...
+environments:
+    - name: "Test1"
+      global_tsigkeys: true
+```
+
 ### Metrics of the proxy
 
 The proxy exposes metrics on the `/metrics` endpoint.

powerdns_api_proxy/config.py

@@ -182,6 +182,12 @@ def check_acme_record_allowed(zone: ProxyConfigZone, rrset: RRSET) -> bool:
     return False
 
 
+def check_pdns_tsigkeys_allowed(environment: ProxyConfigEnvironment) -> bool:
+    if environment.global_tsigkeys:
+        return True
+    return False
+
+
 def ensure_rrsets_request_allowed(zone: ProxyConfigZone, request: RRSETRequest) -> bool:
     '''Raises HTTPException if RRSET is not allowed'''
     if zone.read_only:

powerdns_api_proxy/models.py

@@ -56,6 +56,7 @@ class ProxyConfigEnvironment(BaseModel):
     zones: list[ProxyConfigZone]
     global_read_only: bool = False
     global_search: bool = False
+    global_tsigkeys: bool = False
     _zones_lookup: dict[str, ProxyConfigZone] = {}
     metrics_proxy: bool = False
 
@@ -91,6 +92,7 @@ def __hash__(self):
             + self.token_sha512
             + str(self.global_read_only)
             + str(self.global_search)
+            + str(self.global_tsigkeys)
             + str(self.zones)
         )
 

powerdns_api_proxy/proxy.py

@@ -15,6 +15,7 @@
     check_pdns_search_allowed,
     check_pdns_zone_admin,
     check_pdns_zone_allowed,
+    check_pdns_tsigkeys_allowed,
     dependency_check_token_defined,
     dependency_metrics_proxy_enabled,
     ensure_rrsets_request_allowed,
@@ -442,6 +443,109 @@ async def search_data(
     return data
 
 
+@router_pdns.get('/servers/{server_id}/tsigkeys')
+async def list_tsigkeys(
+    response: Response, server_id: str, X_API_Key: str = Header()
+):
+    '''
+    Get all TSIGKeys on the server, except the actual key.
+
+    <https://doc.powerdns.com/authoritative/http-api/tsigkey.html#get--servers-server_id-tsigkeys>
+    '''
+    environment = get_environment_for_token(config, X_API_Key)
+    if not check_pdns_tsigkeys_allowed(environment):
+        logger.info(f'TSIGKeys not allowed for environment {environment.name}')
+        raise ZoneNotAllowedException()
+    resp = await pdns.get(f'/api/v1/servers/{server_id}/tsigkeys')
+    response.status_code = resp.status
+    data = await response_json_or_text(resp)
+    return data
+
+
+@router_pdns.get('/servers/{server_id}/tsigkeys/{tsigkey_id}')
+async def fetch_tsigkey(
+    response: Response, server_id: str, tsigkey_id: str, X_API_Key: str = Header()
+):
+    '''
+    Get a specific TSIGKeys on the server, including the actual key.
+
+    <https://doc.powerdns.com/authoritative/http-api/tsigkey.html#get--servers-server_id-tsigkeys-tsigkey_id>
+    '''
+    environment = get_environment_for_token(config, X_API_Key)
+    if not check_pdns_tsigkeys_allowed(environment):
+        logger.info(f'TSIGKeys not allowed for environment {environment.name}')
+        raise ZoneNotAllowedException()
+    resp = await pdns.get(f'/api/v1/servers/{server_id}/tsigkeys/{tsigkey_id}')
+    response.status_code = resp.status
+    data = await response_json_or_text(resp)
+    return data
+
+
+@router_pdns.post('/servers/{server_id}/tsigkeys')
+async def create_tsigkey(
+    request: Request, response: Response, server_id: str, X_API_Key: str = Header()
+):
+    '''
+    Add a TSIG key.
+
+    This methods add a new TSIGKey. The actual key can be generated by the server or be provided by the client.
+
+    <https://doc.powerdns.com/authoritative/http-api/tsigkey.html#post--servers-server_id-tsigkeys>
+    '''
+    environment = get_environment_for_token(config, X_API_Key)
+    if not check_pdns_tsigkeys_allowed(environment):
+        logger.info(f'TSIGKeys not allowed for environment {environment.name}')
+        raise ZoneNotAllowedException()
+    resp = await pdns.post(f'/api/v1/servers/{server_id}/tsigkeys', payload=await request.json())
+    response.status_code = resp.status
+    data = await response_json_or_text(resp)
+    return data
+
+
+@router_pdns.put('/servers/{server_id}/tsigkeys/{tsigkey_id}')
+async def update_tsigkey(
+    request: Request, response: Response, server_id: str, tsigkey_id: str, X_API_Key: str = Header()
+):
+    '''
+    The TSIGKey at tsigkey_id can be changed in multiple ways:
+
+    * Changing the Name, this will remove the key with tsigkey_id after adding.
+    * Changing the Algorithm
+    * Changing the Key
+
+    Only the relevant fields have to be provided in the request body.
+
+    <https://doc.powerdns.com/authoritative/http-api/tsigkey.html#put--servers-server_id-tsigkeys-tsigkey_id>
+    '''
+    environment = get_environment_for_token(config, X_API_Key)
+    if not check_pdns_tsigkeys_allowed(environment):
+        logger.info(f'TSIGKeys not allowed for environment {environment.name}')
+        raise ZoneNotAllowedException()
+    resp = await pdns.put(f'/api/v1/servers/{server_id}/tsigkeys/{tsigkey_id}', payload=await request.json())
+    response.status_code = resp.status
+    data = await response_json_or_text(resp)
+    return data
+
+
+@router_pdns.delete('/servers/{server_id}/tsigkeys/{tsigkey_id}')
+async def delete_tsigkey(
+    response: Response, server_id: str, tsigkey_id: str, X_API_Key: str = Header()
+):
+    '''
+    Delete the TSIGKey with tsigkey_id.
+
+    <https://doc.powerdns.com/authoritative/http-api/tsigkey.html#delete--servers-server_id-tsigkeys-tsigkey_id>
+    '''
+    environment = get_environment_for_token(config, X_API_Key)
+    if not check_pdns_tsigkeys_allowed(environment):
+        logger.info(f'TSIGKeys not allowed for environment {environment.name}')
+        raise ZoneNotAllowedException()
+    resp = await pdns.delete(f'/api/v1/servers/{server_id}/tsigkeys/{tsigkey_id}')
+    response.status_code = resp.status
+    data = await response_json_or_text(resp)
+    return data
+
+
 app.include_router(router_proxy)
 app.include_router(router_pdns)
 app.include_router(router_health)

tests/unit/config_test.py

@@ -9,6 +9,7 @@
     check_pdns_search_allowed,
     check_pdns_zone_admin,
     check_pdns_zone_allowed,
+    check_pdns_tsigkeys_allowed,
     check_rrset_allowed,
     check_token_defined,
     ensure_rrsets_request_allowed,
@@ -580,3 +581,15 @@ def test_search_allowed_globally():
     environment = deepcopy(dummy_proxy_environment)
     environment.global_search = True
     assert check_pdns_search_allowed(environment, 'test', 'all') is True
+
+
+def test_tsigkeys_not_allowed():
+    environment = deepcopy(dummy_proxy_environment)
+    environment.global_tsigkeys = False
+    assert check_pdns_tsigkeys_allowed(environment) is False
+
+
+def test_tsigkeys_allowed_globally():
+    environment = deepcopy(dummy_proxy_environment)
+    environment.global_tsigkeys = True
+    assert check_pdns_tsigkeys_allowed(environment) is True