Skip backup paths with symlinked ancestors

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 ## Unreleased
 
+- Skip backup saves and settings reached through a symbolic link anywhere below
+  the home folder, and recheck paths before packing each file. Local synthetic
+  tests cover redirected Proton prefixes and links changed after discovery;
+  concurrent filesystem changes can still race the checks.
+  [Backup layout and limits](docs/backup-and-recovery.md).
+
 - Import a base APK with supported native-library or asset splits from a local
   Android backup or XAPK/APKS/APKM bundle. Check package and version before
   install; refuse resource, code and feature splits and device-variant sets.

docs/backup-and-recovery.md

@@ -100,7 +100,21 @@ copy one by hand if you need it.
 - Steam Cloud's own folders (`userdata/<account>/<appid>`), shader caches and
   screenshots (**Tools → Capture & Live** saves those to Pictures).
 - Android apps themselves, their OBB files and shared storage (`/sdcard`).
-- Symbolic links (never followed), and single files over 2 GiB.
+- Symbolic links, including any ancestor below the home folder of a selected
+  save or settings path, and single files over 2 GiB. Discovery checks each
+  component, and packing checks it again before opening each file.
+
+Steam backups use the normal `~/.local/share/Steam` directory tree. A separate
+`~/.steam/steam` alias does not affect this route. A library, Proton prefix or
+Wine save folder redirected by a symlink below the home folder is skipped,
+even if its target is inside home; it is not resolved and backed up elsewhere.
+The home folder itself is trusted and may use the host's normal path aliases.
+
+These are path checks, not a filesystem snapshot: a concurrent rename or link
+replacement between a check and a read can still race it. Close games and
+avoid moving or relinking these folders during backup. This fix rejects links
+already present when checked, including ones introduced after discovery; it
+does not provide atomic protection against concurrent filesystem changes.
 
 Backups aren't encrypted. Game folders can still hold a launcher's login, and
 Android app data can hold an app's, so keep backups private.
@@ -192,11 +206,18 @@ If the button can't help, or the headset isn't reachable:
 `tests/test_backup.py` runs the headset-side script (`ui/frame_backup_remote.py`)
 for real against a folder laid out like the Frame's home, with only SSH faked:
 which paths are picked, that sign-in files and keys never are (also inside game
-folders), links never followed, the manifest and its checksums, restore checks
+folders), links in save/settings roots and their ancestors refused, the manifest
+and its checksums, restore checks
 after a reset (game not installed, no Proton folders yet, not signed in), that a
 restore replaces only its own files and keeps the old ones, unsafe archives
 (`..`, absolute paths, links, other games' folders) writing nothing, backup
 names that try to leave the backups folder, and the exact Game Mode command.
+Synthetic symlink fixtures cover every component of a Proton save root below
+home, including `compatdata/<appid>` and `steamuser/AppData`, plus directory
+and file links introduced between discovery and packing. Ordinary Proton and
+non-Steam saves, Linux saves, controller layouts, reference settings and home
+aliases still pack with their original bytes. These are local fixture checks;
+this path fix has not been tested on a headset.
 `tests/e2e/test_recovery_kit.py` backs up a Proton save from the fake Frame through
 the server, deletes it and restores it. Neither involves a real headset.
 

tests/test_backup.py

@@ -113,6 +113,108 @@ class Remote(unittest.TestCase):
         os.symlink(str(outside), str(base / 'linked-file'))
         self.assertFalse([f for f in self.files('saves:1234') if 'linked' in f])
 
+    def packed(self, part):
+        out = io.BytesIO()
+        remote.pack(part, out)
+        out.seek(0)
+        with tarfile.open(fileobj=out, mode='r:gz') as archive:
+            return {m.name: archive.extractfile(m).read() for m in archive}
+
+    @unittest.skipIf(os.name == 'nt', 'symlinks need privileges on Windows')
+    def test_backup_rejects_every_ancestor_of_a_save_root(self):
+        save = USER % 1234 + '/AppData/Roaming/Game/fixture.dat'
+        # Each component below HOME, including the selected root itself.
+        components = (USER % 1234 + '/AppData/Roaming').split('/')
+        for end in range(1, len(components) + 1):
+            ancestor = '/'.join(components[:end])
+            with self.subTest(ancestor=ancestor), tempfile.TemporaryDirectory() as tmp:
+                home = Path(tmp) / 'home'
+                manifest(home, 1234, 'Synthetic game')
+                path = write(home, save, b'synthetic fixture')
+                linked = home / ancestor
+                outside = Path(tmp) / 'outside/.ssh'
+                outside.parent.mkdir()
+                linked.rename(outside)
+                linked.symlink_to(outside, target_is_directory=True)
+                # Only synthetic bytes: the lexical archive path hides .ssh.
+                self.assertEqual(path.read_bytes(), b'synthetic fixture')
+                with patch.object(remote, 'HOME', home):
+                    self.assertEqual(self.files('saves:1234'), [])
+                    self.assertNotIn('saves:1234', [p['id'] for p in remote.plan()['parts']])
+                    self.assertEqual(self.packed('saves:1234'), {})
+
+    @unittest.skipIf(os.name == 'nt', 'symlinks need privileges on Windows')
+    def test_pack_rechecks_links_changed_after_discovery(self):
+        save = USER % 1234 + '/Documents/Game/fixture.dat'
+        for ancestor in (remote.COMPAT + '/1234', USER % 1234 + '/Documents', save):
+            with self.subTest(ancestor=ancestor), tempfile.TemporaryDirectory() as tmp:
+                home = Path(tmp) / 'home'
+                write(home, save, b'original save')
+                linked = home / ancestor
+                outside = Path(tmp) / 'outside/.ssh'
+                outside.parent.mkdir()
+                part_files = remote.part_files
+
+                def discover_then_link(part):
+                    result = part_files(part)
+                    self.assertEqual(result[0], [(save, len(b'original save'))])
+                    linked.rename(outside)
+                    linked.symlink_to(outside, target_is_directory=outside.is_dir())
+                    (home / save).write_bytes(b'synthetic redirected bytes')
+                    return result
+
+                with patch.object(remote, 'HOME', home), \
+                        patch.object(remote, 'part_files', side_effect=discover_then_link):
+                    self.assertEqual(self.packed('saves:1234'), {})
+
+    def test_legitimate_steam_layout_and_reference_files_still_pack(self):
+        frame_home(self.home)
+        manifest(self.home, 2800000002, 'Synthetic non-Steam game')
+        save = USER % 2800000002 + '/Saved Games/fixture.dat'
+        write(self.home, save, b'non-Steam save')
+        controller = STEAM + '/steamapps/common/Steam Controller Configs/555/config/layout.vdf'
+        write(self.home, controller, b'controller layout')
+        for part in ('saves:1234', 'saves:2800000002', 'saves:linux',
+                     'settings:steam', 'settings:system'):
+            with self.subTest(part=part):
+                expected = {rel: (self.home / rel).read_bytes() for rel in self.files(part)}
+                self.assertTrue(expected)
+                self.assertEqual(self.packed(part), expected)
+        self.assertEqual(self.packed('saves:2800000002'), {save: b'non-Steam save'})
+        self.assertIn(controller, self.packed('settings:steam'))
+
+    @unittest.skipIf(os.name == 'nt', 'symlinks need privileges on Windows')
+    def test_settings_discovery_and_manifest_reads_reject_links(self):
+        frame_home(self.home)
+        for ancestor, part in (('.config', 'settings:system'),
+                               (STEAM + '/userdata', 'settings:steam')):
+            with self.subTest(ancestor=ancestor):
+                linked = self.home / ancestor
+                moved = linked.with_name(linked.name + '-original')
+                linked.rename(moved)
+                linked.symlink_to(moved, target_is_directory=True)
+                self.assertEqual(self.files(part), [])
+                self.assertEqual(self.packed(part), {})
+                if part == 'settings:steam':
+                    self.assertEqual(remote.accounts(), [])
+        appmanifest = self.home / STEAM / 'steamapps/appmanifest_1234.acf'
+        moved = appmanifest.with_suffix('.original')
+        appmanifest.rename(moved)
+        appmanifest.symlink_to(moved)
+        with patch.object(Path, 'read_text', side_effect=AssertionError('must not read a link')):
+            self.assertIsNone(remote.game_name('1234'))
+
+    @unittest.skipIf(os.name == 'nt', 'symlinks need privileges on Windows')
+    def test_home_alias_and_unused_steam_alias_preserve_normal_backups(self):
+        frame_home(self.home)
+        alias = self.home / 'home-alias'
+        alias.symlink_to(self.home, target_is_directory=True)
+        (self.home / '.steam/steam').symlink_to(self.home / STEAM, target_is_directory=True)
+        expected = self.packed('saves:1234')
+        with patch.object(remote, 'HOME', alias):
+            self.assertEqual(self.packed('saves:1234'), expected)
+            self.assertEqual(remote.game_name('1234'), 'Test Game')
+
     def test_big_files_are_listed_as_skipped(self):
         frame_home(self.home)
         with patch.object(remote, 'MAX_FILE', 3):

ui/frame_backup_remote.py

@@ -19,6 +19,7 @@ import os
 from pathlib import Path, PurePosixPath
 import re
 import shutil
+import stat
 import subprocess
 import sys
 import tarfile
@@ -88,28 +89,57 @@ def _under(rel, root):
     return rel == root or rel.startswith(root + '/')
 
 
+def _backup_stat(rel):
+    """lstat a backup path only if every component below trusted HOME is link-free.
+
+    Rechecked at discovery and before reads; not atomic against concurrent renames.
+    HOME itself may use the host's normal home/temp-directory aliases.
+    """
+    if not safe_rel(rel):
+        return None
+    path = HOME
+    try:
+        for component in rel.split('/'):
+            path = path / component
+            st = path.lstat()
+            if stat.S_ISLNK(st.st_mode):
+                return None
+        return st
+    except OSError:
+        return None
+
+
+def _backup_dir(rel):
+    st = _backup_stat(rel)
+    return st is not None and stat.S_ISDIR(st.st_mode)
+
+
 def _walk(rel_root, skip=()):
     """Regular files under HOME/rel_root, never following links: [(rel, size)], [skipped]."""
     files, skipped = [], []
     base = HOME / rel_root
-    if base.is_symlink() or not base.is_dir():
-        if base.is_file() and not base.is_symlink():
-            return [(rel_root, base.stat().st_size)], skipped
+    st = _backup_stat(rel_root)
+    if st is None or is_secret(rel_root):
+        return files, skipped
+    if stat.S_ISREG(st.st_mode):
+        if st.st_size > MAX_FILE:
+            skipped.append({'path': rel_root, 'reason': 'over 2 GiB'})
+            return files, skipped
+        return [(rel_root, st.st_size)], skipped
+    if not stat.S_ISDIR(st.st_mode):
         return files, skipped
     for top, dirs, names in os.walk(str(base)):
         rel_top = PurePosixPath(rel_root, *Path(top).relative_to(base).parts).as_posix()
-        dirs[:] = sorted(d for d in dirs if not os.path.islink(os.path.join(top, d)) and
+        if not _backup_dir(rel_top):
+            dirs[:] = []
+            continue
+        dirs[:] = sorted(d for d in dirs if _backup_dir(rel_top + '/' + d) and
                          not any(_under(rel_top + '/' + d, s) for s in skip) and
                          not is_secret(rel_top + '/' + d + '/x'))
         for name in sorted(names):
-            path, rel = os.path.join(top, name), rel_top + '/' + name
-            if os.path.islink(path) or is_secret(rel):
-                continue
-            try:
-                st = os.lstat(path)
-            except OSError:
-                continue
-            if not os.path.isfile(path):
+            rel = rel_top + '/' + name
+            st = _backup_stat(rel)
+            if st is None or not stat.S_ISREG(st.st_mode) or is_secret(rel):
                 continue
             if st.st_size > MAX_FILE:
                 skipped.append({'path': rel, 'reason': 'over 2 GiB'})
@@ -124,26 +154,34 @@ def _manifest_field(text, key):
 
 
 def game_name(appid):
+    rel = '%s/steamapps/appmanifest_%s.acf' % (STEAM, appid)
+    st = _backup_stat(rel)
+    if st is None or not stat.S_ISREG(st.st_mode):
+        return None
     try:
-        text = (HOME / STEAM / 'steamapps' / ('appmanifest_%s.acf' % appid)).read_text(errors='replace')
+        text = (HOME / rel).read_text(errors='replace')
     except OSError:
         return None
     return _manifest_field(text, 'name')
 
 
 def accounts():
-    root = HOME / STEAM / 'userdata'
+    rel = STEAM + '/userdata'
+    if not _backup_dir(rel):
+        return []
     try:
-        return sorted(p.name for p in root.iterdir() if p.name.isdigit() and p.is_dir()
-                      and not p.is_symlink() and p.name != '0')
+        return sorted(p.name for p in (HOME / rel).iterdir() if p.name.isdigit() and
+                      p.name != '0' and _backup_dir(rel + '/' + p.name))
     except OSError:
         return []
 
 
 def _prefixes():
+    if not _backup_dir(COMPAT):
+        return []
     try:
         return sorted((p.name for p in (HOME / COMPAT).iterdir()
-                       if p.name.isdigit() and (p / PREFIX_USER).is_dir()), key=int)
+                       if p.name.isdigit() and _backup_dir(COMPAT + '/' + p.name + '/' + PREFIX_USER)), key=int)
     except OSError:
         return []
 
@@ -244,9 +282,14 @@ def pack(part, output):
     with tarfile.open(fileobj=output, mode='w|gz') as archive:
         for rel, _ in files:
             path = HOME / rel
+            st = _backup_stat(rel)
+            if st is None or not stat.S_ISREG(st.st_mode) or is_secret(rel):
+                continue
             try:
                 with open(str(path), 'rb') as src:
                     st = os.fstat(src.fileno())
+                    if not stat.S_ISREG(st.st_mode) or st.st_size > MAX_FILE:
+                        continue
                     member = tarfile.TarInfo(rel)
                     member.size, member.mtime, member.mode = st.st_size, int(st.st_mtime), st.st_mode & 0o755
                     archive.addfile(member, src)