A Nagios compatible script to check your Vertica SSL certificate expiration.

check-vertica-certificate.py

#!/usr/bin/env python

import argparse
import socket
import ssl
import struct
import subprocess
import sys

from datetime import *

try:
    from urlparse import urlparse
except ImportError:
    from urllib.parse import urlparse


# NAGIOS return codes :
# https://nagios-plugins.org/doc/guidelines.html#AEN78
OK       = 0
WARNING  = 1
CRITICAL = 2
UNKNOWN  = 3


def main():
    args = get_args()
    target = get_target_address_from_args(args)
    sock = socket.create_connection(target)
    try:
        certificate_as_pem = get_certificate_from_socket(sock)
        days_until_expiry = days_until_pem_expires(certificate_as_pem)

        if days_until_expiry > 60:
            print "OK - %s days until expiration." % days_until_expiry
            sys.exit(OK)
        elif (days_until_expiry <= 60 and days_until_expiry > 30):
            print "WARNING - %s days until expiration." % days_until_expiry
            sys.exit(WARNING)
        elif (days_until_expiry <= 30):
            print "CRITICAL - %s days until expiration." % days_until_expiry
            sys.exit(CRITICAL)
        else:
            print "UNKNOWN - Unknown days until expiration."
            sys.exit(UNKNOWN)

    except Exception as exc:
        sys.stderr.write('Something failed while fetching certificate: {0}\n'.format(exc))
        sys.exit(3)
    finally:
        sock.close()


def get_args():
    parser = argparse.ArgumentParser()
    parser.add_argument('database', help='Either an IP address, hostname or URL with host and port')
    return parser.parse_args()


def days_until_pem_expires(pem):
    expiry_date = get_expiry_date_from_pem(pem)
    todays_date = date.today()
    days_until_expiry = (expiry_date - todays_date).days
    return days_until_expiry


def get_expiry_date_from_pem(pem):
    # Forking out to openssl to not have to add any dependencies to script,
    # preferably you'd do this with pycrypto or other ssl libraries.
    cmd = ['openssl', 'x509', '-enddate', '-noout']
    pipe = subprocess.PIPE
    process = subprocess.Popen(cmd, stdin=pipe, stdout=pipe, stderr=pipe)
    stdout, stderr = process.communicate(pem)
    if stderr:
        raise Exception('OpenSSL error when fetching date from PEM: {0}'.format(stderr))
    _,expiration_str = stdout.strip().split("=")
    expiration_date = datetime.strptime(expiration_str, '%b %d %X %Y %Z').date()
    return expiration_date


def get_target_address_from_args(args):
    specified_target = args.database
    if '//' not in specified_target:
        specified_target = '//' + specified_target
    parsed = urlparse(specified_target)
    return (parsed.hostname, parsed.port or 5432)


def get_certificate_from_socket(sock):
    request_ssl(sock)
    ssl_context = get_ssl_context()
    sock = ssl_context.wrap_socket(sock)
    sock.do_handshake()
    certificate_as_der = sock.getpeercert(binary_form=True)
    certificate_as_pem = encode_der_as_pem(certificate_as_der)
    return certificate_as_pem


def request_ssl(sock):
    # 1234.5679 is the magic protocol version used to request TLS, defined
    # in pgcomm.h)
    version_ssl = postgres_protocol_version_to_binary(1234, 5679)
    length = struct.pack('!I', 8)
    packet = length + version_ssl

    sock.sendall(packet)
    data = read_n_bytes_from_socket(sock, 1)
    if data != b'S':
        raise Exception('Backend does not support TLS')


def get_ssl_context():
    # Return the strongest SSL context available locally
    for proto in ('PROTOCOL_TLSv1_2', 'PROTOCOL_TLSv1', 'PROTOCOL_SSLv23'):
        protocol = getattr(ssl, proto, None)
        if protocol:
            break
    return ssl.SSLContext(protocol)


def encode_der_as_pem(cert):
    # Forking out to openssl to not have to add any dependencies to script,
    # preferably you'd do this with pycrypto or other ssl libraries.
    cmd = ['openssl', 'x509', '-inform', 'DER']
    pipe = subprocess.PIPE
    process = subprocess.Popen(cmd, stdin=pipe, stdout=pipe, stderr=pipe)
    stdout, stderr = process.communicate(cert)
    if stderr:
        raise Exception('OpenSSL error when converting cert to PEM: {0}'.format(stderr))
    return stdout.strip()


def read_n_bytes_from_socket(sock, n):
    buf = bytearray(n)
    view = memoryview(buf)
    while n:
        nbytes = sock.recv_into(view, n)
        view = view[nbytes:] # slicing views is cheap
        n -= nbytes
    return buf


def postgres_protocol_version_to_binary(major, minor):
    return struct.pack('!I', major << 16 | minor)


if __name__ == '__main__':
    main()