Resolve all known ip addresses from spf record and generate cidr map for postfix

get_subnets_of_spf_record_mynetworks.py

#!/usr/bin/env python3

#
# get_subnets_of_spf_record_mynetworks.py
# Resolve all known ip addresses from spf record and generate cidr map for postfix
# Version 2.0
# Updated by Danny Schmarsel (https://lichtmetzger.de)
# - just a small update for python3
#
# Version 1.0
# Written by Maximilian Thoma (http://www.lanbugs.de)
#
# Checkout blog article:
# https://lanbugs.de/howtos/linux/subnetze-und-ip-adressen-extrahieren-aus-spf-records-z-b-office365-oder-google-apps-for-business/
#
# The generated files can be used in postfix config with for example mynetworks = cidr:/etc/postfix/<generated_file>
#
# This program is free software; you can redistribute it and/or modify it under the terms of the
# GNU General Public License as published by the Free Software Foundation;
# either version 2 of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
# without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
# See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along with this program;
# if not, write to the Free Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston,
# MA 02110, USA
#

#
# Requirements:
# dnspython module  -> pip install dnspython | apt install python3-dnspython
#

import dns.resolver
from dns.exception import DNSException
import re
import sys

# Look for DNS Record at:
#
# "jobname": {
#            "domain": "domainname",
#            "file": "output_file",
#            }
#

#

lookup_spf = {
# Google Apps for Business
"google": {
          "domain": "google.com",
          "file"  : "/etc/postfix/networks/google",
          },

# Office365
"office365": {
          "domain": "spf.protection.outlook.com",
          "file"  : "/etc/postfix/networks/office365",
          },
}

############################################################################################

def getspf(record, filehandler):
    """
    Recursively fetch SPF records for a domain and write IPs to the given file handler.
    """
    # Init resolver
    resolver = dns.resolver.Resolver()

    try:
        # Try to lookup TXT record
        answers = resolver.resolve(record, "TXT")
    except DNSException:
        sys.stderr.write(f"Failed to query record '{record}', SPF broken.\n")
        return

    results = []

    # Get string out of records
    for rdata in answers:
        for txt_bytes in rdata.strings:
            txt = txt_bytes.decode('utf-8')
            # Append to SPF Records buffer if "spf" in string
            if "spf" in txt.lower():
                results.append(txt)

    if not results:
        sys.stderr.write(f"No SPF results found for '{record}'.\n")
        return

    # Work on records
    for spf in results:
        # Split parts
        parts = spf.split()
        # Check parts
        for part in parts:
            s_include = re.match(r"^include:(?P<domain>.+)$", part)
            s_ip4 = re.match(r"^ip4:(?P<ip4>.+)$", part)
            s_ip6 = re.match(r"^ip6:(?P<ip6>.+)$", part)

            # If in part "include" found, next round
            if s_include:
                getspf(s_include.group('domain'), filehandler)
            # elif ip4 found
            elif s_ip4:
                filehandler.write(s_ip4.group('ip4') + " OK\n")
            # elif ip6 found
            elif s_ip6:
                ip6 = s_ip6.group('ip6').replace("/", "]/")
                filehandler.write(f"[{ip6} OK\n")
            # else no valid record

def main():
    # Working on jobs
    for jobname, config in lookup_spf.items():

        print("Working on job %s" % jobname)

        # open file
        filehandler = open(config['file'], 'w')
        # start query spf records
        getspf(config['domain'], filehandler)
        # close file
        filehandler.close()


#getspf(lookup_spf)

if __name__ == "__main__":
    main()