bindmanager/core/adblock_manager.py

import os
import re
import time
import datetime
import subprocess
import urllib.request
import json
import config
from core.file_manager import read_file, write_file, load_adblock_lists, load_custom_rules, load_panel_settings, save_panel_settings
from core.dns_manager import get_zone_block

def is_valid_domain(d):
    if not d or len(d) > 253: return False
    if '..' in d or d.startswith('.'): return False
    # Επιτρέπει μόνο αλφαριθμητικά, παύλες, underscores και τελείες.
    if not re.match(r'^[a-zA-Z0-9\-_.]+$', d): return False
    if '.' not in d: return False
    return True

def generate_and_apply_rpz():
    lists = load_adblock_lists()
    active_lists = [l for l in lists if l.get('enabled')]
    custom_rules = load_custom_rules()
    
    domains_set = set()
    
    try:
        # 1. Fetch Lists
        for lst in active_lists:
            req = urllib.request.Request(lst['url'], headers={'User-Agent': 'Mozilla/5.0'})
            try:
                with urllib.request.urlopen(req, timeout=15, context=config.ssl_ctx) as response:
                    lines = response.read().decode('utf-8', errors='ignore').splitlines()
                    for line in lines:
                        l = line.split('#')[0].strip() # Κόβουμε inline comments
                        
                        # Γρήγορα φίλτρα (Αγνοούμε σχόλια, HTML, και Regex τύπου Pi-hole)
                        if not l or l.startswith(('!', ':', '[', '@')): continue
                        if '/' in l or '*' in l or '|' in l and not l.startswith('||'): continue
                        
                        d = ""
                        if l.startswith('||'):
                            d = re.split(r'[\^/\$:]', l[2:])[0]
                        elif l.startswith(('0.0.0.0 ', '127.0.0.1 ')):
                            parts = l.split()
                            if len(parts) >= 2 and parts[1] not in ('localhost', 'broadcasthost', 'localhost.localdomain'):
                                d = parts[1]
                        elif ' ' not in l:
                            d = re.split(r'[\^/\$:]', l)[0]
                            
                        d = d.strip().rstrip('.')
                        # Αυστηρός Έλεγχος (Strict Validation)
                        if is_valid_domain(d):
                            domains_set.add(d)
            except Exception as e:
                print(f"Failed to fetch {lst['name']}: {e}")

        # 2. Add Custom Deny
        for d in custom_rules.get('deny', []):
            clean_d = d.strip().rstrip('.')
            if clean_d.startswith('*.'): clean_d = clean_d[2:]
            if is_valid_domain(clean_d):
                domains_set.add(clean_d)

        # 3. Create RPZ
        rpz_records = []
        
        # Μετατροπή allowed λίστας σε SET + καθάρισμα (Για ακαριαία ταχύτητα O(1))
        allowed_cleaned = set()
        for d in custom_rules.get('allowed', []):
            clean_d = d.strip().rstrip('.')
            if clean_d.startswith('*.'): clean_d = clean_d[2:]
            if is_valid_domain(clean_d):
                allowed_cleaned.add(clean_d)
        
        # ΠΡΩΤΑ τα Allowed (PASSTHRU)
        for d in allowed_cleaned:
            rpz_records.append(f"{d}\tCNAME\trpz-passthru.")
            rpz_records.append(f"*.{d}\tCNAME\trpz-passthru.")

        # ΜΕΤΑ τα Blocked
        for d in domains_set:
            # Ακαριαίος έλεγχος χάρη στο set()
            if d in allowed_cleaned:
                continue
            rpz_records.append(f"{d}\tCNAME\t.")
            rpz_records.append(f"*.{d}\tCNAME\t.")

        serial = time.strftime('%Y%m%d%H') # Ώρα αντί για 01 για σίγουρο increment
        header = f"$TTL 60\n$ORIGIN rpz.blacklist.\n@ IN SOA localhost. root.localhost. ({serial} 3h 1h 1w 1h)\n IN NS localhost.\n"
        content = header + "\n".join(rpz_records) + "\n"
        
        write_file(config.ADBLOCK_DB_PATH, content)
        subprocess.run(["chown", "bind:bind", config.ADBLOCK_DB_PATH])
        
        # Ενημέρωση named.conf.local
        conf_local = read_file(config.LOCAL_CONF)
        if 'zone "rpz.blacklist"' not in conf_local:
            conf_local += f'\nzone "rpz.blacklist" {{\n    type master;\n    file "{config.ADBLOCK_DB_PATH}";\n    allow-query {{ any; }};\n    allow-transfer {{ any; }};\n}};\n'
            write_file(config.LOCAL_CONF, conf_local)

        # Ενημέρωση named.conf.options
        options = read_file(config.OPTIONS_FILE)
        if "response-policy" not in options:
            options = re.sub(r'(options\s*\{)', r'\1\n    response-policy { zone "rpz.blacklist"; };', options)
            write_file(config.OPTIONS_FILE, options)

        subprocess.run(["rndc", "reconfig"])
        subprocess.run(["rndc", "reload", "rpz.blacklist"])
        
        s = load_panel_settings()
        s['adblock_last_update'] = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        s['adblock_domains_count'] = len(domains_set)
        save_panel_settings(s)
        
        return {"status": "success", "count": len(domains_set), "time": s['adblock_last_update']}
    except Exception as e:
        return {"status": "error", "message": str(e)}

def disable_rpz():
    conf_local = read_file(config.LOCAL_CONF)
    options = read_file(config.OPTIONS_FILE)
    
    block = get_zone_block(conf_local, "rpz.blacklist")
    if block:
        conf_local = conf_local.replace(block, "")
        write_file(config.LOCAL_CONF, conf_local)
        
    options = re.sub(r'\s*response-policy\s*\{\s*zone\s*"rpz\.blacklist";\s*\};\s*', '\n', options)
    write_file(config.OPTIONS_FILE, options)
    
    subprocess.run(["rndc", "reconfig"])
    return {"status": "success", "message": "AdBlock (RPZ) disabled."}