homer.py

#!/usr/bin/env python3

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

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

# 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

# Values that can be changed from the command line
dot = False # DoH by default
post = False
verbose = False
insecure = False
head = False
rtype = 'AAAA'
tests = 1 # Number of repeated tests
ifile = None # Input file
delay = None

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

def error(msg=None):
    if msg is None:
        msg = "Unknown error"
    print(msg,file=sys.stderr)
    sys.exit(1)
    
def usage(msg=None):
    if msg:
        print(msg,file=sys.stderr)
    print("Usage: %s [-P] [-k] url domain-name [DNS type]" % sys.argv[0], file=sys.stderr)

def is_valid_hostname(name):
    name = str(name.encode('idna').lower())
    return re_host.search(name)
    
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

def validate_hostname(hostname, cert):
    hostname = hostname.lower()
    cn = cert.get_subject().commonName.lower()
    if cn.startswith("*."): # Wildcard
        (start, base) = cn.split("*.")
        if hostname.endswith(base):
            return True
    else:
        if hostname == cn:
            return True
    for alt_name in get_certificate_san(cert).split(", "):
        if alt_name.startswith("DNS:"):
            (start, base) = alt_name.split("DNS:")
            base = base.lower()
            if hostname == base:
                return True
        elif alt_name.startswith("IP Address:"):
            (start, base) = alt_name.split("IP Address:")
            base = base.lower()
            if base.endswith("\n"):
                base = base[:-1]
            if hostname == base: # TODO better canonicaliztion of IP addresses with the netaddr module
                return True
        else:
            pass # Ignore unknown alternative name types
    return False

class Connection:
    def __init__(self, server, dot=False, verbose=verbose, insecure=insecure, post=post, head=head):
        if dot and not is_valid_hostname(server):
            error("DoT requires a host name, not \"%s\"" % server)
        if not dot and not is_valid_url(url):
            error("DoH requires a valid HTTPS URL, not \"%s\"" % server)
        self.server = server
        self.dot = dot
        if not self.dot:
            self.post = post
            self.head = head
        self.verbose = verbose
        self.insecure = insecure
        if self.dot:
            addrinfo = socket.getaddrinfo(server, 853)
            # May be loop over the results of getaddrinfo, to test all the IP addresses?
            self.sock = socket.socket(addrinfo[0][0], socket.SOCK_STREAM)
            # 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)
            # TODO set_tlsext_host_name(name) for SNI?
            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.connect((self.server, 853)) 
            self.session.do_handshake()
            self.cert = self.session.get_peer_certificate()
            if not insecure:
                valid = validate_hostname(server, self.cert)
                if not valid:
                    error("Certificate error: \"%s\" is not in the certificate" % server)
                # TODO validate with SPKI?
        else: # DoH
            self.curl = pycurl.Curl()
            self.url = server
            # 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)
            self.curl.setopt(pycurl.HTTPHEADER, ["Content-type: application/dns-message"])
            if self.verbose:
                self.curl.setopt(self.curl.VERBOSE, True)
            if self.insecure:
                self.curl.setopt(pycurl.SSL_VERIFYPEER, False)   
                self.curl.setopt(pycurl.SSL_VERIFYHOST, False)
            if self.head:
                self.curl.setopt(pycurl.NOBODY, True)
            if self.post:
                 self.curl.setopt(pycurl.POST, True)
    def __str__(self):
        return self.server
    def end(self):
        if self.dot:
            self.session.shutdown()
            self.session.close()
        else: # DoH
            self.curl.close()
            
# 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).
def do_test(connection, qname, qtype=rtype):
    message = dns.message.make_query(qname, dns.rdatatype.from_text(qtype))
    if connection.dot:
        # TODO Check what the Query ID is. Really random?
        messagew = message.to_wire() 
        length = len(messagew)
        n = connection.session.send(length.to_bytes(2, byteorder='big') + messagew)
        buf = connection.session.recv(2) 
        received = int.from_bytes(buf, byteorder='big')
        buf = connection.session.recv(received)
        response = dns.message.from_wire(buf) # TODO check the Query ID
        return (True, response)
    else: # DoH
        message.id = 0 # DoH requests that
        if connection.post:
            connection.curl.setopt(connection.curl.URL, connection.server)
            data = message.to_wire()
            connection.curl.setopt(pycurl.POSTFIELDS, data)
        else:
            dns_req = base64.urlsafe_b64encode(message.to_wire()).decode('UTF8').rstrip('=')
            connection.curl.setopt(connection.curl.URL, connection.server + ("?dns=%s" % dns_req))
        buffer = io.BytesIO()
        connection.curl.setopt(connection.curl.WRITEDATA, buffer)
        connection.curl.perform()
        rcode = connection.curl.getinfo(pycurl.RESPONSE_CODE)
        ok = True
        if rcode == 200:
            if not head:
                body = buffer.getvalue()
                try:
                    response = dns.message.from_wire(body)
                except dns.message.TrailingJunk: # Not DNS. 
                    response = "ERROR Not proper DNS data \"%s\"" % body
                    ok = False
            else:
                response = "HEAD successful"
        else:
            ok = False
            body =  buffer.getvalue()
            if len(body) == 0:
                response = "[No details]"
            else:
                response = body
            buffer.close()
        return (rcode, response)
    
# Main program
name = None
message = None
try:
    optlist, args = getopt.getopt (sys.argv[1:], "hvPker:f:d:t",
                                   ["help", "verbose", "dot", "head", "insecure", "POST", "repeat=", "file=", "delay="])
    for option, value in optlist:
        if option == "--help" or option == "-h":
            usage()
            sys.exit(0)
        elif option == "--verbose" or option == "-v":
            verbose = True
        elif option == "--head" or option == "-e":
            head = True
        elif option == "--dot" or option == "-t":
            dot = True
        elif option == "--insecure" or option == "-k":
            insecure = True
        elif option == "--POST" or option == "-P":
            post = True
        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
        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]
ok = True
conn = Connection(url, dot=dot, verbose=verbose, insecure=insecure, post=post, head=head)
start = time.time() 
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:
        line = input.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()
    (rcode, msg) = do_test(conn, name, rtype)
    if (dot and rcode) or (not dot and rcode == 200):
        print(msg)
    else:
        if dot:
            print("Error: %s" % msg, file=sys.stderr)
        else:
            print("HTTP error %i: %s" % (rcode, msg), file=sys.stderr)
        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 = ""
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:
    sys.exit(0)
else:
    sys.exit(1)