Watch history index - Replace per-snapshot realpath()/exists() with a single directory listing (#4562)

Watch.history ran os.path.realpath() twice and os.path.exists() once for every
line in history.txt (~10 syscalls per snapshot), on every watch at startup and
on every history read after that.

basename() already reduces each entry to a plain filename, so the only way out
of the watch data dir is a symlink. One os.scandir() of the data dir now
answers "does it exist", and only symlinks pay for a realpath() containment
check. scandir() never lists '.' or '..', so those are still rejected.

~13x faster on 200 watches x 500 snapshots (1.18s -> 0.09s, warm cache),
identical output. Adds tests for symlinks escaping/inside the data dir,
dangling symlinks, and missing snapshots/blank lines.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

changedetectionio/model/Watch.py

@@ -524,24 +524,35 @@ class model(EntityPersistenceMixin, watch_base):
         fname = os.path.join(self.data_dir, self.history_index_filename)
         if os.path.isfile(fname):
             logger.debug(f"Reading watch history index for {self.get('uuid')}")
+
+            # Entries restored from backup could contain absolute or traversal paths, never trust them.
+            # basename() reduces every entry to a plain name, so the only remaining way out of data_dir
+            # is a symlink. One directory listing replaces the per-line realpath() + exists() calls
+            # (~10 syscalls per snapshot); only symlinks still pay for a realpath() containment check.
+            # Note: scandir() never lists '.' or '..', so those are rejected as "not found".
+            safe_data_dir = os.path.realpath(self.data_dir)
+            with os.scandir(safe_data_dir) as it:
+                dir_entries = {e.name: e for e in it}
+
             with open(fname, "r", encoding='utf-8') as f:
                 for i in f.readlines():
                     if ',' in i:
                         k, v = i.strip().split(',', 2)
 
-                        # Always resolve history entries to within the watch's own data directory.
-                        # Entries restored from backup could contain absolute or traversal paths —
-                        # never trust them. Use realpath to also block symlink-based escapes.
-                        safe_data_dir = os.path.realpath(self.data_dir)
                         snapshot_fname = os.path.basename(v.strip())
-                        resolved_path = os.path.realpath(os.path.join(self.data_dir, snapshot_fname))
-
-                        if not resolved_path.startswith(safe_data_dir + os.sep) and resolved_path != safe_data_dir:
-                            logger.warning(f"Skipping unsafe history entry for {self.get('uuid')}: {v!r}")
+                        entry = dir_entries.get(snapshot_fname)
+                        if entry is None:
                             continue
 
-                        if not os.path.exists(resolved_path):
-                            continue
+                        resolved_path = os.path.join(safe_data_dir, snapshot_fname)
+                        if entry.is_symlink():
+                            resolved_path = os.path.realpath(resolved_path)
+                            if not resolved_path.startswith(safe_data_dir + os.sep):
+                                logger.warning(f"Skipping unsafe history entry for {self.get('uuid')}: {v!r}")
+                                continue
+                            # Dangling symlink, same as the file not existing
+                            if not os.path.exists(resolved_path):
+                                continue
 
                         tmp_history[k] = resolved_path
 

changedetectionio/tests/unit/test_watch_model.py

@@ -402,6 +402,47 @@ class TestHistoryPathTraversal(unittest.TestCase):
             "Resolved path must be inside the watch data directory"
         )
 
+    def test_symlink_pointing_outside_data_dir_is_rejected(self):
+        """A plain-named entry that is a symlink to a file outside data_dir must be rejected."""
+        watch = self._make_watch()
+        with tempfile.NamedTemporaryFile(delete=False) as outside:
+            outside.write(b'secret')
+        try:
+            os.symlink(outside.name, os.path.join(watch.data_dir, 'evil.txt'))
+            self._write_history_txt(watch, ['1000000000,evil.txt\n'])
+            self.assertEqual(watch.history, {}, "Symlink escaping data_dir must be rejected")
+        finally:
+            os.unlink(outside.name)
+
+    def test_symlink_inside_data_dir_is_accepted(self):
+        """A symlink that resolves to a file inside data_dir is fine."""
+        watch = self._make_watch()
+        target = os.path.join(watch.data_dir, 'real.txt')
+        with open(target, 'w') as f:
+            f.write('hello')
+        os.symlink(target, os.path.join(watch.data_dir, 'link.txt'))
+        self._write_history_txt(watch, ['1000000000,link.txt\n'])
+        self.assertEqual(list(watch.history.values()), [os.path.realpath(target)])
+
+    def test_dangling_symlink_is_skipped(self):
+        watch = self._make_watch()
+        os.symlink(os.path.join(watch.data_dir, 'gone.txt'), os.path.join(watch.data_dir, 'dangling.txt'))
+        self._write_history_txt(watch, ['1000000000,dangling.txt\n'])
+        self.assertEqual(watch.history, {})
+
+    def test_missing_snapshot_and_junk_lines_are_skipped(self):
+        """Entries whose snapshot file is gone, blank and whitespace-only lines don't count."""
+        import uuid as uuid_builder
+        watch = self._make_watch()
+        watch.save_history_blob(contents="one", timestamp=1000000001, snapshot_id=str(uuid_builder.uuid4()))
+        watch.save_history_blob(contents="two", timestamp=1000000002, snapshot_id=str(uuid_builder.uuid4()))
+        fname = os.path.join(watch.data_dir, watch.history_index_filename)
+        with open(fname, 'a', encoding='utf-8') as f:
+            f.write('\n   \n1000000003,does-not-exist.txt\n\n')
+        self.assertEqual(list(watch.history.keys()), ['1000000001', '1000000002'])
+        self.assertEqual(watch.history_n, 2)
+        self.assertEqual(watch.newest_history_key, '1000000002')
+
     def test_get_history_snapshot_blocks_outside_path_directly(self):
         """get_history_snapshot(filepath=...) must raise if the path escapes data_dir."""
         watch = self._make_watch()