Merge pull request #3 from akquinet/feat-wildcards

Add the possibility to use regex for zones

README.md

@@ -149,6 +149,24 @@ environments:
           subzones: true
 ```
 
+
+##### Regex
+
+Under a `zone` the option `regex: true` can be set.
+
+That allows use regex in the zone name.
+
+In this example all zones which end with `.example.com` are allowed.
+
+```YAML
+...
+environments:
+  - name: "Test1"
+    zones:
+     - name: ".*\\.example.com"
+       regex: true
+```
+
 #### Global read
 
 Global `read` permissions can be defined under an `environment`.

powerdns_api_proxy/config.py

@@ -17,6 +17,7 @@
     RRSETRequest,
     ZoneNotAllowedException,
 )
+from powerdns_api_proxy.utils import check_zones_equal
 
 
 @lru_cache(maxsize=1)
@@ -88,22 +89,22 @@ def get_only_pdns_zones_allowed(
 
 
 def check_pdns_zone_allowed(environment: ProxyConfigEnvironment, zone: str) -> bool:
+    '''Returns True if zone is allowed in the environment'''
     if environment.global_read_only:
         return True
+
     try:
-        env_zone = get_zone_config(environment, zone)
-        logger.info(f'Zone {env_zone.name} allowed for environment {environment.name}')
+        _ = environment.get_zone_if_allowed(zone)
         return True
-    except HTTPException:
-        pass
-    return False
+    except ZoneNotAllowedException:
+        return False
 
 
 def check_pdns_zone_admin(environment: ProxyConfigEnvironment, zone: str) -> bool:
     try:
-        env_zone = get_zone_config(environment, zone)
+        env_zone = environment.get_zone_if_allowed(zone)
         return env_zone.admin
-    except HTTPException:
+    except ZoneNotAllowedException:
         pass
     return False
 
@@ -124,7 +125,7 @@ def check_rrset_allowed(zone: ProxyConfigZone, rrset: RRSET) -> bool:
         return True
 
     for record in zone.records:
-        if zones_equal(rrset['name'], record):
+        if check_zones_equal(rrset['name'], record):
             return True
 
     if check_acme_record_allowed(zone, rrset):
@@ -143,7 +144,7 @@ def check_acme_record_allowed(zone: ProxyConfigZone, rrset: RRSET) -> bool:
         return False
 
     for record in zone.records:
-        if zones_equal(f'_acme-challenge.{record}', rrset['name']):
+        if check_zones_equal(f'_acme-challenge.{record}', rrset['name']):
             logger.info(f'ACME challenge for record {record} is allowed')
             return True
 
@@ -161,33 +162,3 @@ def ensure_rrsets_request_allowed(zone: ProxyConfigZone, request: RRSETRequest)
             raise HTTPException(403, f'RRSET {rrset["name"]} not allowed')
         logger.info(f'RRSET {rrset["name"]} allowed')
     return True
-
-
-def get_zone_config(environment: ProxyConfigEnvironment, zone: str) -> ProxyConfigZone:
-    for z in environment.zones:
-        if zones_equal(z.name, zone):
-            return z
-        elif z.subzones:
-            if check_subzone(zone, z.name):
-                return ProxyConfigZone(
-                    name=zone,
-                    subzones=z.subzones,
-                    services=z.services,
-                    admin=z.admin,
-                    all_records=z.all_records,
-                    records=z.records,
-                    read_only=z.read_only,
-                )
-    raise ZoneNotAllowedException()
-
-
-def zones_equal(zone1: str, zone2: str) -> bool:
-    '''Checks if zones equal with or without trailing dot'''
-    return zone1.rstrip('.') == zone2.rstrip('.')
-
-
-def check_subzone(zone: str, main_zone: str) -> bool:
-    if zone.rstrip('.').endswith(main_zone.rstrip('.')):
-        logger.debug(f'"{zone}" is a subzone of "{main_zone}"')
-        return True
-    return False

powerdns_api_proxy/models.py

@@ -4,6 +4,11 @@
 from pydantic import BaseModel, field_validator
 
 from powerdns_api_proxy.logging import logger
+from powerdns_api_proxy.utils import (
+    check_subzone,
+    check_zone_in_regex,
+    check_zones_equal,
+)
 
 
 class ProxyConfigServices(BaseModel):
@@ -12,12 +17,16 @@ class ProxyConfigServices(BaseModel):
 
 class ProxyConfigZone(BaseModel):
     '''
+    `name` is the zone name.
+    `regex` should be set to `True` if `name` is a regex.
     `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`.
     '''
 
     name: str
+    regex: bool = False
     description: str = ''
     records: list[str] = []
     services: ProxyConfigServices = ProxyConfigServices(acme=False)
@@ -42,6 +51,7 @@ class ProxyConfigEnvironment(BaseModel):
     zones: list[ProxyConfigZone]
     global_read_only: bool = False
     global_search: bool = False
+    _zones_lookup: dict[str, ProxyConfigZone] = {}
 
     @field_validator('name')
     @classmethod
@@ -66,6 +76,31 @@ def __init__(self, **data):
             for zone in self.zones:
                 zone.read_only = True
 
+                # populate zones lookup
+                self._zones_lookup[zone.name] = zone
+
+    def get_zone_if_allowed(self, zone: str) -> ProxyConfigZone:
+        '''
+        Returns the zone config for the given zone name
+        Raises ZoneNotAllowedException if the zone is not allowed
+        '''
+        if zone in self._zones_lookup:
+            return self._zones_lookup[zone]
+
+        for z in self.zones:
+            if check_zones_equal(zone, z.name):
+                return z
+
+            if z.subzones and check_subzone(zone, z.name):
+                logger.debug(f'"{zone}" is a subzone of "{z.name}"')
+                return z
+
+            if z.regex and check_zone_in_regex(zone, z.name):
+                logger.debug(f'"{zone}" matches regex "{z.name}"')
+                return z
+
+        raise ZoneNotAllowedException()
+
 
 class ProxyConfig(BaseModel):
     pdns_api_url: str
@@ -92,6 +127,12 @@ class ResponseAllowed(BaseModel):
     zones: list[ProxyConfigZone]
 
 
+class ResponseZoneAllowed(BaseModel):
+    zone: str
+    allowed: bool
+    config: ProxyConfigZone | None = None
+
+
 class ZoneNotAllowedException(HTTPException):
     def __init__(self):
         self.status_code = 403

powerdns_api_proxy/proxy.py

@@ -18,13 +18,13 @@
     ensure_rrsets_request_allowed,
     get_environment_for_token,
     get_only_pdns_zones_allowed,
-    get_zone_config,
     load_config,
 )
 from powerdns_api_proxy.logging import logger
 from powerdns_api_proxy.metrics import http_requests_total_environment
 from powerdns_api_proxy.models import (
     ResponseAllowed,
+    ResponseZoneAllowed,
     RessourceNotAllowedException,
     SearchNotAllowedException,
     ZoneAdminNotAllowedException,
@@ -128,6 +128,24 @@ async def get_allowed_ressources(X_API_Key: str = Header()):
     return ResponseAllowed(zones=environment.zones)
 
 
+@router_proxy.get(
+    '/zone-allowed',
+    response_model=ResponseZoneAllowed,
+)
+async def get_zone_allowed(zone: str, X_API_Key: str = Header()):
+    '''
+    Check if the given zone is allowed for the given token.
+    Also returns the zone config that allows the zone.
+    '''
+    logger.debug('Checking if zone is allowed for given api key')
+    environment = get_environment_for_token(config, X_API_Key)
+    if not check_pdns_zone_allowed(environment, zone):
+        return ResponseZoneAllowed(zone=zone, allowed=False)
+
+    zone_config = environment.get_zone_if_allowed(zone)
+    return ResponseZoneAllowed(zone=zone, allowed=True, config=zone_config)
+
+
 @app.get('/api', dependencies=[Depends(dependency_check_token_defined)])
 async def api_root():
     '''Returns the version and a info that this is a proxy.'''
@@ -316,7 +334,7 @@ async def update_zone_rrset(
     if not check_pdns_zone_allowed(environment, zone_id):
         logger.info(f'Zone {zone_id} not allowed for environment {environment.name}')
         raise ZoneNotAllowedException()
-    zone = get_zone_config(environment, zone_id)
+    zone = environment.get_zone_if_allowed(zone_id)
     ensure_rrsets_request_allowed(zone, await request.json())
     resp = await pdns.patch(
         f'/api/v1/servers/{server_id}/zones/{zone_id}',

powerdns_api_proxy/utils.py

@@ -1,4 +1,5 @@
 import json
+import re
 from json.decoder import JSONDecodeError
 from typing import Union
 
@@ -12,3 +13,19 @@ async def response_json_or_text(response: ClientResponse) -> Union[dict, str]:
         return json.loads(text)
     except JSONDecodeError:
         return text
+
+
+def check_subzone(zone: str, main_zone: str) -> bool:
+    if zone.rstrip('.').endswith(main_zone.rstrip('.')):
+        return True
+    return False
+
+
+def check_zone_in_regex(zone: str, regex: str) -> bool:
+    '''Checks if zone is in regex'''
+    return re.match(regex, zone.rstrip('.')) is not None
+
+
+def check_zones_equal(zone1: str, zone2: str) -> bool:
+    '''Checks if zones equal with or without trailing dot'''
+    return zone1.rstrip('.') == zone2.rstrip('.')

tests/fixtures/__init__.py

@@ -0,0 +1,3 @@
+from os import path
+
+FIXTURES_DIR = path.dirname(path.abspath(__file__))

tests/fixtures/test_regex_parsing.yaml

@@ -0,0 +1,4 @@
+---
+# file to test regex parsing with yaml escaping
+name: ".*\\.example\\.com"
+regex: true

tests/unit/config_test.py

@@ -10,12 +10,10 @@
     check_pdns_zone_admin,
     check_pdns_zone_allowed,
     check_rrset_allowed,
-    check_subzone,
     check_token_defined,
     ensure_rrsets_request_allowed,
     get_environment_for_token,
     get_only_pdns_zones_allowed,
-    get_zone_config,
     token_defined,
 )
 from powerdns_api_proxy.models import (
@@ -171,14 +169,14 @@ def test_check_pdns_zone_admin_true_subzone():
 
 def test_get_zone_config():
     env = dummy_proxy_environment
-    zone = get_zone_config(env, dummy_proxy_zone.name)
+    zone = env.get_zone_if_allowed(dummy_proxy_zone.name)
     assert zone.name == dummy_proxy_zone.name
 
 
 def test_get_zone_config_not_allowed():
     env = dummy_proxy_environment
     with pytest.raises(HTTPException) as err:
-        get_zone_config(env, 'blablub_mich_gibtsnicht.example.com.')
+        env.get_zone_if_allowed('blablub_mich_gibtsnicht.example.com.')
     assert err.value.detail == ZoneNotAllowedException().detail
     assert err.value.status_code == ZoneNotAllowedException().status_code
 
@@ -187,21 +185,21 @@ def test_get_zone_config_subzone():
     env = dummy_proxy_environment
     dummy_proxy_environment.zones[0].subzones = True
     subzone = 'blabluuub.' + dummy_proxy_environment.zones[0].name
-    assert get_zone_config(env, subzone)
+    assert env.get_zone_if_allowed(subzone)
 
 
 def test_get_zone_config_subzone_subzone():
     env = dummy_proxy_environment
     dummy_proxy_environment.zones[0].subzones = True
     subzone = 'blabluuub.subzone.' + dummy_proxy_environment.zones[0].name
-    assert get_zone_config(env, subzone)
+    assert env.get_zone_if_allowed(subzone)
 
 
 def test_get_zone_config_subzone_not_allowed():
     env = dummy_proxy_environment
     subzone = 'blabluuub.' + dummy_proxy_environment.zones[0].name + 'test.'
     with pytest.raises(HTTPException) as err:
-        assert get_zone_config(env, subzone)
+        assert env.get_zone_if_allowed(subzone)
     assert err.value.detail == ZoneNotAllowedException().detail
     assert err.value.status_code == ZoneNotAllowedException().status_code
 
@@ -211,7 +209,7 @@ def test_get_zone_config_no_subzone():
     dummy_proxy_environment.zones[0].subzones = False
     subzone = 'blabluuub.' + dummy_proxy_environment.zones[0].name
     with pytest.raises(HTTPException) as err:
-        assert get_zone_config(env, subzone)
+        assert env.get_zone_if_allowed(subzone)
     assert err.value.detail == ZoneNotAllowedException().detail
     assert err.value.status_code == ZoneNotAllowedException().status_code
 
@@ -459,18 +457,6 @@ def test_check_acme_record_not_allowed_false_challenge():
     assert not check_acme_record_allowed(zone, rrset)
 
 
-def test_check_subzone_true():
-    zone = 'myzone.main.example.com'
-    main = 'main.example.com.'
-    assert check_subzone(zone, main)
-
-
-def test_check_subzone_false():
-    zone = 'myzone.test.example.com'
-    main = 'main.example.com.'
-    assert not check_subzone(zone, main)
-
-
 def test_search_not_allowed():
     environment = deepcopy(dummy_proxy_environment)
     environment.global_search = False

tests/unit/proxy_test.py

@@ -79,6 +79,7 @@ def _token_missing_request(client: TestClient, method: str, path: str):
 
 get_routes = [
     '/info/allowed',
+    '/info/zone-allowed',
     '/api',
     '/api/v1/servers',
     '/api/v1/servers/localhost',