Handle and ignore permission errors when attempting to read .pgpass

Fixes: #356

Author: elprans

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

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'):