parsing of DSA keys from X.509 certificates

tlslite/constants.py

@@ -237,6 +237,11 @@ class SignatureScheme(TLSEnum):
     rsa_pss_sha384 = (8, 5)
     rsa_pss_sha512 = (8, 6)
 
+    dsa_sha1 = (2, 2)
+    dsa_sha256 = (4, 2)
+    dsa_sha384 = (5, 2)
+    dsa_sha_512 = (6, 2)
+
     @classmethod
     def toRepr(cls, value, blacklist=None):
         """Convert numeric type to name representation"""
@@ -337,6 +342,14 @@ class AlgorithmOID(TLSEnum):
             SignatureScheme.rsa_pss_rsae_sha384
     oid[bytes(a2b_hex('300b0609608648016503040203'))] = \
             SignatureScheme.rsa_pss_rsae_sha512
+    oid[bytes(a2b_hex('06072A8648CE380403'))] = \
+            SignatureScheme.dsa_sha1
+    oid[bytes(a2b_hex('0609608648016503040302'))] = \
+            SignatureScheme.dsa_sha256
+    oid[bytes(a2b_hex('0609608648016503040303'))] = \
+            SignatureScheme.dsa_sha384
+    oid[bytes(a2b_hex('0609608648016503040304'))] = \
+            SignatureScheme.dsa_sha_512
 
 
 class GroupName(TLSEnum):

tlslite/utils/__init__.py

@@ -22,6 +22,7 @@
            "python_aes",
            "python_rc4",
            "python_rsakey",
+           "python_dsakey",
            "rc4",
            "rijndael",
            "rsakey",

tlslite/utils/dsakey.py

@@ -0,0 +1,33 @@
+"""Abstract class for DSA."""
+
+class DSAKey(object):
+    """This is an abstract base class for DSA keys.
+
+    Particular implementations of DSA keys, such as
+    :py:class:`~.python_dsakey.Python_DSAKey`
+    ... more coming
+    inherit from this.
+
+    To create or parse an DSA key, don't use one of these classes
+    directly.  Instead, use the factory functions in
+    :py:class:`~tlslite.utils.keyfactory`.
+    """
+
+    def __init__(self, p, q, g, x, y):
+        raise NotImplementedError()
+
+    def __len__(self):
+        raise NotImplementedError()
+
+    def hasPrivateKey(self):
+        raise NotImplementedError()
+
+    def hashAndSign(self, data, hAlg):
+        raise NotImplementedError()
+
+    def verify(self, signature, hash_bytes):
+        raise NotImplementedError()
+
+    @staticmethod
+    def generate(L, N):
+        raise NotImplementedError()

tlslite/utils/keyfactory.py

@@ -8,6 +8,7 @@
 from .rsakey import RSAKey
 from .python_rsakey import Python_RSAKey
 from .python_ecdsakey import Python_ECDSAKey
+from .python_dsakey import Python_DSAKey
 from tlslite.utils import cryptomath
 
 if cryptomath.m2cryptoLoaded:
@@ -233,3 +234,28 @@ def _create_public_ecdsa_key(point_x, point_y, curve_name,
         if impl == "python":
             return Python_ECDSAKey(point_x, point_y, curve_name)
     raise ValueError("No acceptable implementation")
+
+def _create_public_dsa_key(p, q, g, y,
+                           implementations=("python",)):
+    """
+    Convert public key parameters into concrete implementation of verifier.
+
+    The public key in DSA consists of four integers.
+
+    :type p: int
+    :param p: domain parameter, prime num defining Gaolis Field
+    :type q: int
+    :param q: domain parameter, prime factor of p-1
+    :type g: int
+    :param g: domain parameter, generator of q-order cyclic group GP(p)
+    :type y: int
+    :param y: public key
+    :type implementations: iterable of str
+    :param implementations: list of implementations that can be used as the
+        concrete implementation of the verifying key (only 'python' is
+        supported currently)
+    """
+    for impl in implementations:
+        if impl == "python":
+            return Python_DSAKey(p=p, q=q, g=g, y=y)
+    raise ValueError("No acceptable implementation")

tlslite/utils/python_dsakey.py

@@ -0,0 +1,99 @@
+from ecdsa.der import encode_sequence, encode_integer,  \
+remove_sequence, remove_integer
+
+from .cryptomath import getRandomNumber, getRandomPrime,    \
+powMod, numBits, bytesToNumber, invMod, secureHash
+
+from .dsakey import DSAKey
+
+class Python_DSAKey(DSAKey):
+
+    def __init__(self, p=0, q=0, g=0, x=0, y=0):
+        self.p = p
+        self.q = q
+        self.g = g
+        self.private_key = x
+        self.public_key = y
+        self.key_type = "dsa"
+
+    def __len__(self):
+        return numBits(self.p)
+
+    def hasPrivateKey(self):
+        return bool(self.private_key)
+
+    @staticmethod
+    def generate(L, N):
+        assert (L, N) in [(1024, 160), (2048, 224), (2048, 256), (3072, 256)]
+        key = Python_DSAKey()
+        (q, p) = Python_DSAKey.generate_qp(L, N)
+
+        index = getRandomNumber(1, (p-1))
+        g = powMod(index, int((p-1)/q), p)
+        x = getRandomNumber(1, q-1)
+        y = powMod(g, x, p)
+        key.q = q
+        key.p = p
+        key.g = g
+        key.private_key = x
+        key.public_key = y
+        return key
+
+    @staticmethod
+    def generate_qp(L, N):
+        assert (L, N) in [(1024, 160), (2048, 224), (2048, 256), (3072, 256)]
+
+        # TODO: optimize, docstring, implement Shawe-Taylor Random_Prime
+        q = int(getRandomPrime(N))
+        while True:
+            p = int(getRandomPrime(L))
+            if (p-1) % q:
+                break
+        return (q, p)
+
+    def hashAndSign(self, data, hAlg="sha1"):
+        # TODO: add assert for hash size < (q-1) constrain, docstring
+        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))
+
+        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))
+
+        # get r, s keys
+        if not signature:
+            return False
+        body, rest = remove_sequence(signature)
+        if rest:
+            return False
+        r, rest = remove_integer(body)
+        s, rest = remove_integer(rest)
+        if rest:
+            return False
+
+        # 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

tlslite/utils/python_key.py

@@ -2,6 +2,7 @@
 
 from .python_rsakey import Python_RSAKey
 from .python_ecdsakey import Python_ECDSAKey
+from .python_dsakey import Python_DSAKey
 from .pem import dePem, pemSniff
 from .asn1parser import ASN1Parser
 from .cryptomath import bytesToNumber
@@ -25,7 +26,10 @@ def parsePEM(s, passwordCallback=None):
             return Python_Key._parse_pkcs8(bytes)
         elif pemSniff(s, "RSA PRIVATE KEY"):
             bytes = dePem(s, "RSA PRIVATE KEY")
-            return Python_Key._parse_ssleay(bytes)
+            return Python_Key._parse_ssleay(bytes, "rsa")
+        elif pemSniff(s, "DSA PRIVATE KEY"):
+            bytes = dePem(s, "DSA PRIVATE KEY")
+            return Python_Key._parse_dsa_ssleay(bytes)
         elif pemSniff(s, "EC PRIVATE KEY"):
             bytes = dePem(s, "EC PRIVATE KEY")
             return Python_Key._parse_ecc_ssleay(bytes)
@@ -51,6 +55,8 @@ def _parse_pkcs8(bytes):
             key_type = "rsa"
         elif list(oid.value) == [42, 134, 72, 134, 247, 13, 1, 1, 10]:
             key_type = "rsa-pss"
+        elif list(oid.value) == [42, 134, 72, 206, 56, 4, 1]:
+            key_type = "dsa"
         elif list(oid.value) == [42, 134, 72, 206, 61, 2, 1]:
             key_type = "ecdsa"
         else:
@@ -64,6 +70,12 @@ def _parse_pkcs8(bytes):
             parameters = alg_ident.getChild(1)
             if parameters.value != bytearray(0):
                 raise SyntaxError("RSA parameters are not NULL")
+        if key_type == "dsa":
+            if seq_len != 2:
+                raise SyntaxError("Invalid encoding of algorithm identifier")
+            parameters = alg_ident.getChild(1)
+            if parameters.value == bytearray(0):
+                parameters = None
         elif key_type == "ecdsa":
             if seq_len != 2:
                 raise SyntaxError("Invalid encoding of algorithm identifier")
@@ -91,20 +103,32 @@ def _parse_pkcs8(bytes):
         if key_type == "ecdsa":
             return Python_Key._parse_ecdsa_private_key(private_key_parser,
                                                        curve)
+        elif key_type == "dsa":
+            return Python_Key._parse_dsa_private_key(private_key_parser, parameters)
         else:
             return Python_Key._parse_asn1_private_key(private_key_parser,
                                                       key_type)
 
     @staticmethod
-    def _parse_ssleay(data):
+    def _parse_ssleay(data, key_type="rsa"):
         """
         Parse binary structure of the old SSLeay file format used by OpenSSL.
 
         For RSA keys.
         """
         private_key_parser = ASN1Parser(data)
         # "rsa" type as old format doesn't support rsa-pss parameters
-        return Python_Key._parse_asn1_private_key(private_key_parser, "rsa")
+        return Python_Key._parse_asn1_private_key(private_key_parser, key_type)
+
+    @staticmethod
+    def _parse_dsa_ssleay(data):
+        """
+        Parse binary structure of the old SSLeay file format used by OpenSSL.
+
+        For DSA keys.
+        """
+        private_key_parser = ASN1Parser(data)
+        return Python_Key._parse_dsa_private_key(private_key_parser)
 
     @staticmethod
     def _parse_ecc_ssleay(data):
@@ -167,3 +191,20 @@ def _parse_asn1_private_key(private_key_parser, key_type):
         qInv = bytesToNumber(private_key_parser.getChild(8).value)
         return Python_RSAKey(n, e, d, p, q, dP, dQ, qInv, key_type)
 
+
+    @staticmethod
+    def _parse_dsa_private_key(private_key_parser, domain_parameters = None):
+        if domain_parameters:
+            p = bytesToNumber(domain_parameters.getChild(0).value)
+            q = bytesToNumber(domain_parameters.getChild(1).value)
+            g = bytesToNumber(domain_parameters.getChild(2).value)
+            x = bytesToNumber(private_key_parser.value)
+            return Python_DSAKey(p, q, g, x)
+        else:
+            p = bytesToNumber(private_key_parser.getChild(1).value)
+            q = bytesToNumber(private_key_parser.getChild(2).value)
+            g = bytesToNumber(private_key_parser.getChild(3).value)
+            y = bytesToNumber(private_key_parser.getChild(4).value)
+            x = bytesToNumber(private_key_parser.getChild(5).value)
+            return Python_DSAKey(p, q, g, x, y)
+

tlslite/x509.py

@@ -1,4 +1,4 @@
-# Authors: 
+# Authors:
 #   Trevor Perrin
 #   Google - parsing subject field
 #
@@ -10,7 +10,8 @@
 
 from .utils.asn1parser import ASN1Parser
 from .utils.cryptomath import *
-from .utils.keyfactory import _createPublicRSAKey, _create_public_ecdsa_key
+from .utils.keyfactory import _createPublicRSAKey, _create_public_ecdsa_key,    \
+_create_public_dsa_key
 from .utils.pem import *
 from .utils.compat import compatHMAC, b2a_hex
 from .constants import AlgorithmOID, RSA_PSS_OID
@@ -135,12 +136,14 @@ def parseBinary(self, cert_bytes):
             self.certAlg = "rsa"
         elif list(alg_oid) == [42, 134, 72, 134, 247, 13, 1, 1, 10]:
             self.certAlg = "rsa-pss"
+        elif list(alg_oid) == [42, 134, 72, 206, 56, 4, 1]:
+            self.certAlg = "dsa"
         elif list(alg_oid) == [42, 134, 72, 206, 61, 2, 1]:
             self.certAlg = "ecdsa"
         else:
             raise SyntaxError("Unrecognized AlgorithmIdentifier")
 
-        # for RSA the parameters of AlgorithmIdentifier should be a NULL
+        # for RSA the parameters of AlgorithmIdentifier shuld be a NULL
         if self.certAlg == "rsa":
             if alg_identifier_len != 2:
                 raise SyntaxError("Missing parameters in AlgorithmIdentifier")
@@ -152,6 +155,9 @@ def parseBinary(self, cert_bytes):
             self._ecdsa_pubkey_parsing(
                 tbs_certificate.getChildBytes(subject_public_key_info_index))
             return
+        elif self.certAlg == "dsa":
+            self._dsa_pubkey_parsing(subject_public_key_info)
+            return
         else:  # rsa-pss
             pass  # ignore parameters, if any - don't apply key restrictions
 
@@ -210,6 +216,31 @@ def _ecdsa_pubkey_parsing(self, subject_public_key_info):
         curve_name = public_key.curve.name
         self.publicKey = _create_public_ecdsa_key(x, y, curve_name)
 
+    def _dsa_pubkey_parsing(self, subject_public_key_info):
+
+        global_parameters = (subject_public_key_info.getChild(0)).getChild(1)
+        # Get the subjectPublicKey
+        public_key = subject_public_key_info.getChild(1)
+
+        # Adjust for BIT STRING encapsulation and get hex value
+        if public_key.value[0]:
+            raise SyntaxError()
+        y = public_key.value[3:]
+
+        # Get the {A, p, q}
+        p = global_parameters.getChild(0)
+        q = global_parameters.getChild(1)
+        g = global_parameters.getChild(2)
+
+        # Decode them into numbers
+        y = bytesToNumber(y)
+        p = bytesToNumber(p.value)
+        q = bytesToNumber(q.value)
+        g = bytesToNumber(g.value)
+
+        # Create a public key instance
+        self.publicKey = _create_public_dsa_key(p, q, g, y)
+
     def getFingerprint(self):
         """
         Get the hex-encoded fingerprint of this certificate.