Loading src/_dns.py +19 −18 Original line number Diff line number Diff line # pylint: disable=import-error,invalid-name import dns.resolver import dns.resolver # pylint: disable=import-error import utils Loading @@ -11,7 +10,9 @@ def get_resolvers(): """Return a list of resolvers for each nameserver in the DNS zone.""" default_resolver = dns.resolver.Resolver() resolvers = [] for ns in default_resolver.query(DOMAIN, "NS"): for ns in default_resolver.query( # pylint: disable=invalid-name DOMAIN, "NS" ): for address in default_resolver.query(ns.to_text()): resolver = dns.resolver.Resolver(configure=False) resolver.nameservers = [address.to_text()] Loading @@ -25,9 +26,9 @@ RESOLVERS = get_resolvers() def cross_query(qname, rdtype="A"): """Return all of the answers from all nameservers.""" answers = set() for r in RESOLVERS: for a in r.query(qname, rdtype): answers.add(a) for resolver in RESOLVERS: for address in resolver.query(qname, rdtype): answers.add(address) return answers Loading @@ -39,9 +40,9 @@ def validate_soa(): return [False, "SOA records don't match."] try: r = soas.pop() r.mname.to_text() except Exception as e: # pylint: disable=broad-except 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."] Loading @@ -55,11 +56,11 @@ def validate_mx(): return [False, "MX records don't match."] try: r = mxs.pop() ips = cross_query(r.exchange.to_text()) record = mxs.pop() ips = cross_query(record.exchange.to_text()) if len(ips) > 1: return [False, "MX records don't match."] except Exception as e: # pylint: disable=broad-except except Exception as e: # pylint: disable=broad-except,invalid-name print(str(e)) return [False, "MX record is invalid."] Loading @@ -68,12 +69,12 @@ def validate_mx(): def validate_subdomains(): """Validate important subdomains.""" for d in SUBDOMAINS: for domain in SUBDOMAINS: try: ips = cross_query(d) ips = cross_query(domain) if len(ips) > 1: return [True, f"Domain {d} records don't match."] except Exception as e: # pylint: disable=broad-except 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."] Loading @@ -81,8 +82,8 @@ def validate_subdomains(): def handler(event, context): # pylint: disable=unused-argument """Lambda event handler.""" for c in [validate_soa, validate_mx, validate_subdomains]: success, message = c.__call__() for check in [validate_soa, validate_mx, validate_subdomains]: success, message = check.__call__() print(message) if not success: utils.publish(message) Loading src/mta_sts.py +0 −1 Original line number Diff line number Diff line # pylint: disable=invalid-name from utils import check_urls Loading src/utils.py +2 −3 Original line number Diff line number Diff line # pylint: disable=import-error import os import boto3 import requests import boto3 # pylint: disable=import-error import requests # pylint: disable=import-error TOPIC_ARN = os.getenv("TOPIC_ARN") Loading Loading
src/_dns.py +19 −18 Original line number Diff line number Diff line # pylint: disable=import-error,invalid-name import dns.resolver import dns.resolver # pylint: disable=import-error import utils Loading @@ -11,7 +10,9 @@ def get_resolvers(): """Return a list of resolvers for each nameserver in the DNS zone.""" default_resolver = dns.resolver.Resolver() resolvers = [] for ns in default_resolver.query(DOMAIN, "NS"): for ns in default_resolver.query( # pylint: disable=invalid-name DOMAIN, "NS" ): for address in default_resolver.query(ns.to_text()): resolver = dns.resolver.Resolver(configure=False) resolver.nameservers = [address.to_text()] Loading @@ -25,9 +26,9 @@ RESOLVERS = get_resolvers() def cross_query(qname, rdtype="A"): """Return all of the answers from all nameservers.""" answers = set() for r in RESOLVERS: for a in r.query(qname, rdtype): answers.add(a) for resolver in RESOLVERS: for address in resolver.query(qname, rdtype): answers.add(address) return answers Loading @@ -39,9 +40,9 @@ def validate_soa(): return [False, "SOA records don't match."] try: r = soas.pop() r.mname.to_text() except Exception as e: # pylint: disable=broad-except 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."] Loading @@ -55,11 +56,11 @@ def validate_mx(): return [False, "MX records don't match."] try: r = mxs.pop() ips = cross_query(r.exchange.to_text()) record = mxs.pop() ips = cross_query(record.exchange.to_text()) if len(ips) > 1: return [False, "MX records don't match."] except Exception as e: # pylint: disable=broad-except except Exception as e: # pylint: disable=broad-except,invalid-name print(str(e)) return [False, "MX record is invalid."] Loading @@ -68,12 +69,12 @@ def validate_mx(): def validate_subdomains(): """Validate important subdomains.""" for d in SUBDOMAINS: for domain in SUBDOMAINS: try: ips = cross_query(d) ips = cross_query(domain) if len(ips) > 1: return [True, f"Domain {d} records don't match."] except Exception as e: # pylint: disable=broad-except 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."] Loading @@ -81,8 +82,8 @@ def validate_subdomains(): def handler(event, context): # pylint: disable=unused-argument """Lambda event handler.""" for c in [validate_soa, validate_mx, validate_subdomains]: success, message = c.__call__() for check in [validate_soa, validate_mx, validate_subdomains]: success, message = check.__call__() print(message) if not success: utils.publish(message) Loading
src/mta_sts.py +0 −1 Original line number Diff line number Diff line # pylint: disable=invalid-name from utils import check_urls Loading
src/utils.py +2 −3 Original line number Diff line number Diff line # pylint: disable=import-error import os import boto3 import requests import boto3 # pylint: disable=import-error import requests # pylint: disable=import-error TOPIC_ARN = os.getenv("TOPIC_ARN") Loading