check_caa_records.py

import yaml

import dns.resolver

VALID_NS = ('ns0.wikimedia.org.', 'ns1.wikimedia.org.', 'ns2.wikimedia.org.')
VALID_CAA_RECORDS = (
    '0 issue "globalsign.com',
    '0 issue "digicert.com',
    '0 issue "letsencrypt.org',
    '0 iodef "mailto:dns-admin@wikimedia.org',
)


def main():
    with open('/tmp/snis.yaml') as f:
        data = f.read()

    snis = yaml.safe_load(data)
    print('[*] checking {} SNIs'.format(len(snis)))

    unique_snis = set()

    for sni in snis:
        unique_snis.add(sni.strip('*.'))

    print('[*] checking {} unique SNIs'.format(len(unique_snis)))

    for sni in unique_snis:
        print('[*] checking {}'.format(sni))
        invalid_ns = False
        try:
            answers = dns.resolver.query(sni, 'NS')
            for rrset in answers.rrset:
                ns_server = rrset.to_text().strip('"')
                if ns_server not in VALID_NS:
                    invalid_ns = True
                    print('[!] unexpected ns server {} found in {}'.format(ns_server, sni))
        except:
            print('[!] unable to fetch NS servers for {}'.format(sni))

        if invalid_ns:
            continue

        try:
            answers = dns.resolver.query(sni, 'CAA')
            for rrset in answers.rrset:
                caa_record = rrset.to_text().strip('"')
                if caa_record not in VALID_CAA_RECORDS:
                    print('[!] invalid CAA record {} found in {}'.format(caa_record, sni))
        except:
            print('[!] unable to fetch CAA records for {}'.format(sni))


if __name__ == '__main__':
    main()