Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 33 additions & 0 deletions hermes_state.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
import asyncio
import json
import logging
import os
import random
import re
import sqlite3
Expand Down Expand Up @@ -932,6 +933,26 @@ def __init__(self, db_path: Path = None, read_only: bool = False):
self.db_path.parent.mkdir(parents=True, exist_ok=True)

def _connect_and_init():

# Pre-create state.db with 0o600 before SQLite touches it (TOCTOU-safe)
# Mirrors the pattern in hermes_cli/auth.py for auth.json
if not self.db_path.exists():
try:
fd = os.open(str(self.db_path), os.O_CREAT | os.O_EXCL, 0o600)
os.close(fd)
except FileExistsError:
# Race: another process created it; chmod to be safe
try:
os.chmod(str(self.db_path), 0o600)
except OSError:
pass # Best effort; logged if critical
else:
# Already exists (e.g., repaired); ensure permissions
try:
os.chmod(str(self.db_path), 0o600)
except OSError:
pass

self._conn = sqlite3.connect(
str(self.db_path),
check_same_thread=False,
Expand All @@ -946,6 +967,18 @@ def _connect_and_init():
)
self._conn.row_factory = sqlite3.Row
apply_wal_with_fallback(self._conn, db_label="state.db")

# Apply 0o600 to WAL/SHM sidecars created by SQLite (umask does not affect them)
# These hold recent uncommitted writes and should not be world-readable
wal_path = str(self.db_path) + "-wal"
shm_path = str(self.db_path) + "-shm"
for sidecar in (wal_path, shm_path):
if os.path.exists(sidecar):
try:
os.chmod(sidecar, 0o600)
except OSError:
pass # Best effort; logged if critical

self._conn.execute("PRAGMA foreign_keys=ON")
self._init_schema()

Expand Down
123 changes: 123 additions & 0 deletions tests/hermes_state/test_permissions.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,123 @@
"""
Test that state.db and its WAL/SHM sidecars are created with 0o600 permissions
(regression test for issue #59706).
"""
import os
import sqlite3
import stat
import tempfile
from pathlib import Path

import pytest

from hermes_state import SessionDB


def test_state_db_created_with_0600_permissions():
"""
state.db should be created with 0o600 permissions, not 0o644 (umask 022 default).

This is a security hardening fix: even if ~/.hermes has wider permissions
(e.g., HERMES_HOME_MODE=0755 for a web-server traversal use case, or NixOS
managed mode sets it to 0750), the database file itself should not be
world-readable.

Regression test for issue #59706.
"""
with tempfile.TemporaryDirectory() as tmpdir:
db_path = Path(tmpdir) / "state.db"

# Set umask to 0o022 (the default on most distros) to ensure we're not
# getting a false negative because the test runner happens to use 0o077
old_umask = os.umask(0o022)
try:
# Create a SessionDB instance, which triggers the file creation
db = SessionDB(db_path=db_path)

# Check that state.db has 0o600 permissions
mode = stat.S_IMODE(os.stat(db_path).st_mode)
assert mode == 0o600, f"state.db should be 0o600, got {oct(mode)}"

# Check that WAL/SHM sidecars (if they exist) also have 0o600 permissions
# WAL mode is the default; these files should be created
wal_path = db_path.with_suffix(".db-wal")
shm_path = db_path.with_suffix(".db-shm")

# WAL/SHM may not exist if WAL fell back to DELETE mode (NFS/SMB incompatibility)
if wal_path.exists():
wal_mode = stat.S_IMODE(os.stat(wal_path).st_mode)
assert wal_mode == 0o600, f"state.db-wal should be 0o600, got {oct(wal_mode)}"

if shm_path.exists():
shm_mode = stat.S_IMODE(os.stat(shm_path).st_mode)
assert shm_mode == 0o600, f"state.db-shm should be 0o600, got {oct(shm_mode)}"

db.close()
finally:
os.umask(old_umask)


def test_state_db_permissions_on_existing_file():
"""
If state.db already exists (e.g., after a repair or from a previous run),
SessionDB.__init__ should still ensure it has 0o600 permissions.
"""
with tempfile.TemporaryDirectory() as tmpdir:
db_path = Path(tmpdir) / "state.db"

# Create a state.db with wrong permissions (simulating the bug)
conn = sqlite3.connect(str(db_path))
conn.execute("CREATE TABLE t (x int)")
conn.commit()
conn.close()

# Set wrong permissions (0o644, the umask default)
os.chmod(db_path, 0o644)

# Verify it's wrong before the fix
mode_before = stat.S_IMODE(os.stat(db_path).st_mode)
assert mode_before == 0o644, "Setup failed: should start with 0o644"

# Open SessionDB (should chmod to 0o600)
db = SessionDB(db_path=db_path)

# Check that it now has correct permissions
mode_after = stat.S_IMODE(os.stat(db_path).st_mode)
assert mode_after == 0o600, f"After SessionDB init, state.db should be 0o600, got {oct(mode_after)}"

db.close()


def test_state_db_permissions_race_condition():
"""
If state.db is created by another process between our exists() check and
os.open(), we should still chmod it to 0o600 (best-effort).

This tests the FileExistsError branch in the pre-creation logic.
"""
with tempfile.TemporaryDirectory() as tmpdir:
db_path = Path(tmpdir) / "state.db"

old_umask = os.umask(0o022)
try:
# Pre-create the file with wrong permissions (simulating a race)
conn = sqlite3.connect(str(db_path))
conn.execute("CREATE TABLE t (x int)")
conn.commit()
conn.close()
os.chmod(db_path, 0o644)

# Verify it's wrong before SessionDB
mode_before = stat.S_IMODE(os.stat(db_path).st_mode)
assert mode_before == 0o644, "Setup failed: should start with 0o644"

# Open SessionDB (should chmod to 0o600 despite the race)
db = SessionDB(db_path=db_path)

# Check that it now has correct permissions
mode_after = stat.S_IMODE(os.stat(db_path).st_mode)
assert mode_after == 0o600, f"After SessionDB init, state.db should be 0o600, got {oct(mode_after)}"

db.close()
finally:
os.umask(old_umask)
Loading