homer.py

#!/usr/bin/env python3

# Homer is a DoH (DNS-over-HTTPS) and DoT (DNS-over-TLS) client. Its
# main purpose is to test DoH and DoT resolvers. Reference site is
# <https://framagit.org/bortzmeyer/homer/> See author, documentation,
# etc, there, or in the README.md included with the distribution.

# http://pycurl.io/docs/latest
import pycurl

# http://www.dnspython.org/
import dns.message

# https://github.com/drkjam/netaddr/
import netaddr

# Octobre 2019: the Python GnuTLS bindings don't work with Python 3. So we use OpenSSL.
# https://www.pyopenssl.org/
# https://pyopenssl.readthedocs.io/
import OpenSSL

import io
import sys
import base64
import getopt
import urllib.parse
import time
import socket
import ctypes
import re
import os.path
import hashlib
import base64

# Values that can be changed from the command line
dot = False # DoH by default
verbose = False
insecure = False
post = False
head = False
dnssec = False
edns = True
rtype = 'AAAA'
vhostname = None
tests = 1 # Number of repeated tests
ifile = None # Input file
delay = None
forceIPv4 = False
forceIPv6 = False
connectTo = None
# Monitoring plugin only:
host = None
path = None

# Do not change these
re_host = re.compile(r'^([0-9a-z][0-9a-z-\.]*)|([0-9:]+)|([0-9\.])$')

# For the monitoring plugin
STATE_OK = 0
STATE_WARNING = 1
STATE_CRITICAL = 2
STATE_UNKNOWN = 3
STATE_DEPENDENT = 4

def error(msg=None):
    if msg is None:
        msg = "Unknown error"
    if monitoring:
        print("%s: %s" % (url, msg))
        sys.exit(STATE_CRITICAL)
    else:
        print(msg,file=sys.stderr)
        sys.exit(1)

def usage(msg=None):
    if msg:
        print(msg,file=sys.stderr)
    print("Usage: %s [--dot] url-or-servername domain-name [DNS type]" % sys.argv[0], file=sys.stderr)
    print("See the README.md for more details.", file=sys.stderr)

def is_valid_hostname(name):
    name = canonicalize(name)
    return re_host.search(name)

def canonicalize(hostname):
    result = hostname.lower()
    # TODO handle properly the case where it fails with UnicodeError
    # (two consecutive dots for instance) to get a custom exception
    result = result.encode('idna').decode()
    if result[len(result)-1] == '.':
        result = result[:-1]
    return result

def is_valid_ip_address(addr):
    try:
        baddr = netaddr.IPAddress(addr)
    except netaddr.core.AddrFormatError:
        return (False, None)
    return (True, baddr.version)

def is_valid_url(url):
  try:
    result = urllib.parse.urlparse(url) # A very poor validation, many
    # errors (for instance whitespaces, IPv6 address litterals without
    # brackets...) are ignored.
    return (result.scheme=="https" and result.netloc != "")
  except ValueError:
    return False

def get_certificate_san(x509cert):
    san = ""
    ext_count = x509cert.get_extension_count()
    for i in range(0, ext_count):
        ext = x509cert.get_extension(i)
        if "subjectAltName" in str(ext.get_short_name()):
            san = str(ext)
    return san

# Try one possible name. Names must be already canonicalized.
def match_hostname(hostname, possibleMatch):
    if possibleMatch.startswith("*."): # Wildcard
        base = possibleMatch[1:] # Skip the star
        # RFC 6125 says that we MAY accept left-most labels with
        # wildcards included (foo*bar). We don't do it here.
        try:
            (first, rest) = hostname.split(".", maxsplit=1)
        except ValueError: # One-label name
            rest = hostname
        if rest == base[1:]:
            return True
        if hostname == base[1:]:
            return True
        return False
    else:
        return hostname == possibleMatch

# Try all the names in the certificate
def validate_hostname(hostname, cert):
    # Complete specification is in RFC 6125. It is long and
    # complicated and I'm not sure we do it perfectly.
    (is_addr, family) = is_valid_ip_address(hostname)
    hostname = canonicalize(hostname)
    for alt_name in get_certificate_san(cert).split(", "):
        if alt_name.startswith("DNS:") and not is_addr:
            (start, base) = alt_name.split("DNS:")
            base = canonicalize(base)
            found = match_hostname(hostname, base)
            if found:
                return True
        elif alt_name.startswith("IP Address:") and is_addr:
            host_i = netaddr.IPAddress(hostname)
            (start, base) = alt_name.split("IP Address:")
            if base.endswith("\n"):
                base = base[:-1]
            try:
                base_i = netaddr.IPAddress(base)
            except netaddr.core.AddrFormatError:
                continue # Ignore broken IP addresses in certificates. Are we too liberal?
            if host_i == base_i:
                return True
        else:
            pass # Ignore unknown alternative name types. May be
                 # accept URI alternative names for DoH,
    # According to RFC 6125, we MUST NOT try the Common Name before the Subject Alternative Names.
    cn = canonicalize(cert.get_subject().commonName)
    found = match_hostname(hostname, cn)
    if found:
        return True
    return False


class CustomException(Exception):
    pass


class Request:
    def __init__(self, qname, qtype=rtype, use_edns=edns, want_dnssec=dnssec):
        self.message = dns.message.make_query(qname, dns.rdatatype.from_text(qtype), use_edns=use_edns, want_dnssec=want_dnssec)
        self.message.flags |= dns.flags.AD # Ask for validation

    def to_wire(self):
        self.data = self.message.to_wire()


class RequestDoT(Request):
    def check_response(self):
        if self.response.id != self.message.id:
            raise Exception("The ID in the answer does not match the one in the query")


class RequestDoH(Request):
    def __init__(self, qname, qtype=rtype, use_edns=edns, want_dnssec=dnssec):
        Request.__init__(self, qname, qtype=rtype, use_edns=edns, want_dnssec=dnssec)
        self.message.id = 0 # DoH requests that
        self.post = False
        self.head = False

    def check_response(self):
        ok = True
        if self.rcode == 200:
            if self.ctype != "application/dns-message":
                self.response = "Content type of the response (\"%s\") invalid" % self.ctype
                ok = False
            else:
                if not self.head:
                    try:
                        self.response = dns.message.from_wire(self.body)
                    except dns.message.TrailingJunk: # Not DNS. Should
                        # not happen for a content type
                        # application/dns-message but who knows?
                        self.response = "ERROR Not proper DNS data, trailing junk \"%s\"" % self.body
                        ok = False
                    except dns.name.BadLabelType: # Not DNS.
                        self.response = "ERROR Not proper DNS data (wrong path in the URL?) \"%s\"" % self.body[:100]
                        ok = False
                else:
                    if self.body_size == 0:
                        self.response = "HEAD successful"
                    else:
                        self.response = "ERROR Body length is not null \"%s\"" % self.body[:100]
                        ok = False
        else:
            ok = False
            if self.body_size == 0:
                self.response = "[No details]"
            else:
                self.response = self.body
        return ok


class Connection:
    def __init__(self, server, servername=None, connect=None, forceIPv4=False, forceIPv6=False,
                 dot=False, verbose=verbose, insecure=insecure):
        if dot and not is_valid_hostname(server):
            error("DoT requires a host name or IP address, not \"%s\"" % server)
        if not dot and not is_valid_url(url):
            error("DoH requires a valid HTTPS URL, not \"%s\"" % server)
        if forceIPv4 and forceIPv6:
            raise CustomException("Force IPv4 *or* IPv6 but not both")
        self.server = server
        self.servername = servername
        if self.servername is not None:
            self.check = self.servername
        else:
            self.check = self.server
        self.dot = dot
        self.verbose = verbose
        self.insecure = insecure

    def __str__(self):
        return self.server

    def check_ip_address(self, addr):
        (is_addr, self.family) = is_valid_ip_address(addr)
        if not is_addr and not self.dot:
            raise CustomException("%s is not IPv4 and not IPv6" % addr)
        if forceIPv4 and self.family == 6:
            raise CustomException("You cannot force IPv4 with a litteral IPv6 address (%s)" % addr)
        elif forceIPv6 and self.family == 4:
            raise CustomException("You cannot force IPv6 with a litteral IPv4 address (%s)" % addr)
        if forceIPv4 or self.family == 4:
            self.family = socket.AF_INET
            self.repraddress = addr
        elif forceIPv6 or self.family == 6:
            self.family = socket.AF_INET6
            self.repraddress = f'[{addr}]'
        else:
            self.family = 0

    def do_test(self, qname, qtype=rtype):
        # Routine doing one actual test. Returns a tuple, first member is a
        # result (boolean indicating success for DoT, HTTP status code for
        # DoH), second member is a DNS message (or a string if there is an
        # error), third member is the size of the DNS message (or None if no
        # proper response).
        pass

    def print_result(self, rcode, msg, size):
        ok = True
        if (self.dot and rcode) or (not self.dot and rcode == 200):
            if not monitoring:
                print(msg)
            else:
                if size is not None and size > 0:
                    print("%s OK - %s" % (self.server, "No error for %s/%s, %i bytes received" % (name, rtype, size)))
                else:
                    print("%s OK - %s" % (self.server, "No error"))
                sys.exit(STATE_OK)
        else:
            if not monitoring:
                if dot:
                    print("Error: %s" % msg, file=sys.stderr)
                else:
                   try:
                       msg = msg.decode()
                   except (UnicodeDecodeError, AttributeError):
                       pass # Sometimes, msg can be binary, or Latin-1
                   print("HTTP error %i: %s" % (rcode, msg), file=sys.stderr)
            else:
                if not dot:
                    print("%s HTTP error - %i: %s" % (self.server, rcode, msg))
                else:
                    print("%s Error - %i: %s" % (self.server, rcode, msg))
                sys.exit(STATE_CRITICAL)
            ok = False
        return ok


class ConnectionDoT(Connection):
    def __init__(self, server, servername=None, connect=None, forceIPv4=False, forceIPv6=False,
                 dot=False, verbose=verbose, insecure=insecure):
        Connection.__init__(self, server, servername=servername, connect=connect,
                forceIPv4=forceIPv4, forceIPv6=forceIPv6, dot=dot,
                verbose=verbose, insecure=insecure)
        self.check_ip_address(self.server)
        self.hasher = hashlib.sha256()
        addrinfo = socket.getaddrinfo(server, 853, self.family)
        # May be loop over the results of getaddrinfo, to test all
        # the IP addresses? See #13.
        self.sock = socket.socket(addrinfo[0][0], socket.SOCK_STREAM)
        self.addr = addrinfo[0][4]
        if self.verbose:
            print("Connecting to %s ..." % str(self.addr))
        # With typical DoT servers, we *must* use TLS 1.2 (otherwise,
        # do_handshake fails with "OpenSSL.SSL.SysCallError: (-1, 'Unexpected
        # EOF')" Typical HTTP servers are more lax.
        self.context = OpenSSL.SSL.Context(OpenSSL.SSL.TLSv1_2_METHOD)
        if self.insecure:
            self.context.set_verify(OpenSSL.SSL.VERIFY_NONE, lambda *x: True)
        else:
            self.context.set_default_verify_paths()
            self.context.set_verify_depth(4) # Seems ignored
            self.context.set_verify(OpenSSL.SSL.VERIFY_PEER | OpenSSL.SSL.VERIFY_FAIL_IF_NO_PEER_CERT | \
                                    OpenSSL.SSL.VERIFY_CLIENT_ONCE,
                                    lambda conn, cert, errno, depth, preverify_ok: preverify_ok)
        self.session = OpenSSL.SSL.Connection(self.context, self.sock)
        self.session.set_tlsext_host_name(canonicalize(self.check).encode()) # Server Name Indication (SNI)
        self.session.connect((self.addr))
        # TODO We may here have exceptions such as OpenSSL.SSL.ZeroReturnError
        self.session.do_handshake()
        self.cert = self.session.get_peer_certificate()
        # RFC 7858, section 4.2 and appendix A
        self.publickey = self.cert.get_pubkey()
        if verbose:
            print("Certificate #%x for \"%s\", delivered by \"%s\"" % \
                  (self.cert.get_serial_number(),
                   self.cert.get_subject().commonName,
                   self.cert.get_issuer().commonName))
            self.hasher.update(OpenSSL.crypto.dump_publickey(OpenSSL.crypto.FILETYPE_ASN1,
                                                  self.publickey))
            self.digest = self.hasher.digest()
            print("Public key is pin-sha256=\"%s\"" % \
                  base64.standard_b64encode(self.digest).decode())
        if not insecure:
            valid = validate_hostname(self.check, self.cert)
            if not valid:
                error("Certificate error: \"%s\" is not in the certificate" % (self.check))

    def end(self):
        self.session.shutdown()
        self.session.close()

    def send_data(self, data):
        length = len(data)
        self.session.send(length.to_bytes(2, byteorder='big') + data)

    def receive_data(self, request):
        buf = self.session.recv(2)
        request.response_size = int.from_bytes(buf, byteorder='big')
        buf = self.session.recv(request.response_size)
        request.response = dns.message.from_wire(buf)
        request.rcode = True

    def send_and_receive(self, request):
        request.to_wire()
        self.send_data(request.data)
        self.receive_data(request)

    def do_test(self, qname, qtype=rtype):
        request = RequestDoT(qname, qtype, want_dnssec=dnssec, use_edns=edns)
        self.send_and_receive(request)
        request.check_response()
        return (request.rcode, request.response, request.response_size)


class ConnectionDoH(Connection):
    def __init__(self, server, servername=None, connect=None, forceIPv4=False, forceIPv6=False,
                 dot=False, verbose=verbose, insecure=insecure):
        Connection.__init__(self, server, servername=servername, connect=connect,
                forceIPv4=forceIPv4, forceIPv6=forceIPv6, dot=dot,
                verbose=verbose, insecure=insecure)
        self.url = server
        self.connect = connect

    def create_handle(self):
        self.curl = pycurl.Curl()
        # Does not work if pycurl was not compiled with nghttp2 (recent Debian
        # packages are OK) https://github.com/pycurl/pycurl/issues/477
        self.curl.setopt(pycurl.HTTP_VERSION, pycurl.CURL_HTTP_VERSION_2)
        if self.verbose:
            self.curl.setopt(pycurl.VERBOSE, True)
        if self.insecure:
            self.curl.setopt(pycurl.SSL_VERIFYPEER, False)
            self.curl.setopt(pycurl.SSL_VERIFYHOST, False)
        if forceIPv4:
            self.curl.setopt(pycurl.IPRESOLVE, pycurl.IPRESOLVE_V4)
        if forceIPv6:
            self.curl.setopt(pycurl.IPRESOLVE, pycurl.IPRESOLVE_V6)
        if self.connect is not None:
            self.check_ip_address(self.connect)
            self.curl.setopt(pycurl.CONNECT_TO, [f'::{self.repraddress}:443',])
        self.curl.setopt(pycurl.HTTPHEADER, ["Content-type: application/dns-message"])

    def end(self):
        self.curl.close()

    def set_opt(self, opt, value):
        self.curl.setopt(opt, value)

    def reset_opt_default(self):
        opts = {
                pycurl.NOBODY: False,
                pycurl.POST: False,
                pycurl.POSTFIELDS: '',
                pycurl.URL: ''
               }
        for opt, value in opts.items():
            self.set_opt(opt, value)

    def prepare(self, request):
        try:
            self.reset_opt_default()
        except AttributeError:
            self.create_handle()
        if request.post:
            self.prepare_post(request)
        elif request.head:
            self.prepare_head(request)
        else:
            self.prepare_get(request)

    def prepare_get(self, request):
        self.set_opt(pycurl.HTTPGET, True)
        dns_req = base64.urlsafe_b64encode(request.data).decode('UTF8').rstrip('=')
        self.set_opt(pycurl.URL, self.server + ("?dns=%s" % dns_req))

    def prepare_post(self, request):
        request.post = True
        self.set_opt(pycurl.POST, True)
        self.set_opt(pycurl.POSTFIELDS, request.data)
        self.set_opt(pycurl.URL, self.server)

    def prepare_head(self, request):
        request.head = True
        self.prepare_get(request)
        self.set_opt(pycurl.NOBODY, True)

    def perform(self):
        self.buffer = io.BytesIO()
        self.set_opt(pycurl.WRITEDATA, self.buffer)
        self.curl.perform()

    def receive(self, request):
        body = self.buffer.getvalue()
        body_size = len(body)
        http_code = self.curl.getinfo(pycurl.RESPONSE_CODE)
        content_type = self.curl.getinfo(pycurl.CONTENT_TYPE)
        request.body = body
        request.body_size = body_size
        request.rcode = http_code
        request.ctype = content_type
        self.buffer.close()

    def send_and_receive(self, request):
        request.to_wire()
        self.prepare(request)
        self.perform()
        self.receive(request)

    def do_test(self, qname, qtype=rtype):
        request = RequestDoH(qname, qtype, want_dnssec=dnssec, use_edns=edns)
        request.head = head
        request.post = post
        self.send_and_receive(request)
        request.check_response()
        return (request.rcode, request.response, request.body_size)


def get_next_domain(input_file):
    name, rtype = 'framagit.org', 'AAAA'
    line = input_file.readline()
    if line[:-1] == "":
        error("Not enough data in %s for the %i tests" % (ifile, tests))
    if line.find(' ') == -1:
        name = line[:-1]
        rtype = 'AAAA'
    else:
        (name, rtype) = line.split()
    return name, rtype

# Main program
me = os.path.basename(sys.argv[0])
monitoring = (me == "check_doh" or me == "check_dot")
if not monitoring:
    name = None
    message = None
    try:
        optlist, args = getopt.getopt (sys.argv[1:], "hvPkeV:r:f:d:t46",
                                       ["help", "verbose", "dot", "head",
                                        "insecure", "POST", "vhost=",
                                        "dnssec", "noedns","repeat=", "file=", "delay=", "v4only", "v6only"])
        for option, value in optlist:
            if option == "--help" or option == "-h":
                usage()
                sys.exit(0)
            elif option == "--dot" or option == "-t":
                dot = True
            elif option == "--verbose" or option == "-v":
                verbose = True
            elif option == "--HEAD" or option == "--head" or option == "-e":
                head = True
            elif option == "--POST" or option == "--post" or option == "-P":
                post = True
            elif option == "--vhost" or option == "-V":
                vhostname = value
            elif option == "--insecure" or option == "-k":
                insecure = True
            elif option == "--dnssec":
                dnssec = True
            elif option == "--noedns":
                edns = False
            elif option == "--repeat" or option == "-r":
                tests = int(value)
                if tests <= 1:
                    error("--repeat needs a value > 1")
            elif option == "--delay" or option == "-d":
                delay = float(value)
                if delay <= 0:
                    error("--delay needs a value > 0")
            elif option == "--file" or option == "-f":
                ifile = value
            elif option == "-4" or option == "v4only":
                forceIPv4 = True
            elif option == "-6" or option == "v6only":
                forceIPv6 = True
            else:
                error("Unknown option %s" % option)
    except getopt.error as reason:
        usage(reason)
        sys.exit(1)
    if tests <= 1 and delay is not None:
        error("--delay makes no sense if there is no repetition")
    if post and head:
        usage("POST or HEAD but not both")
        sys.exit(1)
    if dot and (post or head):
        usage("POST or HEAD makes non sense for DoT")
        sys.exit(1)
    if ifile is None and (len(args) != 2 and len(args) != 3):
        usage("Wrong number of arguments")
        sys.exit(1)
    if ifile is not None and len(args) != 1:
        usage("Wrong number of arguments (if --file is used, do not indicate the domain name)")
        sys.exit(1)
    url = args[0]
    if ifile is None:
        name = args[1]
        if len(args) == 3:
            rtype = args[2]
else: # Monitoring plugin
    dot = (me == "check_dot")
    name = None
    try:
        optlist, args = getopt.getopt (sys.argv[1:], "H:n:p:V:t:Pih46")
        for option, value in optlist:
            if option == "-H":
                host = value
            elif option == "-V":
                vhostname = value
            elif option == "-n":
                name = value
            elif option == "-t":
                rtype = value
            elif option == "-p":
                path = value
            elif option == "-P":
                post = True
            elif option == "-h":
                head = True
            elif option == "-i":
                insecure = True
            elif option == "-4":
                forceIPv4 = True
            elif option == "-6":
                forceIPv6 = True
            else:
                # Should never occur, it is trapped by getopt
                print("Unknown option %s" % option)
                sys.exit(STATE_UNKNOWN)
    except getopt.error as reason:
        print("Option parsing problem %s" % reason)
        sys.exit(STATE_UNKNOWN)
    if len(args) > 0:
        print("Too many arguments (\"%s\")" % args)
        sys.exit(STATE_UNKNOWN)
    if host is None or name is None:
        print("Host (-H) and name to lookup (-n) are necessary")
        sys.exit(STATE_UNKNOWN)
    if post and head:
        print("POST or HEAD but not both")
        sys.exit(STATE_UNKNOWN)
    if dot and (post or head):
        print("POST or HEAD makes no sense for DoT")
        sys.exit(STATE_UNKNOWN)
    if dot and path:
        print("URL path makes no sense for DoT")
        sys.exit(STATE_UNKNOWN)
    if dot:
        url = host
    else:
        if vhostname is None or vhostname == host:
            connectTo = None
            url = "https://%s/" % host
        else:
            connectTo = host
            url = "https://%s/" % vhostname
        if path is not None:
            if path.startswith("/"):
                path = path[1:]
            url += path
ok = True
start = time.time()
try:
    if dot and vhostname is not None:
        extracheck = vhostname
    else:
        extracheck = None
    if dot:
        conn = ConnectionDoT(url, dot=dot, servername=extracheck, connect=connectTo, verbose=verbose,
                          forceIPv4=forceIPv4, forceIPv6=forceIPv6,
                          insecure=insecure)
    else:
        conn = ConnectionDoH(url, dot=dot, servername=extracheck, connect=connectTo, verbose=verbose,
                          forceIPv4=forceIPv4, forceIPv6=forceIPv6,
                          insecure=insecure)
except TimeoutError:
    error("timeout")
except ConnectionRefusedError:
    error("Connection to server refused")
except CustomException as e:
    error(e)
if ifile is not None:
    input = open(ifile)
for i in range (0, tests):
    if tests > 1:
        print("\nTest %i" % i)
    if ifile is not None:
        name, rtype = get_next_domain(input)
    try:
        (rcode, msg, size) = conn.do_test(name, rtype)
    except (OpenSSL.SSL.Error, CustomException) as e:
        ok = False
        error(e)
        break
    if not conn.print_result(rcode, msg, size):
        ok = False
    if tests > 1 and i == 0:
        start2 = time.time()
    if delay is not None:
        time.sleep(delay)
stop = time.time()
if tests > 1:
    extra = ", %.2f ms/request if we ignore the first one" % ((stop-start2)*1000/(tests-1))
else:
    extra = ""
if not monitoring:
    print("\nTotal elapsed time: %.2f seconds (%.2f ms/request %s)" % (stop-start, (stop-start)*1000/tests, extra))
if ifile is not None:
    input.close()
conn.end()
if ok:
    if not monitoring:
        sys.exit(0)
    else:
        sys.exit(STATE_OK)
else:
    if not monitoring:
        sys.exit(1)
    else:
        sys.exit(STATE_CRITICAL)