DH subgroup order (q) (#3369)

* Support DH q (subgroup order)

* Change RFC5114.txt to NIST format

* Add tests for DH q

* Update docs for DH q

* Fix pep8

* Improve test covergae for DH q

* Create _dh_params_dup that copy q if DHparams_dup don't

On OpenSSL < 1.0.2 DHparams_dup don't copy q. _dh_params_dup
call DHparams_dup and if the version is smaller than 1.0.2
copy q manually

* Copy q manually on libressl

* Add to test vectors serialized RFC5114 2048 bit DH parameters with 224 bit subgroup

* Support serialization of DH with q

* Add tests for serialization of DH with q

* Support DH serialization with q only if Cryptography_HAS_EVP_PKEY_DHX is true

* Raise exception when trying to serialize DH X9.42 when not supported

* raise unsupported key type when deserilizing DH X9.42 if not supported

* pep8 fixes

* Fix test_serialization

* Add dhx_serialization_supported method to DHBacked

* document q in dh_parameters_supported

* Rename dhx_serialization_supported to dh_x942_serialization_supported

4 files changed, 243 insertions, 33 deletions

diff --git a/tests/hazmat/backends/test_multibackend.py b/tests/hazmat/backends/test_multibackend.py
index bd806731..9370387c 100644
--- a/tests/hazmat/backends/test_multibackend.py
+++ b/tests/hazmat/backends/test_multibackend.py
@@ -265,7 +265,10 @@ class DummyDHBackend(object):
     def generate_dh_private_key_and_parameters(self, generator, key_size):
         pass
 
-    def dh_parameters_supported(self, p, g):
+    def dh_parameters_supported(self, p, g, q=None):
+        pass
+
+    def dh_x942_serialization_supported(self):
         pass
 
 
@@ -638,6 +641,8 @@ class TestMultiBackend(object):
             backend.generate_dh_private_key_and_parameters(2, 512)
         with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_DIFFIE_HELLMAN):
             backend.dh_parameters_supported(2, 3)
+        with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_DIFFIE_HELLMAN):
+            backend.dh_x942_serialization_supported()
 
     def test_scrypt(self):
         backend = MultiBackend([DummyScryptBackend()])
diff --git a/tests/hazmat/backends/test_openssl.py b/tests/hazmat/backends/test_openssl.py
index ff8a42ef..f561c793 100644
--- a/tests/hazmat/backends/test_openssl.py
+++ b/tests/hazmat/backends/test_openssl.py
@@ -16,13 +16,13 @@ import pytest
 
 from cryptography import utils, x509
 from cryptography.exceptions import InternalError, _Reasons
-from cryptography.hazmat.backends.interfaces import RSABackend
+from cryptography.hazmat.backends.interfaces import DHBackend, RSABackend
 from cryptography.hazmat.backends.openssl.backend import (
     Backend, backend
 )
 from cryptography.hazmat.backends.openssl.ec import _sn_to_elliptic_curve
 from cryptography.hazmat.primitives import hashes, serialization
-from cryptography.hazmat.primitives.asymmetric import dsa, ec, padding
+from cryptography.hazmat.primitives.asymmetric import dh, dsa, ec, padding
 from cryptography.hazmat.primitives.ciphers import Cipher
 from cryptography.hazmat.primitives.ciphers.algorithms import AES
 from cryptography.hazmat.primitives.ciphers.modes import CBC
@@ -32,7 +32,9 @@ from ...doubles import (
     DummyAsymmetricPadding, DummyCipherAlgorithm, DummyHashAlgorithm, DummyMode
 )
 from ...test_x509 import _load_cert
-from ...utils import load_vectors_from_file, raises_unsupported_algorithm
+from ...utils import (
+    load_nist_vectors, load_vectors_from_file, raises_unsupported_algorithm
+)
 
 
 def skip_if_libre_ssl(openssl_version):
@@ -611,3 +613,78 @@ class TestGOSTCertificate(object):
             assert cert.subject.get_attributes_for_oid(
                 x509.ObjectIdentifier("1.2.643.3.131.1.1")
             )[0].value == "007710474375"
+
+
+@pytest.mark.skipif(
+    backend._lib.Cryptography_HAS_EVP_PKEY_DHX == 1,
+    reason="Requires OpenSSL without EVP_PKEY_DHX (1.0.2-)")
+@pytest.mark.requires_backend_interface(interface=DHBackend)
+class TestOpenSSLDHSerialization(object):
+
+    @pytest.mark.parametrize(
+        "vector",
+        load_vectors_from_file(
+            os.path.join("asymmetric", "DH", "RFC5114.txt"),
+            load_nist_vectors))
+    def test_dh_serialization_with_q_unsupported(self, backend, vector):
+        parameters = dh.DHParameterNumbers(int(vector["p"], 16),
+                                           int(vector["g"], 16),
+                                           int(vector["q"], 16))
+        public = dh.DHPublicNumbers(int(vector["ystatcavs"], 16), parameters)
+        private = dh.DHPrivateNumbers(int(vector["xstatcavs"], 16), public)
+        private_key = private.private_key(backend)
+        public_key = private_key.public_key()
+        with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_SERIALIZATION):
+            private_key.private_bytes(serialization.Encoding.PEM,
+                                      serialization.PrivateFormat.PKCS8,
+                                      serialization.NoEncryption())
+        with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_SERIALIZATION):
+            public_key.public_bytes(
+                serialization.Encoding.PEM,
+                serialization.PublicFormat.SubjectPublicKeyInfo)
+
+    @pytest.mark.parametrize(
+        ("key_path", "loader_func"),
+        [
+            (
+                os.path.join("asymmetric", "DH", "dhkey_rfc5114_2.pem"),
+                serialization.load_pem_private_key,
+            ),
+            (
+                os.path.join("asymmetric", "DH", "dhkey_rfc5114_2.der"),
+                serialization.load_der_private_key,
+            )
+        ]
+    )
+    def test_private_load_dhx_unsupported(self, key_path, loader_func,
+                                          backend):
+        key_bytes = load_vectors_from_file(
+            key_path,
+            lambda pemfile: pemfile.read(), mode="rb"
+        )
+        with raises_unsupported_algorithm(
+                _Reasons.UNSUPPORTED_PUBLIC_KEY_ALGORITHM):
+            loader_func(key_bytes, None, backend)
+
+    @pytest.mark.parametrize(
+        ("key_path", "loader_func"),
+        [
+            (
+                os.path.join("asymmetric", "DH", "dhpub_rfc5114_2.pem"),
+                serialization.load_pem_public_key,
+            ),
+            (
+                os.path.join("asymmetric", "DH", "dhpub_rfc5114_2.der"),
+                serialization.load_der_public_key,
+            )
+        ]
+    )
+    def test_public_load_dhx_unsupported(self, key_path, loader_func,
+                                         backend):
+        key_bytes = load_vectors_from_file(
+            key_path,
+            lambda pemfile: pemfile.read(), mode="rb"
+        )
+        with raises_unsupported_algorithm(
+                _Reasons.UNSUPPORTED_PUBLIC_KEY_ALGORITHM):
+            loader_func(key_bytes, backend)
diff --git a/tests/hazmat/primitives/test_dh.py b/tests/hazmat/primitives/test_dh.py
index 1086630d..5b35fe51 100644
--- a/tests/hazmat/primitives/test_dh.py
+++ b/tests/hazmat/primitives/test_dh.py
@@ -18,6 +18,13 @@ from ...doubles import DummyKeySerializationEncryption
 from ...utils import load_nist_vectors, load_vectors_from_file
 
 
+def _skip_dhx_unsupported(backend):
+    if not backend.dh_x942_serialization_supported():
+        pytest.skip(
+            "DH x9.42 serialization is not supported"
+        )
+
+
 def test_dh_parameternumbers():
     params = dh.DHParameterNumbers(
         65537, 2
@@ -46,6 +53,19 @@ def test_dh_parameternumbers():
             65537, 7
         )
 
+    params = dh.DHParameterNumbers(
+        65537, 7, 1245
+    )
+
+    assert params.p == 65537
+    assert params.g == 7
+    assert params.q == 1245
+
+    with pytest.raises(TypeError):
+        dh.DHParameterNumbers(
+            65537, 2, "hello"
+        )
+
 
 def test_dh_numbers():
     params = dh.DHParameterNumbers(
@@ -89,7 +109,11 @@ def test_dh_numbers():
 
 def test_dh_parameter_numbers_equality():
     assert dh.DHParameterNumbers(65537, 2) == dh.DHParameterNumbers(65537, 2)
+    assert dh.DHParameterNumbers(65537, 7, 12345) == dh.DHParameterNumbers(
+        65537, 7, 12345)
     assert dh.DHParameterNumbers(6, 2) != dh.DHParameterNumbers(65537, 2)
+    assert dh.DHParameterNumbers(65537, 2, 123) != dh.DHParameterNumbers(
+        65537, 2, 456)
     assert dh.DHParameterNumbers(65537, 5) != dh.DHParameterNumbers(65537, 2)
     assert dh.DHParameterNumbers(65537, 2) != object()
 
@@ -132,15 +156,35 @@ class TestDH(object):
         assert backend.dh_parameters_supported(23, 5)
         assert not backend.dh_parameters_supported(23, 18)
 
-    def test_convert_to_numbers(self, backend):
-        parameters = backend.generate_dh_private_key_and_parameters(2, 512)
+    @pytest.mark.parametrize(
+        "vector",
+        load_vectors_from_file(
+            os.path.join("asymmetric", "DH", "RFC5114.txt"),
+            load_nist_vectors))
+    def test_dh_parameters_supported_with_q(self, backend, vector):
+        assert backend.dh_parameters_supported(int(vector["p"], 16),
+                                               int(vector["g"], 16),
+                                               int(vector["q"], 16))
+
+    @pytest.mark.parametrize("with_q", [False, True])
+    def test_convert_to_numbers(self, backend, with_q):
+        if with_q:
+            vector = load_vectors_from_file(
+                os.path.join("asymmetric", "DH", "RFC5114.txt"),
+                load_nist_vectors)[0]
+            p = int(vector["p"], 16)
+            g = int(vector["g"], 16)
+            q = int(vector["q"], 16)
+        else:
+            parameters = backend.generate_dh_private_key_and_parameters(2, 512)
 
-        private = parameters.private_numbers()
+            private = parameters.private_numbers()
 
-        p = private.public_numbers.parameter_numbers.p
-        g = private.public_numbers.parameter_numbers.g
+            p = private.public_numbers.parameter_numbers.p
+            g = private.public_numbers.parameter_numbers.g
+            q = None
 
-        params = dh.DHParameterNumbers(p, g)
+        params = dh.DHParameterNumbers(p, g, q)
         public = dh.DHPublicNumbers(1, params)
         private = dh.DHPrivateNumbers(2, public)
 
@@ -163,11 +207,22 @@ class TestDH(object):
         with pytest.raises(ValueError):
             private.private_key(backend)
 
-    def test_generate_dh(self, backend):
-        generator = 2
-        key_size = 512
+    @pytest.mark.parametrize("with_q", [False, True])
+    def test_generate_dh(self, backend, with_q):
+        if with_q:
+            vector = load_vectors_from_file(
+                os.path.join("asymmetric", "DH", "RFC5114.txt"),
+                load_nist_vectors)[0]
+            p = int(vector["p"], 16)
+            g = int(vector["g"], 16)
+            q = int(vector["q"], 16)
+            parameters = dh.DHParameterNumbers(p, g, q).parameters(backend)
+            key_size = 1024
+        else:
+            generator = 2
+            key_size = 512
 
-        parameters = dh.generate_parameters(generator, key_size, backend)
+            parameters = dh.generate_parameters(generator, key_size, backend)
         assert isinstance(parameters, dh.DHParameters)
 
         key = parameters.generate_private_key()
@@ -290,6 +345,27 @@ class TestDH(object):
 
         assert int_from_bytes(symkey, 'big') == int(vector["k"], 16)
 
+    @pytest.mark.parametrize(
+        "vector",
+        load_vectors_from_file(
+            os.path.join("asymmetric", "DH", "RFC5114.txt"),
+            load_nist_vectors))
+    def test_dh_vectors_with_q(self, backend, vector):
+        parameters = dh.DHParameterNumbers(int(vector["p"], 16),
+                                           int(vector["g"], 16),
+                                           int(vector["q"], 16))
+        public1 = dh.DHPublicNumbers(int(vector["ystatcavs"], 16), parameters)
+        private1 = dh.DHPrivateNumbers(int(vector["xstatcavs"], 16), public1)
+        public2 = dh.DHPublicNumbers(int(vector["ystatiut"], 16), parameters)
+        private2 = dh.DHPrivateNumbers(int(vector["xstatiut"], 16), public2)
+        key1 = private1.private_key(backend)
+        key2 = private2.private_key(backend)
+        symkey1 = key1.exchange(public2.public_key(backend))
+        symkey2 = key2.exchange(public1.public_key(backend))
+
+        assert int_from_bytes(symkey1, 'big') == int(vector["z"], 16)
+        assert int_from_bytes(symkey2, 'big') == int(vector["z"], 16)
+
 
 @pytest.mark.requires_backend_interface(interface=DHBackend)
 @pytest.mark.requires_backend_interface(interface=PEMSerializationBackend)
@@ -332,11 +408,20 @@ class TestDHPrivateKeySerialization(object):
                 os.path.join("asymmetric", "DH", "dhkey.der"),
                 serialization.load_der_private_key,
                 serialization.Encoding.DER,
+            ), (
+                os.path.join("asymmetric", "DH", "dhkey_rfc5114_2.pem"),
+                serialization.load_pem_private_key,
+                serialization.Encoding.PEM,
+            ), (
+                os.path.join("asymmetric", "DH", "dhkey_rfc5114_2.der"),
+                serialization.load_der_private_key,
+                serialization.Encoding.DER,
             )
         ]
     )
     def test_private_bytes_match(self, key_path, loader_func,
                                  encoding, backend):
+        _skip_dhx_unsupported(backend)
         key_bytes = load_vectors_from_file(
             key_path,
             lambda pemfile: pemfile.read(), mode="rb"
@@ -349,34 +434,48 @@ class TestDHPrivateKeySerialization(object):
         assert serialized == key_bytes
 
     @pytest.mark.parametrize(
-        ("key_path", "loader_func"),
+        ("key_path", "loader_func", "vec_path"),
         [
             (
                 os.path.join("asymmetric", "DH", "dhkey.pem"),
                 serialization.load_pem_private_key,
+                os.path.join("asymmetric", "DH", "dhkey.txt")
             ), (
                 os.path.join("asymmetric", "DH", "dhkey.der"),
                 serialization.load_der_private_key,
+                os.path.join("asymmetric", "DH", "dhkey.txt")
+            ), (
+                os.path.join("asymmetric", "DH", "dhkey_rfc5114_2.pem"),
+                serialization.load_pem_private_key,
+                os.path.join("asymmetric", "DH", "dhkey_rfc5114_2.txt")
+            ), (
+                os.path.join("asymmetric", "DH", "dhkey_rfc5114_2.der"),
+                serialization.load_der_private_key,
+                os.path.join("asymmetric", "DH", "dhkey_rfc5114_2.txt")
             )
         ]
     )
     def test_private_bytes_values(self, key_path, loader_func,
-                                  backend):
+                                  vec_path, backend):
+        _skip_dhx_unsupported(backend)
         key_bytes = load_vectors_from_file(
             key_path,
             lambda pemfile: pemfile.read(), mode="rb"
         )
-        vec = load_vectors_from_file(
-            os.path.join("asymmetric", "DH", "dhkey.txt"),
-            load_nist_vectors)[0]
+        vec = load_vectors_from_file(vec_path, load_nist_vectors)[0]
         key = loader_func(key_bytes, None, backend)
         private_numbers = key.private_numbers()
         assert private_numbers.x == int(vec["x"], 16)
         assert private_numbers.public_numbers.y == int(vec["y"], 16)
         assert private_numbers.public_numbers.parameter_numbers.g == int(
-            vec["g"])
+            vec["g"], 16)
         assert private_numbers.public_numbers.parameter_numbers.p == int(
             vec["p"], 16)
+        if "q" in vec:
+            assert private_numbers.public_numbers.parameter_numbers.q == int(
+                vec["q"], 16)
+        else:
+            assert private_numbers.public_numbers.parameter_numbers.q is None
 
     def test_private_bytes_traditional_openssl_invalid(self, backend):
         parameters = dh.generate_parameters(2, 512, backend)
@@ -469,11 +568,20 @@ class TestDHPublicKeySerialization(object):
                 os.path.join("asymmetric", "DH", "dhpub.der"),
                 serialization.load_der_public_key,
                 serialization.Encoding.DER,
+            ), (
+                os.path.join("asymmetric", "DH", "dhpub_rfc5114_2.pem"),
+                serialization.load_pem_public_key,
+                serialization.Encoding.PEM,
+            ), (
+                os.path.join("asymmetric", "DH", "dhpub_rfc5114_2.der"),
+                serialization.load_der_public_key,
+                serialization.Encoding.DER,
             )
         ]
     )
     def test_public_bytes_match(self, key_path, loader_func,
                                 encoding, backend):
+        _skip_dhx_unsupported(backend)
         key_bytes = load_vectors_from_file(
             key_path,
             lambda pemfile: pemfile.read(), mode="rb"
@@ -486,31 +594,44 @@ class TestDHPublicKeySerialization(object):
         assert serialized == key_bytes
 
     @pytest.mark.parametrize(
-        ("key_path", "loader_func"),
+        ("key_path", "loader_func", "vec_path"),
         [
             (
                 os.path.join("asymmetric", "DH", "dhpub.pem"),
                 serialization.load_pem_public_key,
+                os.path.join("asymmetric", "DH", "dhkey.txt"),
             ), (
                 os.path.join("asymmetric", "DH", "dhpub.der"),
                 serialization.load_der_public_key,
+                os.path.join("asymmetric", "DH", "dhkey.txt"),
+            ), (
+                os.path.join("asymmetric", "DH", "dhpub_rfc5114_2.pem"),
+                serialization.load_pem_public_key,
+                os.path.join("asymmetric", "DH", "dhkey_rfc5114_2.txt"),
+            ), (
+                os.path.join("asymmetric", "DH", "dhpub_rfc5114_2.der"),
+                serialization.load_der_public_key,
+                os.path.join("asymmetric", "DH", "dhkey_rfc5114_2.txt"),
             )
         ]
     )
     def test_public_bytes_values(self, key_path, loader_func,
-                                 backend):
+                                 vec_path, backend):
+        _skip_dhx_unsupported(backend)
         key_bytes = load_vectors_from_file(
             key_path,
             lambda pemfile: pemfile.read(), mode="rb"
         )
-        vec = load_vectors_from_file(
-            os.path.join("asymmetric", "DH", "dhkey.txt"),
-            load_nist_vectors)[0]
+        vec = load_vectors_from_file(vec_path, load_nist_vectors)[0]
         pub_key = loader_func(key_bytes, backend)
         public_numbers = pub_key.public_numbers()
         assert public_numbers.y == int(vec["y"], 16)
-        assert public_numbers.parameter_numbers.g == int(vec["g"])
+        assert public_numbers.parameter_numbers.g == int(vec["g"], 16)
         assert public_numbers.parameter_numbers.p == int(vec["p"], 16)
+        if "q" in vec:
+            assert public_numbers.parameter_numbers.q == int(vec["q"], 16)
+        else:
+            assert public_numbers.parameter_numbers.q is None
 
     def test_public_bytes_invalid_encoding(self, backend):
         parameters = dh.generate_parameters(2, 512, backend)
diff --git a/tests/hazmat/primitives/test_serialization.py b/tests/hazmat/primitives/test_serialization.py
index dad056c6..bc16b5f8 100644
--- a/tests/hazmat/primitives/test_serialization.py
+++ b/tests/hazmat/primitives/test_serialization.py
@@ -236,10 +236,12 @@ class TestDERSerialization(object):
         """).encode()
         bad_der = base64.b64decode(b"".join(key_data.splitlines()))
 
-        with pytest.raises(ValueError):
+        with raises_unsupported_algorithm(
+                _Reasons.UNSUPPORTED_PUBLIC_KEY_ALGORITHM):
             load_pem_private_key(bad_der, None, backend)
 
-        with pytest.raises(ValueError):
+        with raises_unsupported_algorithm(
+                _Reasons.UNSUPPORTED_PUBLIC_KEY_ALGORITHM):
             load_pem_private_key(
                 bad_der, b"this password will not be used", backend
             )
@@ -575,12 +577,14 @@ class TestPEMSerialization(object):
     def test_wrong_private_format(self, backend):
         key_data = b"---- NOT A KEY ----\n"
 
-        with pytest.raises(ValueError):
+        with raises_unsupported_algorithm(
+                _Reasons.UNSUPPORTED_PUBLIC_KEY_ALGORITHM):
             load_pem_private_key(
                 key_data, None, backend
             )
 
-        with pytest.raises(ValueError):
+        with raises_unsupported_algorithm(
+                _Reasons.UNSUPPORTED_PUBLIC_KEY_ALGORITHM):
             load_pem_private_key(
                 key_data, b"this password will not be used", backend
             )
@@ -588,7 +592,8 @@ class TestPEMSerialization(object):
     def test_wrong_public_format(self, backend):
         key_data = b"---- NOT A KEY ----\n"
 
-        with pytest.raises(ValueError):
+        with raises_unsupported_algorithm(
+                _Reasons.UNSUPPORTED_PUBLIC_KEY_ALGORITHM):
             load_pem_public_key(key_data, backend)
 
     def test_corrupt_traditional_format(self, backend):
@@ -720,12 +725,14 @@ class TestPEMSerialization(object):
 
         password = b"this password is wrong"
 
-        with pytest.raises(ValueError):
+        with raises_unsupported_algorithm(
+                _Reasons.UNSUPPORTED_PUBLIC_KEY_ALGORITHM):
             load_pem_private_key(
                 key_data, None, backend
             )
 
-        with pytest.raises(ValueError):
+        with raises_unsupported_algorithm(
+                _Reasons.UNSUPPORTED_PUBLIC_KEY_ALGORITHM):
             load_pem_private_key(
                 key_data, password, backend
             )