Skip to content

Commit

Permalink
verifying DSA signatures in SKE
Browse files Browse the repository at this point in the history
  • Loading branch information
@FrantisekKrenzelok
FrantisekKrenzelok committed Oct 23, 2020
1 parent 9951ec1 commit bcf080b
Show file tree
Hide file tree
Showing 7 changed files with 439 additions and 51 deletions.
15 changes: 13 additions & 2 deletions tlslite/constants.py
View file
Expand Up @@ -304,7 +304,7 @@ def getHash(scheme):
kType, _, hName = vals
else:
kType, _, _, hName = vals
assert kType in ('rsa', 'ecdsa')
assert kType in ('rsa', 'ecdsa', 'dsa')
return hName


Expand Down Expand Up @@ -958,6 +958,7 @@ class CipherSuite:
tripleDESSuites.append(TLS_DHE_DSS_WITH_3DES_EDE_CBC_SHA) # unsupported
tripleDESSuites.append(TLS_SRP_SHA_DSS_WITH_3DES_EDE_CBC_SHA) # unsupp


#: AES-128 CBC ciphers
aes128Suites = []
aes128Suites.append(TLS_SRP_SHA_WITH_AES_128_CBC_SHA)
Expand Down Expand Up @@ -1452,6 +1453,16 @@ def getEcdheCertSuites(cls, settings, version=None):
ecdheEcdsaSuites.append(TLS_ECDHE_ECDSA_WITH_RC4_128_SHA)
ecdheEcdsaSuites.append(TLS_ECDHE_ECDSA_WITH_NULL_SHA)

#: DHE key exchange, DSA authentication
dheDsaSuites = []
dheDsaSuites.append(TLS_DHE_DSS_WITH_3DES_EDE_CBC_SHA)
dheDsaSuites.append(TLS_DHE_DSS_WITH_AES_128_CBC_SHA)
dheDsaSuites.append(TLS_DHE_DSS_WITH_AES_256_CBC_SHA)
dheDsaSuites.append(TLS_DHE_DSS_WITH_AES_128_CBC_SHA256)
dheDsaSuites.append(TLS_DHE_DSS_WITH_AES_256_CBC_SHA256)
dheDsaSuites.append(TLS_DHE_DSS_WITH_AES_128_GCM_SHA256)
dheDsaSuites.append(TLS_DHE_DSS_WITH_AES_256_GCM_SHA384)

@classmethod
def getEcdsaSuites(cls, settings, version=None):
"""Provide ECDSA authenticated ciphersuites matching settings"""
Expand All @@ -1474,7 +1485,7 @@ def getAnonSuites(cls, settings, version=None):
"""Provide anonymous DH ciphersuites matching settings"""
return cls._filterSuites(CipherSuite.anonSuites, settings, version)

dhAllSuites = dheCertSuites + anonSuites
dhAllSuites = dheCertSuites + anonSuites + dheDsaSuites

#: anon ECDHE key exchange
ecdhAnonSuites = []
Expand Down
21 changes: 21 additions & 0 deletions tlslite/keyexchange.py
View file
Expand Up @@ -183,6 +183,19 @@ def _tls12_verify_ecdsa_SKE(serverKeyExchange, publicKey, clientRandom,
raise TLSDecryptionFailed("Server Key Exchange signature "
"invalid")

@staticmethod
def _tls12_verify_dsa_SKE(serverKeyExchange, publicKey, clientRandom,
serverRandom, validSigAlgs):
hashName = HashAlgorithm.toRepr(serverKeyExchange.hashAlg)
if not hashName:
raise TLSIllegalParameterException("Unknown hash algorithm")

hashBytes = serverKeyExchange.hash(clientRandom, serverRandom)

if not publicKey.verify(serverKeyExchange.signature, hashBytes):
raise TLSDecryptionFailed("Server Key Exchange signature "
"invalid")

@staticmethod
def _tls12_verify_SKE(serverKeyExchange, publicKey, clientRandom,
serverRandom, validSigAlgs):
Expand All @@ -198,6 +211,14 @@ def _tls12_verify_SKE(serverKeyExchange, publicKey, clientRandom,
clientRandom,
serverRandom,
validSigAlgs)

elif serverKeyExchange.signAlg == SignatureAlgorithm.dsa:
return KeyExchange._tls12_verify_dsa_SKE(serverKeyExchange,
publicKey,
clientRandom,
serverRandom,
validSigAlgs)

schemeID = (serverKeyExchange.hashAlg,
serverKeyExchange.signAlg)
scheme = SignatureScheme.toRepr(schemeID)
Expand Down
11 changes: 7 additions & 4 deletions tlslite/messages.py
View file
Expand Up @@ -1517,7 +1517,8 @@ def parse(self, parser):
raise AssertionError()

if self.cipherSuite in CipherSuite.certAllSuites or\
self.cipherSuite in CipherSuite.ecdheEcdsaSuites:
self.cipherSuite in CipherSuite.ecdheEcdsaSuites or\
self.cipherSuite in CipherSuite.dheDsaSuites:
if self.version == (3, 3):
self.hashAlg = parser.get(1)
self.signAlg = parser.get(1)
Expand Down Expand Up @@ -1566,7 +1567,8 @@ def write(self):
writer = Writer()
writer.bytes += self.writeParams()
if self.cipherSuite in CipherSuite.certAllSuites or \
self.cipherSuite in CipherSuite.ecdheEcdsaSuites:
self.cipherSuite in CipherSuite.ecdheEcdsaSuites or \
self.cipherSuite in CipherSuite.dheDsaSuites:
if self.version >= (3, 3):
assert self.hashAlg != 0 and self.signAlg != 0
writer.add(self.hashAlg, 1)
Expand All @@ -1591,9 +1593,10 @@ def hash(self, clientRandom, serverRandom):
else:
hashAlg = SignatureScheme.getHash(sigScheme)
return secureHash(bytesToHash, hashAlg)
# ECDSA ciphers in TLS 1.1 and earlier sign the messages using
# DSA and ECDSA ciphers in TLS 1.1 and earlier sign the messages using
# SHA-1 only
if self.cipherSuite in CipherSuite.ecdheEcdsaSuites:
if self.cipherSuite in CipherSuite.ecdheEcdsaSuites or\
self.cipherSuite in CipherSuite.dheDsaSuites:
return SHA1(bytesToHash)
return MD5(bytesToHash) + SHA1(bytesToHash)

Expand Down
35 changes: 17 additions & 18 deletions tlslite/utils/python_dsakey.py
View file
@@ -1,5 +1,4 @@
# Author: Frantisek Krenzelok

"""Pure-Python RSA implementation."""
from ecdsa.der import encode_sequence, encode_integer, \
remove_sequence, remove_integer
Expand All @@ -14,6 +13,7 @@
from gmpy import mpz

from .dsakey import DSAKey
import math

class Python_DSAKey(DSAKey):
"""
Expand Down Expand Up @@ -78,29 +78,21 @@ def generate_qp(L, N):
return (q, p)

def hashAndSign(self, data, hAlg="sha1"):
digest = bytesToNumber(secureHash(bytearray(data), hAlg))
digest_size = numBits(digest)

# extract min(|hAlg|, N) left bits of digest
N = numBits(self.q)
if N < digest_size:
digest &= ~(~0 << (digest_size - N))
hashData = (secureHash(bytearray(data), hAlg))
N = int(numBits(self.q) / 8)
hashData = hashData[:N]
digest = bytesToNumber(hashData)

k = getRandomNumber(1, (self.q-1))
r = powMod(self.g, k, self.p) % self.q
s = invMod(k, self.q) * (digest + self.private_key * r) % self.q

return encode_sequence(encode_integer(r), encode_integer(s))

def hashAndVerify(self, signature, data, hAlg="sha1"):
# Get r, s components from signature
digest = bytesToNumber(secureHash(bytearray(data), hAlg))
digest_size = numBits(digest)

# extract min(|hAlg|, N) left bits of digest
N = numBits(self.q)
if N < digest_size:
digest &= ~(~0 << (digest_size - N))
def verify(self, signature, hashData):
N = int(numBits(self.q) / 8)
hashData = hashData[:N]
digest = bytesToNumber(hashData)

# get r, s keys
if not signature:
Expand All @@ -113,13 +105,20 @@ def hashAndVerify(self, signature, data, hAlg="sha1"):
if rest:
return False

if gmpyLoaded or GMPY2_LOADED:
r = mpz(r)
s = mpz(s)

# check the signature
if 0 < r < self.q and 0 < s < self.q:
w = invMod(s, self.q)
u1 = (digest * w) % self.q
u2 = (r * w) % self.q
v = ((powMod(self.g, u1, self.p) * \
powMod(self.public_key, u2, self.p)) % self.p) % self.q

return r == v
return False

def hashAndVerify(self, signature, data, hAlg="sha1"):
digest = secureHash(bytearray(data), hAlg)
return self.verify(signature, digest)
126 changes: 126 additions & 0 deletions unit_tests/test_tlslite_keyexchange.py
View file
Expand Up @@ -361,6 +361,132 @@ def test_verifyServerKeyExchange_with_damaged_signature_in_TLS1_1(self):
bytearray(32),
None)

class TestServerKeyExchangeDSA(unittest.TestCase):
@classmethod
def setUpClass(cls):

certificate = (
"-----BEGIN CERTIFICATE-----\n"
"MIIB1zCCAZOgAwIBAgIBAjALBglghkgBZQMEAwIwFTETMBEGA1UECgwKRXhhbXBs\n"
"ZSBDQTAeFw0yMDEwMTUxMzUxNDhaFw0yMTEwMTUxMzUxNDhaMCkxEzARBgNVBAoM\n"
"CmRzYSBzZXJ2ZXIxEjAQBgNVBAMMCWxvY2FsaG9zdDCBkDBoBgcqhkjOOAQBMF0C\n"
"IQD39iS3O596+YVqFqG6UfOjCIBY5BebyusVBKYwOxittwIVAOeyxSjVHwNGs7gG\n"
"hiSS9ptu7OwZAiEArSNSuMtXZiCjKeGl3a9l+4GqUpQNY/ZxGRxFKKNysooDJAAC\n"
"IQCsT1RpxKHjf3XLkjxPXs9Pg+bd4ANjGeD0kG8KvCvv+aOBhjCBgzAOBgNVHQ8B\n"
"Af8EBAMCA6gwEwYDVR0lBAwwCgYIKwYBBQUHAwEwHQYDVR0OBBYEFLMOo+D36RMZ\n"
"2HOD7nlMMAADHD6pMD0GA1UdIwQ2MDSAFOgsxs+YbaXS234a8lnTDs/Crrt9oRmk\n"
"FzAVMRMwEQYDVQQKDApFeGFtcGxlIENBggEBMAsGCWCGSAFlAwQDAgMxADAuAhUA\n"
"qzq6O701fT/xLbwQtTj8xF7bin4CFQCGkqFn+2WScdlR2+7pDY5+EzcfJQ==\n"
"-----END CERTIFICATE-----")

x509 = X509()
x509.parse(certificate)
cls.x509 = x509

def test_verify_dsa_signature_in_TLS1_1_SHA1(self):
skemsg = a2b_hex(
"0001b70080b10b8f96a080e01dde92de5eae5d54ec52c99fbcfb06a3c69a"
"6a9dca52d23b616073e28675a23d189838ef1e2ee652c013ecb4aea9061123"
"24975c3cd49b83bfaccbdd7d90c4bd7098488e9c219a73724effd6fae56447"
"38faa31a4ff55bccc0a151af5f0dc8b4bd45bf37df365c1a65e68cfda76d4d"
"a708df1fb2bc2e4a43710080a4d1cbd5c3fd34126765a442efb99905f8104d"
"d258ac507fd6406cff14266d31266fea1e5c41564b777e690f5504f2131602"
"17b4b01b886a5e91547f9e2749f4d7fbd7d3b9a92ee1909d0d2263f80a76a6"
"a24c087a091f531dbf0a0169b6a28ad662a4d18e73afa32d779d5918d08bc8"
"858f4dcef97c2a24855e6eeb22b3b2e500807380851fe7c5c2d64477633ddc"
"02fa20fa5f643890fa066f1c0f91d093bbed22b19838bb75481d8e6a329ac4"
"4f36474fd368b58c549f1b2a5f8482527fd5e66bdaf7f303100abf06a82434"
"506761396cf9fa5e50b2d403cc3015b34ad16072ca9ec3147a5f43c7895cfd"
"fd4ec01dee567017b1b54bcef8e8de5d2d25e5d1ad52002f302d02147337e4"
"506ebe94158d8b05661f92ce8824c3ab28021500d19d4c637c650d8eafaef3"
"13d832c5f206338d7a")
parser = Parser(skemsg)

ske = ServerKeyExchange(
CipherSuite.TLS_DHE_DSS_WITH_AES_128_CBC_SHA,
(3, 1))
ske.parse(parser)
client_random = a2b_hex("3866b841a109ef41b69dc87e39972250c94d5a6f9f213d"
"b501fdee8148abb56a")
server_random = a2b_hex("c921b34964e0b13293ed8c4c1b102fc7c1b2ee21ca2726"
"0b444f574e47524400")

KeyExchange.verifyServerKeyExchange(ske,
self.x509.publicKey,
client_random,
server_random,
[(HashAlgorithm.sha256,
SignatureAlgorithm.dsa)])

def test_verify_dsa_signature_in_TLS1_2_SHA1(self):
skemsg = a2b_hex(
"0001b90080b10b8f96a080e01dde92de5eae5d54ec52c99fbcfb06a3c69a"
"6a9dca52d23b616073e28675a23d189838ef1e2ee652c013ecb4aea9061123"
"24975c3cd49b83bfaccbdd7d90c4bd7098488e9c219a73724effd6fae56447"
"38faa31a4ff55bccc0a151af5f0dc8b4bd45bf37df365c1a65e68cfda76d4d"
"a708df1fb2bc2e4a43710080a4d1cbd5c3fd34126765a442efb99905f8104d"
"d258ac507fd6406cff14266d31266fea1e5c41564b777e690f5504f2131602"
"17b4b01b886a5e91547f9e2749f4d7fbd7d3b9a92ee1909d0d2263f80a76a6"
"a24c087a091f531dbf0a0169b6a28ad662a4d18e73afa32d779d5918d08bc8"
"858f4dcef97c2a24855e6eeb22b3b2e5008093e6b9d650c32a077f9eeff284"
"d986a100614a2eb7ed588b8f7808ad9bad09945c781ea81956e5192e505098"
"f8d97478946d14bfe77b5c96674ea848fce0f394ff1206fd34c9684ead202f"
"c70f66038589e55d035ce043042d41b9ab84a7feffbb456511f54a579de1e0"
"01ce39ee30c75f1b33c2db6e949ef901b4adfb8762c10202002f302d02147e"
"2f0fc391f3636c3cdfdc9c40cac26b0fd527b3021500af5a5ecf105c42937e"
"5dd0d477c71840cd6ad5c6")
parser = Parser(skemsg)

ske = ServerKeyExchange(
CipherSuite.TLS_DHE_DSS_WITH_AES_128_CBC_SHA,
(3, 3))
ske.parse(parser)
client_random = a2b_hex("eddcc21c1c52aa53af5d428d73a8eb47182ccb04d4c3be"
"4b3e4fc9079e9d6dd9")
server_random = a2b_hex("a943c57c21dc242e67465f5c2b630952a7d83d137112a4"
"621f61f668c29cd3bc")

KeyExchange.verifyServerKeyExchange(ske,
self.x509.publicKey,
client_random,
server_random,
[(HashAlgorithm.sha1,
SignatureAlgorithm.dsa)])

def test_verify_dsa_signature_in_TLS1_2_SHA256(self):
skemsg = a2b_hex(
"0001b90080b10b8f96a080e01dde92de5eae5d54ec52c99fbcfb06a3c69a6a"
"9dca52d23b616073e28675a23d189838ef1e2ee652c013ecb4aea906112324"
"975c3cd49b83bfaccbdd7d90c4bd7098488e9c219a73724effd6fae5644738"
"faa31a4ff55bccc0a151af5f0dc8b4bd45bf37df365c1a65e68cfda76d4da7"
"08df1fb2bc2e4a43710080a4d1cbd5c3fd34126765a442efb99905f8104dd2"
"58ac507fd6406cff14266d31266fea1e5c41564b777e690f5504f213160217"
"b4b01b886a5e91547f9e2749f4d7fbd7d3b9a92ee1909d0d2263f80a76a6a2"
"4c087a091f531dbf0a0169b6a28ad662a4d18e73afa32d779d5918d08bc885"
"8f4dcef97c2a24855e6eeb22b3b2e50080424880aa4d93add83c1ce00ca6ef"
"2c923e7e9c3ede6be506e5ed978aef958e150652891fd2f28c5d366b47962b"
"3fddeae3988273eff87d4d0dbe8a7945ceb7760c78f535bb173f8f78558994"
"6ed06c1b4de16c12b97f2c34a6698a238f2ad0d9126de8923e720243406371"
"ad6b382013b6bc6ed3044fbbab0b2f9d26bcce3bae0402002f302d021500b5"
"dd272c8ca096c5d5ea999ea0d369aecd492441021440bd8a5f340991b41c17"
"f9d0724f3ca8ded4f5d7")
parser = Parser(skemsg)

ske = ServerKeyExchange(
CipherSuite.TLS_DHE_DSS_WITH_AES_128_GCM_SHA256,
(3, 3))
ske.parse(parser)
client_random = a2b_hex("93b4ce37563b8c4fba7aa594b242fa13c9c14cca3c6922"
"c878f8e254ad453d5f")
server_random = a2b_hex("2b27c3c3b5be6412e5434dceb76189ca9b17368c5d609e"
"4c2e6e938af2b12609")

KeyExchange.verifyServerKeyExchange(ske,
self.x509.publicKey,
client_random,
server_random,
[(HashAlgorithm.sha256,
SignatureAlgorithm.dsa)])

class TestServerKeyExchangeP256(unittest.TestCase):
@classmethod
Expand Down
13 changes: 13 additions & 0 deletions unit_tests/test_tlslite_messages.py
View file
Expand Up @@ -2391,6 +2391,19 @@ def test_hash_with_ecdsa_in_tls1_1(self):
bytearray(b' \xa0\xc1P5\xf7K/\xednd'
b'\xbaQ\xedo\xa13Z\xa5}'))

def test_hash_with_dsa_in_tls1_2(self):
ske = ServerKeyExchange(
CipherSuite.TLS_DHE_DSS_WITH_AES_128_CBC_SHA256,
(3, 3))

ske.createDH(dh_p=31, dh_g=2, dh_Ys=16)
ske.hashAlg, ske.signAlg = SignatureScheme.dsa_sha1

hash1 = ske.hash(bytearray(32), bytearray(32))

self.assertEqual(hash1,
bytearray(b'\x4a\x95\x6f\x85\x4e\xed\x6c\x5f\x84\x17\x92\xd6\xa8\x8d\x8c\xd7\xb2\xd0\xf7\xfb'))

def test_hash_with_rsa_pss_sha256(self):
ske = ServerKeyExchange(
CipherSuite.TLS_DHE_RSA_WITH_AES_128_CBC_SHA,
Expand Down

0 comments on commit bcf080b

Please sign in to comment.