locking: treat a lock object we are not permitted to read as unreadable

src/borg/storelocking.py

@@ -12,8 +12,9 @@
 them): wrapped in the repository key's envelope, encrypted and authenticated (just authenticated in the
 authenticated-* modes). The object's name is the store hash of the sealed content, so neither the
 content nor the name tells who uses the repository. A lock object that can not be unsealed (corrupt,
-tampered with, or not written by a borg with the repository key) is "unreadable": nothing is known
-about it but its store-side mtime, so it is treated as a foreign exclusive lock (see Staleness).
+tampered with, not written by a borg with the repository key, or not permitted to be read by us) is
+"unreadable": nothing is known about it but its store-side mtime, so it is treated as a foreign
+exclusive lock (see Staleness).
 Where the storage backend provides object timestamps (file, sftp, s3, current rest servers - not
 rclone), a lock object additionally carries a store-side mtime, stamped by the *storage's* clock at
 the same write instant; borgstore reports it as ItemInfo.mtime (0 if unavailable).
@@ -102,6 +103,7 @@
 import time
 from collections import namedtuple
 
+from borgstore.backends.errors import PermissionDenied
 from borgstore.store import ObjectNotFound
 
 from . import platform
@@ -465,7 +467,11 @@ def _get_locks(self):
                 # the lock vanished between our listing and loading it, e.g. it was released
                 # by its owner or another client killed it as stale - so just ignore it.
                 continue
-            lock = self._parse_lock(content)
+            except (PermissionError, PermissionDenied):
+                # e.g. created by another user: we can not tell what it is, so it is unreadable as well.
+                lock = None
+            else:
+                lock = self._parse_lock(content)
             if lock is None:
                 if not self.unreadable_warned:
                     self.unreadable_warned = True

src/borg/testsuite/storelocking_test.py

@@ -7,6 +7,7 @@
 
 import pytest
 
+from borgstore.backends.errors import PermissionDenied
 from borgstore.store import ObjectNotFound, Store
 
 from .fslocking_test import free_pid  # NOQA
@@ -565,3 +566,23 @@ def test_break_lock(self, lockstore, caplog):
         assert list(lockstore.list("locks")) == []
         with Lock(lockstore, exclusive=True, id=ID1):
             pass
+
+    @pytest.mark.parametrize("error", [PermissionError, PermissionDenied])
+    def test_permission_denied(self, lockstore, monkeypatch, error):
+        # e.g. a lock object created by root, listed by a non-root user.
+        key = write_unreadable_lock(lockstore, b"garbage")
+        orig_load = lockstore.load
+
+        def load_denied(name, *args, **kwargs):
+            if name == f"locks/{key}":
+                raise error(name)
+            return orig_load(name, *args, **kwargs)
+
+        monkeypatch.setattr(lockstore, "load", load_denied)
+        lock = Lock(lockstore, exclusive=False, id=ID1, timeout=0.2)
+        lock.retry_delay_min = lock.retry_delay_max = 0.05
+        with pytest.raises(LockTimeout, match=f"unreadable lock object locks/{key}"):
+            lock.acquire()
+        assert [info.name for info in lockstore.list("locks")] == [key]  # no lock of ours left behind
+        Lock(lockstore, id=ID1).break_lock()
+        assert list(lockstore.list("locks")) == []