zonemaster-tools/zmtools/blacklists.py

"""Classic DNS blacklists (DNSBL / RHSBL / URIBL).

Domain lists are asked about the domain itself; IP lists about every public
IPv4 address of the domain's web and mail hosts. Queries go straight to each
list's authoritative servers: Spamhaus, URIBL and SURBL refuse (or answer
with a sentinel code) anything relayed through big public resolvers.

Each list's return codes are mapped to a reason; known "you may not query"
codes are reported as unavailable, never as listed."""

from concurrent.futures import ThreadPoolExecutor

from . import dnsutil

_pool = ThreadPoolExecutor(max_workers=48, thread_name_prefix='dnsbl')

# (id, name, zone, info url, {code: reason}, {codes meaning 'query refused'}).
# Integer keys mean the list answers a bitmask in the last octet (SURBL, URIBL).
DOMAIN_LISTS = [
    ('spamhaus_dbl', 'Spamhaus DBL', 'dbl.spamhaus.org',
     'https://check.spamhaus.org/',
     {'127.0.1.2': 'spam', '127.0.1.4': 'phish', '127.0.1.5': 'malware',
      '127.0.1.6': 'botnet_cc', '127.0.1.102': 'abused_legit_spam',
      '127.0.1.103': 'abused_redirector', '127.0.1.104': 'abused_legit_phish',
      '127.0.1.105': 'abused_legit_malware', '127.0.1.106': 'abused_legit_botnet_cc'},
     {'127.255.255.252', '127.255.255.254', '127.255.255.255'}),
    ('surbl', 'SURBL multi', 'multi.surbl.org', 'https://www.surbl.org/surbl-analysis',
     {8: 'phish', 16: 'malware', 64: 'abuse', 128: 'cracked'},
     {'127.0.0.1', '127.0.0.254'}),
    ('uribl', 'URIBL multi', 'multi.uribl.com', 'https://admin.uribl.com/',
     {2: 'black', 4: 'grey', 8: 'red'}, {'127.0.0.1'}),
    ('sem_uribl', 'SEM URIBL', 'uribl.spameatingmonkey.net',
     'https://spameatingmonkey.com/', {'127.0.0.2': 'spam'}, set()),
    ('sem_fresh', 'SEM FRESH (new domains)', 'fresh.spameatingmonkey.net',
     'https://spameatingmonkey.com/', {'127.0.0.2': 'recently_registered'}, set()),
    ('nordspam', 'Nordspam DBL', 'dbl.nordspam.com', 'https://www.nordspam.com/',
     {'127.0.0.2': 'spam'}, set()),
]

IP_LISTS = [
    ('spamhaus_zen', 'Spamhaus ZEN', 'zen.spamhaus.org', 'https://check.spamhaus.org/',
     {'127.0.0.2': 'sbl', '127.0.0.3': 'sbl_css', '127.0.0.4': 'xbl', '127.0.0.9': 'drop',
      '127.0.0.10': 'pbl_isp', '127.0.0.11': 'pbl'},
     {'127.255.255.252', '127.255.255.254', '127.255.255.255'}),
    ('spamcop', 'SpamCop', 'bl.spamcop.net', 'https://www.spamcop.net/bl.shtml',
     {'127.0.0.2': 'spam'}, set()),
    ('barracuda', 'Barracuda BRBL', 'b.barracudacentral.org',
     'https://www.barracudacentral.org/lookups', {'127.0.0.2': 'spam'}, set()),
    ('psbl', 'PSBL', 'psbl.surriel.com', 'https://psbl.org/', {'127.0.0.2': 'spam'}, set()),
    ('mailspike', 'Mailspike', 'bl.mailspike.net', 'https://mailspike.org/',
     {'127.0.0.2': 'spam', '127.0.0.10': 'worst', '127.0.0.11': 'very_bad', '127.0.0.12': 'bad',
      '127.0.0.13': 'suspicious'}, set()),
    ('uceprotect1', 'UCEPROTECT L1', 'dnsbl-1.uceprotect.net',
     'https://www.uceprotect.net/', {'127.0.0.2': 'spam'}, set()),
    ('dronebl', 'DroneBL', 'dnsbl.dronebl.org', 'https://dronebl.org/lookup',
     {}, set()),
    ('s5h', 'S5H', 'all.s5h.net', 'https://www.usenix.org.uk/content/rbl.html',
     {'127.0.0.2': 'spam'}, set()),
    ('blocklist_de', 'blocklist.de', 'bl.blocklist.de', 'https://www.blocklist.de/',
     {'127.0.0.2': 'attacks'}, set()),
    ('anonmails', 'Anonmails', 'spam.dnsbl.anonmails.de', 'https://anonmails.de/dnsbl.php',
     {'127.0.0.2': 'spam'}, set()),
]

MAX_IPS = 6


def _lookup(spec, qname):
    lid, name, zone, url, codes, refused = spec
    rcode, vals = dnsutil.query_authoritative(zone, f'{qname}.{zone}', 'A')
    row = {'id': lid, 'name': name, 'zone': zone, 'url': url}
    if rcode is None:
        return {**row, 'state': 'error'}
    if rcode == 'NXDOMAIN' or not vals:
        return {**row, 'state': 'clean'}
    if set(vals) & refused:
        return {**row, 'state': 'unavailable', 'codes': vals}
    if not all(v.startswith('127.') for v in vals):
        # A wildcard / parked list answering junk is not a listing.
        return {**row, 'state': 'error', 'codes': vals}
    if codes and isinstance(next(iter(codes)), int):
        bits = 0
        for v in vals:
            bits |= int(v.rsplit('.', 1)[1])
        reasons = sorted(r for b, r in codes.items() if bits & b) or ['listed']
    else:
        reasons = sorted({codes.get(v, 'listed') for v in vals})
    # Spamhaus PBL only says "this is end-user address space" — not a sin for
    # a web server, but a mail server there will be rejected.
    return {**row, 'state': 'listed', 'codes': vals, 'reasons': reasons}


def check(domain):
    if dnsutil.resolve_full(domain, 'SOA')[0] == 'NXDOMAIN':
        raise dnsutil.BadInput('domain does not exist')
    # Which IPs to check, and why.
    targets = {}
    for ip in dnsutil.resolve(domain, 'A'):
        targets.setdefault(ip, set()).add('web')
    for ip in dnsutil.resolve('www.' + domain, 'A'):
        targets.setdefault(ip, set()).add('web')
    for v in dnsutil.resolve(domain, 'MX'):
        host = v.split(None, 1)[1].rstrip('.')
        if host:
            for ip in dnsutil.resolve(host, 'A'):
                targets.setdefault(ip, set()).add('mail')
    ips = [ip for ip in targets if dnsutil.is_public_ip(ip)][:MAX_IPS]

    dom_jobs = [_pool.submit(_lookup, s, domain) for s in DOMAIN_LISTS]
    ip_jobs = {ip: [_pool.submit(_lookup, s, '.'.join(reversed(ip.split('.'))))
                    for s in IP_LISTS] for ip in ips}
    domain_rows = [j.result() for j in dom_jobs]
    ip_rows = [{'ip': ip, 'roles': sorted(targets[ip]), 'ptr': dnsutil.ptr(ip),
                'lists': [j.result() for j in jobs]} for ip, jobs in ip_jobs.items()]
    listed = sum(r['state'] == 'listed' for r in domain_rows) + \
        sum(l['state'] == 'listed' for r in ip_rows for l in r['lists'])
    return {'domain': domain, 'domain_lists': domain_rows, 'ips': ip_rows,
            'skipped_ips': max(0, len(targets) - len(ips)),
            'summary': {'listed': listed,
                        'checked': len(domain_rows) + sum(len(r['lists']) for r in ip_rows)}}