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
22 changes: 22 additions & 0 deletions cron/lifecycle_guard.py
Original file line number Diff line number Diff line change
Expand Up @@ -1009,6 +1009,28 @@ def _has_binary_magic(data: bytes) -> bool:

def _read_referenced_script(
path: Path, *, max_bytes: Optional[int] = None
) -> tuple[Optional[str], bool]:
"""Read a referenced script without racing SQLite connection lifecycle.

The registry check must cover the complete ``open``/``read``/``close``
sequence. A separate ``has_live_connection`` check would leave a race in
which another thread opens SQLite after the check but before this function
closes its descriptor, cancelling that connection's POSIX locks.
"""
from hermes_cli.sqlite_safe_read import LiveConnectionError, offline_file_access

try:
with offline_file_access(path, what="read referenced script"):
return _read_referenced_script_unlocked(path, max_bytes=max_bytes)
except LiveConnectionError:
return None, True
except (OSError, ValueError):
# Invalid path values, including embedded NULs, are not scripts.
return None, False


def _read_referenced_script_unlocked(
path: Path, *, max_bytes: Optional[int] = None
) -> tuple[Optional[str], bool]:
"""Return ``(text, unsafe)`` using bounded, regular-file-only reads.

Expand Down
44 changes: 44 additions & 0 deletions tests/cron/test_lifecycle_guard_live_db.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
"""Regression tests for raw lifecycle-guard reads of live SQLite files."""

from __future__ import annotations

from pathlib import Path

import cron.lifecycle_guard as lifecycle_guard
from hermes_cli.sqlite_safe_read import connect_tracked


def test_referenced_script_read_refuses_live_sqlite_connection(tmp_path, monkeypatch):
db = tmp_path / "state.db"
conn = connect_tracked(db)
try:
def must_not_open(*_args, **_kwargs):
raise AssertionError("raw-opened a live SQLite database")

monkeypatch.setattr(lifecycle_guard.os, "open", must_not_open)

text, unsafe = lifecycle_guard._read_referenced_script(db)

assert text is None
assert unsafe is True
finally:
conn.close()


def test_referenced_script_read_preserves_normal_script_behavior(tmp_path):
script = tmp_path / "script.sh"
script.write_text("echo ok\n", encoding="utf-8")

text, unsafe = lifecycle_guard._read_referenced_script(script)

assert text == "echo ok\n"
assert unsafe is False


def test_invalid_referenced_script_path_is_still_tolerated():
text, unsafe = lifecycle_guard._read_referenced_script(
Path("/tmp/hermes\x00binary")
)

assert text is None
assert unsafe is False
Loading