Commit a50746ca authored by nimrod's avatar nimrod
Browse files

Major refactor. 

- Check base class, all checks derive from it and have a common calling
convention.
- Add retries.
- Add a DNS zone transfer check. Turns out if fails on ns1. WIP.
- A wrapper around checking web sites. Just provide a list of URLs.
parent 8ad9776a

src/_dns.py

+91 −73
Original line number Diff line number Diff line 
from functools import lru_cache

import dns.resolver  # pylint: disable=import-error

import utils

import dns.query  # pylint: disable=import-error

import dns.zone  # pylint: disable=import-error

from utils import Check





DOMAIN = "shore.co.il"

SUBDOMAINS = [DOMAIN, f"www.{DOMAIN}"]

class CheckDNS(Check):  # pylint: disable=abstract-method

    _domain = "shore.co.il"

    _subdomains = [_domain, f"www.{_domain}"]





def get_resolvers():

    @lru_cache(3)

    def _get_resolvers(self):

        """Return a list of resolvers for each nameserver in the DNS zone."""

        default_resolver = dns.resolver.Resolver()

        resolvers = []

        for ns in default_resolver.query(  # pylint: disable=invalid-name

        DOMAIN, "NS"

            self._domain, "NS"

        ):

            for address in default_resolver.query(ns.to_text()):

                resolver = dns.resolver.Resolver(configure=False)
@@ -19,74 +23,88 @@ def get_resolvers(): 
                resolvers.append(resolver)

        return resolvers





RESOLVERS = get_resolvers()





def cross_query(qname, rdtype="A"):

    def _cross_query(self, qname, rdtype="A"):

        """Return all of the answers from all nameservers."""

        answers = set()

    for resolver in RESOLVERS:

        for resolver in self._resolvers:

            for address in resolver.query(qname, rdtype):

                answers.add(address)

        return answers



    def __init__(self):

        self._resolvers = self._get_resolvers()





def validate_soa():

class CheckSOA(CheckDNS):

    def _check(self):

        """Validate the SOA record."""

    soas = cross_query(DOMAIN, "SOA")

        soas = self._cross_query(self._domain, "SOA")



        if len(soas) > 1:

        return [False, "SOA records don't match."]

            return False, "SOA records don't match."



        try:

            record = soas.pop()

            record.mname.to_text()

        except Exception as e:  # pylint: disable=broad-except,invalid-name

        print(str(e))

        return [False, "SOA record is invalid."]

            return False, str(e)



    return [True, "SOA record validated."]

        return True, "SOA record validated."





def validate_mx():

class CheckMX(CheckDNS):

    def _check(self):

        """Validate the MX record."""

    mxs = cross_query(DOMAIN, "MX")

        mxs = self._cross_query(self._domain, "MX")

        if len(mxs) > 1:

        return [False, "MX records don't match."]

            return False, "MX records don't match."



        try:

            record = mxs.pop()

        ips = cross_query(record.exchange.to_text())

            ips = self._cross_query(record.exchange.to_text())

            if len(ips) > 1:

            return [False, "MX records don't match."]

                return False, "MX records don't match."

        except Exception as e:  # pylint: disable=broad-except,invalid-name

        print(str(e))

        return [False, "MX record is invalid."]

            return [False, str(e)]



    return [True, "MX record validated."]

        return True, "MX record validated."





def validate_subdomains():

class CheckSubDomains(CheckDNS):

    def _check(self):

        """Validate important subdomains."""

    for domain in SUBDOMAINS:

        for domain in self._subdomains:

            try:

            ips = cross_query(domain)

                ips = self._cross_query(domain)

                if len(ips) > 1:

                    return [True, f"Domain {domain} records don't match."]

            except Exception as e:  # pylint: disable=broad-except,invalid-name

            print(str(e))

            return [False, "Failed to validate domain {d}."]

    return [True, "Subdomains validated."]

                return [False, str(e)]

        return True, "Subdomains validated."





class CheckTransfer(CheckDNS):

    def _check(self):

        """Validate zone transfer, to check the TCP transport."""

        zones = []

        for resolver in self._resolvers:

            try:

                zone = dns.zone.from_xfr(

                    dns.query.xfr(resolver.nameservers[0], self._domain)

                )

                zones.append(zone)

            except Exception as e:  # pylint: disable=broad-except,invalid-name

                return False, str(e)



        return True, "Zone transfer validated."





def handler(event, context):  # pylint: disable=unused-argument

    """Lambda event handler."""

    for check in [validate_soa, validate_mx, validate_subdomains]:

        success, message = check.__call__()

        print(message)

        if not success:

            utils.publish(message)

    CheckSOA().run()

    CheckMX().run()

    CheckSubDomains().run()

    # CheckTransfer().run()  # AXFR is blocked on ns1, need to figure out why.





if __name__ == "__main__":

src/autoconfig.py

+5 −7
Original line number Diff line number Diff line 
from utils import check_urls

from utils import website_handler





def handler(event, context):  # pylint: disable=unused-argument

    """Lambda event handler."""

    checks = [

URLS = [

    {"url": "https://autoconfig.shore.co.il/mail/config-v1.1.xml"},

]

    check_urls(checks)



handler = website_handler(URLS)





if __name__ == "__main__":

src/gitlab.py

+7 −9
Original line number Diff line number Diff line 
from utils import check_urls

from utils import website_handler





def handler(event, context):  # pylint: disable=unused-argument

    """Lambda event handler."""

    checks = [

URLS = [

    {"url": "http://git.shore.co.il/", "codes": [301, 302]},

    {"url": "https://git.shore.co.il/", "codes": [301, 302]},

    {"url": "https://git.shore.co.il/explore/"},

]

    check_urls(checks)



handler = website_handler(URLS)





if __name__ == "__main__":

src/imap.py

+16 −18
Original line number Diff line number Diff line 
from imaplib import IMAP4_SSL

from utils import publish

from utils import Check





def check_imap():

class CheckIMAP(Check):

    def _check(self):

        """Check the IMAP port."""

        try:

            imap = IMAP4_SSL("imap.shore.co.il", 993)
@@ -19,10 +20,7 @@ def check_imap(): 


def handler(event, context):  # pylint: disable=unused-argument

    """Lambda event handler."""

    success, message = check_imap()

    print(message)

    if not success:

        publish(message)

    CheckIMAP().run()





if __name__ == "__main__":

src/kodi.py

+6 −8
Original line number Diff line number Diff line 
from utils import check_urls

from utils import website_handler





def handler(event, context):  # pylint: disable=unused-argument

    """Lambda event handler."""

    checks = [

URLS = [

    {"url": "http://kodi.shore.co.il/", "codes": [301, 302]},

    {"url": "https://kodi.shore.co.il/", "codes": [401]},

]

    check_urls(checks)



handler = website_handler(URLS)





if __name__ == "__main__":