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