Handle and ignore permission errors when attempting to read .pgpass

elprans · elprans · commit df7830f0503d · 2018-09-06T10:53:55.000-04:00
Fixes: MagicStack#356
diff --git a/asyncpg/connect_utils.py b/asyncpg/connect_utils.py
@@ -59,23 +59,27 @@
 def _read_password_file(passfile: pathlib.Path) \
 -> typing.List[typing.Tuple[str, ...]]:
 
- if not passfile.is_file():
- warnings.warn(
- 'password file{!r} is not a plain file'.format(passfile))
+ passtab = []
 
- return None
+ try:
+ if not passfile.exists():
+ return []
 
- if _system != 'Windows':
- if passfile.stat().st_mode & (stat.S_IRWXG | stat.S_IRWXO):
+ if not passfile.is_file():
 warnings.warn(
- 'password file{!r} has group or world access; '
- 'permissions should be u=rw (0600) or less'.format(passfile))
+ 'password file{!r} is not a plain file'.format(passfile))
 
- return None
+ return []
 
- passtab = []
+ if _system != 'Windows':
+ if passfile.stat().st_mode & (stat.S_IRWXG | stat.S_IRWXO):
+ warnings.warn(
+ 'password file{!r} has group or world access; '
+ 'permissions should be u=rw (0600) or less'.format(
+ passfile))
+
+ return []
 
- try:
 with passfile.open('rt') as f:
 for line in f:
 line = line.strip()
@@ -105,9 +109,6 @@ def _read_password_from_pgpass(
 Password string, if found, ``None`` otherwise.
 """
 
- if not passfile.exists():
- return None
-
 passtab = _read_password_file(passfile)
 if not passtab:
 return None
diff --git a/tests/test_connect.py b/tests/test_connect.py
@@ -661,6 +661,50 @@ def test_connect_pgpass_nonexistent(self):
 )
 })
 
+ @unittest.skipIf(_system == 'Windows', 'no mode checking on Windows')
+ def test_connect_pgpass_inaccessible_file(self):
+ with tempfile.NamedTemporaryFile('w+t') as passfile:
+ os.chmod(passfile.name, stat.S_IWUSR)
+
+ # nonexistent passfile is OK
+ self.run_testcase({
+ 'host': 'abc',
+ 'user': 'user',
+ 'database': 'db',
+ 'passfile': passfile.name,
+ 'result': (
+ [('abc', 5432)],
+{
+ 'user': 'user',
+ 'database': 'db',
+ }
+ )
+ })
+
+ @unittest.skipIf(_system == 'Windows', 'no mode checking on Windows')
+ def test_connect_pgpass_inaccessible_directory(self):
+ with tempfile.TemporaryDirectory() as passdir:
+ with tempfile.NamedTemporaryFile('w+t', dir=passdir) as passfile:
+ os.chmod(passdir, stat.S_IWUSR)
+
+ try:
+ # nonexistent passfile is OK
+ self.run_testcase({
+ 'host': 'abc',
+ 'user': 'user',
+ 'database': 'db',
+ 'passfile': passfile.name,
+ 'result': (
+ [('abc', 5432)],
+{
+ 'user': 'user',
+ 'database': 'db',
+ }
+ )
+ })
+ finally:
+ os.chmod(passdir, stat.S_IRWXU)
+
 async def test_connect_args_validation(self):
 for val in{-1, 'a', True, False, 0}:
 with self.assertRaisesRegex(ValueError, 'greater than 0'):

-Original file line number
+Diff line change
 def_read_password_file(passfile: pathlib.Path) \
 ->typing.List[typing.Tuple[str, ...]]:
 -ifnotpassfile.is_file():
 -warnings.warn(
 -'password file{!r} is not a plain file'.format(passfile))
 +passtab= []
 -returnNone
 +try:
 +ifnotpassfile.exists():
 +return []
 -if_system!='Windows':
 -ifpassfile.stat().st_mode& (stat.S_IRWXG|stat.S_IRWXO):
 +ifnotpassfile.is_file():
 warnings.warn(
 -'password file{!r} has group or world access; '
 -'permissions should be u=rw (0600) or less'.format(passfile))
 +'password file{!r} is not a plain file'.format(passfile))
 -returnNone
 +return[]
 -passtab= []
 +if_system!='Windows':
 +ifpassfile.stat().st_mode& (stat.S_IRWXG|stat.S_IRWXO):
 +warnings.warn(
 +'password file{!r} has group or world access; '
 +'permissions should be u=rw (0600) or less'.format(
 +passfile))
++
 +return []
 -try:
 withpassfile.open('rt') asf:
 forlineinf:
 line=line.strip()
  Password string, if found, ``None`` otherwise.
  """
 -ifnotpassfile.exists():
 -returnNone
+-
 passtab=_read_password_file(passfile)
 ifnotpasstab:
 returnNone