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

hermes_state.py

@@ -229,6 +229,10 @@ def _secure_state_db_files(db_path: Path, *, create_main: bool = False) -> None:
     """
     if os.name == "nt":
         return
+    if sys.platform == "darwin":
+        from hermes_state_dbfile import secure_state_db_files_macos
+        secure_state_db_files_macos(db_path, create_main=create_main)
+        return
 
     for index, path in enumerate(
         (

hermes_state_dbfile.py

@@ -10,12 +10,14 @@
 from __future__ import annotations
 
 import contextlib
+import errno
 import hashlib
 import json
 import logging
 import os
 import shutil
 import sqlite3
+import stat
 import struct
 import sys
 import threading
@@ -31,6 +33,32 @@
 logger = logging.getLogger("hermes_state")
 
 
+def secure_state_db_files_macos(db_path: Path, *, create_main: bool = False) -> None:
+    """Closing even an O_EVTONLY fd cancels SQLite's POSIX locks on macOS.
+
+    Change existing files by pathname without following symlinks. Only an
+    exclusively created, not-yet-opened database gets a descriptor to close.
+    """
+    if create_main:
+        try:
+            fd = os.open(db_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600)
+        except FileExistsError:
+            pass
+        else:
+            os.close(fd)
+    for path in (db_path, db_path.with_name(db_path.name + "-wal"), db_path.with_name(db_path.name + "-shm")):
+        try:
+            mode = path.lstat().st_mode
+            if stat.S_ISDIR(mode):
+                continue  # sqlite3.connect owns the canonical directory error.
+            if not stat.S_ISREG(mode):
+                code = errno.ELOOP if stat.S_ISLNK(mode) else errno.EINVAL
+                raise OSError(code, "Refusing a non-regular SQLite database file", str(path))
+            os.chmod(path, 0o600, follow_symlinks=False)
+        except FileNotFoundError:
+            continue
+
+
 def _prepare_connection_retirement():
     """Bind a non-finalizing reference before opening a writable SQLite handle.
 

tests/test_state_permission_locks_macos.py

@@ -0,0 +1,81 @@
+"""Permission hardening must retain live SQLite locks on macOS."""
+
+import os
+from pathlib import Path
+import sqlite3
+import stat
+import subprocess
+import sys
+
+import pytest
+
+from hermes_state import _secure_state_db_files
+
+
+@pytest.mark.macos_only
+@pytest.mark.parametrize("journal_mode", ["wal", "delete"])
+def test_permission_hardening_preserves_live_sqlite_write_lock(tmp_path, journal_mode):
+    path = tmp_path / "state.db"
+    conn = sqlite3.connect(path)
+    conn.execute(f"PRAGMA journal_mode={journal_mode}")
+    conn.execute("CREATE TABLE evidence(value)")
+    conn.commit()
+    conn.execute("BEGIN IMMEDIATE")
+    probe = """
+import sqlite3, sys
+conn = sqlite3.connect(sys.argv[1], timeout=0)
+try:
+    conn.execute('BEGIN IMMEDIATE')
+except sqlite3.OperationalError as exc:
+    assert 'locked' in str(exc), str(exc)
+    print('blocked')
+else:
+    conn.rollback()
+    print('admitted')
+finally:
+    conn.close()
+"""
+
+    def competing_writer():
+        return subprocess.check_output(
+            [sys.executable, "-c", probe, str(path)], text=True, timeout=10).strip()
+
+    try:
+        assert competing_writer() == "blocked"
+        # Both constructor call sites can run while another handle holds locks.
+        _secure_state_db_files(path, create_main=True)
+        _secure_state_db_files(path)
+        assert competing_writer() == "blocked"
+    finally:
+        conn.rollback()
+        conn.close()
+
+
+@pytest.mark.macos_only
+def test_permission_hardening_keeps_private_creation_and_rejects_symlinks(tmp_path):
+    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
+    files = [path, Path(str(path) + "-wal"), Path(str(path) + "-shm")]
+    for file in files:
+        file.touch()
+        file.chmod(0o666)
+    _secure_state_db_files(path, create_main=True)
+    assert all(stat.S_IMODE(file.stat().st_mode) == 0o600 for file in files)
+
+    target = tmp_path / "unrelated"
+    target.write_text("unchanged", encoding="utf-8")
+    target.chmod(0o644)
+    for file in files:
+        file.unlink()
+        file.symlink_to(target)
+        with pytest.raises(OSError):
+            _secure_state_db_files(path, create_main=True)
+        assert target.read_text(encoding="utf-8") == "unchanged"
+        assert stat.S_IMODE(target.stat().st_mode) == 0o644
+        file.unlink()
+        file.touch(mode=0o600)