Handle environments without home dir

LeonardBesson · LeonardBesson · commit c3372614c157 · 2023-03-08T21:37:24.000+01:00
diff --git a/asyncpg/compat.py b/asyncpg/compat.py
@@ -8,6 +8,7 @@
 import asyncio
 import pathlib
 import platform
+import typing
 
 
 SYSTEM = platform.uname().system
@@ -18,7 +19,7 @@
 
 CSIDL_APPDATA = 0x001a
 
- def get_pg_home_directory() -> pathlib.Path:
+ def get_pg_home_directory() -> typing.Optional[pathlib.Path]:
 # We cannot simply use expanduser() as that returns the user's
 # home directory, whereas Postgres stores its config in
 # %AppData% on Windows.
@@ -30,8 +31,11 @@ def get_pg_home_directory() -> pathlib.Path:
 return pathlib.Path(buf.value) / 'postgresql'
 
 else:
- def get_pg_home_directory() -> pathlib.Path:
- return pathlib.Path.home()
+ def get_pg_home_directory() -> typing.Optional[pathlib.Path]:
+ try:
+ return pathlib.Path.home()
+ except (RuntimeError, KeyError):
+ return None
 
 
 async def wait_closed(stream):
diff --git a/asyncpg/connect_utils.py b/asyncpg/connect_utils.py
@@ -249,8 +249,12 @@ def _parse_tls_version(tls_version):
 )
 
 
-def _dot_postgresql_path(filename) -> pathlib.Path:
- return (pathlib.Path.home() / '.postgresql' / filename).resolve()
+def _dot_postgresql_path(filename) -> typing.Optional[pathlib.Path]:
+ homedir = compat.get_pg_home_directory()
+ if homedir is None:
+ return None
+
+ return (homedir / '.postgresql' / filename).resolve()
 
 
 def _parse_connect_dsn_and_args(*, dsn, host, port, user,
@@ -501,11 +505,16 @@ 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 = _dot_postgresql_path('root.crt')
 try:
+ sslrootcert = _dot_postgresql_path('root.crt')
+ assert sslrootcert is not None
 ssl.load_verify_locations(cafile=sslrootcert)
- except FileNotFoundError:
+ except (AssertionError, FileNotFoundError):
 if sslmode > SSLMode.require:
+ if sslrootcert is None:
+ raise RuntimeError(
+ 'Cannot determine home directory'
+ )
 raise ValueError(
 f'root certificate file "{sslrootcert}" does '
 f'not exist\nEither provide the file or '
@@ -526,18 +535,20 @@ def _parse_connect_dsn_and_args(*, dsn, host, port, user,
 ssl.verify_flags |= ssl_module.VERIFY_CRL_CHECK_CHAIN
 else:
 sslcrl = _dot_postgresql_path('root.crl')
- try:
- ssl.load_verify_locations(cafile=sslcrl)
- except FileNotFoundError:
- pass
- else:
- ssl.verify_flags |= ssl_module.VERIFY_CRL_CHECK_CHAIN
+ if sslcrl is not None:
+ try:
+ ssl.load_verify_locations(cafile=sslcrl)
+ except FileNotFoundError:
+ pass
+ else:
+ ssl.verify_flags |= \
+ ssl_module.VERIFY_CRL_CHECK_CHAIN
 
 if sslkey is None:
 sslkey = os.getenv('PGSSLKEY')
 if not sslkey:
 sslkey = _dot_postgresql_path('postgresql.key')
- if not sslkey.exists():
+ if sslkey is not None and not sslkey.exists():
 sslkey = None
 if not sslpassword:
 sslpassword = ''
@@ -549,12 +560,15 @@ def _parse_connect_dsn_and_args(*, dsn, host, port, user,
 )
 else:
 sslcert = _dot_postgresql_path('postgresql.crt')
- try:
- ssl.load_cert_chain(
- sslcert, keyfile=sslkey, password=lambda: sslpassword
- )
- except FileNotFoundError:
- pass
+ if sslcert is not None:
+ try:
+ ssl.load_cert_chain(
+ sslcert,
+ keyfile=sslkey,
+ password=lambda: sslpassword
+ )
+ except FileNotFoundError:
+ pass
 
 # OpenSSL 1.1.1 keylog file, copied from create_default_context()
 if hasattr(ssl, 'keylog_filename'):
diff --git a/tests/test_connect.py b/tests/test_connect.py
@@ -71,6 +71,14 @@ def mock_dot_postgresql(*, ca=True, crl=False, client=False, protected=False):
 yield
 
 
+@contextlib.contextmanager
+def mock_no_home_dir():
+ with unittest.mock.patch(
+ 'pathlib.Path.home', unittest.mock.Mock(side_effect=RuntimeError)
+ ):
+ yield
+
+
 class TestSettings(tb.ConnectedTestCase):
 
 async def test_get_settings_01(self):
@@ -1257,6 +1265,27 @@ async def test_connection_implicit_host(self):
 user=conn_spec.get('user'))
 await con.close()
 
+ @unittest.skipIf(os.environ.get('PGHOST'), 'unmanaged cluster')
+ async def test_connection_no_home_dir(self):
+ with mock_no_home_dir():
+ con = await self.connect(
+ dsn='postgresql://foo/',
+ user='postgres',
+ database='postgres',
+ host='localhost')
+ await con.fetchval('SELECT 42')
+ await con.close()
+
+ with self.assertRaisesRegex(
+ RuntimeError,
+ 'Cannot determine home directory'
+ ):
+ with mock_no_home_dir():
+ await self.connect(
+ host='localhost',
+ user='ssl_user',
+ ssl='verify-full')
+
 
 class BaseTestSSLConnection(tb.ConnectedTestCase):
 @classmethod

-Original file line number
+Diff line change
 importasyncio
 importpathlib
 importplatform
 +importtyping
 SYSTEM=platform.uname().system
 CSIDL_APPDATA=0x001a
 -defget_pg_home_directory() ->pathlib.Path:
 +defget_pg_home_directory() ->typing.Optional[pathlib.Path]:
 # We cannot simply use expanduser() as that returns the user's
 # home directory, whereas Postgres stores its config in
 # %AppData% on Windows.
 returnpathlib.Path(buf.value) /'postgresql'
 else:
 -defget_pg_home_directory() ->pathlib.Path:
 -returnpathlib.Path.home()
 +defget_pg_home_directory() ->typing.Optional[pathlib.Path]:
 +try:
 +returnpathlib.Path.home()
 +except (RuntimeError, KeyError):
 +returnNone
 asyncdefwait_closed(stream):
-Original file line number
+Diff line change
+ )
 -def_dot_postgresql_path(filename) ->pathlib.Path:
 -return (pathlib.Path.home() /'.postgresql'/filename).resolve()
 +def_dot_postgresql_path(filename) ->typing.Optional[pathlib.Path]:
 +homedir=compat.get_pg_home_directory()
 +ifhomedirisNone:
 +returnNone
++
 +return (homedir/'.postgresql'/filename).resolve()
 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=_dot_postgresql_path('root.crt')
 try:
 +sslrootcert=_dot_postgresql_path('root.crt')
 +assertsslrootcertisnotNone
 ssl.load_verify_locations(cafile=sslrootcert)
 -exceptFileNotFoundError:
 +except(AssertionError, FileNotFoundError):
 ifsslmode>SSLMode.require:
 +ifsslrootcertisNone:
 +raiseRuntimeError(
 +'Cannot determine home directory'
 + )
 raiseValueError(
 f'root certificate file "{sslrootcert}" does '
 f'not exist\nEither provide the file or '
 ssl.verify_flags|=ssl_module.VERIFY_CRL_CHECK_CHAIN
 else:
 sslcrl=_dot_postgresql_path('root.crl')
 -try:
 -ssl.load_verify_locations(cafile=sslcrl)
 -exceptFileNotFoundError:
 -pass
 -else:
 -ssl.verify_flags|=ssl_module.VERIFY_CRL_CHECK_CHAIN
 +ifsslcrlisnotNone:
 +try:
 +ssl.load_verify_locations(cafile=sslcrl)
 +exceptFileNotFoundError:
 +pass
 +else:
 +ssl.verify_flags|= \
 +ssl_module.VERIFY_CRL_CHECK_CHAIN
 ifsslkeyisNone:
 sslkey=os.getenv('PGSSLKEY')
 ifnotsslkey:
 sslkey=_dot_postgresql_path('postgresql.key')
 -ifnotsslkey.exists():
 +ifsslkeyisnotNoneandnotsslkey.exists():
 sslkey=None
 ifnotsslpassword:
 sslpassword=''
+ )
 else:
 sslcert=_dot_postgresql_path('postgresql.crt')
 -try:
 -ssl.load_cert_chain(
 -sslcert, keyfile=sslkey, password=lambda: sslpassword
 - )
 -exceptFileNotFoundError:
 -pass
 +ifsslcertisnotNone:
 +try:
 +ssl.load_cert_chain(
 +sslcert,
 +keyfile=sslkey,
 +password=lambda: sslpassword
 + )
 +exceptFileNotFoundError:
 +pass
 # OpenSSL 1.1.1 keylog file, copied from create_default_context()
 ifhasattr(ssl, 'keylog_filename'):