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
import signal

# Values that can be changed from the command line
dot = False # DoH by default
verbose = False
debug = False
insecure = False
post = False
head = False
dnssec = False
edns = True
no_ecs = True
sni = True
rtype = 'AAAA'
vhostname = None
tests = 1 # Number of repeated tests
key = None # SPKI
ifile = None # Input file
delay = None
forceIPv4 = False
forceIPv6 = False
connectTo = None
multistreams = False
sync = False
display_results = True
show_time = False
check = False
mandatory_level = None
check_additional = True
# Monitoring plugin only:
host = None
path = None
expect = 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

# For the check option
DOH_GET = 0
DOH_POST = 1
DOH_HEAD = 2
# Is the test mandatory?
mandatory_levels = {"legal": 30, "necessary": 20, "nicetohave": 10}

TIMEOUT_CONN = 2

def error(msg=None, exit=True):
    if msg is None:
        msg = "Unknown error"
    if monitoring:
        print("%s: %s" % (url, msg))
        if exit:
            sys.exit(STATE_CRITICAL)
    else:
        print(msg, file=sys.stderr)
        if check:
            print('KO')
        if exit:
            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

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

def dump_data(data, text="data"):
    pref = ' ' * (len(text) - 4)
    print(f'{text}: ', data)
    print(pref, 'hex:', " ".join(format(c, '02x') for c in data))
    print(pref, 'bin:', " ".join(format(c, '08b') for c in data))

def timeout_connection(signum, frame):
    raise TimeoutConnectionError('Connection timeout')

class TimeoutConnectionError(Exception):
    pass


class CustomException(Exception):
    pass


class Request:
    def __init__(self, qname, qtype=rtype, use_edns=edns, want_dnssec=dnssec):
        if no_ecs:
             opt = dns.edns.ECSOption(address='', srclen=0) # Disable ECS (RFC 7871, section 7.1.2)
             options = [opt]
        else:
            options = None
        self.message = dns.message.make_query(qname, dns.rdatatype.from_text(qtype),
                                              use_edns=use_edns, want_dnssec=want_dnssec, options=options)
        self.message.flags |= dns.flags.AD # Ask for validation
        self.ok = True

    def trunc_data(self):
        self.data = self.message.to_wire()
        half = round(len(self.data) / 2)
        self.data = self.data[:half]

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


class RequestDoT(Request):
    def check_response(self, debug=False):
        ok = self.ok
        if not self.rcode:
            self.ok = False
            return False
        if self.response.id != self.message.id:
            self.response = "The ID in the answer does not match the one in the query"
            if debug:
                self.response += f'"(query id: {self.message.id}) (response id: {self.response.id})'
            self.ok = False
            return False
        return self.ok


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, debug=False):
        ok = self.ok
        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:
                        response = dns.message.from_wire(self.response)
                    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"
                        if debug:
                            self.response += " \"%s\"" % response
                        ok = False
                    except dns.name.BadLabelType: # Not DNS.
                        self.response = "ERROR Not proper DNS data (wrong path in the URL?)"
                        if debug:
                            self.response += " \"%s\"" % response[:100]
                        ok = False
                    else:
                        self.response = response
                else:
                    if self.response_size == 0:
                        self.response = "HEAD successful"
                    else:
                        data = self.response
                        self.response = "ERROR Body length is not null"
                        if debug:
                            self.response += "\"%s\"" % data[:100]
                        ok = False
        else:
            ok = False
            if self.response_size == 0:
                self.response = "[No details]"
            else:
                self.response = self.response
        self.ok = ok
        return ok


class Connection:
    def __init__(self, server, servername=None, connect=None, forceIPv4=False, forceIPv6=False,
                 dot=dot, verbose=verbose, debug=debug, 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(server):
            error("DoH requires a valid HTTPS URL, not \"%s\"" % server)
        if forceIPv4 and forceIPv6:
            raise CustomException("Force IPv4 *or* IPv6 but not both")
        self.dot = dot
        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.debug = debug
        self.insecure = insecure
        self.forceIPv4 = forceIPv4
        self.forceIPv6 = forceIPv6
        self.connect_to = connect

    def __str__(self):
        return self.server

    def do_test(self, request):
        # Routine doing one actual test. Returns nothing
        pass


class ConnectionDoT(Connection):
    def __init__(self, server, servername=None, connect=None, forceIPv4=False, forceIPv6=False,
                 verbose=verbose, debug=debug, insecure=insecure):
        Connection.__init__(self, server, servername=servername, connect=connect,
                forceIPv4=forceIPv4, forceIPv6=forceIPv6, dot=True,
                verbose=verbose, debug=debug, insecure=insecure)
        if connect is not None:
            addr = connect
        else:
            addr = self.server
        self.family, self.repraddress = check_ip_address(self.server, dot=True)
        addrinfo_list = socket.getaddrinfo(addr, 853, self.family)
        addrinfo_set = { (addrinfo[4], addrinfo[0]) for addrinfo in addrinfo_list }
        signal.signal(signal.SIGALRM, timeout_connection)
        self.success = False
        for addrinfo in addrinfo_set:
            self.hasher = hashlib.sha256()
            if self.connect(addrinfo[0], addrinfo[1]):
                self.success = True
                break
            if self.verbose and connect is None:
                print("Trying another IP address")
        if not self.success:
            if self.verbose and connect is None:
                print("No other IP address")
            if connect is None:
                error(f'Could not connect to "{server}"')
            else:
                print(f'Could not connect to "{server}" on {connect}')


    def connect(self, addr, sock_family):
        signal.alarm(TIMEOUT_CONN)
        self.addr = addr
        self.sock = socket.socket(sock_family, socket.SOCK_STREAM)
        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)
        if sni:
            self.session.set_tlsext_host_name(canonicalize(self.check).encode()) # Server Name Indication (SNI)
        try:
            self.session.connect((self.addr))
            # TODO We may here have exceptions such as OpenSSL.SSL.ZeroReturnError
            self.session.do_handshake()
        except TimeoutConnectionError:
            if self.verbose:
                print("Timeout")
            return False
        except OSError:
            if self.verbose:
                print("Cannot connect")
            return False
        except OpenSSL.SSL.Error as e:
            if self.verbose:
                print(f"OpenSSL error: {', '.join(err[0][2] for err in e.args)}")
            return False
        # RFC 7858, section 4.2 and appendix A
        self.cert = self.session.get_peer_certificate()
        self.publickey = self.cert.get_pubkey()
        if debug or key is not None:
            self.hasher.update(OpenSSL.crypto.dump_publickey(OpenSSL.crypto.FILETYPE_ASN1,
                                                  self.publickey))
            self.digest = self.hasher.digest()
            key_string = base64.standard_b64encode(self.digest).decode()
        if debug:
            print("Certificate #%x for \"%s\", delivered by \"%s\"" % \
                  (self.cert.get_serial_number(),
                   self.cert.get_subject().commonName,
                   self.cert.get_issuer().commonName))
            print("Public key is pin-sha256=\"%s\"" % \
                  key_string)
        if not insecure:
            if key is None:
                valid = validate_hostname(self.check, self.cert)
                if not valid:
                    error("Certificate error: \"%s\" is not in the certificate" % (self.check))
            else:
                if key_string != key:
                    error("Key error: expected \"%s\", got \"%s\"" % (key, key_string))
        signal.alarm(0)
        return True

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

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

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

    def send_and_receive(self, request, dump=False):
        self.send_data(request.data, dump=dump)
        self.receive_data(request, dump=dump)

    def do_test(self, request, synchronous=True):
        self.send_and_receive(request)
        request.check_response(self.debug)


def create_handle(connection):
    def reset_opt_default(handle):
        opts = {
                pycurl.NOBODY: False,
                pycurl.POST: False,
                pycurl.POSTFIELDS: '',
                pycurl.URL: ''
               }
        for opt, value in opts.items():
            handle.setopt(opt, value)

    def prepare(handle, connection, request):
        if not connection.multistreams:
            handle.reset_opt_default(handle)
        if request.post:
            handle.setopt(pycurl.POST, True)
            handle.setopt(pycurl.POSTFIELDS, request.data)
            handle.setopt(pycurl.URL, connection.server)
        else:
            handle.setopt(pycurl.HTTPGET, True) # automatically sets CURLOPT_NOBODY to 0
            if request.head:
                handle.setopt(pycurl.NOBODY, True)
            dns_req = base64.urlsafe_b64encode(request.data).decode('UTF8').rstrip('=')
            handle.setopt(pycurl.URL, connection.server + ("?dns=%s" % dns_req))
        handle.buffer = io.BytesIO()
        handle.setopt(pycurl.WRITEDATA, handle.buffer)
        handle.request = request

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


class ConnectionDoH(Connection):
    def __init__(self, server, servername=None, connect=None, forceIPv4=False, forceIPv6=False,
                 multistreams=False, verbose=verbose, debug=debug, insecure=insecure):
        Connection.__init__(self, server, servername=servername, connect=connect,
                forceIPv4=forceIPv4, forceIPv6=forceIPv6, dot=False,
                verbose=verbose, debug=debug, insecure=insecure)
        self.url = server
        self.connect = connect
        self.multistreams = multistreams
        if self.multistreams:
            self.multi = self.create_multi()
            self.all_handles = []
        else:
            self.curl_handle = create_handle(self)

    def create_multi(self):
        multi = pycurl.CurlMulti()
        multi.setopt(pycurl.M_MAX_HOST_CONNECTIONS, 1)
        return multi

    def end(self):
        if not self.multistreams:
            self.curl_handle.close()
        else:
            self.remove_handles()
            self.multi.close()

    def remove_handles(self):
        n, handle_success, handle_fail = self.multi.info_read()
        handles = handle_success + handle_fail
        for h in handles:
            h.close()
            self.multi.remove_handle(h)

    def perform_multi(self):
        while 1:
            ret, num_handles = self.multi.perform()
            if ret != pycurl.E_CALL_MULTI_PERFORM:
                break
        while num_handles:
            ret = self.multi.select(1.0)
            if ret == -1:
                continue
            while 1:
                ret, num_handles = self.multi.perform()
                if not sync:
                    n, handle_pass, handle_fail = self.multi.info_read()
                    for handle in handle_pass:
                        self.read_result_handle(handle)
                if ret != pycurl.E_CALL_MULTI_PERFORM:
                    break
        if not sync:
            n, handle_pass, handle_fail = self.multi.info_read()
            for handle in handle_pass:
                self.read_result_handle(handle)

    def send(self, handle):
        handle.buffer = io.BytesIO()
        handle.setopt(pycurl.WRITEDATA, handle.buffer)
        try:
            handle.perform()
        except pycurl.error as e:
            error(e.args[1])

    def receive(self, handle):
        request = handle.request
        body = handle.buffer.getvalue()
        body_size = len(body)
        http_code = handle.getinfo(pycurl.RESPONSE_CODE)
        handle.time = handle.getinfo(pycurl.TOTAL_TIME)
        handle.pretime = handle.getinfo(pycurl.PRETRANSFER_TIME)
        try:
            content_type = handle.getinfo(pycurl.CONTENT_TYPE)
        except TypeError: # This is the exception we get if there is no Content-Type: (for intance in response to HEAD requests)
            content_type = None
        request.response = body
        request.response_size = body_size
        request.rcode = http_code
        request.ctype = content_type
        handle.buffer.close()

    def send_and_receive(self, handle, dump=False):
        self.send(handle)
        self.receive(handle)

    def read_result_handle(self, handle):
        self.receive(handle)
        handle.request.check_response()
        if show_time:
            print(f'{handle.request.i:3d}', end='   ')
            print(f'({handle.request.rcode})', end='   ')
            print(f'{handle.pretime * 1000:8.3f} ms', end='  ')
            print(f'{handle.time * 1000:8.3f} ms', end='  ')
            print(f'{(handle.time - handle.pretime) * 1000:8.3f} ms')
        if display_results:
            print("Return code %s (%.2f ms):" % (handle.request.rcode,
                (handle.time - handle.pretime) * 1000))
            print(f"{handle.request.response}\n")
        handle.close()
        self.multi.remove_handle(handle)

    def read_results(self):
        for handle in self.all_handles:
            self.read_result_handle(handle)

    def do_test(self, request, synchronous=True):
        if synchronous:
            handle = self.curl_handle
        else:
            handle = create_handle(self)
            self.all_handles.append(handle)
        handle.prepare(handle, self, request)
        if synchronous:
            self.send_and_receive(handle)
            request.check_response(self.debug)
        else:
            self.multi.add_handle(handle)


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

def print_result(connection, request, prefix=None, display_err=True):
    ok = request.ok
    dot = connection.dot
    server = connection.server
    rcode = request.rcode
    msg = request.response
    size = request.response_size
    if (dot and rcode) or (not dot and rcode == 200):
        if not monitoring:
            if display_results and (not check or verbose):
                print(msg)
        else:
            if expect is not None and expect not in str(request.response):
                ok = False
                print("%s Cannot find \"%s\" in response" % (server, expect))
                sys.exit(STATE_CRITICAL)
            if ok and size is not None and size > 0:
                print("%s OK - %s" % (server, "No error for %s/%s, %i bytes received" % (name, rtype, size)))
            else:
                print("%s OK - %s" % (server, "No error"))
            sys.exit(STATE_OK)
    else:
        if not monitoring:
            if display_err:
                if check:
                    print(connection.connect_to, end=': ', file=sys.stderr)
                if prefix:
                    print(prefix, end=': ', file=sys.stderr)
                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" % (server, rcode, msg))
            else:
                print("%s Error - %i: %s" % (server, rcode, msg))
            sys.exit(STATE_CRITICAL)
        ok = False
    return ok

def create_request(qname, qtype=rtype, use_edns=edns, want_dnssec=dnssec, dot=dot, trunc=False):
    if dot:
        request = RequestDoT(qname, rtype, use_edns, want_dnssec)
    else:
        request = RequestDoH(qname, rtype, use_edns, want_dnssec)
    if trunc:
        request.trunc_data()
    else:
        request.to_wire()
    return request

def create_requests_list(dot=dot, **req_args):
    requests = []
    if dot:
        requests.append(('Test 1', create_request(dot=dot, **req_args),
                        mandatory_levels["legal"]))
        requests.append(('Test 2', create_request(dot=dot, **req_args),
                         mandatory_levels["necessary"])) # RFC 7858,
        # section 3.3, SHOULD accept several requests on one connection.
        # TODO we miss the tests of pipelining and out-of-order.
    else:
        requests.append(('Test GET', create_request(**req_args), DOH_GET,
                         mandatory_levels["legal"])) # RFC 8484, section 4.1
        requests.append(('Test POST', create_request(**req_args), DOH_POST,
                         mandatory_levels["legal"])) # RFC 8484, section 4.1
        requests.append(('Test HEAD', create_request(**req_args), DOH_HEAD,
                         mandatory_levels["nicetohave"])) # HEAD
        # method is not mentioned in RFC 8484 (see section 4.1), so
        # just "nice to have".
    return requests

def run_check_default(connection):
    ok = True
    req_args = { 'qname': name, 'qtype': rtype, 'use_edns': edns, 'want_dnssec': dnssec }
    requests = create_requests_list(dot=dot, **req_args)
    for request_pack in requests:
        if dot:
            test_name, request, mandatory = request_pack
        else:
            test_name, request, method, mandatory = request_pack
        if verbose:
            print(test_name)
        if dot:
            bundle = request
        else:
            if method == DOH_POST:
                request.post = True
            elif method == DOH_HEAD:
                request.head = True
            handle = connection.curl_handle
            handle.prepare(handle, connection, request)
            bundle = handle
        try:
            connection.send_and_receive(bundle)
        except CustomException as e:
            ok = False
            error(e)
            break
        request.check_response(debug)
        if not print_result(connection, request, prefix=test_name, display_err=False):
            if mandatory >= mandatory_level:
                print_result(connection, request, prefix=test_name, display_err=True)
                ok = False
            if verbose:
                print()
            break
        if verbose:
            print()
    return ok

def run_check_mime(connection, accept="application/dns-message", content_type="application/dns-message"):
    if dot:
        return True
    ok = True
    header = [f"Accept: {accept}", f"Content-type: {content_type}"]
    if verbose:
        test_name = f'Test mime: {", ".join(h for h in header)}'
        print(test_name)
    req_args = { 'qname': name, 'qtype': rtype, 'use_edns': edns, 'want_dnssec': dnssec }
    request = create_request(**req_args)
    handle = connection.curl_handle
    handle.setopt(pycurl.HTTPHEADER, header)
    handle.prepare(handle, connection, request)
    try:
        connection.send_and_receive(handle)
    except CustomException as e:
        ok = False
        error(e)
    request.check_response(debug)
    if not print_result(connection, request, prefix=f"Test Header {', '.join(header)}"):
        ok = False
    default = "application/dns-message"
    default_header = [f"Accept: {default}", f"Content-type: {default}"]
    handle.setopt(pycurl.HTTPHEADER, default_header)
    if verbose:
        print()
    return ok

def run_check_trunc(connection):
    ok = True
    test_name = 'Test truncated data'
    if verbose:
        print(test_name)
    req_args = { 'qname': name, 'qtype': rtype, 'use_edns': edns, 'want_dnssec': dnssec }
    if dot:
        request = create_request(dot=dot, trunc=True, **req_args)
        bundle = request
    else:
        request = create_request(trunc=True, **req_args)
        request.post = True
        handle = connection.curl_handle
        handle.prepare(handle, connection, request)
        bundle = handle
    try:
        # 8.8.8.8 replies FORMERR but most DoT servers violently shut down the connection (which is legal)
        connection.send_and_receive(bundle, dump=debug)
    except CustomException as e:
        ok = False
        error(e)
    except OpenSSL.SSL.ZeroReturnError: # This is acceptable
        return ok
    except dns.exception.FormError: # This is also acceptable
        # Some DSN resolvers will echo mangled requests with
        # the RCODE set to FORMERR
        # so response can not be parsed in this case
        return ok
    if request.check_response(debug): # FORMERR is expected
        if dot:
            ok = request.rcode == dns.rcode.FORMERR
        else:
            ok = (request.response.rcode() == dns.rcode.FORMERR)
    else:
        if dot:
            ok = False
        else: # a 400 response's status is acceptable
            ok = (request.rcode >= 400 and request.rcode < 500)
    print_result(connection, request, prefix=test_name, display_err=not ok)
    if verbose:
        print()
    return ok

def run_check_additionals(connection):
    if not run_check_trunc(connection):
        return False
    # The DoH server is right to reject these (Example: 'HTTP
    # error 415: only Content-Type: application/dns-message is
    # supported')
    run_check_mime(connection, accept="text/html")
    run_check_mime(connection, content_type="text/html")
    return True

def run_check(connection):
    if not run_check_default(connection):
        return False
    if check_additional and not run_check_additionals(connection):
        return False
    return True

def resolved_ips(host, port, family, dot=dot):
    try:
        addr_list = socket.getaddrinfo(host, port, family)
    except socket.gaierror:
        error(f'Could not resolve "{url}"')
    ip_set = { addr[4][0] for addr in addr_list }
    return ip_set

# 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", "debug", "dot", "head",
                                        "insecure", "POST", "vhost=", "multistreams",
                                        "sync", "no-display-results", "time",
                                        "dnssec", "noedns", "ecs", "repeat=", "file=", "delay=",
                                        "key=", "nosni",
                                        "v4only", "v6only",
                                        "check", "mandatory-level="])
        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 == "--debug":
                debug = True
                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 == "--multistreams":
                multistreams = True
            elif option == "--sync":
                sync = True
            elif option == "--no-display-results":
                display_results = False
            elif option == "--time":
                show_time = True
            elif option == "--dnssec":
                dnssec = True
            elif option == "--nosni":
                sni = False