Issue #13634: Add support for querying and disabling SSL compression.

3 files changed, 38 insertions, 0 deletions

diff --git a/Lib/ssl.py b/Lib/ssl.py
index d244104..0b2f743 100644
--- a/Lib/ssl.py
+++ b/Lib/ssl.py
@@ -70,6 +70,10 @@ from _ssl import (
     OP_ALL, OP_NO_SSLv2, OP_NO_SSLv3, OP_NO_TLSv1,
     OP_CIPHER_SERVER_PREFERENCE, OP_SINGLE_ECDH_USE,
     )
+try:
+    from _ssl import OP_NO_COMPRESSION
+except ImportError:
+    pass
 from _ssl import RAND_status, RAND_egd, RAND_add, RAND_bytes, RAND_pseudo_bytes
 from _ssl import (
     SSL_ERROR_ZERO_RETURN,
@@ -330,6 +334,13 @@ class SSLSocket(socket):
         else:
             return self._sslobj.cipher()
 
+    def compression(self):
+        self._checkClosed()
+        if not self._sslobj:
+            return None
+        else:
+            return self._sslobj.compression()
+
     def send(self, data, flags=0):
         self._checkClosed()
         if self._sslobj:
diff --git a/Lib/test/ssl_servers.py b/Lib/test/ssl_servers.py
index 86bc950..becbfab 100644
--- a/Lib/test/ssl_servers.py
+++ b/Lib/test/ssl_servers.py
@@ -97,6 +97,7 @@ class StatsRequestHandler(BaseHTTPRequestHandler):
         stats = {
             'session_cache': context.session_stats(),
             'cipher': sock.cipher(),
+            'compression': sock.compression(),
             }
         body = pprint.pformat(stats)
         body = body.encode('utf-8')
diff --git a/Lib/test/test_ssl.py b/Lib/test/test_ssl.py
index 505550f..76fb3e7 100644
--- a/Lib/test/test_ssl.py
+++ b/Lib/test/test_ssl.py
@@ -100,6 +100,8 @@ class BasicSocketTests(unittest.TestCase):
         ssl.CERT_REQUIRED
         ssl.OP_CIPHER_SERVER_PREFERENCE
         ssl.OP_SINGLE_ECDH_USE
+        if ssl.OPENSSL_VERSION_INFO >= (1, 0):
+            ssl.OP_NO_COMPRESSION
         self.assertIn(ssl.HAS_SNI, {True, False})
 
     def test_random(self):
@@ -1185,7 +1187,12 @@ else:
             if connectionchatty:
                 if support.verbose:
                     sys.stdout.write(" client:  closing connection.\n")
+            stats = {
+                'compression': s.compression(),
+                'cipher': s.cipher(),
+            }
             s.close()
+            return stats
         finally:
             server.stop()
             server.join()
@@ -1814,6 +1821,25 @@ else:
                 server.stop()
                 server.join()
 
+        def test_compression(self):
+            context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
+            context.load_cert_chain(CERTFILE)
+            stats = server_params_test(context, context,
+                                       chatty=True, connectionchatty=True)
+            if support.verbose:
+                sys.stdout.write(" got compression: {!r}\n".format(stats['compression']))
+            self.assertIn(stats['compression'], { None, 'ZLIB', 'RLE' })
+
+        @unittest.skipUnless(hasattr(ssl, 'OP_NO_COMPRESSION'),
+                             "ssl.OP_NO_COMPRESSION needed for this test")
+        def test_compression_disabled(self):
+            context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
+            context.load_cert_chain(CERTFILE)
+            stats = server_params_test(context, context,
+                                       chatty=True, connectionchatty=True)
+            self.assertIs(stats['compression'], None)
+
+
 def test_main(verbose=False):
     if support.verbose:
         plats = {