enh: Enforce Record Restrictions in API (#1089)

Co-authored-by: Tom <tom@tom.com>

powerdnsadmin/decorators.py

@@ -5,9 +5,10 @@ from flask import g, request, abort, current_app, Response
 from flask_login import current_user
 
 from .models import User, ApiKey, Setting, Domain, Setting
-from .lib.errors import RequestIsNotJSON, NotEnoughPrivileges
+from .lib.errors import RequestIsNotJSON, NotEnoughPrivileges, RecordTTLNotAllowed, RecordTypeNotAllowed
 from .lib.errors import DomainAccessForbidden, DomainOverrideForbidden
 
+
 def admin_role_required(f):
     """
     Grant access if user is in Administrator role
@@ -384,6 +385,60 @@ def apikey_can_configure_dnssec(http_methods=[]):
         return decorated_function
     return decorator
 
+def allowed_record_types(f):
+    @wraps(f)
+    def decorated_function(*args, **kwargs):
+        if request.method == 'GET':
+            return f(*args, **kwargs)
+
+        if g.apikey.role.name in ['Administrator', 'Operator']:
+            return f(*args, **kwargs)
+
+        records_allowed_to_edit = Setting().get_records_allow_to_edit()
+        content = request.get_json()
+        try:
+            for record in content['rrsets']:
+                if 'type' not in record:
+                    raise RecordTypeNotAllowed()
+
+                if record['type'] not in records_allowed_to_edit:
+                    current_app.logger.error(f"Error: Record type not allowed: {record['type']}")
+                    raise RecordTypeNotAllowed(message=f"Record type not allowed: {record['type']}")
+        except (TypeError, KeyError) as e:
+            raise e
+        return f(*args, **kwargs)
+
+    return decorated_function
+
+def allowed_record_ttl(f):
+    @wraps(f)
+    def decorated_function(*args, **kwargs):
+        if not Setting().get('enforce_api_ttl'):
+            return f(*args, **kwargs)
+
+        if request.method == 'GET':
+            return f(*args, **kwargs)
+            
+        if g.apikey.role.name in ['Administrator', 'Operator']:
+            return f(*args, **kwargs)
+
+        allowed_ttls = Setting().get_ttl_options()
+        allowed_numeric_ttls = [ ttl[0] for ttl in allowed_ttls ]
+        content = request.get_json()
+        try:
+            for record in content['rrsets']:
+                if 'ttl' not in record:
+                    raise RecordTTLNotAllowed()
+
+                if record['ttl'] not in allowed_numeric_ttls:
+                    current_app.logger.error(f"Error: Record TTL not allowed: {record['ttl']}")
+                    raise RecordTTLNotAllowed(message=f"Record TTL not allowed: {record['ttl']}")
+        except (TypeError, KeyError) as e:
+            raise e
+        return f(*args, **kwargs)
+
+    return decorated_function
+
 
 def apikey_auth(f):
     @wraps(f)

powerdnsadmin/lib/errors.py

@@ -170,7 +170,6 @@ class UserUpdateFailEmail(StructuredException):
         self.message = message
         self.name = name
 
-
 class UserDeleteFail(StructuredException):
     status_code = 500
 
@@ -178,3 +177,19 @@ class UserDeleteFail(StructuredException):
         StructuredException.__init__(self)
         self.message = message
         self.name = name
+
+class RecordTypeNotAllowed(StructuredException):
+    status_code = 400
+
+    def __init__(self, name=None, message="Record type not allowed or does not present"):
+        StructuredException.__init__(self)
+        self.message = message
+        self.name = name
+
+class RecordTTLNotAllowed(StructuredException):
+    status_code = 400
+
+    def __init__(self, name=None, message="Record TTL not allowed or does not present"):
+        StructuredException.__init__(self)
+        self.message = message
+        self.name = name

powerdnsadmin/models/setting.py

@@ -28,7 +28,7 @@ class Setting(db.Model):
         'allow_user_create_domain': False,
         'allow_user_remove_domain': False,
         'allow_user_view_history': False,
-	'delete_sso_accounts': False,
+	    'delete_sso_accounts': False,
         'bg_domain_updates': False,
         'enable_api_rr_history': True,
         'site_name': 'PowerDNS-Admin',
@@ -110,6 +110,7 @@ class Setting(db.Model):
         'oidc_oauth_email': 'email',
         'oidc_oauth_account_name_property': '',
         'oidc_oauth_account_description_property': '',
+        'enforce_api_ttl': False,
         'forward_records_allow_edit': {
             'A': True,
             'AAAA': True,

powerdnsadmin/routes/admin.py

@@ -1268,7 +1268,7 @@ def setting_basic():
             'allow_user_create_domain', 'allow_user_remove_domain', 'allow_user_view_history', 'bg_domain_updates', 'site_name',
             'session_timeout', 'warn_session_timeout', 'ttl_options',
             'pdns_api_timeout', 'verify_ssl_connections', 'verify_user_email',
-	          'delete_sso_accounts', 'otp_field_enabled', 'custom_css', 'enable_api_rr_history', 'max_history_records', 'otp_force', 'deny_domain_override'
+	          'delete_sso_accounts', 'otp_field_enabled', 'custom_css', 'enable_api_rr_history', 'max_history_records', 'otp_force', 'deny_domain_override', 'enforce_api_ttl'
         ]
 
         return render_template('admin_setting_basic.html', settings=settings)

powerdnsadmin/routes/api.py

@@ -30,7 +30,7 @@ from ..decorators import (
     apikey_can_create_domain, apikey_can_remove_domain,
     apikey_is_admin, apikey_can_access_domain, apikey_can_configure_dnssec,
     api_role_can, apikey_or_basic_auth,
-    callback_if_request_body_contains_key,
+    callback_if_request_body_contains_key, allowed_record_types, allowed_record_ttl
 )
 import secrets
 import string
@@ -1079,6 +1079,8 @@ def api_zone_subpath_forward(server_id, zone_id, subpath):
 @api_bp.route('/servers/<string:server_id>/zones/<string:zone_id>',
               methods=['GET', 'PUT', 'PATCH', 'DELETE'])
 @apikey_auth
+@allowed_record_types
+@allowed_record_ttl
 @apikey_can_access_domain
 @apikey_can_remove_domain(http_methods=['DELETE'])
 @callback_if_request_body_contains_key(apikey_can_configure_dnssec()(),