Merge pull request tlsfuzzer#974 from tlsfuzzer/dsa-sig-flex

dsa-sig-flexibility: verify that DSA can be used with all the different hashes

.github/workflows/ci.yml

@@ -199,6 +199,9 @@ jobs:
           export PYTHONPATH=`echo /usr/share/python-wheels/pip-*py2*.whl`
           sudo --preserve-env=PYTHONPATH python -m pip install --upgrade pip setuptools wheel
           sudo chown -R $USER /usr/local/lib/python2.7
+          # workaround https://bugs.launchpad.net/ubuntu/+source/python2.7/+bug/2089136
+          wget https://bugs.launchpad.net/ubuntu/+source/python2.7/+bug/2089136/+attachment/5838791/+files/tarfile.py
+          sudo mv tarfile.py /usr/lib/python2.7/tarfile.py
       - name: Display Python version
         run: python -c "import sys; print(sys.version)"
       - name: Display installed python package versions

scripts/test-dsa-sig-flexibility.py

@@ -0,0 +1,344 @@
+# Author: Alicja Kario, (c) 2024
+# Released under Gnu GPL v2.0, see LICENSE file for details
+
+from __future__ import print_function
+import traceback
+import sys
+import getopt
+from itertools import chain
+from random import sample
+
+from tlsfuzzer.runner import Runner
+from tlsfuzzer.messages import Connect, ClientHelloGenerator, \
+        ClientKeyExchangeGenerator, ChangeCipherSpecGenerator, \
+        FinishedGenerator, ApplicationDataGenerator, AlertGenerator
+from tlsfuzzer.expect import ExpectServerHello, ExpectCertificate, \
+        ExpectServerHelloDone, ExpectChangeCipherSpec, ExpectFinished, \
+        ExpectAlert, ExpectApplicationData, ExpectClose, \
+        ExpectServerKeyExchange
+
+
+from tlslite.constants import CipherSuite, AlertLevel, AlertDescription, \
+        GroupName, ExtensionType, HashAlgorithm, SignatureAlgorithm
+from tlslite.extensions import SupportedGroupsExtension, \
+        SignatureAlgorithmsExtension, SignatureAlgorithmsCertExtension
+from tlsfuzzer.utils.lists import natural_sort_keys
+from tlsfuzzer.helpers import SIG_ALL, AutoEmptyExtension
+
+
+version = 1
+
+
+def help_msg():
+    print("Usage: <script-name> [-h hostname] [-p port] [[probe-name] ...]")
+    print(" -h hostname    name of the host to run the test against")
+    print("                localhost by default")
+    print(" -p port        port number to use for connection, 4433 by default")
+    print(" probe-name     if present, will run only the probes with given")
+    print("                names and not all of them, e.g \"sanity\"")
+    print(" -e probe-name  exclude the probe from the list of the ones run")
+    print("                may be specified multiple times")
+    print(" -x probe-name  expect the probe to fail. When such probe passes despite being marked like this")
+    print("                it will be reported in the test summary and the whole script will fail.")
+    print("                May be specified multiple times.")
+    print(" -X message     expect the `message` substring in exception raised during")
+    print("                execution of preceding expected failure probe")
+    print("                usage: [-x probe-name] [-X exception], order is compulsory!")
+    print(" -n num         run 'num' or all(if 0) tests instead of default(all)")
+    print("                (\"sanity\" tests are always executed)")
+    print(" -C ciph        Use specified ciphersuite. Either numerical value or")
+    print("                IETF name.")
+    print(" -M | --ems     Advertise support for Extended Master Secret")
+    print(" --help         this message")
+    # already used single-letter options:
+    # -m test-large-hello.py - min extension number for fuzz testing
+    # -s signature algorithms sent by server
+    # -k client key
+    # -c client certificate
+    # -z don't expect 1/n-1 record split in TLS1.0
+    # -a override for expected alert description
+    # -l override the expected alert level
+    # -C explicit cipher for connection
+    # -T expected certificates types in CertificateRequest
+    # -b server is expected to have multiple (both) certificate types available
+    #    at the same time
+    # -t timeout to wait for messages (also count of NSTs in
+    #    test-tls13-count-tickets.py)
+    # -r perform renegotation multiple times
+    # -S signature algorithms sent by client
+    # -E additional extensions to be sent by client
+    #
+    # reserved:
+    # -x expected fail for probe (alternative to -e)
+    # -X expected failure message for probe (to be used together with -x)
+    # -i enables timing the test using the specified interface
+    # -o output directory for files related to collection of timing information
+
+
+def main():
+    host = "localhost"
+    port = 4433
+    num_limit = None
+    run_exclude = set()
+    expected_failures = {}
+    last_exp_tmp = None
+    ciphers = None
+    ems = False
+
+    argv = sys.argv[1:]
+    opts, args = getopt.getopt(argv, "h:p:e:x:X:n:C:M", ["help", "ems"])
+    for opt, arg in opts:
+        if opt == '-h':
+            host = arg
+        elif opt == '-p':
+            port = int(arg)
+        elif opt == '-e':
+            run_exclude.add(arg)
+        elif opt == '-x':
+            expected_failures[arg] = None
+            last_exp_tmp = str(arg)
+        elif opt == '-X':
+            if not last_exp_tmp:
+                raise ValueError("-x has to be specified before -X")
+            expected_failures[last_exp_tmp] = str(arg)
+        elif opt == '-n':
+            num_limit = int(arg)
+        elif opt == '-C':
+            if arg[:2] == '0x':
+                ciphers = [int(arg, 16)]
+            else:
+                try:
+                    ciphers = [getattr(CipherSuite, arg)]
+                except AttributeError:
+                    ciphers = [int(arg)]
+        elif opt == '-M' or opt == '--ems':
+            ems = True
+        elif opt == '--help':
+            help_msg()
+            sys.exit(0)
+        else:
+            raise ValueError("Unknown option: {0}".format(opt))
+
+    if args:
+        run_only = set(args)
+    else:
+        run_only = None
+
+    if ciphers:
+        if ciphers[0] not in CipherSuite.dheDsaSuites:
+            raise ValueError("Ciphersuite not supported or does not use DHE_DSS key exchange")
+    else:
+        ciphers = [CipherSuite.TLS_DHE_DSS_WITH_AES_128_CBC_SHA]
+
+
+    conversations = {}
+
+    conversation = Connect(host, port)
+    node = conversation
+    ext = {}
+    if ems:
+        ext[ExtensionType.extended_master_secret] = AutoEmptyExtension()
+    groups = [GroupName.secp256r1,
+              GroupName.ffdhe2048]
+    ext[ExtensionType.supported_groups] = SupportedGroupsExtension()\
+        .create(groups)
+    ext[ExtensionType.signature_algorithms] = \
+        SignatureAlgorithmsExtension().create(SIG_ALL)
+    ext[ExtensionType.signature_algorithms_cert] = \
+        SignatureAlgorithmsCertExtension().create(SIG_ALL)
+    node = node.add_child(ClientHelloGenerator(
+        ciphers + [CipherSuite.TLS_EMPTY_RENEGOTIATION_INFO_SCSV],
+        extensions=ext))
+    node = node.add_child(ExpectServerHello())
+    node = node.add_child(ExpectCertificate())
+    node = node.add_child(ExpectServerKeyExchange())
+    node = node.add_child(ExpectServerHelloDone())
+    node = node.add_child(ClientKeyExchangeGenerator())
+    node = node.add_child(ChangeCipherSpecGenerator())
+    node = node.add_child(FinishedGenerator())
+    node = node.add_child(ExpectChangeCipherSpec())
+    node = node.add_child(ExpectFinished())
+    node = node.add_child(ApplicationDataGenerator(
+        bytearray(b"GET / HTTP/1.0\r\n\r\n")))
+    node = node.add_child(ExpectApplicationData())
+    node = node.add_child(AlertGenerator(AlertLevel.warning,
+                                         AlertDescription.close_notify))
+    node = node.add_child(ExpectAlert())
+    node.next_sibling = ExpectClose()
+    conversations["sanity"] = conversation
+
+    for h_alg in ("sha512", "sha384", "sha256", "sha224", "sha1"):
+        conversation = Connect(host, port)
+        node = conversation
+        ext = {}
+        if ems:
+            ext[ExtensionType.extended_master_secret] = AutoEmptyExtension()
+        groups = [GroupName.secp256r1,
+                  GroupName.ffdhe2048]
+        ext[ExtensionType.supported_groups] = SupportedGroupsExtension()\
+            .create(groups)
+        sig_alg = [(getattr(HashAlgorithm, h_alg), SignatureAlgorithm.dsa)]
+        ext[ExtensionType.signature_algorithms] = \
+            SignatureAlgorithmsExtension().create(sig_alg)
+        ext[ExtensionType.signature_algorithms_cert] = \
+            SignatureAlgorithmsCertExtension().create(SIG_ALL)
+        node = node.add_child(ClientHelloGenerator(
+            ciphers + [CipherSuite.TLS_EMPTY_RENEGOTIATION_INFO_SCSV],
+            extensions=ext))
+        node = node.add_child(ExpectServerHello())
+        node = node.add_child(ExpectCertificate())
+        node = node.add_child(ExpectServerKeyExchange())
+        node = node.add_child(ExpectServerHelloDone())
+        node = node.add_child(ClientKeyExchangeGenerator())
+        node = node.add_child(ChangeCipherSpecGenerator())
+        node = node.add_child(FinishedGenerator())
+        node = node.add_child(ExpectChangeCipherSpec())
+        node = node.add_child(ExpectFinished())
+        node = node.add_child(ApplicationDataGenerator(
+            bytearray(b"GET / HTTP/1.0\r\n\r\n")))
+        node = node.add_child(ExpectApplicationData())
+        node = node.add_child(AlertGenerator(AlertLevel.warning,
+                                             AlertDescription.close_notify))
+        node = node.add_child(ExpectAlert())
+        node.next_sibling = ExpectClose()
+        conversations["{0}+DSA only".format(h_alg)] = conversation
+
+    conversation = Connect(host, port)
+    node = conversation
+    ext = {}
+    if ems:
+        ext[ExtensionType.extended_master_secret] = AutoEmptyExtension()
+    groups = [GroupName.secp256r1,
+              GroupName.ffdhe2048]
+    ext[ExtensionType.supported_groups] = SupportedGroupsExtension()\
+        .create(groups)
+    sig_alg = [(HashAlgorithm.md5, SignatureAlgorithm.dsa)]
+    ext[ExtensionType.signature_algorithms] = \
+        SignatureAlgorithmsExtension().create(sig_alg)
+    ext[ExtensionType.signature_algorithms_cert] = \
+        SignatureAlgorithmsCertExtension().create(SIG_ALL)
+    node = node.add_child(ClientHelloGenerator(
+        ciphers + [CipherSuite.TLS_EMPTY_RENEGOTIATION_INFO_SCSV],
+        extensions=ext))
+    node = node.add_child(ExpectAlert(AlertLevel.fatal, AlertDescription.handshake_failure))
+    node.add_child(ExpectClose())
+    conversations["MD5+DSA only"] = conversation
+
+    conversation = Connect(host, port)
+    node = conversation
+    ext = {}
+    if ems:
+        ext[ExtensionType.extended_master_secret] = AutoEmptyExtension()
+    groups = [GroupName.secp256r1,
+              GroupName.ffdhe2048]
+    ext[ExtensionType.supported_groups] = SupportedGroupsExtension()\
+        .create(groups)
+    node = node.add_child(ClientHelloGenerator(
+        ciphers + [CipherSuite.TLS_EMPTY_RENEGOTIATION_INFO_SCSV],
+        extensions=ext))
+    node = node.add_child(ExpectServerHello())
+    node = node.add_child(ExpectCertificate())
+    node = node.add_child(ExpectServerKeyExchange())
+    node = node.add_child(ExpectServerHelloDone())
+    node = node.add_child(ClientKeyExchangeGenerator())
+    node = node.add_child(ChangeCipherSpecGenerator())
+    node = node.add_child(FinishedGenerator())
+    node = node.add_child(ExpectChangeCipherSpec())
+    node = node.add_child(ExpectFinished())
+    node = node.add_child(ApplicationDataGenerator(
+        bytearray(b"GET / HTTP/1.0\r\n\r\n")))
+    node = node.add_child(ExpectApplicationData())
+    node = node.add_child(AlertGenerator(AlertLevel.warning,
+                                         AlertDescription.close_notify))
+    node = node.add_child(ExpectAlert())
+    node.next_sibling = ExpectClose()
+    conversations["SHA1+DSA implicit".format(h_alg)] = conversation
+
+    # run the conversation
+    good = 0
+    bad = 0
+    xfail = 0
+    xpass = 0
+    failed = []
+    xpassed = []
+    if not num_limit:
+        num_limit = len(conversations)
+
+    # make sure that sanity test is run first and last
+    # to verify that server was running and kept running throughout
+    sanity_tests = [('sanity', conversations['sanity'])]
+    if run_only:
+        if num_limit > len(run_only):
+            num_limit = len(run_only)
+        regular_tests = [(k, v) for k, v in conversations.items() if k in run_only]
+    else:
+        regular_tests = [(k, v) for k, v in conversations.items() if
+                         (k != 'sanity') and k not in run_exclude]
+    sampled_tests = sample(regular_tests, min(num_limit, len(regular_tests)))
+    ordered_tests = chain(sanity_tests, sampled_tests, sanity_tests)
+
+    for c_name, c_test in ordered_tests:
+        print("{0} ...".format(c_name))
+
+        runner = Runner(c_test)
+
+        res = True
+        exception = None
+        try:
+            runner.run()
+        except Exception as exp:
+            exception = exp
+            print("Error while processing")
+            print(traceback.format_exc())
+            res = False
+
+        if c_name in expected_failures:
+            if res:
+                xpass += 1
+                xpassed.append(c_name)
+                print("XPASS-expected failure but test passed\n")
+            else:
+                if expected_failures[c_name] is not None and  \
+                    expected_failures[c_name] not in str(exception):
+                        bad += 1
+                        failed.append(c_name)
+                        print("Expected error message: {0}\n"
+                            .format(expected_failures[c_name]))
+                else:
+                    xfail += 1
+                    print("OK-expected failure\n")
+        else:
+            if res:
+                good += 1
+                print("OK\n")
+            else:
+                bad += 1
+                failed.append(c_name)
+
+    print("Basic conversation script; check basic communication with typical")
+    print("cipher, TLS 1.2 or earlier and RSA key exchange (or (EC)DHE if")
+    print("-d option is used)\n")
+
+    print("Test end")
+    print(20 * '=')
+    print("version: {0}".format(version))
+    print(20 * '=')
+    print("TOTAL: {0}".format(len(sampled_tests) + 2*len(sanity_tests)))
+    print("SKIP: {0}".format(len(run_exclude.intersection(conversations.keys()))))
+    print("PASS: {0}".format(good))
+    print("XFAIL: {0}".format(xfail))
+    print("FAIL: {0}".format(bad))
+    print("XPASS: {0}".format(xpass))
+    print(20 * '=')
+    sort = sorted(xpassed ,key=natural_sort_keys)
+    if len(sort):
+        print("XPASSED:\n\t{0}".format('\n\t'.join(repr(i) for i in sort)))
+    sort = sorted(failed, key=natural_sort_keys)
+    if len(sort):
+        print("FAILED:\n\t{0}".format('\n\t'.join(repr(i) for i in sort)))
+
+    if bad or xpass:
+        sys.exit(1)
+
+if __name__ == "__main__":
+    main()

tests/tlslite-ng-py3.8.json

@@ -642,6 +642,17 @@
           "arguments": ["-p", "4434", "-n", "0"]}
      ]
     },
+    {"server_command": ["{python}", "-u", "{command}", "server",
+                "-k", "tests/serverDSAKey.pem",
+                "-c", "tests/serverDSACert.pem",
+                "localhost:4433"],
+     "server_hostname": "localhost",
+     "server_port": 4433,
+     "environment": {"PYTHONPATH": "."},
+     "tests" : [
+         {"name": "test-dsa-sig-flexibility.py"}
+     ]
+    },
     {"server_command": ["{python}", "-u", "{command}", "server",
                 "--reqcert",
                 "-k", "tests/serverDSAKey.pem",

tests/tlslite-ng-random-subset-py3.8.json

@@ -638,6 +638,17 @@
           "arguments": ["-p", "4434", "-n", "200"]}
      ]
     },
+    {"server_command": ["{python}", "-u", "{command}", "server",
+                "-k", "tests/serverDSAKey.pem",
+                "-c", "tests/serverDSACert.pem",
+                "localhost:4433"],
+     "server_hostname": "localhost",
+     "server_port": 4433,
+     "environment": {"PYTHONPATH": "."},
+     "tests" : [
+         {"name": "test-dsa-sig-flexibility.py"}
+     ]
+    },
     {"server_command": ["{python}", "-u", "{command}", "server",
                 "--reqcert",
                 "-k", "tests/serverDSAKey.pem",