From 64ade669054b4de0a501fd6c4e2ed0a04dfe9562 Mon Sep 17 00:00:00 2001 From: Brice Le Blevennec <7482674+bricelb@users.noreply.github.com> Date: Sun, 13 Sep 2026 09:28:29 +0200 Subject: [PATCH 1/2] fix(state): preserve SQLite locks during Linux permission hardening --- hermes_state.py | 33 +++++++++++++-- tests/test_state_permission_locks.py | 63 ++++++++++++++++++++++++++++ 2 files changed, 93 insertions(+), 3 deletions(-) create mode 100644 tests/test_state_permission_locks.py diff --git a/hermes_state.py b/hermes_state.py index 8bec8e91f1395..e67a804d43e0b 100644 --- a/hermes_state.py +++ b/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,6 +232,7 @@ 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, @@ -237,13 +240,26 @@ def _secure_state_db_files(db_path: Path, *, create_main: bool = False) -> None: 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) diff --git a/tests/test_state_permission_locks.py b/tests/test_state_permission_locks.py new file mode 100644 index 0000000000000..d7d9f64894893 --- /dev/null +++ b/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") + 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() == "keep this private test content unchanged" From bdcec6b6718dd406ef346d97d39dc02c39da8f66 Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Sun, 13 Sep 2026 01:05:15 -0700 Subject: [PATCH 2/2] test: pass encoding to read_text/write_text in permission-lock test scripts/check-windows-footguns.py --all runs in CI lint and flags the bare calls; no behaviour change. --- tests/test_state_permission_locks.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_state_permission_locks.py b/tests/test_state_permission_locks.py index d7d9f64894893..240aa2a0efceb 100644 --- a/tests/test_state_permission_locks.py +++ b/tests/test_state_permission_locks.py @@ -53,11 +53,11 @@ def test_hardening_keeps_files_private_and_refuses_symlink_targets(tmp_path, suf if sidecar.exists(): sidecar.unlink() target = tmp_path / "unrelated-file" - target.write_text("keep this private test content unchanged") + 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() == "keep this private test content unchanged" + assert target.read_text(encoding="utf-8") == "keep this private test content unchanged"