Development

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

@@ -1,7 +1,8 @@
 from pathlib import Path
 import copy
 import re
-from flask import Blueprint, request, redirect, flash, send_file
+import sqlite3
+from flask import Blueprint, request, redirect, flash, send_file, jsonify
 import auth
 import config_utils
 import sanitize
@@ -69,6 +70,98 @@ def _parse_fields():
     return {'name': name, 'description': description, 'bl_type': 'community', 'url': url}, None
 
 
+@bp.route('/api/dnsblocking/search', methods=['GET'])
+@auth.require_level('viewer')
+def api_blocklist_search():
+    term  = request.args.get('term', '').strip()
+    match = request.args.get('match', 'partial')
+    if not term:
+        return jsonify({'results': [], 'count': 0, 'truncated': False})
+    if match not in ('exact', 'starts_with', 'ends_with', 'partial'):
+        match = 'partial'
+
+    escaped = term.replace('\\', '\\\\').replace('%', '\\%').replace('_', '\\_')
+    if match == 'exact':
+        sql_where = 'WHERE d.domain = ?'
+        param     = term
+    elif match == 'starts_with':
+        sql_where = "WHERE d.domain LIKE ? ESCAPE '\\'"
+        param     = escaped + '%'
+    elif match == 'ends_with':
+        sql_where = "WHERE d.domain LIKE ? ESCAPE '\\'"
+        param     = '%' + escaped
+    else:
+        sql_where = "WHERE d.domain LIKE ? ESCAPE '\\'"
+        param     = '%' + escaped + '%'
+
+    db_path = str(Path(config_utils.BLOCKLISTS_DIR) / 'domains.db')
+    try:
+        con  = sqlite3.connect(db_path)
+        rows = con.execute(f"""
+            SELECT d.domain, GROUP_CONCAT(b.name, '|')
+            FROM domains d
+            JOIN blocklists b ON b.id = d.blocklist_id
+            {sql_where}
+            GROUP BY d.domain
+            ORDER BY d.domain
+            LIMIT 501
+        """, (param,)).fetchall()
+        capped_rows = rows[:500]
+        domain_list = [r[0] for r in capped_rows]
+        phs         = ','.join('?' * len(domain_list))
+        overridden  = set(
+            r[0] for r in con.execute(
+                f"SELECT domain FROM overrides WHERE domain IN ({phs})", domain_list
+            ).fetchall()
+        ) if domain_list else set()
+        con.close()
+    except Exception:
+        return jsonify({'error': 'Database unavailable. Run merge-blocklists first.'})
+
+    cfg     = config_utils.load_config()
+    bl_vlans = {}
+    for vlan in cfg.get('vlans', []):
+        for bl_name in vlan.get('use_blocklists', []):
+            bl_vlans.setdefault(bl_name, []).append(vlan['name'])
+
+    truncated = len(rows) > 500
+
+    results = []
+    for domain, bl_str in capped_rows:
+        bl_names = bl_str.split('|') if bl_str else []
+        vlans    = []
+        for bl in bl_names:
+            for v in bl_vlans.get(bl, []):
+                if v not in vlans:
+                    vlans.append(v)
+        results.append({'domain': domain, 'blocklists': bl_names, 'vlans': vlans, 'overridden': domain in overridden})
+
+    return jsonify({'results': results, 'count': len(results), 'truncated': truncated})
+
+
+@bp.route('/api/dnsblocking/override', methods=['POST'])
+@auth.require_level('administrator')
+def api_blocklist_override():
+    data   = request.get_json(silent=True) or {}
+    domain = str(data.get('domain', '')).strip().lower()
+    allow  = bool(data.get('allow', False))
+    if not domain:
+        return jsonify({'error': 'domain required'})
+    db_path = str(Path(config_utils.BLOCKLISTS_DIR) / 'domains.db')
+    try:
+        con = sqlite3.connect(db_path)
+        if allow:
+            con.execute("INSERT OR IGNORE INTO overrides (domain) VALUES (?)", (domain,))
+        else:
+            con.execute("DELETE FROM overrides WHERE domain = ?", (domain,))
+        con.commit()
+        con.close()
+    except Exception as e:
+        return jsonify({'error': str(e)})
+    config_utils.queue_command('merge blocklists')
+    return jsonify({'ok': True})
+
+
 @bp.route('/action/dnsblocking/blocklists_delete', methods=['POST'])
 @auth.require_level('administrator')
 def blocklists_delete():