A script to check the SOA serial numbers of all nameservers of a domain and notify via Zephyr

dnscheck.py

#!/usr/bin/env python3

from __future__ import print_function
import sys
import time
import dns.resolver
import zwrite

DEFAULT_ARGS = {"classname": "message",
                "instance": "dns",
                "sender": "bot",
                "zsig": "dnscheck.py",
                "opcode": "auto"}

TEMPLATE = "{:15} SOA Serial: @b({})"

def find_nameservers(domain):
    nameservers = []
    ns = dns.resolver.query(domain, "NS")
    for rrset in ns.response.additional:
        for a in rrset:
            nameservers.append(a.address)
    return nameservers

def spawn_resolvers(nameservers):
    resolvers = []
    for ns in nameservers:
        resolvers.append(dns.resolver.Resolver())
        resolvers[-1].nameservers = [ns]
    return resolvers

def serials_gt(a, b):
    # TODO: actually use the algorithm specified in RFC 1982
    return int(a) > int(b)

def serials_eq(a, b):
    return int(a) == int(b)

def check_soa_serial(domain, resolvers, serials):
    answers = []
    new_serials = {}
    updated = False
    for r in resolvers:
        soa = r.query(domain, "SOA")
        n = r.nameservers[0]
        s = soa[0].serial
        if serials_gt(s, serials[n]):
            answers.append(TEMPLATE.format(n, s) + " @b(@color(green)UPDATE)")
            updated = True
        elif serials_eq(s, serials[n]):
            if updated:
                answers.append(TEMPLATE.format(n, s) + " @b(@color(yellow)UNCHANGED)")
        else:
            answers.append(TEMPLATE.format(n, s) + " @b(@color(red)REGRESSION)")
            updated = True
        new_serials[n] = s
    serials.update(new_serials)
    return "\n".join(answers)

def main():
    domain = sys.argv[1]
    print(domain)
    resolvers = spawn_resolvers(find_nameservers(domain))
    serials = {}
    for r in resolvers:
        serials[r.nameservers[0]] = "0"
    while True:
        message = check_soa_serial(domain, resolvers, serials)
        if message:
            zwrite.zwrite("\n".join((domain, message)), **DEFAULT_ARGS)
        time.sleep(300)

if __name__ == "__main__":
    main()