Checks NSEC3PARAM for a given zone

check-nsec3param

#! /usr/bin/env python3
#
# Copyright (c) 2019 Karol Babioch <karol@babioch.de>
#
# 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 <http://www.gnu.org/licenses/>.

import argparse
import datetime
import logging
import sqlite3
import subprocess
import sys

# Valid exit codes (Nagios / Icinga)
EXIT_OK = 0
EXIT_WARNING = 1
EXIT_CRITICAL = 2
EXIT_UNKNOWN = 3

# TODO Make this configurable
DB = './nsec3params.db'

# Setup argument parser
parser = argparse.ArgumentParser(description='Checks the NSEC3PARAM record for a zone')
parser.add_argument('zone', metavar='ZONE', help='Name of zone that NSEC3PARAM record should be checked', type=str)
parser.add_argument('-w', '--warning', dest='warning', help='Exit with warning if record has not changed for this amount of days (default: 0)', type=int)
parser.add_argument('-c', '--critical', dest='critical', help='Exit with critical if record has not changed for this amount of days (default: 0)', type=int)
parser.add_argument('-ns', '--nameserver', dest='nameserver', help='Specify a resolver explicitly (default: resolver used by system)', type=str)
parser.add_argument('-d', '--debug', dest='debug', help='Enable debugging output', action='store_true')

# Parse arguments
args = parser.parse_args()

# Setup logging
logger = logging.getLogger()
handler = logging.StreamHandler()
logger.addHandler(handler)
if args.debug:
    logger.setLevel(logging.DEBUG)

# Construct command to execute
cmd = ['dig']
if args.nameserver:
    cmd.extend(['@{}'.format(args.nameserver)])
cmd.extend(['+short', args.zone, 'NSEC3PARAM']) 

# Run command
cmd = subprocess.run(cmd, capture_output=True, encoding=sys.getdefaultencoding())

# Some debug output
logger.debug('args: {}'.format(args))
logger.debug('cmd: {}'.format(cmd))

# Check if command was executed successfully
if cmd.returncode != 0:
    print('Query for NSEC3PARAMS unsuccessful')
    sys.exit(EXIT_UNKNOWN)

record = cmd.stdout.rstrip()
logger.debug('NSEC3PARAM: \'{}\''.format(record))

if not record:
    print('No NSEC3PARAM record found for {}'.format(args.zone))
    sys.exit(EXIT_CRITICAL)

try:

    # Connect to database (using context manager)
    with sqlite3.connect(DB, detect_types=sqlite3.PARSE_DECLTYPES) as con:
        #con = sqlite3.connect(DB, detect_types=sqlite3.PARSE_DECLTYPES)
        cur = con.cursor()

        # Initialize database (not totally normalized to keep it simple)
        sql = '''
            CREATE TABLE IF NOT EXISTS nsec3params (
                id INTEGER PRIMARY KEY,
                name VARCHAR(255) NOT NULL,
                record VARCHAR(255),
                timestamp TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
            );
        '''
        cur.execute(sql)

        # Get last NSEC3PARAM record for this domain from database
        cur.execute('SELECT record, timestamp FROM nsec3params WHERE name = ? ORDER BY timestamp DESC LIMIT 1;', (args.zone,))
        result = cur.fetchone()

        # Exit code in case nothing goes wrong
        exitcode = EXIT_OK

        # There is (at least) one NSEC3PARAM for this zone in the database
        if result:

            (lastrecord, lasttimestamp) = result
            logger.debug('Last NSEC3PARAM entry for {} in database: \'{}\''.format(args.zone, lastrecord))

            # Check if record has changed since last invocation
            if lastrecord != record:
                logger.debug('Last NSEC3PARAM differs from current one, inserting current one into database')
                cur.execute('INSERT INTO nsec3params (name, record) VALUES (?, ?)', (args.zone, record))

            # Record remained the same
            else:
                logger.debug('Last NSEC3PARAM has not changed since: {}'.format(lasttimestamp))

                # Check whether warning threshold has been reached
                if args.warning and (lasttimestamp + datetime.timedelta(args.warning) < datetime.datetime.now()):
                    logger.debug('Warning threshold of {} days reached'.format(args.warning))
                    exitcode = EXIT_WARNING

                # Check whether critical threshold has been reached
                if args.critical and (lasttimestamp + datetime.timedelta(args.critical) < datetime.datetime.now()):
                    logger.debug('Critical threshold of {} days reached'.format(args.critical))
                    exitcode = EXIT_CRITICAL

        # No NSEC3PARAM record for this zone yet
        else:

            logger.debug('Found no NSEC3PARAM entry for {} in database, inserting current one'.format(args.zone))
            cur.execute('INSERT INTO nsec3params (name, record) VALUES (?, ?)', (args.zone, record))

# Catch any error (most likely database related)
except Exception as e:
    logger.debug('Exception: {}'.format(e))
    exitcode = EXIT_UNKNOWN

# Quit with the correct exit code
logger.debug('Exit code: {}'.format(exitcode))
sys.exit(exitcode)