Issue #13626: Add support for SSL Diffie-Hellman key exchange, through the

SSLContext.load_dh_params() method and the ssl.OP_SINGLE_DH_USE option.

3 files changed, 34 insertions, 1 deletions

diff --git a/Lib/ssl.py b/Lib/ssl.py
index d43d255..b56a8c8 100644
--- a/Lib/ssl.py
+++ b/Lib/ssl.py
@@ -68,7 +68,7 @@ from _ssl import (
 from _ssl import CERT_NONE, CERT_OPTIONAL, CERT_REQUIRED
 from _ssl import (
     OP_ALL, OP_NO_SSLv2, OP_NO_SSLv3, OP_NO_TLSv1,
-    OP_CIPHER_SERVER_PREFERENCE, OP_SINGLE_ECDH_USE,
+    OP_CIPHER_SERVER_PREFERENCE, OP_SINGLE_DH_USE, OP_SINGLE_ECDH_USE,
     )
 try:
     from _ssl import OP_NO_COMPRESSION
diff --git a/Lib/test/ssl_servers.py b/Lib/test/ssl_servers.py
index becbfab..8686153 100644
--- a/Lib/test/ssl_servers.py
+++ b/Lib/test/ssl_servers.py
@@ -180,6 +180,8 @@ if __name__ == "__main__":
     parser.add_argument('--curve-name', dest='curve_name', type=str,
                         action='store',
                         help='curve name for EC-based Diffie-Hellman')
+    parser.add_argument('--dh', dest='dh_file', type=str, action='store',
+                        help='PEM file containing DH parameters')
     args = parser.parse_args()
 
     support.verbose = args.verbose
@@ -192,6 +194,8 @@ if __name__ == "__main__":
     context.load_cert_chain(CERTFILE)
     if args.curve_name:
         context.set_ecdh_curve(args.curve_name)
+    if args.dh_file:
+        context.load_dh_params(args.dh_file)
 
     server = HTTPSServer(("", args.port), handler_class, context)
     if args.verbose:
diff --git a/Lib/test/test_ssl.py b/Lib/test/test_ssl.py
index d549799..a4bcdd0 100644
--- a/Lib/test/test_ssl.py
+++ b/Lib/test/test_ssl.py
@@ -56,6 +56,8 @@ WRONGCERT = data_file("XXXnonexisting.pem")
 BADKEY = data_file("badkey.pem")
 NOKIACERT = data_file("nokia.pem")
 
+DHFILE = data_file("dh512.pem")
+BYTES_DHFILE = os.fsencode(DHFILE)
 
 def handle_error(prefix):
     exc_format = ' '.join(traceback.format_exception(*sys.exc_info()))
@@ -99,6 +101,7 @@ class BasicSocketTests(unittest.TestCase):
         ssl.CERT_OPTIONAL
         ssl.CERT_REQUIRED
         ssl.OP_CIPHER_SERVER_PREFERENCE
+        ssl.OP_SINGLE_DH_USE
         ssl.OP_SINGLE_ECDH_USE
         if ssl.OPENSSL_VERSION_INFO >= (1, 0):
             ssl.OP_NO_COMPRESSION
@@ -538,6 +541,19 @@ class ContextTests(unittest.TestCase):
         # Issue #10989: crash if the second argument type is invalid
         self.assertRaises(TypeError, ctx.load_verify_locations, None, True)
 
+    def test_load_dh_params(self):
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
+        ctx.load_dh_params(DHFILE)
+        if os.name != 'nt':
+            ctx.load_dh_params(BYTES_DHFILE)
+        self.assertRaises(TypeError, ctx.load_dh_params)
+        self.assertRaises(TypeError, ctx.load_dh_params, None)
+        with self.assertRaises(FileNotFoundError) as cm:
+            ctx.load_dh_params(WRONGCERT)
+        self.assertEqual(cm.exception.errno, errno.ENOENT)
+        with self.assertRaisesRegex(ssl.SSLError, "PEM routines"):
+            ctx.load_dh_params(CERTFILE)
+
     @skip_if_broken_ubuntu_ssl
     def test_session_stats(self):
         for proto in PROTOCOLS:
@@ -1802,6 +1818,19 @@ else:
                                        chatty=True, connectionchatty=True)
             self.assertIs(stats['compression'], None)
 
+        def test_dh_params(self):
+            # Check we can get a connection with ephemeral Diffie-Hellman
+            context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
+            context.load_cert_chain(CERTFILE)
+            context.load_dh_params(DHFILE)
+            context.set_ciphers("kEDH")
+            stats = server_params_test(context, context,
+                                       chatty=True, connectionchatty=True)
+            cipher = stats["cipher"][0]
+            parts = cipher.split("-")
+            if "ADH" not in parts and "EDH" not in parts and "DHE" not in parts:
+                self.fail("Non-DH cipher: " + cipher[0])
+
 
 def test_main(verbose=False):
     if support.verbose: