zonemaster-tools/zmtools/blocklists.py

"""Is the domain blocked by the lists people load into Pi-hole and AdGuard?

Two independent answers:

1. **Lists.** The popular lists are downloaded every 12 h into SQLite
   (`/var/db/zmtools/blocklists.sqlite`, rebuilt beside and swapped in
   atomically) and matched the way each blocker matches them:
   - hosts-format lists (Pi-hole's default, StevenBlack): exact name only;
   - adblock-format lists (`||name^`, AdGuard / Pi-hole v6 ABP style):
     the name *or any parent*, with `@@||name^` exceptions honoured.
2. **Live filtering resolvers.** AdGuard DNS, Cloudflare for Families,
   Quad9, OpenDNS FamilyShield, CleanBrowsing and Control D are asked
   directly; a sinkhole address or an NXDOMAIN that the unfiltered resolver
   does not give means "blocked".
"""

import os
import re
import sqlite3
import threading
import time
import urllib.request
from concurrent.futures import ThreadPoolExecutor

from . import dnsutil

DB_PATH = os.environ.get('ZMTOOLS_BLOCKLIST_DB', '/var/db/zmtools/blocklists.sqlite')
REFRESH = 12 * 3600

# (id, name, used by, format, url)
LISTS = [
    ('adguard_dns', 'AdGuard DNS filter', 'adguard', 'adblock',
     'https://adguardteam.github.io/AdGuardSDNSFilter/Filters/filter.txt'),
    ('stevenblack', 'StevenBlack Unified (Pi-hole default)', 'pihole', 'hosts',
     'https://raw.githubusercontent.com/StevenBlack/hosts/master/hosts'),
    ('hagezi_pro', 'HaGeZi Multi PRO', 'both', 'adblock',
     'https://raw.githubusercontent.com/hagezi/dns-blocklists/main/adblock/pro.txt'),
    ('oisd_small', 'OISD small', 'both', 'adblock', 'https://small.oisd.nl/'),
    ('urlhaus', 'URLhaus malware (abuse.ch)', 'both', 'hosts',
     'https://urlhaus.abuse.ch/downloads/hostfile/'),
    ('phishing_army', 'Phishing Army', 'both', 'domains',
     'https://phishing.army/download/phishing_army_blocklist.txt'),
    ('hagezi_tif', 'HaGeZi Threat Intelligence', 'both', 'adblock',
     'https://raw.githubusercontent.com/hagezi/dns-blocklists/main/adblock/tif.mini.txt'),
]

# (id, name, ip, how it signals a block)
FILTERING_RESOLVERS = [
    ('adguard', 'AdGuard DNS', '94.140.14.14', 'sinkhole'),
    ('adguard_family', 'AdGuard DNS Family', '94.140.14.15', 'sinkhole'),
    ('cf_malware', 'Cloudflare 1.1.1.2 (malware)', '1.1.1.2', 'sinkhole'),
    ('cf_family', 'Cloudflare 1.1.1.3 (family)', '1.1.1.3', 'sinkhole'),
    ('quad9', 'Quad9 9.9.9.9 (threat blocking)', '9.9.9.9', 'nxdomain'),
    ('opendns_family', 'OpenDNS FamilyShield', '208.67.222.123', 'sinkhole'),
    ('cleanbrowsing', 'CleanBrowsing Security', '185.228.168.9', 'sinkhole'),
    ('controld_malware', 'Control D (malware)', '76.76.2.1', 'sinkhole'),
]
SINKHOLES = {'0.0.0.0', '127.0.0.1', '94.140.14.33', '94.140.14.35',
             '146.112.61.104', '146.112.61.105', '146.112.61.106', '146.112.61.107',
             '146.112.61.108', '146.112.61.109', '146.112.61.110'}

_pool = ThreadPoolExecutor(max_workers=16, thread_name_prefix='bl')
_lock = threading.Lock()
_ADB = re.compile(r'^(@@)?\|\|([a-z0-9._-]+)\^(\$(.*))?$')
_HOSTS = re.compile(r'^(?:0\.0\.0\.0|127\.0\.0\.1|::1?|::)\s+([a-z0-9._-]+)')
_SKIP_HOSTS = {'localhost', 'localhost.localdomain', 'local', 'broadcasthost',
               'ip6-localhost', 'ip6-loopback', '0.0.0.0'}


def parse(fmt, text):
    """Yield (domain, rule, is_exception). Rules with modifiers that change
    matching beyond 'block this name' ($denyallow, $client, $ctag…) or
    wildcards/regex are skipped: we cannot evaluate them for an arbitrary
    visitor, and guessing would mislabel domains."""
    for line in text.splitlines():
        line = line.strip().lower()
        if not line or line[0] in '!#[':
            continue
        if fmt == 'adblock':
            m = _ADB.match(line)
            if not m:
                continue
            mods = m.group(4) or ''
            if mods and any(k in mods for k in ('denyallow', 'client', 'ctag', 'dnstype', 'dnsrewrite', 'badfilter')):
                continue
            yield m.group(2).strip('.'), line, bool(m.group(1))
        elif fmt == 'hosts':
            m = _HOSTS.match(line)
            if m and m.group(1) not in _SKIP_HOSTS:
                yield m.group(1).strip('.'), line, False
        else:  # plain domain per line
            d = line.split()[0]
            if '.' in d and not d.startswith(('*', '/')):
                yield d.strip('.'), line, False


def _fetch(url):
    req = urllib.request.Request(url, headers={'User-Agent': dnsutil.UA})
    with urllib.request.urlopen(req, timeout=60) as r:
        return r.read(64 * 1024 * 1024).decode('utf-8', 'replace')


def refresh(force=False):
    """Rebuild the database if it is older than REFRESH. Safe to call often;
    only one rebuild runs at a time, and readers keep using the old file."""
    if not force and os.path.exists(DB_PATH) and time.time() - os.path.getmtime(DB_PATH) < REFRESH:
        return False
    if not _lock.acquire(blocking=False):
        return False
    try:
        os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
        tmp = DB_PATH + '.new'
        if os.path.exists(tmp):
            os.remove(tmp)
        con = sqlite3.connect(tmp)
        con.execute('CREATE TABLE rules (list TEXT, domain TEXT, rule TEXT, exc INTEGER)')
        con.execute('CREATE TABLE lists (id TEXT PRIMARY KEY, entries INTEGER, fetched REAL, error TEXT)')
        old = _old_rows()
        for lid, _, _, fmt, url in LISTS:
            try:
                rows = [(lid, d, r, int(e)) for d, r, e in parse(fmt, _fetch(url))]
                con.executemany('INSERT INTO rules VALUES (?,?,?,?)', rows)
                con.execute('INSERT INTO lists VALUES (?,?,?,NULL)', (lid, len(rows), time.time()))
            except Exception as e:  # keep yesterday's copy of a list that failed today
                kept = old.get(lid)
                if kept:
                    con.executemany('INSERT INTO rules VALUES (?,?,?,?)', kept['rules'])
                con.execute('INSERT INTO lists VALUES (?,?,?,?)',
                            (lid, len(kept['rules']) if kept else 0,
                             kept['fetched'] if kept else None, type(e).__name__))
            con.commit()
        con.execute('CREATE INDEX rules_domain ON rules (domain)')
        con.commit()
        con.close()
        os.replace(tmp, DB_PATH)
        return True
    finally:
        _lock.release()


def _old_rows():
    if not os.path.exists(DB_PATH):
        return {}
    out = {}
    con = sqlite3.connect(DB_PATH)
    try:
        for lid, fetched, err in con.execute('SELECT id, fetched, error FROM lists'):
            out[lid] = {'fetched': fetched,
                        'rules': con.execute('SELECT list, domain, rule, exc FROM rules WHERE list=?',
                                             (lid,)).fetchall()}
    finally:
        con.close()
    return out


def parents(domain):
    labels = domain.split('.')
    return ['.'.join(labels[i:]) for i in range(len(labels) - 1)]


def check_lists(domain):
    if not os.path.exists(DB_PATH):
        return {'ready': False, 'lists': []}
    con = sqlite3.connect(f'file:{DB_PATH}?mode=ro', uri=True)
    try:
        names = parents(domain)
        q = 'SELECT list, domain, rule, exc FROM rules WHERE domain IN (%s)' % ','.join('?' * len(names))
        hits = con.execute(q, names).fetchall()
        meta = {r[0]: r[1:] for r in con.execute('SELECT id, entries, fetched, error FROM lists')}
    finally:
        con.close()
    out = []
    for lid, name, used_by, fmt, url in LISTS:
        rows = [h for h in hits if h[0] == lid]
        if fmt in ('hosts', 'domains'):
            rows = [h for h in rows if h[1] == domain]  # exact match only
        exc = [h for h in rows if h[3]]
        blk = [h for h in rows if not h[3]]
        entries, fetched, err = meta.get(lid, (0, None, 'missing'))
        out.append({
            'id': lid, 'name': name, 'used_by': used_by, 'format': fmt,
            'blocked': bool(blk) and not exc,
            'rule': blk[0][2] if blk else None,
            'matched': blk[0][1] if blk else None,
            'exception': exc[0][2] if exc else None,
            'entries': entries, 'updated': fetched, 'error': err,
        })
    return {'ready': True, 'lists': out}


def check_resolvers(domain):
    base_rcode, base_vals, _ = dnsutil.resolve_full(domain, 'A')

    def ask(spec):
        rid, name, ip, how = spec
        r, ms, err = dnsutil.query(ip, domain, 'A')
        vals = [a['value'] for a in dnsutil.answers(r) if a['type'] == 'A']
        rcode = dnsutil.rcode_text(r)
        blocked = False
        if r is not None:
            if set(vals) & SINKHOLES:
                blocked = True
            elif rcode in ('NXDOMAIN', 'REFUSED') and base_rcode == 'NOERROR' and base_vals:
                blocked = True
            elif how == 'sinkhole' and vals and base_vals and not set(vals) & set(base_vals) \
                    and any(v.startswith(('146.112.61.', '94.140.14.', '185.228.168.')) for v in vals):
                blocked = True
        return {'id': rid, 'name': name, 'ip': ip, 'rcode': rcode, 'values': vals,
                'blocked': blocked, 'error': err, 'ms': ms}

    return {'baseline': {'rcode': base_rcode, 'values': base_vals},
            'resolvers': list(_pool.map(ask, FILTERING_RESOLVERS))}


def check(domain):
    lists_f = _pool.submit(check_lists, domain)
    res = check_resolvers(domain)
    lists = lists_f.result()
    blocked_lists = [l['id'] for l in lists['lists'] if l['blocked']]
    blocked_res = [r['id'] for r in res['resolvers'] if r['blocked']]
    return {'domain': domain, **lists, **res,
            'summary': {'lists_blocking': len(blocked_lists), 'lists_total': len(LISTS),
                        'resolvers_blocking': len(blocked_res),
                        'resolvers_total': len(FILTERING_RESOLVERS)}}