Revision dbd56c149072e656ca8d6a43a59588f3e7513da2 authored by Hubert Kario on 29 September 2021, 13:05:34 UTC, committed by GitHub on 29 September 2021, 13:05:34 UTC
ExpectAlert - add __repr__
test-tls13-eddsa-in-certificate-verify.py
# Author: Hubert Kario, (c) 2021
# Released under Gnu GPL v2.0, see LICENSE file for details
"""Test of ECDSA in CertificateVerify"""
from __future__ import print_function
import traceback
import sys
import getopt
import re
from itertools import chain, islice
from random import sample
from tlsfuzzer.runner import Runner
from tlsfuzzer.messages import Connect, ClientHelloGenerator, \
ClientKeyExchangeGenerator, ChangeCipherSpecGenerator, \
FinishedGenerator, ApplicationDataGenerator, \
CertificateGenerator, CertificateVerifyGenerator, \
AlertGenerator
from tlsfuzzer.expect import ExpectServerHello, ExpectCertificate, \
ExpectServerHelloDone, ExpectChangeCipherSpec, ExpectFinished, \
ExpectAlert, ExpectClose, ExpectCertificateRequest, \
ExpectApplicationData, ExpectEncryptedExtensions, \
ExpectCertificateVerify, ExpectNewSessionTicket
from tlsfuzzer.utils.lists import natural_sort_keys
from tlsfuzzer.helpers import key_share_ext_gen, sig_algs_to_ids, RSA_SIG_ALL,\
ECDSA_SIG_ALL, EDDSA_SIG_ALL
from tlslite.extensions import SignatureAlgorithmsExtension, \
SignatureAlgorithmsCertExtension, ClientKeyShareExtension, \
SupportedVersionsExtension, SupportedGroupsExtension
from tlslite.constants import CipherSuite, AlertLevel, AlertDescription, \
HashAlgorithm, SignatureAlgorithm, ExtensionType, SignatureScheme, \
GroupName
from tlslite.utils import tlshashlib
from tlslite.utils.cryptomath import numBytes
from tlslite.utils.keyfactory import parsePEMKey
from tlslite.x509 import X509
from tlslite.x509certchain import X509CertChain
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(10)")
print(" (excluding \"sanity\" tests)")
print(" -s sigalgs hash and signature algorithm pairs that the server")
print(" is expected to support. Either pairs of algorithms")
print(" (\"sha1+ecdsa\"), pairs of identifiers (\"1+1\")")
print(" or the TLS 1.3 names (\"rsa_pss_rsae_sha256\").")
print(" Multiple values separated by spaces.")
print(" --hash-order the order in which hashes are preferred in some")
print(" tests, can be used to test different combinations")
print(" -k keyfile file with private key of client")
print(" -c certfile file with the certificate of client")
print(" --help this message")
def hashes_to_list(arg):
hlist = []
for h in arg.split():
name = None
hnum = getattr(HashAlgorithm, h, None)
if not hnum:
try:
hnum = int(h)
except ValueError:
pass
if hnum:
name = HashAlgorithm.toRepr(hnum)
if not name:
raise ValueError("Invalid Hash Id or Name {0}".format(h))
hlist.append(name)
if len(hlist) < 2:
raise ValueError(
"The ordered list of hashes must contain at least 2 elements," +
" found {0}".format(len(hlist)))
return hlist
def main():
"""Check that server properly rejects malformed signatures in TLS 1.3"""
hostname = "localhost"
port = 4433
num_limit = 120
run_exclude = set()
expected_failures = {}
last_exp_tmp = None
cert = None
private_key = None
# algorithms to expect from server in Certificate Request
cr_sigalgs = [SignatureScheme.ed25519,
SignatureScheme.ed448,
SignatureScheme.ecdsa_secp521r1_sha512,
SignatureScheme.ecdsa_secp384r1_sha384,
SignatureScheme.ecdsa_secp256r1_sha256,
(HashAlgorithm.sha224, SignatureAlgorithm.ecdsa),
(HashAlgorithm.sha1, SignatureAlgorithm.ecdsa),
SignatureScheme.rsa_pss_rsae_sha512,
SignatureScheme.rsa_pss_pss_sha512,
SignatureScheme.rsa_pss_rsae_sha384,
SignatureScheme.rsa_pss_pss_sha384,
SignatureScheme.rsa_pss_rsae_sha256,
SignatureScheme.rsa_pss_pss_sha256,
SignatureScheme.rsa_pkcs1_sha512,
SignatureScheme.rsa_pkcs1_sha384,
SignatureScheme.rsa_pkcs1_sha256,
SignatureScheme.rsa_pkcs1_sha224,
SignatureScheme.rsa_pkcs1_sha1]
# algorithms to advertise in ClientHello
sig_algs = [SignatureScheme.ecdsa_secp521r1_sha512,
SignatureScheme.ecdsa_secp384r1_sha384,
SignatureScheme.ecdsa_secp256r1_sha256,
SignatureScheme.rsa_pss_rsae_sha256,
SignatureScheme.rsa_pss_pss_sha256,
SignatureScheme.rsa_pss_rsae_sha384,
SignatureScheme.rsa_pss_pss_sha384,
SignatureScheme.ed25519,
SignatureScheme.ed448]
hashalgs = hashes_to_list("sha256 sha384 sha512")
argv = sys.argv[1:]
opts, args = getopt.getopt(argv, "h:p:e:x:X:n:s:k:c:", ["help", "hash-order="])
for opt, arg in opts:
if opt == '-h':
hostname = 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 == '--help':
help_msg()
sys.exit(0)
elif opt == '-s':
cr_sigalgs = sig_algs_to_ids(arg)
elif opt == '--hash-order':
hashalgs = hashes_to_list(arg)
elif opt == '-k':
text_key = open(arg, 'rb').read()
if sys.version_info[0] >= 3:
text_key = str(text_key, 'utf-8')
private_key = parsePEMKey(text_key, private=True)
elif opt == '-c':
text_cert = open(arg, 'rb').read()
if sys.version_info[0] >= 3:
text_cert = str(text_cert, 'utf-8')
cert = X509()
cert.parse(text_cert)
else:
raise ValueError("Unknown option: {0}".format(opt))
if args:
run_only = set(args)
else:
run_only = None
if not cert or not private_key:
raise Exception("A Client certificate and a private key are required")
certType = cert.certAlg
conversations = {}
conversations_long = {}
# sanity check for Client Certificates
conversation = Connect(hostname, port)
node = conversation
ciphers = [CipherSuite.TLS_AES_128_GCM_SHA256,
CipherSuite.TLS_EMPTY_RENEGOTIATION_INFO_SCSV]
ext = {}
groups = [GroupName.secp256r1]
ext[ExtensionType.key_share] = key_share_ext_gen(groups)
ext[ExtensionType.supported_versions] = \
SupportedVersionsExtension().create([(3, 4), (3, 3)])
ext[ExtensionType.supported_groups] = \
SupportedGroupsExtension().create(groups)
ext[ExtensionType.signature_algorithms] = \
SignatureAlgorithmsExtension().create(sig_algs)
ext[ExtensionType.signature_algorithms_cert] = \
SignatureAlgorithmsCertExtension().create(
ECDSA_SIG_ALL + RSA_SIG_ALL + EDDSA_SIG_ALL)
node = node.add_child(ClientHelloGenerator(ciphers, extensions=ext))
node = node.add_child(ExpectServerHello())
node = node.add_child(ExpectChangeCipherSpec())
node = node.add_child(ExpectEncryptedExtensions())
node = node.add_child(ExpectCertificateRequest())
node = node.add_child(ExpectCertificate())
node = node.add_child(ExpectCertificateVerify())
node = node.add_child(ExpectFinished())
node = node.add_child(CertificateGenerator(X509CertChain([cert])))
node = node.add_child(CertificateVerifyGenerator(private_key))
node = node.add_child(FinishedGenerator())
node = node.add_child(ApplicationDataGenerator(
bytearray(b"GET / HTTP/1.0\r\n\r\n")))
# This message is optional and may show up 0 to many times
cycle = ExpectNewSessionTicket()
node = node.add_child(cycle)
node.add_child(cycle)
node.next_sibling = ExpectApplicationData()
node = node.next_sibling.add_child(AlertGenerator(AlertLevel.warning,
AlertDescription.close_notify))
node = node.add_child(ExpectAlert())
node.next_sibling = ExpectClose()
conversations["sanity"] = conversation
# verify the advertised hashes
conversation = Connect(hostname, port)
node = conversation
ciphers = [CipherSuite.TLS_AES_128_GCM_SHA256,
CipherSuite.TLS_EMPTY_RENEGOTIATION_INFO_SCSV]
ext = {}
groups = [GroupName.secp256r1]
ext[ExtensionType.key_share] = key_share_ext_gen(groups)
ext[ExtensionType.supported_versions] = \
SupportedVersionsExtension().create([(3, 4), (3, 3)])
ext[ExtensionType.supported_groups] = \
SupportedGroupsExtension().create(groups)
ext[ExtensionType.signature_algorithms] = \
SignatureAlgorithmsExtension().create(sig_algs)
ext[ExtensionType.signature_algorithms_cert] = \
SignatureAlgorithmsCertExtension().create(
ECDSA_SIG_ALL + RSA_SIG_ALL + EDDSA_SIG_ALL)
node = node.add_child(ClientHelloGenerator(ciphers, extensions=ext))
node = node.add_child(ExpectServerHello())
node = node.add_child(ExpectChangeCipherSpec())
node = node.add_child(ExpectEncryptedExtensions())
node = node.add_child(ExpectCertificateRequest(cr_sigalgs))
node = node.add_child(ExpectCertificate())
node = node.add_child(ExpectCertificateVerify())
node = node.add_child(ExpectFinished())
node = node.add_child(CertificateGenerator(X509CertChain([cert])))
node = node.add_child(CertificateVerifyGenerator(private_key))
node = node.add_child(FinishedGenerator())
node = node.add_child(ApplicationDataGenerator(
bytearray(b"GET / HTTP/1.0\r\n\r\n")))
# This message is optional and may show up 0 to many times
cycle = ExpectNewSessionTicket()
node = node.add_child(cycle)
node.add_child(cycle)
node.next_sibling = ExpectApplicationData()
node = node.next_sibling.add_child(AlertGenerator(AlertLevel.warning,
AlertDescription.close_notify))
node = node.add_child(ExpectAlert())
node.next_sibling = ExpectClose()
conversations["check sigalgs in cert request"] = conversation
for sigalg in ECDSA_SIG_ALL:
real_sig = getattr(SignatureScheme, private_key.key_type.lower())
conversation = Connect(hostname, port)
node = conversation
ciphers = [CipherSuite.TLS_AES_128_GCM_SHA256,
CipherSuite.TLS_EMPTY_RENEGOTIATION_INFO_SCSV]
ext = {}
# NOTE: groups do NOT influence the signature negotiation
groups = [GroupName.secp256r1]
ext[ExtensionType.key_share] = key_share_ext_gen(groups)
ext[ExtensionType.supported_versions] = \
SupportedVersionsExtension().create([(3, 4), (3, 3)])
ext[ExtensionType.supported_groups] = \
SupportedGroupsExtension().create(groups)
ext[ExtensionType.signature_algorithms] = \
SignatureAlgorithmsExtension().create(sig_algs)
ext[ExtensionType.signature_algorithms_cert] = \
SignatureAlgorithmsCertExtension().create(
ECDSA_SIG_ALL + RSA_SIG_ALL + EDDSA_SIG_ALL)
node = node.add_child(ClientHelloGenerator(ciphers, extensions=ext))
node = node.add_child(ExpectServerHello())
node = node.add_child(ExpectChangeCipherSpec())
node = node.add_child(ExpectEncryptedExtensions())
node = node.add_child(ExpectCertificateRequest())
node = node.add_child(ExpectCertificate())
node = node.add_child(ExpectCertificateVerify())
node = node.add_child(ExpectFinished())
node = node.add_child(CertificateGenerator(X509CertChain([cert])))
# force sigalg
node = node.add_child(CertificateVerifyGenerator(private_key, msg_alg=
sigalg, sig_alg=real_sig))
node = node.add_child(FinishedGenerator())
node = node.add_child(ExpectAlert(
AlertLevel.fatal, AlertDescription.illegal_parameter))
node.add_child(ExpectClose())
result = "is refused"
name = SignatureScheme.toRepr(sigalg)
if not name:
name = "{0}+{1}".format(HashAlgorithm.toStr(sigalg[0]),
SignatureAlgorithm.toStr(sigalg[1]))
conversations["check {0} signature {1}".format(
name, result)] = conversation
# verify that an ECDSA signature with mismatched message hash fails
if private_key.key_type == "Ed25519":
sig_alg = SignatureScheme.ed25519
msg_alg = SignatureScheme.ed448
else:
assert private_key.key_type == "Ed448"
sig_alg = SignatureScheme.ed448
msg_alg = SignatureScheme.ed25519
conversation = Connect(hostname, port)
node = conversation
ciphers = [CipherSuite.TLS_AES_128_GCM_SHA256,
CipherSuite.TLS_EMPTY_RENEGOTIATION_INFO_SCSV]
ext = {}
groups = [GroupName.secp256r1]
ext[ExtensionType.key_share] = key_share_ext_gen(groups)
ext[ExtensionType.supported_versions] = \
SupportedVersionsExtension().create([(3, 4), (3, 3)])
ext[ExtensionType.supported_groups] = \
SupportedGroupsExtension().create(groups)
ext[ExtensionType.signature_algorithms] = \
SignatureAlgorithmsExtension().create(sig_algs)
ext[ExtensionType.signature_algorithms_cert] = \
SignatureAlgorithmsCertExtension().create(
ECDSA_SIG_ALL + RSA_SIG_ALL + EDDSA_SIG_ALL)
node = node.add_child(ClientHelloGenerator(ciphers, extensions=ext))
node = node.add_child(ExpectServerHello())
node = node.add_child(ExpectChangeCipherSpec())
node = node.add_child(ExpectEncryptedExtensions())
node = node.add_child(ExpectCertificateRequest())
node = node.add_child(ExpectCertificate())
node = node.add_child(ExpectCertificateVerify())
node = node.add_child(ExpectFinished())
node = node.add_child(CertificateGenerator(X509CertChain([cert])))
node = node.add_child(CertificateVerifyGenerator(
private_key, sig_alg=sig_alg, msg_alg=msg_alg))
node = node.add_child(FinishedGenerator())
node = node.add_child(ExpectAlert(
AlertLevel.fatal, AlertDescription.illegal_parameter))
node.add_child(ExpectClose())
conversations["check eddsa signature with mismatched scheme fails"] = \
conversation
# check that fuzzed signatures are rejected
if private_key.key_type == "Ed25519":
siglen = 64
else:
assert private_key.key_type == "Ed448"
siglen = 114
for pos in range(siglen):
for xor in [0x01, 0x02, 0x04, 0x08, 0x10, 0x20, 0x40, 0x80]:
conversation = Connect(hostname, port)
node = conversation
ciphers = [CipherSuite.TLS_AES_128_GCM_SHA256,
CipherSuite.TLS_EMPTY_RENEGOTIATION_INFO_SCSV]
ext = {}
groups = [GroupName.secp256r1]
ext[ExtensionType.key_share] = key_share_ext_gen(groups)
ext[ExtensionType.supported_versions] = \
SupportedVersionsExtension().create([(3, 4), (3, 3)])
ext[ExtensionType.supported_groups] = \
SupportedGroupsExtension().create(groups)
ext[ExtensionType.signature_algorithms] = \
SignatureAlgorithmsExtension().create(sig_algs)
ext[ExtensionType.signature_algorithms_cert] = \
SignatureAlgorithmsCertExtension().create(
ECDSA_SIG_ALL + RSA_SIG_ALL + EDDSA_SIG_ALL)
node = node.add_child(ClientHelloGenerator(
ciphers, extensions=ext))
node = node.add_child(ExpectServerHello())
node = node.add_child(ExpectChangeCipherSpec())
node = node.add_child(ExpectEncryptedExtensions())
node = node.add_child(ExpectCertificateRequest())
node = node.add_child(ExpectCertificate())
node = node.add_child(ExpectCertificateVerify())
node = node.add_child(ExpectFinished())
node = node.add_child(CertificateGenerator(X509CertChain([cert])))
node = node.add_child(
CertificateVerifyGenerator(private_key,
padding_xors={pos:xor}))
node = node.add_child(FinishedGenerator())
node = node.add_child(ExpectAlert(
AlertLevel.fatal, AlertDescription.decrypt_error))
node.add_child(ExpectClose())
conversations_long["check that fuzzed signatures are rejected." +
" Malformed {0} - xor {1} at {2}".format(
certType, hex(xor), pos)] = conversation
# check if all zero values are rejected
for name, subs in [
("All-zero R value in signature is rejected",
dict((i, 0) for i in range(siglen // 2))),
("All-zero S value in signature is rejected",
dict((i + siglen // 2, 0) for i in range(siglen // 2))),
("All-zero bytes signature is rejected",
dict((i, 0) for i in range(siglen))),
]:
conversation = Connect(hostname, port)
node = conversation
ciphers = [CipherSuite.TLS_AES_128_GCM_SHA256,
CipherSuite.TLS_EMPTY_RENEGOTIATION_INFO_SCSV]
ext = {}
groups = [GroupName.secp256r1]
ext[ExtensionType.key_share] = key_share_ext_gen(groups)
ext[ExtensionType.supported_versions] = \
SupportedVersionsExtension().create([(3, 4), (3, 3)])
ext[ExtensionType.supported_groups] = \
SupportedGroupsExtension().create(groups)
ext[ExtensionType.signature_algorithms] = \
SignatureAlgorithmsExtension().create(sig_algs)
ext[ExtensionType.signature_algorithms_cert] = \
SignatureAlgorithmsCertExtension().create(
ECDSA_SIG_ALL + RSA_SIG_ALL + EDDSA_SIG_ALL)
node = node.add_child(ClientHelloGenerator(
ciphers, extensions=ext))
node = node.add_child(ExpectServerHello())
node = node.add_child(ExpectChangeCipherSpec())
node = node.add_child(ExpectEncryptedExtensions())
node = node.add_child(ExpectCertificateRequest())
node = node.add_child(ExpectCertificate())
node = node.add_child(ExpectCertificateVerify())
node = node.add_child(ExpectFinished())
node = node.add_child(CertificateGenerator(X509CertChain([cert])))
node = node.add_child(
CertificateVerifyGenerator(private_key,
padding_subs=subs))
node = node.add_child(FinishedGenerator())
node = node.add_child(ExpectAlert(
AlertLevel.fatal, AlertDescription.decrypt_error))
node.add_child(ExpectClose())
conversations_long[name] = conversation
# empty signature field
conversation = Connect(hostname, port)
node = conversation
ciphers = [CipherSuite.TLS_AES_128_GCM_SHA256,
CipherSuite.TLS_EMPTY_RENEGOTIATION_INFO_SCSV]
ext = {}
groups = [GroupName.secp256r1]
ext[ExtensionType.key_share] = key_share_ext_gen(groups)
ext[ExtensionType.supported_versions] = \
SupportedVersionsExtension().create([(3, 4), (3, 3)])
ext[ExtensionType.supported_groups] = \
SupportedGroupsExtension().create(groups)
ext[ExtensionType.signature_algorithms] = \
SignatureAlgorithmsExtension().create(sig_algs)
ext[ExtensionType.signature_algorithms_cert] = \
SignatureAlgorithmsCertExtension().create(
ECDSA_SIG_ALL + RSA_SIG_ALL + EDDSA_SIG_ALL)
node = node.add_child(ClientHelloGenerator(
ciphers, extensions=ext))
node = node.add_child(ExpectServerHello())
node = node.add_child(ExpectChangeCipherSpec())
node = node.add_child(ExpectEncryptedExtensions())
node = node.add_child(ExpectCertificateRequest())
node = node.add_child(ExpectCertificate())
node = node.add_child(ExpectCertificateVerify())
node = node.add_child(ExpectFinished())
node = node.add_child(CertificateGenerator(X509CertChain([cert])))
node = node.add_child(
CertificateVerifyGenerator(private_key,
signature=bytearray(0)))
node = node.add_child(FinishedGenerator())
node = node.add_child(ExpectAlert(
AlertLevel.fatal, AlertDescription.decrypt_error))
node.add_child(ExpectClose())
conversations_long["empty signature field"] = conversation
# run the conversation
good = 0
bad = 0
xfail = 0
xpass = 0
failed = []
xpassed = []
if not num_limit:
num_limit = len(conversations_long)
# 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)
long_tests = [(k, v) for k, v in conversations_long.items() if k in run_only]
short_tests = [(k, v) for k, v in conversations.items() if (k != 'sanity') and k in run_only]
else:
long_tests = [(k, v) for k, v in conversations_long.items() if
k not in run_exclude]
short_tests = [(k, v) for k, v in conversations.items() if
(k != 'sanity') and k not in run_exclude]
sampled_tests = sample(long_tests, min(num_limit, len(long_tests)))
ordered_tests = chain(sanity_tests, short_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("Test to verify that server properly accepts or refuses")
print("EdDSA signatures in TLS1.3")
print("Test should be executed two times, once with Ed25519 key and")
print("once with Ed448 key (if both are supported by the server).\n")
print("Test end")
print(20 * '=')
print("version: {0}".format(version))
print(20 * '=')
print("TOTAL: {0}".format(len(sampled_tests) + len(short_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()

Computing file changes ...