Fix for Python 3.6, win32 and lack of TLSv1.1

Author: fantix

asyncpg/connect_utils.py

@@ -222,6 +222,10 @@ def _parse_hostlist(hostlist, port, *, unquote=False):
 
 
 def _parse_tls_version(tls_version):
+    if not hasattr(ssl_module, 'TLSVersion'):
+        raise ValueError(
+            "TLSVersion is not supported in this version of Python"
+        )
     if tls_version.startswith('SSL'):
         raise ValueError(
             f"Unsupported TLS version: {tls_version}"
@@ -234,6 +238,10 @@ def _parse_tls_version(tls_version):
         )
 
 
+def _dot_postgresql_path(filename) -> pathlib.Path:
+    return (pathlib.Path.home() / '.postgresql' / filename).resolve()
+
+
 def _parse_connect_dsn_and_args(*, dsn, host, port, user,
                                 password, passfile, database, ssl,
                                 connect_timeout, server_settings):
@@ -485,7 +493,7 @@ def _parse_connect_dsn_and_args(*, dsn, host, port, user,
                     ssl.load_verify_locations(cafile=sslrootcert)
                     ssl.verify_mode = ssl_module.CERT_REQUIRED
                 else:
-                    sslrootcert = os.path.expanduser('~/.postgresql/root.crt')
+                    sslrootcert = _dot_postgresql_path('root.crt')
                     try:
                         ssl.load_verify_locations(cafile=sslrootcert)
                     except FileNotFoundError:
@@ -509,7 +517,7 @@ def _parse_connect_dsn_and_args(*, dsn, host, port, user,
                     ssl.load_verify_locations(cafile=sslcrl)
                     ssl.verify_flags |= ssl_module.VERIFY_CRL_CHECK_CHAIN
                 else:
-                    sslcrl = os.path.expanduser('~/.postgresql/root.crl')
+                    sslcrl = _dot_postgresql_path('root.crl')
                     try:
                         ssl.load_verify_locations(cafile=sslcrl)
                     except FileNotFoundError:
@@ -520,8 +528,8 @@ def _parse_connect_dsn_and_args(*, dsn, host, port, user,
             if sslkey is None:
                 sslkey = os.getenv('PGSSLKEY')
             if not sslkey:
-                sslkey = os.path.expanduser('~/.postgresql/postgresql.key')
-                if not os.path.exists(sslkey):
+                sslkey = _dot_postgresql_path('postgresql.key')
+                if not sslkey.exists():
                     sslkey = None
             if not sslpassword:
                 sslpassword = ''
@@ -532,7 +540,7 @@ def _parse_connect_dsn_and_args(*, dsn, host, port, user,
                     sslcert, keyfile=sslkey, password=lambda: sslpassword
                 )
             else:
-                sslcert = os.path.expanduser('~/.postgresql/postgresql.crt')
+                sslcert = _dot_postgresql_path('postgresql.crt')
                 try:
                     ssl.load_cert_chain(
                         sslcert, keyfile=sslkey, password=lambda: sslpassword
@@ -552,13 +560,17 @@ def _parse_connect_dsn_and_args(*, dsn, host, port, user,
                 ssl.options &= ~ssl_module.OP_NO_COMPRESSION
 
             if ssl_min_protocol_version is None:
-                ssl_min_protocol_version = os.getenv(
-                    'PGSSLMINPROTOCOLVERSION', 'TLSv1.2'
-                )
+                ssl_min_protocol_version = os.getenv('PGSSLMINPROTOCOLVERSION')
             if ssl_min_protocol_version:
                 ssl.minimum_version = _parse_tls_version(
                     ssl_min_protocol_version
                 )
+            else:
+                try:
+                    ssl.minimum_version = _parse_tls_version('TLSv1.2')
+                except ValueError:
+                    # Python 3.6 does not have ssl.TLSVersion
+                    pass
 
             if ssl_max_protocol_version is None:
                 ssl_max_protocol_version = os.getenv('PGSSLMAXPROTOCOLVERSION')

tests/test_connect.py

@@ -9,10 +9,12 @@
 import contextlib
 import ipaddress
 import os
+import pathlib
 import platform
 import shutil
 import ssl
 import stat
+import sys
 import tempfile
 import textwrap
 import unittest
@@ -46,31 +48,26 @@
 @contextlib.contextmanager
 def mock_dot_postgresql(*, ca=True, crl=False, client=False, protected=False):
     with tempfile.TemporaryDirectory() as temp_dir:
-        pg_home = os.path.join(temp_dir, '.postgresql')
-        os.mkdir(pg_home)
+        home = pathlib.Path(temp_dir)
+        pg_home = home / '.postgresql'
+        pg_home.mkdir()
         if ca:
-            shutil.copyfile(
-                SSL_CA_CERT_FILE, os.path.join(pg_home, 'root.crt')
-            )
+            shutil.copyfile(SSL_CA_CERT_FILE, pg_home / 'root.crt')
         if crl:
-            shutil.copyfile(
-                SSL_CA_CRL_FILE, os.path.join(pg_home, 'root.crl')
-            )
+            shutil.copyfile(SSL_CA_CRL_FILE, pg_home / 'root.crl')
         if client:
-            shutil.copyfile(
-                CLIENT_SSL_CERT_FILE, os.path.join(pg_home, 'postgresql.crt')
-            )
+            shutil.copyfile(CLIENT_SSL_CERT_FILE, pg_home / 'postgresql.crt')
             if protected:
                 shutil.copyfile(
-                    CLIENT_SSL_PROTECTED_KEY_FILE,
-                    os.path.join(pg_home, 'postgresql.key'),
+                    CLIENT_SSL_PROTECTED_KEY_FILE, pg_home / 'postgresql.key'
                 )
             else:
                 shutil.copyfile(
-                    CLIENT_SSL_KEY_FILE,
-                    os.path.join(pg_home, 'postgresql.key'),
+                    CLIENT_SSL_KEY_FILE, pg_home / 'postgresql.key'
                 )
-        with unittest.mock.patch.dict('os.environ', {'HOME': temp_dir}):
+        with unittest.mock.patch(
+            'pathlib.Path.home', unittest.mock.Mock(return_value=home)
+        ):
             yield
 
 
@@ -1340,13 +1337,13 @@ async def verify_fails(sslmode, *, host='localhost', exn_type):
             await verify_works('allow')
             await verify_works('prefer')
             await verify_fails('require',
-                               exn_type=ssl.CertificateError)
+                               exn_type=ssl.SSLError)
             await verify_fails('verify-ca',
-                               exn_type=ssl.CertificateError)
+                               exn_type=ssl.SSLError)
             await verify_fails('verify-ca', host='127.0.0.1',
-                               exn_type=ssl.CertificateError)
+                               exn_type=ssl.SSLError)
             await verify_fails('verify-full',
-                               exn_type=ssl.CertificateError)
+                               exn_type=ssl.SSLError)
 
     async def test_ssl_connection_default_context(self):
         # XXX: uvloop artifact
@@ -1410,6 +1407,9 @@ async def test_executemany_uvloop_ssl_issue_700(self):
             finally:
                 await con.close()
 
+    @unittest.skipIf(
+        sys.version_info < (3, 7), "Python < 3.7 doesn't have ssl.TLSVersion"
+    )
     async def test_tls_version(self):
         # XXX: uvloop artifact
         old_handler = self.loop.get_exception_handler()
@@ -1420,7 +1420,7 @@ async def test_tls_version(self):
                     dsn='postgresql://ssl_user@localhost/postgres'
                         '?sslmode=require&ssl_min_protocol_version=TLSv1.3'
                 )
-            with self.assertRaisesRegex(ssl.SSLError, 'protocol version'):
+            with self.assertRaises(ssl.SSLError):
                 await self.connect(
                     dsn='postgresql://ssl_user@localhost/postgres'
                         '?sslmode=require'