check_dane_smtp

   1 #!/usr/bin/python3
   2 #
   3 #
   4
   5 from __future__ import print_function
   6
   7 import sys
   8 import argparse
   9 import logging
  10
  11 from socket import socket, AF_INET6, AF_INET, create_connection
  12 from ssl import SSLContext, PROTOCOL_TLSv1_2, CERT_REQUIRED, cert_time_to_seconds, SSLError, CertificateError, create_default_context
  13 from unbound import ub_ctx, idn2dname, ub_strerror
  14
  15 from check_dane.tlsa import verify_tlsa_record
  16
  17 def init_connection(sslcontext, args):
  18     host = args.Host
  19
  20     if args.ssl:
  21         port = 465 if args.port == 0 else args.port
  22         connection = sslcontext.wrap_socket(socket(AF_INET),
  23                                             server_hostname=host)
  24         connection.connect((host, port))
  25
  26     else:
  27         port = 25 if args.port == 0 else args.port
  28
  29         connection = create_connection((host, port))
  30         answer = connection.recv(512)
  31         logging.debug(answer)
  32
  33         connection.send(b"EHLO localhost\r\n")
  34         answer = connection.recv(512)
  35         logging.debug(answer)
  36
  37         connection.send(b"STARTTLS\r\n")
  38         answer = connection.recv(512)
  39         logging.debug(answer)
  40
  41         connection = sslcontext.wrap_socket(connection, server_hostname=host)
  42         connection.do_handshake()
  43
  44     return connection
  45
  46
  47 def close_connection(connection):
  48     connection.send(b"QUIT\r\n")
  49     answer = connection.recv(512)
  50     logging.debug(answer)
  51
  52
  53 def init(args):
  54     sslcontext = SSLContext(PROTOCOL_TLSv1_2)
  55     sslcontext.verify_mode = CERT_REQUIRED
  56     sslcontext.load_verify_locations(args.castore)
  57
  58     resolver = ub_ctx()
  59     resolver.add_ta_file(args.ancor)
  60
  61     return sslcontext, resolver
  62
  63
  64 def main():
  65     logging.basicConfig(format='%(levelname)5s %(message)s')
  66     parser = argparse.ArgumentParser()
  67     parser.add_argument("Host")
  68
  69     parser.add_argument("--verbose", action="store_true")
  70     parser.add_argument("--quiet", action="store_true")
  71     parser.add_argument("-p", "--port",
  72                         action="store", type=int, default=0,
  73                         help="SMTP port")
  74     parser.add_argument("--ssl",
  75                         action="store_true",
  76                         help="Use direct TLS connection instead of starttls (default: disabled)")
  77     parser.add_argument("--check-dane",
  78                         action="store_false",
  79                         help="Verify presented certificate via DANE (default: enabled)")
  80     parser.add_argument("--check-ca",
  81                         action="store_false",
  82                         help="Verify presented certificate via the CA system (default: enabled)")
  83     parser.add_argument("--check-expire",
  84                         action="store_false",
  85                         help="Verify presented certificate for expiration (default: enabled)")
  86
  87     parser.add_argument("-a", "--ancor",
  88                         action="store", type=str, default="/etc/unbound/root.key",
  89                         help="DNSSEC root ancor")
  90     parser.add_argument("--castore", action="store", type=str,
  91                         default="/etc/ssl/certs/ca-certificates.crt",
  92                         help="ca certificate bundle")
  93
  94     group = parser.add_mutually_exclusive_group()
  95     group.add_argument("-6", "--6", action="store_true", help="check via IPv6 only")
  96     group.add_argument("-4", "--4", action="store_true", help="check via IPv4 only")
  97     group.add_argument("--64", action="store_false", help="check via IPv4 and IPv6 (default)")
  98
  99     args = parser.parse_args()
 100
 101     if args.verbose:
 102         logging.getLogger().setLevel(logging.DEBUG)
 103     elif args.quiet:
 104         logging.getLogger().setLevel(logging.WARNING)
 105     else:
 106         logging.getLogger().setLevel(logging.INFO)
 107
 108     port = args.port
 109     if port == 0:
 110         port = 465 if args.ssl else 25
 111     host = args.Host.encode('idna').decode()
 112
 113     sslcontext, resolver = init(args)
 114     try:
 115         connection = init_connection(sslcontext, args)
 116     except ConnectionRefusedError:
 117         logging.error("Connection refused")
 118         return 2
 119
 120     retval = verify_tlsa_record(resolver, "_%d._tcp.%s" % (port, host), connection.getpeercert(binary_form=True))
 121
 122     close_connection(connection)
 123     return retval
 124
 125
 126 if __name__ == '__main__':
 127     sys.exit(main())