zonemaster-tools/zmtools/security.py

"""DNS security scan. Every check returns a language-neutral result:

    {'id': 'spf', 'status': 'pass'|'warn'|'fail'|'info', 'data': {...}}

The GUI turns (id, status, data) into prose in each locale, so nothing here
is user-facing text. Results are kept per domain for the tracker view."""

import random
import re
import string
from concurrent.futures import ThreadPoolExecutor

import dns.flags
import dns.query
import dns.rdatatype
import dns.zone

from . import dnsutil

_pool = ThreadPoolExecutor(max_workers=32, thread_name_prefix='sec')

WEIGHTS = {'dnssec': 15, 'ns': 10, 'soa': 5, 'axfr': 10, 'spf': 12, 'dmarc': 12,
           'dkim': 6, 'mx': 5, 'caa': 7, 'mta_sts': 3, 'tls_rpt': 2,
           'dangling': 13}
POINTS = {'pass': 1.0, 'warn': 0.5, 'fail': 0.0}

DKIM_SELECTORS = ['default', 'dkim', 'mail', 'selector1', 'selector2', 'google',
                  'k1', 'k2', 'k3', 's1', 's2', 'smtp', 'mx', 'zoho', 'zmail',
                  'protonmail', 'protonmail2', 'protonmail3', 'fm1', 'fm2', 'fm3',
                  'mandrill', 'mailjet', 'sendgrid', 'smtpapi', 'everlytic', 'sig1',
                  'ionos', 'key1', 'key2', 'dk', 'mailchimp', 'pm', 'hs1', 'hs2']

SUBDOMAINS = ['www', 'mail', 'webmail', 'blog', 'shop', 'tienda', 'store', 'app', 'api',
              'dev', 'staging', 'test', 'beta', 'cdn', 'static', 'img', 'assets', 'docs',
              'help', 'support', 'status', 'portal', 'admin', 'm', 'autodiscover',
              'news', 'landing', 'go', 'promo', 'campus', 'crm']

# CNAME targets whose owner name can be claimed by anyone once the resource is
# deleted (see can-i-take-over-xyz). A CNAME to one of these that no longer
# resolves is a subdomain takeover waiting to happen.
TAKEOVER_SUFFIXES = ['github.io', 'herokuapp.com', 'herokudns.com', 'azurewebsites.net',
                     'cloudapp.net', 'cloudapp.azure.com', 'trafficmanager.net',
                     'blob.core.windows.net', 'azureedge.net', 's3.amazonaws.com',
                     's3-website', 'elasticbeanstalk.com', 'netlify.app', 'netlify.com',
                     'vercel.app', 'surge.sh', 'ghost.io', 'myshopify.com',
                     'pantheonsite.io', 'readthedocs.io', 'bitbucket.io',
                     'wordpress.com', 'zendesk.com', 'helpscoutdocs.com', 'tumblr.com',
                     'webflow.io', 'fly.dev', 'agilecrm.com', 'unbouncepages.com',
                     'strikinglydns.com', 'wixdns.net', 'uservoice.com', 'teamwork.com',
                     'pages.dev', 'onrender.com', 'smartjob.com']


def _check(cid, status, **data):
    return {'id': cid, 'status': status, 'data': data}


VALIDATORS = [('cloudflare', '1.1.1.1'), ('google', '8.8.8.8'), ('quad9', '9.9.9.9'),
              ('opendns', '208.67.222.222')]


def check_dnssec(domain):
    ds = dnsutil.resolve(domain, 'DS')
    keys = dnsutil.resolve(domain, 'DNSKEY')
    algs = sorted({int(k.split()[2]) for k in keys}) if keys else []

    # Ask several validators: they do not all implement every algorithm
    # (Cloudflare and Google treat Ed448, alg 16, as insecure), so one
    # resolver's missing AD bit does not mean the zone is unsigned.
    # Broken validation shows as SERVFAIL that goes away with CD set.
    def ask(v):
        name, ip = v
        r, _, _ = dnsutil.query(ip, domain, 'SOA', dnssec=True)
        if r is None:
            return name, None
        if dnsutil.rcode_text(r) == 'SERVFAIL':
            rcd, _, _ = dnsutil.query(ip, domain, 'SOA', dnssec=True, cd=True)
            if rcd is not None and dnsutil.rcode_text(rcd) == 'NOERROR':
                return name, 'bogus'
        return name, 'secure' if r.flags & dns.flags.AD else 'insecure'

    verdicts = {n: v for n, v in _pool.map(ask, VALIDATORS) if v}
    secure = sorted(n for n, v in verdicts.items() if v == 'secure')
    insecure = sorted(n for n, v in verdicts.items() if v == 'insecure')
    bogus = sorted(n for n, v in verdicts.items() if v == 'bogus')
    data = dict(ds=len(ds), dnskey=len(keys), validated=bool(secure), algorithms=algs,
                validators=verdicts)
    if bogus:
        return _check('dnssec', 'fail', bogus=True, bogus_on=bogus, **data)
    if ds and secure:
        weak = [a for a in algs if a in (1, 3, 5, 6, 7, 12)]
        status = 'warn' if weak or insecure else 'pass'
        return _check('dnssec', status, weak_algorithms=weak, not_validated_by=insecure, **data)
    if ds and keys:
        return _check('dnssec', 'warn', not_validated_by=insecure, **data)
    if keys and not ds:
        return _check('dnssec', 'warn', missing_ds=True, **data)
    return _check('dnssec', 'fail', unsigned=True, **data)


def _ns_info(domain):
    names = sorted(n.rstrip('.') for n in dnsutil.resolve(domain, 'NS'))
    info = []
    for n in names:
        v4 = [i for i in dnsutil.resolve(n, 'A') if dnsutil.is_public_ip(i)]
        v6 = dnsutil.resolve(n, 'AAAA')
        info.append({'name': n, 'ipv4': v4, 'ipv6': v6})
    return info


def check_ns(domain, ns):
    if not ns:
        return _check('ns', 'fail', count=0)
    nets = {'.'.join(i.split('.')[:3]) for n in ns for i in n['ipv4']}
    has_v6 = any(n['ipv6'] for n in ns)
    status = 'pass'
    if len(ns) < 2 or len(nets) < 2:
        status = 'fail' if len(ns) < 2 else 'warn'
    elif not has_v6:
        status = 'warn'
    return _check('ns', status, count=len(ns), subnets=len(nets), ipv6=has_v6,
                  servers=[n['name'] for n in ns])


def check_soa(domain, ns):
    serials, lame = {}, []

    def ask(n):
        if not n['ipv4']:
            return n['name'], None
        r, _, _ = dnsutil.query(n['ipv4'][0], domain, 'SOA', rd=False)
        if r is None or not (r.flags & dns.flags.AA):
            return n['name'], None
        a = [x for x in dnsutil.answers(r) if x['type'] == 'SOA']
        return n['name'], int(a[0]['value'].split()[2]) if a else None

    for name, serial in _pool.map(ask, ns):
        if serial is None:
            lame.append(name)
        else:
            serials[name] = serial
    if not serials:
        return _check('soa', 'fail', lame=lame, serials={})
    status = 'pass' if len(set(serials.values())) == 1 and not lame else 'warn'
    return _check('soa', status, lame=lame, serials=serials)


def check_axfr(domain, ns):
    open_on = []

    def try_xfr(n):
        for ip in n['ipv4'][:1]:
            try:
                z = dns.zone.from_xfr(dns.query.xfr(ip, domain, timeout=4, lifetime=8))
                return n['name'], len(z.nodes)
            except Exception:
                return n['name'], None
        return n['name'], None

    for name, nodes in _pool.map(try_xfr, ns[:6]):
        if nodes:
            open_on.append({'server': name, 'names': nodes})
    return _check('axfr', 'fail' if open_on else 'pass', open=open_on)


_SPF_MECH = re.compile(r'^([+\-~?]?)(include|a|mx|ptr|exists|redirect)(?:[:=]|$)', re.I)


def spf_lookups(record, depth=0, seen=None):
    """Count DNS-querying mechanisms per RFC 7208 §4.6.4, following includes.
    Returns (count, [include domains that do not resolve])."""
    seen = seen if seen is not None else set()
    count, broken = 0, []
    for term in record.split()[1:]:
        m = _SPF_MECH.match(term)
        if not m:
            continue
        count += 1
        mech = m.group(2).lower()
        if mech in ('include', 'redirect') and depth < 10:
            target = term.split(':' if mech == 'include' else '=', 1)[1].lower().rstrip('.')
            if target in seen:
                continue
            seen.add(target)
            sub = [t for t in dnsutil.txt_values(target) if t.lower().startswith('v=spf1')]
            if not sub:
                broken.append(target)
                continue
            c, b = spf_lookups(sub[0], depth + 1, seen)
            count += c
            broken += b
    return count, broken


def check_spf(domain):
    recs = [t for t in dnsutil.txt_values(domain) if t.lower().startswith('v=spf1')]
    if not recs:
        return _check('spf', 'fail', present=False)
    if len(recs) > 1:
        return _check('spf', 'fail', present=True, multiple=len(recs), record=recs[0])
    rec = recs[0]
    lookups, broken = spf_lookups(rec)
    all_q = re.search(r'\s([+\-~?]?)all\b', ' ' + rec)
    qual = all_q.group(1) or '+' if all_q else None
    issues = []
    if qual in ('+', '?'):
        issues.append('permissive_all')
    if qual is None and 'redirect=' not in rec:
        issues.append('no_all')
    if lookups > 10:
        issues.append('too_many_lookups')
    if re.search(r'\bptr\b', rec):
        issues.append('ptr')
    if broken:
        issues.append('broken_include')
    status = 'pass'
    if 'permissive_all' in issues or 'too_many_lookups' in issues:
        status = 'fail'
    elif issues or qual == '~':
        status = 'warn' if issues else 'pass'
    return _check('spf', status, present=True, record=rec, lookups=lookups,
                  all=qual, issues=issues, broken=broken)


def check_dmarc(domain):
    recs = [t for t in dnsutil.txt_values('_dmarc.' + domain) if t.lower().startswith('v=dmarc1')]
    if not recs:
        return _check('dmarc', 'fail', present=False)
    tags = dict(kv.strip().split('=', 1) for kv in recs[0].split(';') if '=' in kv)
    tags = {k.strip().lower(): v.strip() for k, v in tags.items()}
    policy = tags.get('p', '').lower()
    pct = int(tags.get('pct', '100')) if tags.get('pct', '100').isdigit() else 100
    status = {'reject': 'pass', 'quarantine': 'pass', 'none': 'warn'}.get(policy, 'fail')
    if status == 'pass' and pct < 100:
        status = 'warn'
    return _check('dmarc', status, present=True, record=recs[0], policy=policy,
                  subdomain_policy=tags.get('sp'), pct=pct, rua='rua' in tags)


def check_dkim(domain):
    def probe(sel):
        vals = dnsutil.txt_values(f'{sel}._domainkey.{domain}')
        return sel if any('p=' in v for v in vals) else None
    found = [s for s in _pool.map(probe, DKIM_SELECTORS) if s]
    # Not finding a selector proves nothing — they are not enumerable.
    return _check('dkim', 'pass' if found else 'info', selectors=found,
                  probed=len(DKIM_SELECTORS))


def check_mx(domain):
    mx = []
    for v in dnsutil.resolve(domain, 'MX'):
        pref, host = v.split(None, 1)
        mx.append({'pref': int(pref), 'host': host.rstrip('.')})
    mx.sort(key=lambda m: m['pref'])
    if not mx:
        return _check('mx', 'info', count=0)
    if len(mx) == 1 and mx[0]['host'] in ('', '.'):
        return _check('mx', 'info', null_mx=True, count=0)
    bad = []
    for m in mx:
        if dnsutil.resolve(m['host'], 'CNAME'):
            bad.append({'host': m['host'], 'problem': 'cname'})
        elif not dnsutil.addresses(m['host']):
            bad.append({'host': m['host'], 'problem': 'no_address'})
    status = 'fail' if len(bad) == len(mx) else 'warn' if bad else 'pass'
    return _check('mx', status, count=len(mx), hosts=mx, problems=bad)


def check_caa(domain):
    # CAA is inherited from the closest parent that has one (RFC 8659).
    labels = domain.split('.')
    for i in range(len(labels) - 1):
        name = '.'.join(labels[i:])
        recs = dnsutil.resolve(name, 'CAA')
        if recs:
            issuers = sorted({r.split(None, 2)[2].strip('"') for r in recs
                              if r.split()[1] in ('issue', 'issuewild')})
            iodef = any(r.split()[1] == 'iodef' for r in recs)
            return _check('caa', 'pass', at=name, issuers=issuers, iodef=iodef)
    return _check('caa', 'warn', present=False)


def check_mta_sts(domain):
    recs = [t for t in dnsutil.txt_values('_mta-sts.' + domain) if t.lower().startswith('v=stsv1')]
    return _check('mta_sts', 'pass' if recs else 'info', present=bool(recs))


def check_tls_rpt(domain):
    recs = [t for t in dnsutil.txt_values('_smtp._tls.' + domain) if t.lower().startswith('v=tlsrptv1')]
    return _check('tls_rpt', 'pass' if recs else 'info', present=bool(recs))


def check_bimi(domain):
    recs = [t for t in dnsutil.txt_values('default._bimi.' + domain) if t.lower().startswith('v=bimi1')]
    return _check('bimi', 'pass' if recs else 'info', present=bool(recs))


def check_dangling(domain):
    def probe(sub):
        name = f'{sub}.{domain}'
        chain = dnsutil.cname_chain(name)
        if not chain:
            return None
        target = chain[-1]
        rcode, vals, _ = dnsutil.resolve_full(target, 'A')
        if not vals:
            rcode6, vals6, _ = dnsutil.resolve_full(target, 'AAAA')
            vals = vals6
        risky = any(target.endswith(s) or s in target for s in TAKEOVER_SUFFIXES)
        if vals:
            return {'name': name, 'target': target, 'state': 'ok', 'service': risky}
        return {'name': name, 'target': target, 'state': 'dangling', 'rcode': rcode,
                'service': risky}

    rows = [r for r in _pool.map(probe, SUBDOMAINS) if r]
    dangling = [r for r in rows if r['state'] == 'dangling']
    status = 'fail' if any(r['service'] for r in dangling) else 'warn' if dangling else 'pass'
    return _check('dangling', status, checked=len(SUBDOMAINS), cnames=rows,
                  dangling=dangling)


def check_wildcard(domain):
    label = ''.join(random.choices(string.ascii_lowercase + string.digits, k=20))
    vals = dnsutil.addresses(f'{label}.{domain}', v6=False)
    return _check('wildcard', 'info', wildcard=bool(vals), answers=vals[:4])


def scan(domain):
    rcode, soa, _ = dnsutil.resolve_full(domain, 'SOA')
    if rcode == 'NXDOMAIN':
        raise dnsutil.BadInput('domain does not exist')
    ns = _ns_info(domain)
    jobs = [
        _pool.submit(check_dnssec, domain), _pool.submit(check_ns, domain, ns),
        _pool.submit(check_soa, domain, ns), _pool.submit(check_axfr, domain, ns),
        _pool.submit(check_spf, domain), _pool.submit(check_dmarc, domain),
        _pool.submit(check_dkim, domain), _pool.submit(check_mx, domain),
        _pool.submit(check_caa, domain), _pool.submit(check_mta_sts, domain),
        _pool.submit(check_tls_rpt, domain), _pool.submit(check_bimi, domain),
        _pool.submit(check_dangling, domain), _pool.submit(check_wildcard, domain),
    ]
    checks = [j.result() for j in jobs]
    # A domain with no mail (no MX, or null MX) is not penalised on mail checks
    # beyond SPF/DMARC, which still matter: they stop others spoofing it.
    mx = next(c for c in checks if c['id'] == 'mx')
    no_mail = mx['data'].get('count', 0) == 0
    total = got = 0.0
    for c in checks:
        w = WEIGHTS.get(c['id'])
        # 'info' means "could not tell / optional": it neither helps nor hurts.
        if not w or c['status'] == 'info' or (no_mail and c['id'] in ('dkim', 'mta_sts', 'tls_rpt')):
            continue
        total += w
        got += w * POINTS[c['status']]
    score = round(100 * got / total) if total else 0
    grade = 'A' if score >= 90 else 'B' if score >= 75 else 'C' if score >= 60 else 'D' if score >= 40 else 'F'
    return {'domain': domain, 'score': score, 'grade': grade, 'no_mail': no_mail,
            'checks': checks}