fix(state): preserve SQLite locks during Linux permission hardening

hermes_state.py

@@ -7,6 +7,7 @@
 
 import asyncio
 import atexit
+import errno
 import hashlib
 import json
 import logging
@@ -15,6 +16,7 @@
 import random
 import re
 import sqlite3
+import stat
 import sys
 import threading
 import time
@@ -230,20 +232,34 @@ def _secure_state_db_files(db_path: Path, *, create_main: bool = False) -> None:
     if os.name == "nt":
         return
 
+    path_only = sys.platform.startswith("linux") and hasattr(os, "O_PATH")
     for index, path in enumerate(
         (
             db_path,
             db_path.with_name(db_path.name + "-wal"),
             db_path.with_name(db_path.name + "-shm"),
         )
     ):
-        flags = os.O_RDONLY
-        if index == 0 and create_main:
+        # A normal open/fchmod/close drops this process's SQLite POSIX locks.
+        # O_PATH pins the inode without participating in those locks.
+        flags = os.O_PATH if path_only else os.O_RDONLY
+        if index == 0 and create_main and not path_only:
             flags = os.O_WRONLY | os.O_CREAT
         if hasattr(os, "O_NOFOLLOW"):
             flags |= os.O_NOFOLLOW
         if hasattr(os, "O_CLOEXEC"):
             flags |= os.O_CLOEXEC
+        if path_only and index == 0 and create_main:
+            try:
+                created = os.open(
+                    path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC,
+                    0o600,
+                )
+            except FileExistsError:
+                pass
+            else:
+                # Only a newly created inode is closed, before SQLite opens it.
+                os.close(created)
         try:
             fd = os.open(path, flags, 0o600)
         except FileNotFoundError:
@@ -253,7 +269,18 @@ def _secure_state_db_files(db_path: Path, *, create_main: bool = False) -> None:
             # canonical error for this, and a directory leaks no row data.
             continue
         try:
-            os.fchmod(fd, 0o600)
+            if path_only:
+                mode = os.fstat(fd).st_mode
+                if stat.S_ISDIR(mode):
+                    continue
+                if stat.S_ISLNK(mode):
+                    raise OSError(errno.ELOOP, "Refusing a symlink database file", str(path))
+                if not stat.S_ISREG(mode):
+                    raise OSError(errno.EINVAL, "Database file is not regular", str(path))
+                # fchmod rejects O_PATH; procfs resolves this exact pinned inode.
+                os.chmod(f"/proc/self/fd/{fd}", 0o600)
+            else:
+                os.fchmod(fd, 0o600)
         finally:
             os.close(fd)
 

tests/test_state_permission_locks.py

@@ -0,0 +1,63 @@
+"""Permission hardening must preserve live SQLite locks and reject symlinks."""
+
+import errno
+import os
+import stat
+import subprocess
+import sys
+
+import pytest
+
+from hermes_state import SessionDB, _secure_state_db_files
+from hermes_state_dbfile import iter_deleted_sqlite_sidecar_holders
+
+
+@pytest.mark.linux_only
+def test_hardening_preserves_live_wal_generation_across_other_openers(tmp_path):
+    path = tmp_path / "state.db"
+    first = SessionDB(path)
+    second = SessionDB(path)
+    try:
+        for _ in range(3):
+            _secure_state_db_files(path, create_main=True)
+            subprocess.run(
+                [sys.executable, "-c",
+                 "import sqlite3,sys; c=sqlite3.connect(sys.argv[1]); "
+                 "c.execute('SELECT count(*) FROM sessions').fetchone(); c.close()",
+                 str(path)],
+                check=True, capture_output=True, text=True, timeout=10,
+            )
+            assert iter_deleted_sqlite_sidecar_holders(path) == []
+            first._raise_if_db_replaced()
+            first._conn.execute("BEGIN IMMEDIATE")
+            first._conn.rollback()
+    finally:
+        second.close()
+        first.close()
+    assert iter_deleted_sqlite_sidecar_holders(path) == []
+    reopened = SessionDB(path)
+    reopened.close()
+
+
+@pytest.mark.linux_only
+@pytest.mark.parametrize("suffix", ["", "-wal", "-shm"])
+def test_hardening_keeps_files_private_and_refuses_symlink_targets(tmp_path, suffix):
+    path = tmp_path / "state.db"
+    old_umask = os.umask(0)
+    try:
+        _secure_state_db_files(path, create_main=True)
+    finally:
+        os.umask(old_umask)
+    assert stat.S_IMODE(path.stat().st_mode) == 0o600
+    sidecar = path.with_name(path.name + suffix)
+    if sidecar.exists():
+        sidecar.unlink()
+    target = tmp_path / "unrelated-file"
+    target.write_text("keep this private test content unchanged", encoding="utf-8")
+    target.chmod(0o644)
+    sidecar.symlink_to(target)
+    with pytest.raises(OSError) as error:
+        _secure_state_db_files(path, create_main=True)
+    assert error.value.errno == errno.ELOOP
+    assert stat.S_IMODE(target.stat().st_mode) == 0o644
+    assert target.read_text(encoding="utf-8") == "keep this private test content unchanged"