POSIX owner-only checks: repair our own folders whose mode drifted, note drives without modes, and name the chown command for folders another account owns

strata_cli/bench.py

@@ -34,6 +34,7 @@ from pathlib import Path
 from strata_cli.client import ClientError, Endpoint, _NoRedirect
 from strata_control.profiles import ProfileError
 from strata_control import atomic as _atomic_mod
+from strata_control import posix_owner
 
 
 SCHEMA = 1
@@ -698,9 +699,8 @@ def progress(path: Path) -> dict:
     try:
         details = target.lstat()
         if not stat.S_ISREG(details.st_mode) or details.st_size > 4096 or \
-                (os.name == "posix" and (details.st_uid != os.getuid() or
-                                         stat.S_IMODE(details.st_mode) & 0o077)):
-            raise ProfileError("benchmark progress must be an owner-only regular file under 4 KiB", 5)
+                (os.name == "posix" and (posix_owner.owner_only_wrong(target, details))):
+            raise ProfileError(posix_owner.owner_only_text("benchmark progress must be an owner-only regular file under 4 KiB", target), 5)
         record = json.loads(target.read_text(encoding="utf-8"))
     except (OSError, UnicodeError, ValueError) as exc:
         if isinstance(exc, ProfileError):

strata_cli/connections.py

@@ -13,6 +13,14 @@ except ImportError:                                     # a thin bundle (monitor
     def _private_mkdir(path, *, parents=False, exist_ok=True):
         path.mkdir(mode=0o700, parents=parents, exist_ok=exist_ok)
         return path
+try:
+    from strata_control import posix_owner
+except ImportError:                                     # the same thin bundle: the old plain check
+    import stat as _stat
+
+    class posix_owner:                                  # noqa: N801
+        owner_only_wrong = staticmethod(lambda path, details: os.name == "posix" and (details.st_uid != os.getuid() or _stat.S_IMODE(details.st_mode) & 0o077))
+        owner_only_text = staticmethod(lambda text, path: text)
 try:
     from strata_control import atomic as _atomic_mod
 except ImportError:                                     # the same thin bundle: a plain replace
@@ -77,8 +85,8 @@ def connection_dir(*, create: bool = False) -> Path:
     if directory.exists() or directory.is_symlink():
         details = directory.lstat()
         if not stat.S_ISDIR(details.st_mode) or (os.name == "posix" and
-                (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077)):
-            raise ConnectionError("connection directory must be owner-only", 5)
+                (posix_owner.owner_only_wrong(directory, details))):
+            raise ConnectionError(posix_owner.owner_only_text("connection directory must be owner-only", directory), 5)
     return directory
 
 
@@ -122,9 +130,8 @@ def load(name: str) -> tuple[dict, str]:
     try:
         details = path.lstat()
         if not stat.S_ISREG(details.st_mode) or details.st_size > MAX_BYTES or \
-                (os.name == "posix" and (details.st_uid != os.getuid() or
-                                         stat.S_IMODE(details.st_mode) & 0o077)):
-            raise ConnectionError("connection file must be a private regular file under 4 KiB", 5)
+                (os.name == "posix" and (posix_owner.owner_only_wrong(path, details))):
+            raise ConnectionError(posix_owner.owner_only_text("connection file must be a private regular file under 4 KiB", path), 5)
         data = path.read_bytes()
         value = json.loads(data.decode("utf-8"))
     except FileNotFoundError:
@@ -187,8 +194,8 @@ def _lock(name: str):
     with os.fdopen(descriptor, "r+b") as stream:
         details = os.fstat(stream.fileno())
         if not stat.S_ISREG(details.st_mode) or (os.name == "posix" and
-                (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077)):
-            raise ConnectionError("connection lock must be owner-only", 5)
+                (posix_owner.owner_only_wrong(path, details))):
+            raise ConnectionError(posix_owner.owner_only_text("connection lock must be owner-only", path), 5)
         if os.name == "nt":
             import msvcrt
             stream.seek(0)

strata_control/benchmarking.py

@@ -17,6 +17,7 @@ from strata_cli.cli import _launch_plan, _source
 from strata_cli.client import ClientError, Endpoint
 from strata_control import journal, profiles, runtime
 from strata_control.profiles import ProfileError
+from strata_control import posix_owner
 
 
 #: Windows byte-range locks are mandatory: a locked byte cannot be read through any other handle. The lock
@@ -33,8 +34,8 @@ def _lease_file(profile: str, *, create: bool) -> Path | None:
         return None
     details = directory.lstat()
     if not stat.S_ISDIR(details.st_mode) or (os.name == "posix" and
-            (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077)):
-        raise ProfileError("benchmark lease directory must be owner-only", 5)
+            (posix_owner.owner_only_wrong(directory, details))):
+        raise ProfileError(posix_owner.owner_only_text("benchmark lease directory must be owner-only", directory), 5)
     return directory / f".bench-{profile}.lock"
 
 
@@ -45,9 +46,9 @@ def _lease_descriptor(path: Path, *, create: bool) -> int:
     descriptor = os.open(path, flags, 0o600)
     details = os.fstat(descriptor)
     if not stat.S_ISREG(details.st_mode) or details.st_nlink != 1 or details.st_size > 65 or (os.name == "posix" and
-            (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077)):
+            (posix_owner.owner_only_wrong(path, details))):
         os.close(descriptor)
-        raise ProfileError("benchmark lease file must be owner-only", 5)
+        raise ProfileError(posix_owner.owner_only_text("benchmark lease file must be owner-only", path), 5)
     return descriptor
 
 
@@ -139,8 +140,8 @@ def request_cancel(profile: str, path: Path) -> dict:
     directory = path.parent if path.name in ("results.json", "progress.json") else path
     details = directory.lstat()
     if not stat.S_ISDIR(details.st_mode) or (os.name == "posix" and
-            (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077)):
-        raise ProfileError("benchmark output directory must be owner-only", 5)
+            (posix_owner.owner_only_wrong(directory, details))):
+        raise ProfileError(posix_owner.owner_only_text("benchmark output directory must be owner-only", directory), 5)
     if benchmark_activity(profile, directory) != "active":
         raise ProfileError("the exact benchmark run is not active", 6)
     if bench.progress(directory)["state"] != "running":
@@ -152,7 +153,7 @@ def request_cancel(profile: str, path: Path) -> dict:
     except FileExistsError:
         details = marker.lstat()
         if not stat.S_ISREG(details.st_mode) or details.st_nlink != 1 or details.st_size != 1 or (os.name == "posix" and
-                (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077)):
+                (posix_owner.owner_only_wrong(marker, details))):
             raise ProfileError("benchmark cancellation marker is invalid", 5)
     else:
         try:
@@ -175,7 +176,7 @@ def cancel_requested(directory: Path) -> bool:
     except FileNotFoundError:
         return False
     if not stat.S_ISREG(details.st_mode) or details.st_nlink != 1 or details.st_size != 1 or (os.name == "posix" and
-            (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077)):
+            (posix_owner.owner_only_wrong(marker, details))):
         raise ProfileError("benchmark cancellation marker is invalid", 5)
     return True
 
@@ -321,8 +322,8 @@ def job_status(profile: str, job_id: str) -> dict:
     if directory.exists() or directory.is_symlink():
         details = directory.lstat()
         if not stat.S_ISDIR(details.st_mode) or (os.name == "posix" and
-                (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077)):
-            raise ProfileError("benchmark output directory must be owner-only", 5)
+                (posix_owner.owner_only_wrong(directory, details))):
+            raise ProfileError(posix_owner.owner_only_text("benchmark output directory must be owner-only", directory), 5)
     activity = benchmark_activity(profile, directory)
     try:
         phase = bench.progress(directory)

strata_control/calibration.py

@@ -27,6 +27,7 @@ from strata_control import profiles
 from strata_control.doctor import inspect as inspect_components
 from strata_control.profiles import ProfileError
 from strata_control import atomic as _atomic_mod
+from strata_control import posix_owner
 
 
 SUITE = "legacy-decode-v1"
@@ -461,9 +462,8 @@ def _backup_directory(path: Path, *, create: bool = False) -> Path:
     except OSError as exc:
         raise ProfileError(f"cannot access calibration backup directory ({type(exc).__name__})", 5) from None
     if not stat.S_ISDIR(details.st_mode) or (os.name == "posix" and
-                                            (details.st_uid != os.getuid() or
-                                             stat.S_IMODE(details.st_mode) & 0o077)):
-        raise ProfileError("calibration backup directory must be owner-only and not a symlink", 5)
+                                            (posix_owner.owner_only_wrong(directory, details))):
+        raise ProfileError(posix_owner.owner_only_text("calibration backup directory must be owner-only and not a symlink", directory), 5)
     return directory
 
 
@@ -483,9 +483,8 @@ def _load_application(path: Path, report: dict) -> dict:
     try:
         details = manifest.lstat()
         if not stat.S_ISREG(details.st_mode) or details.st_size > MAX_APPLICATION_BYTES or \
-                (os.name == "posix" and (details.st_uid != os.getuid() or
-                                         stat.S_IMODE(details.st_mode) & 0o077)):
-            raise ProfileError("calibration application record is not an owner-only regular file", 5)
+                (os.name == "posix" and (posix_owner.owner_only_wrong(manifest, details))):
+            raise ProfileError(posix_owner.owner_only_text("calibration application record is not an owner-only regular file", manifest), 5)
         data = json.loads(manifest.read_text(encoding="utf-8"))
     except FileNotFoundError:
         raise ProfileError("calibration application record was not found", 5) from None
@@ -634,8 +633,8 @@ def _saved_bytes(path: Path, *, required: bool = True, owner_only: bool = False)
         raise ProfileError("calibration rollback file is missing", 5) from None
     if not stat.S_ISREG(details.st_mode) or details.st_size > MAX_REPORT_BYTES or \
             (owner_only and os.name == "posix" and
-             (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077)):
-        raise ProfileError("calibration rollback file must be a regular owner-only file under 2 MiB", 5)
+             (posix_owner.owner_only_wrong(path, details))):
+        raise ProfileError(posix_owner.owner_only_text("calibration rollback file must be a regular owner-only file under 2 MiB", path), 5)
     try:
         return path.read_bytes()
     except OSError as exc:

strata_control/calibration_jobs.py

@@ -14,6 +14,7 @@ from pathlib import Path
 
 from strata_control import calibration, journal, profiles
 from strata_control.profiles import ProfileError
+from strata_control import posix_owner
 
 
 KINDS = {"measure": "engine.calibrate.measure", "apply": "engine.calibrate.apply",
@@ -43,9 +44,9 @@ def _open_lease(profile: str, *, create: bool) -> int | None:
     details = os.fstat(descriptor)
     if not stat.S_ISREG(details.st_mode) or details.st_nlink != 1 or details.st_size > 32 or \
             (os.name == "posix" and
-             (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077)):
+             (posix_owner.owner_only_wrong(_lease_path(profile), details))):
         os.close(descriptor)
-        raise ProfileError("calibration worker lease must be owner-only", 5)
+        raise ProfileError(posix_owner.owner_only_text("calibration worker lease must be owner-only", _lease_path(profile)), 5)
     return descriptor
 
 
@@ -129,7 +130,7 @@ def _cancel_requested(profile: str, job_id: str) -> bool:
         return False
     if not stat.S_ISREG(details.st_mode) or details.st_nlink != 1 or details.st_size != 1 or \
             (os.name == "posix" and
-             (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077)):
+             (posix_owner.owner_only_wrong(path, details))):
         raise ProfileError("calibration cancellation marker is invalid", 5)
     descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
     try:
@@ -187,8 +188,8 @@ def _load_progress(profile: str, job_id: str) -> dict | None:
         return None
     if not stat.S_ISREG(details.st_mode) or details.st_nlink != 1 or details.st_size > 1024 or \
             (os.name == "posix" and
-             (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077)):
-        raise ProfileError("calibration progress file must be owner-only", 5)
+             (posix_owner.owner_only_wrong(path, details))):
+        raise ProfileError(posix_owner.owner_only_text("calibration progress file must be owner-only", path), 5)
     try:
         value = json.loads(path.read_text(encoding="utf-8"))
     except (OSError, UnicodeError, ValueError):

strata_control/journal.py

@@ -19,6 +19,7 @@ from pathlib import Path
 
 from strata_control import profiles, recipes
 from strata_control.profiles import ProfileError
+from strata_control import posix_owner
 
 
 MAX_RECORDS = 500
@@ -51,8 +52,8 @@ def _directory(profile: str, *, create: bool = False) -> Path:
         details = directory.lstat()
         if not stat.S_ISDIR(details.st_mode):
             raise ProfileError("operation journal directory must not be a symlink", 5)
-        if os.name == "posix" and (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077):
-            raise ProfileError("operation journal directory must be owner-only", 5)
+        if os.name == "posix" and (posix_owner.owner_only_wrong(directory, details)):
+            raise ProfileError(posix_owner.owner_only_text("operation journal directory must be owner-only", directory), 5)
     return directory
 
 
@@ -186,8 +187,8 @@ def _load_file(path: Path, profile: str) -> dict:
         details = path.lstat()
         if not stat.S_ISREG(details.st_mode) or details.st_size > MAX_RECORD_BYTES:
             raise ProfileError("operation record must be a regular file under 64 KiB", 5)
-        if os.name == "posix" and (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077):
-            raise ProfileError("operation record must be owner-only", 5)
+        if os.name == "posix" and (posix_owner.owner_only_wrong(path, details)):
+            raise ProfileError(posix_owner.owner_only_text("operation record must be owner-only", path), 5)
         record = json.loads(path.read_text(encoding="utf-8"))
     except FileNotFoundError:
         raise ProfileError("operation record was not found", 5) from None

strata_control/posix_owner.py

@@ -0,0 +1,100 @@
+"""POSIX 'owner-only' checks that repair what is ours and explain what is not.
+
+Strata keeps its private state in folders (0700) and files (0600) of the user. The checks used to refuse as soon as a mode had drifted
+(a `chmod -R`, an rsync, a restored backup) or the owner was not the current user (one `sudo strata ...` leaves root-owned folders behind).
+Now:
+
+* the folder or file is ours and only the mode bits are wrong: the mode is put back (0700 / 0600) and the command goes on;
+* the filesystem cannot hold modes at all (a Windows drive seen from WSL, vfat, some network shares): one note, then it goes on;
+* another account owns it (root after sudo): the check still refuses, because that is the real boundary, but the message names the
+  cause and the exact command that fixes it.
+
+Windows has no mode bits; its own check lives in `private_files` and is not touched here.
+"""
+
+from __future__ import annotations
+
+import os
+import stat
+import sys
+from pathlib import Path
+
+POSIX = os.name == "posix"
+_NOTED: set[str] = set()
+
+
+def owner_only_wrong(path: str | os.PathLike, details: os.stat_result) -> bool:
+    """True when `path` (whose lstat or fstat is `details`) is not private to this user and could not be made so.
+
+    Replaces `details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077`. Never follows a link and never touches
+    a file that another account owns.
+    """
+    if not POSIX:
+        return False
+    if details.st_uid != os.getuid():
+        return True
+    if not stat.S_IMODE(details.st_mode) & 0o077:
+        return False
+    if stat.S_ISLNK(details.st_mode):
+        return True
+    want = 0o700 if stat.S_ISDIR(details.st_mode) else 0o600
+    try:
+        os.chmod(path, want)                              # not a link (checked above), so this is the file itself
+        after = os.lstat(path)
+    except (OSError, NotImplementedError):
+        after = None
+    if after is not None and after.st_uid == os.getuid() and not stat.S_IMODE(after.st_mode) & 0o077:
+        return False
+    # The mode did not stick although the owner is right: the filesystem has no real modes. Say it once and carry on.
+    folder = str(path if stat.S_ISDIR(details.st_mode) else Path(path).parent)
+    if folder not in _NOTED:
+        _NOTED.add(folder)
+        print(f"strata: {folder} is on a drive that cannot keep files private to your account (a Windows drive seen from Linux, "
+              "a USB stick or a network share). Strata goes on; keep your data on the Linux disk to make it private.", file=sys.stderr)
+    return False
+
+
+def _owner_name(uid: int) -> str:
+    try:
+        import pwd
+        return pwd.getpwuid(uid).pw_name
+    except (ImportError, KeyError):
+        return f"user {uid}"
+
+
+def fix_target(path: str | os.PathLike) -> Path:
+    """The highest folder above `path` (below the home folder) that another account owns: one `chown` there fixes the whole tree."""
+    target = Path(path)
+    home = Path.home()
+    uid = os.getuid()
+    best = target
+    current = target
+    for _ in range(6):
+        parent = current.parent
+        if parent == current or parent == home or parent == Path(parent.anchor):
+            break
+        try:
+            if parent.lstat().st_uid == uid:
+                break
+        except OSError:
+            break
+        best = parent
+        current = parent
+    return best
+
+
+def owner_only_text(text: str, path: str | os.PathLike) -> str:
+    """The message for a refused check. `text` is the old sentence ("recipe directory must be owner-only"); when another account owns
+    `path` it becomes a sentence with the cause and the exact command, otherwise it is returned unchanged."""
+    if not POSIX:
+        return text
+    try:
+        owner = os.lstat(path).st_uid
+    except OSError:
+        return text
+    if owner == os.getuid():
+        return text
+    subject = text.split(" must ")[0].split(" is not ")[0].strip()
+    target = fix_target(path)
+    return (f"{subject} belongs to {_owner_name(owner)}, not to you (usually from running Strata once with sudo). "
+            f"Run this once, then try again: sudo chown -R \"$USER\" {target}")

strata_control/recipes.py

@@ -22,6 +22,7 @@ from pathlib import Path
 
 from strata_control import profiles
 from strata_control.profiles import ProfileError
+from strata_control import posix_owner
 
 
 KINDS = frozenset({"runtime.restart", "engine.calibrate.measure", "components.verify"})
@@ -44,8 +45,8 @@ def _directory(profile: str, *, create: bool = False) -> Path:
         details = path.lstat()
         if not stat.S_ISDIR(details.st_mode):
             raise ProfileError("recipe directory must not be a symlink", 5)
-        if os.name == "posix" and (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077):
-            raise ProfileError("recipe directory must be owner-only (mode 0700)", 5)
+        if os.name == "posix" and (posix_owner.owner_only_wrong(path, details)):
+            raise ProfileError(posix_owner.owner_only_text("recipe directory must be owner-only (mode 0700)", path), 5)
     if create:
         _private_mkdir(directory.parent, exist_ok=True)
         _private_mkdir(directory, exist_ok=True)
@@ -53,8 +54,8 @@ def _directory(profile: str, *, create: bool = False) -> Path:
         details = directory.lstat()
         if not stat.S_ISDIR(details.st_mode):
             raise ProfileError("recipe directory must not be a symlink", 5)
-        if os.name == "posix" and (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077):
-            raise ProfileError("recipe directory must be owner-only (mode 0700)", 5)
+        if os.name == "posix" and (posix_owner.owner_only_wrong(directory, details)):
+            raise ProfileError(posix_owner.owner_only_text("recipe directory must be owner-only (mode 0700)", directory), 5)
     return directory
 
 
@@ -80,8 +81,8 @@ def _history_dir(profile: str, name: str, *, create: bool = False) -> Path:
         if path.exists():
             details = path.lstat()
             if not stat.S_ISDIR(details.st_mode) or (os.name == "posix" and
-                    (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077)):
-                raise ProfileError("recipe history directory must be owner-only", 5)
+                    (posix_owner.owner_only_wrong(path, details))):
+                raise ProfileError(posix_owner.owner_only_text("recipe history directory must be owner-only", path), 5)
     return directory
 
 
@@ -182,8 +183,8 @@ def load(profile: str, name: str) -> tuple[dict, str]:
         details = path.lstat()
         if not stat.S_ISREG(details.st_mode) or details.st_size > MAX_BYTES:
             raise ProfileError("recipe must be a regular file under 64 KiB", 5)
-        if os.name == "posix" and (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077):
-            raise ProfileError("recipe file must be owner-only (mode 0600)", 5)
+        if os.name == "posix" and (posix_owner.owner_only_wrong(path, details)):
+            raise ProfileError(posix_owner.owner_only_text("recipe file must be owner-only (mode 0600)", path), 5)
         data = path.read_bytes()
         value = json.loads(data.decode("utf-8-sig"))
     except FileNotFoundError:
@@ -227,7 +228,7 @@ def save(profile: str, name: str, proposed: dict, expected_revision: str) -> tup
         if backup.exists() or backup.is_symlink():
             details = backup.lstat()
             if not stat.S_ISREG(details.st_mode) or (os.name == "posix" and
-                    (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077)) or \
+                    (posix_owner.owner_only_wrong(backup, details))) or \
                     backup.read_bytes() != old:
                 raise ProfileError("recipe history conflicts with the saved revision", 5)
         else:
@@ -258,8 +259,8 @@ def rollback(profile: str, name: str, target_revision: str) -> tuple[dict, str,
         details = path.lstat()
         if not stat.S_ISREG(details.st_mode) or details.st_size > MAX_BYTES:
             raise ProfileError("recipe history must be a regular file under 64 KiB", 5)
-        if os.name == "posix" and (details.st_uid != os.getuid() or stat.S_IMODE(details.st_mode) & 0o077):
-            raise ProfileError("recipe history file must be owner-only", 5)
+        if os.name == "posix" and (posix_owner.owner_only_wrong(path, details)):
+            raise ProfileError(posix_owner.owner_only_text("recipe history file must be owner-only", path), 5)
         data = path.read_bytes()
         if profiles._revision(data) != target_revision:
             raise ProfileError("recipe history content does not match its revision", 5)

strata_control/runtime.py

@@ -38,7 +38,7 @@ try:
 except ImportError:  # Windows has no qualified owner transport yet.
     fcntl = None
 
-from strata_control import owner_transport, paths, private_files, procs, profiles, readiness, winproc
+from strata_control import owner_transport, paths, posix_owner, private_files, procs, profiles, readiness, winproc
 from serve import launch_identity
 from strata_control.profiles import ProfileError
 
@@ -56,7 +56,7 @@ def _is_private(details: os.stat_result, path: Path) -> bool:
             return private_files.is_unshared(path)
         except OSError:
             return False
-    return details.st_uid == os.getuid() and not stat.S_IMODE(details.st_mode) & 0o077
+    return not posix_owner.owner_only_wrong(path, details)
 
 
 def _lock_file(descriptor: int, *, exclusive: bool) -> bool:

tests/terminal/test_connections.py

@@ -3,6 +3,7 @@
 import io
 import json
 import os
+import stat
 import subprocess
 import sys
 import tempfile
@@ -149,10 +150,12 @@ class ConnectionTests(unittest.TestCase):
             connections.create("local", "http://127.0.0.1:8095")
         if os.name == "posix":
             record = self.directory / "local.json"
-            record.chmod(0o644)
-            with self.assertRaisesRegex(connections.ConnectionError, "private regular"):
-                connections.load("local")
-            record.chmod(0o600)
+            record.chmod(0o644)                          # our own file with drifted modes is put back, not refused
+            connections.load("local")
+            self.assertEqual(stat.S_IMODE(record.stat().st_mode), 0o600)
+            with mock.patch("os.getuid", return_value=os.getuid() + 1):      # a file that belongs to another account is refused
+                with self.assertRaisesRegex(connections.ConnectionError, "belongs to .*sudo chown"):
+                    connections.load("local")
         record = self.directory / "local.json"
         corrupted = json.loads(record.read_text())
         corrupted["url"] = 42

tests/terminal/test_posix_owner.py

@@ -0,0 +1,88 @@
+"""POSIX owner-only checks: our own folders with drifted modes are repaired; another account's folders get a message with the exact fix."""
+
+import os
+import stat
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+from strata_control import posix_owner
+
+posix_only = unittest.skipUnless(os.name == "posix", "POSIX modes")
+
+
+@posix_only
+class Repair(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.root = Path(self.tmp.name)
+        self.addCleanup(self.tmp.cleanup)
+
+    def test_our_folder_with_open_modes_is_put_back_to_0700(self):
+        d = self.root / "recipes"
+        d.mkdir()
+        os.chmod(d, 0o755)
+        self.assertFalse(posix_owner.owner_only_wrong(d, d.lstat()))
+        self.assertEqual(stat.S_IMODE(d.lstat().st_mode), 0o700)
+
+    def test_our_file_with_open_modes_is_put_back_to_0600(self):
+        f = self.root / "x.json"
+        f.write_text("{}")
+        os.chmod(f, 0o666)
+        self.assertFalse(posix_owner.owner_only_wrong(f, f.lstat()))
+        self.assertEqual(stat.S_IMODE(f.lstat().st_mode), 0o600)
+
+    def test_a_private_one_is_left_alone(self):
+        d = self.root / "ok"
+        d.mkdir(mode=0o700)
+        self.assertFalse(posix_owner.owner_only_wrong(d, d.lstat()))
+
+    def test_a_link_is_never_followed_or_repaired(self):
+        target = self.root / "t"
+        target.mkdir()
+        os.chmod(target, 0o755)
+        link = self.root / "l"
+        os.symlink(target, link)
+        self.assertTrue(posix_owner.owner_only_wrong(link, link.lstat()))
+        self.assertEqual(stat.S_IMODE(target.lstat().st_mode), 0o755)
+
+    def test_a_folder_owned_by_someone_else_is_refused_and_not_touched(self):
+        d = self.root / "root-made"
+        d.mkdir()
+        os.chmod(d, 0o755)
+        with mock.patch.object(posix_owner.os, "getuid", return_value=os.getuid() + 1):
+            self.assertTrue(posix_owner.owner_only_wrong(d, d.lstat()))
+            text = posix_owner.owner_only_text("recipe directory must be owner-only (mode 0700)", d)
+        self.assertEqual(stat.S_IMODE(d.lstat().st_mode), 0o755)
+        self.assertIn("recipe directory belongs to", text)
+        self.assertIn("sudo", text)
+        self.assertIn("sudo chown -R \"$USER\"", text)
+        self.assertNotIn("0700", text)
+
+    def test_a_filesystem_that_keeps_no_modes_warns_once_and_goes_on(self):
+        d = self.root / "drvfs"
+        d.mkdir()
+        os.chmod(d, 0o777)
+        posix_owner._NOTED.discard(str(d))
+        with mock.patch.object(posix_owner.os, "chmod"), mock.patch("sys.stderr") as err:
+            self.assertFalse(posix_owner.owner_only_wrong(d, d.lstat()))
+            self.assertFalse(posix_owner.owner_only_wrong(d, d.lstat()))
+        printed = "".join(str(c.args[0]) for c in err.write.call_args_list)
+        self.assertEqual(printed.count("cannot keep files private"), 1)
+
+    def test_the_text_of_our_own_folder_is_unchanged(self):
+        d = self.root / "mine"
+        d.mkdir()
+        self.assertEqual(posix_owner.owner_only_text("x must be owner-only", d), "x must be owner-only")
+
+
+class Windows(unittest.TestCase):
+    def test_nothing_changes_off_posix(self):
+        with mock.patch.object(posix_owner, "POSIX", False):
+            self.assertFalse(posix_owner.owner_only_wrong(".", os.stat(".")))
+            self.assertEqual(posix_owner.owner_only_text("x", "."), "x")
+
+
+if __name__ == "__main__":
+    unittest.main()