check_dane_ssh

   1 #!/usr/bin/python3
   2 #
   3 #
   4
   5 from __future__ import print_function
   6
   7 import sys
   8 import argparse
   9 import logging
  10 import hashlib
  11 import codecs
  12
  13 from unbound import ub_ctx, ub_strerror
  14 import paramiko
  15
  16 try:
  17     from unbound import RR_TYPE_SSHFP
  18 except ImportError:
  19     RR_TYPE_SSHFP = 44
  20
  21
  22 class HostKeyMatchSSHFP(BaseException):
  23     pass
  24
  25
  26 class HostKeyMismatchSSHFP(BaseException):
  27     pass
  28
  29
  30 class HostKeyLookup(paramiko.client.MissingHostKeyPolicy):
  31     def __init__(self, args):
  32         self._args = args
  33         self._resolver = ub_ctx()
  34         self._resolver.add_ta_file(args.ancor)
  35
  36
  37     def missing_host_key(self, client, hostname, key):
  38         actualhostkey = key.asbytes()
  39         actualkeytype = key.get_name()
  40         hexencoder = codecs.getencoder('hex')
  41
  42         s, r = self._resolver.resolve(hostname, RR_TYPE_SSHFP)
  43         if 0 != s:
  44             ub_strerror(s)
  45             return
  46
  47         if r.data is None:
  48             logging.error("No SSHFP record returned")
  49             return 2
  50
  51         for record in r.data.data:
  52             keytype = record[0]
  53             hashtype = record[1]
  54             data = record[2:]
  55
  56             if hashtype == 1:
  57                 actualhash = hashlib.sha1(actualhostkey).digest()
  58             elif hashtype == 2:
  59                 actualhash = hashlib.sha256(actualhostkey).digest()
  60             else:
  61                 logging.warning("Only hashtypes 1 and 2 supported")
  62
  63             if keytype == 1 and actualkeytype == 'ssh-rsa':
  64                 if data == actualhash:
  65                     logging.info("Found matching record: `SSHFP %d %d %s`",
  66                                  keytype, hashtype, hexencoder(data)[0].decode())
  67                     raise HostKeyMatchSSHFP
  68
  69             elif keytype == 2 and actualkeytype == 'ssh-dss':
  70                 if data == actualhash:
  71                     logging.info("Found matching record: `SSHFP %d %d %s`",
  72                                  keytype, hashtype, hexencoder(data)[0].decode())
  73                     raise HostKeyMatchSSHFP
  74
  75             elif keytype == 3 and actualkeytype == 'ssh-ecdsa':
  76                 if data == actualhash:
  77                     logging.info("Found matching record: `SSHFP %d %d %s`",
  78                                  keytype, hashtype, hexencoder(data)[0].decode())
  79                     raise HostKeyMatchSSHFP
  80
  81             elif keytype == 4 and actualkeytype == 'ssh-ed25519':
  82                 if data == actualhash:
  83                     logging.info("Found matching record: `SSHFP %d %d %s`",
  84                                  keytype, hashtype, hexencoder(data)[0].decode())
  85                     raise HostKeyMatchSSHFP
  86
  87         logging.error("No matching SSHFP record found")
  88         raise HostKeyMismatchSSHFP
  89
  90
  91 def init_connection(args):
  92     connection = paramiko.client.SSHClient()
  93     connection.set_missing_host_key_policy(HostKeyLookup(args))
  94
  95     return connection
  96
  97
  98 def main():
  99     logging.basicConfig(format='%(levelname)5s %(message)s')
 100     parser = argparse.ArgumentParser()
 101     parser.add_argument("Host")
 102
 103     parser.add_argument("--verbose", action="store_true")
 104     parser.add_argument("--quiet", action="store_true")
 105     parser.add_argument("-p", "--port",
 106                         action="store", type=int, default=22,
 107                         help="SMTP port")
 108
 109     parser.add_argument("-a", "--ancor",
 110                         action="store", type=str, default="/etc/unbound/root.key",
 111                         help="DNSSEC root ancor")
 112
 113     group = parser.add_mutually_exclusive_group()
 114     group.add_argument("-6", "--6", action="store_true", help="check via IPv6 only")
 115     group.add_argument("-4", "--4", action="store_true", help="check via IPv4 only")
 116     group.add_argument("--64", action="store_false", help="check via IPv4 and IPv6 (default)")
 117
 118     args = parser.parse_args()
 119
 120     if args.verbose:
 121         logging.getLogger().setLevel(logging.DEBUG)
 122     elif args.quiet:
 123         logging.getLogger().setLevel(logging.WARNING)
 124     else:
 125         logging.getLogger().setLevel(logging.INFO)
 126
 127     connection = init_connection(args)
 128
 129     try:
 130         connection.connect(args.Host)
 131     except HostKeyMatchSSHFP:
 132         return 0
 133     except HostKeyMismatchSSHFP:
 134         return 2
 135
 136
 137 if __name__ == '__main__':
 138     sys.exit(main())