Development

docker/routlin-dash/app/pages/dnsblocking/action.py

@@ -13,8 +13,6 @@ _PAGE = Path(__file__).parent.name
 
 bp = Blueprint(_PAGE, __name__)
 
-_VALID_FORMATS_STR = ', '.join(sorted(validate.VALID_BLOCKLIST_FORMATS))
-
 
 def _row_index():
     try:
@@ -30,29 +28,45 @@ def _hash_ok():
     return True
 
 
-def _save_as_from_name(name):
+def _save_as_from_name(name, ext):
     slug = re.sub(r'[^a-z0-9_-]', '_', name.lower()).strip('_')
-    return f'{slug}.conf'
+    return f'{slug}.{ext}'
+
+
+def _write_local_file(save_as, lines):
+    """Write domain list to blocklists dir. Returns error string or None."""
+    try:
+        bl_path = Path(config_utils.BLOCKLISTS_DIR) / save_as
+        bl_path.parent.mkdir(parents=True, exist_ok=True)
+        bl_path.write_text('\n'.join(lines))
+    except Exception as ex:
+        return str(ex)
+    return None
 
 
 def _parse_fields():
+    bl_type     = sanitize.filtervalue(request.form.get('bl_type', ''), {'community', 'local'})
     name        = sanitize.name(request.form.get('name', ''))
     description = sanitize.description(request.form.get('description', ''))
-    fmt         = sanitize.filtervalue(request.form.get('format', ''), validate.VALID_BLOCKLIST_FORMATS)
-    url         = sanitize.url(request.form.get('url', ''))
 
     if not name:
         flash('The configuration has not been saved because a name is required.', 'error')
         return None, True
+    if not bl_type:
+        flash('The configuration has not been saved because a type is required.', 'error')
+        return None, True
+
+    if bl_type == 'local':
+        raw = request.form.get('local_entries', '')
+        local_lines = [ln.strip() for ln in raw.splitlines() if ln.strip()]
+        return {'name': name, 'description': description, 'bl_type': 'local',
+                'local_lines': local_lines}, None
+
+    url = sanitize.url(request.form.get('url', ''))
     if not url:
         flash('The configuration has not been saved because a URL is required.', 'error')
         return None, True
-    if not fmt:
-        flash(f'The configuration has not been saved because the format is invalid. '
-              f'Accepted formats: {_VALID_FORMATS_STR}.', 'error')
-        return None, True
-
-    return {'name': name, 'description': description, 'format': fmt, 'url': url}, None
+    return {'name': name, 'description': description, 'bl_type': 'community', 'url': url}, None
 
 
 @bp.route('/action/dnsblocking/blocklists_delete', methods=['POST'])
@@ -108,18 +122,36 @@ def blocklists_edit():
 
     before = copy.deepcopy(items[idx])
 
-    # Blocklist name must be unique - it is the lookup key for VLAN use_blocklists references
     err = validate.check_blocklist_name_unique(items, fields['name'], exclude_idx=idx)
     if err:
         flash(err, 'error')
         return redirect(f'/{_PAGE}')
 
-    items[idx].update({
-        'name':        fields['name'],
-        'description': fields['description'],
-        'format':      fields['format'],
-        'url':         fields['url'],
-    })
+    if fields['bl_type'] == 'local':
+        save_as = items[idx].get('save_as') or _save_as_from_name(fields['name'], 'txt')
+        write_err = _write_local_file(save_as, fields['local_lines'])
+        if write_err:
+            flash(f'Could not save local blocklist file: {write_err}', 'error')
+            return redirect(f'/{_PAGE}')
+        items[idx].update({
+            'name':        fields['name'],
+            'description': fields['description'],
+            'bl_type':     'local',
+            'save_as':     save_as,
+        })
+        items[idx].pop('format', None)
+        items[idx].pop('url', None)
+    else:
+        items[idx].update({
+            'name':        fields['name'],
+            'description': fields['description'],
+            'bl_type':     'community',
+            'url':         fields['url'],
+        })
+        if not items[idx].get('save_as'):
+            items[idx]['save_as'] = _save_as_from_name(fields['name'], 'conf')
+        items[idx].pop('local_lines', None)
+
     errors = validate.validate_config(cfg)
     if errors:
         for msg in errors:
@@ -143,19 +175,32 @@ def addblocklist_add():
     cfg        = config_utils.load_config()
     blocklists = cfg.setdefault('dns_blocking', {}).setdefault('blocklists', [])
 
-    # Blocklist name must be unique - it is the lookup key for VLAN use_blocklists references
     err = validate.check_blocklist_name_unique(blocklists, fields['name'])
     if err:
         flash(err, 'error')
         return redirect(f'/{_PAGE}')
 
-    entry = {
-        'name':        fields['name'],
-        'description': fields['description'],
-        'format':      fields['format'],
-        'url':         fields['url'],
-        'save_as':     _save_as_from_name(fields['name']),
-    }
+    if fields['bl_type'] == 'local':
+        save_as = _save_as_from_name(fields['name'], 'txt')
+        write_err = _write_local_file(save_as, fields['local_lines'])
+        if write_err:
+            flash(f'Could not save local blocklist file: {write_err}', 'error')
+            return redirect(f'/{_PAGE}')
+        entry = {
+            'name':        fields['name'],
+            'description': fields['description'],
+            'bl_type':     'local',
+            'save_as':     save_as,
+        }
+    else:
+        entry = {
+            'name':        fields['name'],
+            'description': fields['description'],
+            'bl_type':     'community',
+            'url':         fields['url'],
+            'save_as':     _save_as_from_name(fields['name'], 'conf'),
+        }
+
     blocklists.append(entry)
     errors = validate.validate_config(cfg)
     if errors: