unbound-dnsbl-updater.py.j2

#! /usr/bin/env python3
# vim: tabstop=4 expandtab shiftwidth=4 softtabstop=4 smartindent nu ft=python
# pylint: disable=invalid-name,protected-access,too-many-locals,too-many-branches,too-many-statements



# Ansible managed

# this file is part of
# https://github.com/Rosa-Luxemburgstiftung-Berlin/ansible-role-unbound
# (c) 2022 Klaus Zerwes for Rosa-Luxemburgstiftung-Berlin
#
#    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 3 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, see <https://www.gnu.org/licenses/>.

__doc__ = """
simple script to fetch a defined list of dnsbl uri's
and convert them to unbound compatible format
"""

import sys
import os
import shutil
import subprocess
import argparse
import logging
import logging.handlers
import datetime
import tempfile
import http.client as http_client
from lockfile import locked
import requests

# list of dnsbl uri's to fetch
DNSBL_LISTS = [
    {% for dnsbluri in dnsbl_lists %}
        '{{ dnsbluri }}',
    {% endfor %}
    ]

# list of domains to exclude
DNSBL_EXCLUDE = (
    {% for domain in dnsbl_exclude_domains %}
    '{{ domain }}',
    {% endfor %}
    )

class LoggingAction(argparse.Action):
    """simple helper class for argparse and logging"""
    def __call__(self, parser, namespace, values, option_string=None):
        logger = logging.getLogger()
        logger.setLevel(values)
        setattr(namespace, self.dest, values)

def validatednsbl(filename):
    """call unbound-checkconf on a file and return result"""
    try:
        logging.info('unbound-checkconf %s ...', filename)
        subprocess.check_output(
                ['/usr/sbin/unbound-checkconf', filename],
                stderr=subprocess.STDOUT
            )
        logging.info('unbound-checkconf %s: OK', filename)
        return (0, 'OK')
    except subprocess.CalledProcessError as e:
        logging.warning(
                'unbound-checkconf %s : ecode %s "%s"',
                filename, e.returncode, e.output
            )
        return (e.returncode, e.output)

def writednsbl2file(filep, domains):
    """write list to file in unbound format"""
    # dnsbl_match_subdomains: {{ dnsbl_match_subdomains|bool }}
    # dnsbl_use_always_null: {{ dnsbl_use_always_null|bool }}
{% if dnsbl_match_subdomains|bool %}
{% if dnsbl_use_always_null|bool %}
    filep.write(
        '\n'.join(
            [d if d.startswith('#') else 'local-zone: "%s" always_null' % d for d in domains]
            )
        )
{% else %}
    filep.write(
        '\n'.join(
            [d if d.startswith('#') else 'local-zone: "%s" redirect\nlocal-data: "%s A 0.0.0.0"' % (d, d) for d in domains]
            )
        )
{% endif %}{# end dnsbl_use_always_null #}
{% else %}
    filep.write(
        '\n'.join(
            [d if d.startswith('#') else 'local-data: "%s A 0.0.0.0"' % d for d in domains]
            )
        )
{% endif %}
    filep.write('\n')

@locked('/var/lock/dnsbl', timeout=30)
def main(args):
    """main function"""
    logger = logging.getLogger()
    lformat='%(levelname)s\t[%(name)s] %(funcName)s: %(message)s'

    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        '-l', '--loglevel',
        help='set loglevel',
        type=str,
        choices=[k for k in list(logging._nameToLevel.keys()) if isinstance(k, str)],
        action=LoggingAction
    )
    parser.add_argument(
        '-s', '--syslog', help='log to syslog',
        dest='syslog', action='store_true', default=False
    )
    parser.add_argument(
        '-n', '--dry-run', help='dry run, no final action including the dnsbl into unbound',
        dest='dryrun', action='store_true', default=False
    )
    parser.add_argument(
        '-g', '--no-garbage-cleanup', help='do not delete temporary files',
        dest='garbage', action='store_true', default=False
    )
    parser.add_argument(
        '-t', '--tempdir', help='temp directory to store files (defaults: /tmp)',
        dest='tempdir', action='store', default='/tmp'
    )
    parser.add_argument(
        '-c', '--no-checkconf', help='do not validate files using unbound-checkconf',
        dest='nocheckconf', action='store_true', default=False
    )
    args = parser.parse_args()

    if args.syslog:
        syslogh = logging.handlers.SysLogHandler(address = '/dev/log')
        syslogh.setFormatter(
                logging.Formatter('%(module)s %(levelname)s: %(funcName)s: %(message)s')
            )
        logger.setLevel(logging.INFO)
        if logger.isEnabledFor(logging.DEBUG):
            syslogh.setLevel(logging.DEBUG)
        else:
            syslogh.setLevel(logging.INFO)
        logger.addHandler(syslogh)
        stdouth = logging.StreamHandler(sys.stdout)
        stdouth.setLevel(logging.WARN)
        stdouth.setFormatter(logging.Formatter(lformat))
        logger.addHandler(stdouth)

    logging.basicConfig()

    if logger.isEnabledFor(logging.DEBUG):
        http_client.HTTPConnection.debuglevel = 1
        requests_log = logging.getLogger("requests.packages.urllib3")
        requests_log.setLevel(logging.DEBUG)
        requests_log.propagate = True

    # start fetching
    logger.info(
            'start %s for %s dnsbl lists at %s',
            sys.argv[0], len(DNSBL_LISTS), datetime.datetime.now()
        )
    errorcount = 0
    domainlist = set()
    dnsbltmp = tempfile.NamedTemporaryFile(
            dir=args.tempdir,
            prefix='unbound-dnsbl',
            delete=False,
            mode='w+',
            newline='\n'
        )
    logger.info('using %s as main  temp file', dnsbltmp.name)
    dnsbltmp.write(
            '# dnsbl file created by %s at %s\n' % (
                    os.path.basename(__file__), datetime.datetime.now(),)
        )
    dnsbltmp.write('server:\n')
    for dnsbluri in DNSBL_LISTS:
        logger.info('start: %s', dnsbluri)
        try:
            r = requests.get(dnsbluri)
            logger.info('fetched %s w/ %s', dnsbluri, r.status_code)
            if r.status_code < 200 or r.status_code >= 300:
                logger.warning('failed to fetched %s : http code: %s', dnsbluri, r.status_code)
                errorcount += 1
            else:
                currentdomainlist = ['# start %s at %s\n' % (dnsbluri, datetime.datetime.now(),)]
                for line in r.text.splitlines():
                    line = line.strip()
                    if not line:
                        continue
                    if line.startswith('#'):
                        continue
                    lineparts = line.split()
                    # no need for big ipv6 matching regex here
                    if '::' in lineparts[0]:
                        continue
                    if lineparts[0] in ['127.0.0.1', '0.0.0.0']:
                        domain = lineparts[1]
                    else:
                        domain = lineparts[0]
                    if domain.endswith(DNSBL_EXCLUDE):
                        continue
                    # '*.' is considered a valid wildcard prefix,
                    #   but domains/zones starting with '.' as error
                    domain = domain.removeprefix('*.')
                    if domain.startswith('.'):
                        logger.info('ignoring invalid domain "%s"', domain)
                        continue
                    if domain not in domainlist:
                        currentdomainlist.append(domain)
                currentdomainlist.append('# end %s at %s\n' % (dnsbluri, datetime.datetime.now(),))

                if not args.nocheckconf:
                    tmpf = tempfile.NamedTemporaryFile(
                            dir=args.tempdir,
                            prefix='unbound-dnsbl',
                            delete=False,
                            mode='w+',
                            newline='\n'
                        )
                    tmpf.write('# %s\n' % dnsbluri)
                    tmpf.write('server:\n')
                    writednsbl2file(tmpf, currentdomainlist)
                    tmpf.close()
                    logger.info('%s : check file: %s', dnsbluri, tmpf.name)
                    (validatecode, validatemsg) = validatednsbl(tmpf.name)
                    if validatecode > 0:
                        logger.warning('validation of %s failed with %s', tmpf.name, validatecode)
                        logger.warning(validatemsg)
                        errorcount += 1
                        continue
                    if not args.garbage:
                        os.unlink(tmpf.name)

                writednsbl2file(dnsbltmp, currentdomainlist)
                domainlist = domainlist.union(currentdomainlist)
                logger.info(
                        'got %s domains from %s (current total domains: %s)',
                        len(currentdomainlist), dnsbluri, len(domainlist)
                    )
        except requests.Timeout as e:
            logger.fatal('requests.Timeout error: %s', e)
            errorcount += 1
        except requests.RequestException as e:
            logger.fatal('requests.RequestException: %s', e)
            errorcount += 1

    dnsbltmp.close()

    if not args.nocheckconf:
        (validatecode, validatemsg) = validatednsbl(dnsbltmp.name)
        if validatecode > 0:
            logger.fatal('validation of %s failed with %s', tmpf.name, validatecode)
            logger.warning(validatemsg)
            sys.exit(126)

    # copy temp file into unbound config dir and reload unbound
    if args.dryrun:
        logger.warning('skip replacing conf file and unbound restart due to dry-run mode!')
    else:
        shutil.move(dnsbltmp.name, '{{ unbound_cfg_dir_incl }}/dnsbl.conf')
        try:
            if not args.nocheckconf:
                subprocess.check_output(['/usr/sbin/unbound-checkconf'], stderr=subprocess.STDOUT)
            subprocess.check_output(
                    ['/usr/sbin/service', 'unbound', 'restart'],
                    stderr=subprocess.STDOUT
                )
        except subprocess.CalledProcessError as e:
            logger.fatal('command %s exited with %s', e.cmd, e.returncode)
            logger.warning(e.output)
            sys.exit(126)

    logger.info(
            'done with %s errors and %s domains at %s',
            errorcount, len(domainlist), datetime.datetime.now()
        )
    sys.exit(errorcount)

if __name__ == "__main__":
    main(sys.argv[1:])