fix(memory): secure built-in memory lock files
This commit is contained in:
committed by
Teknium
parent
baf1200e0a
commit
c445987559
@@ -1,6 +1,8 @@
|
||||
"""Tests for tools/memory_tool.py — MemoryStore, security scanning, and tool dispatcher."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
|
||||
@@ -108,6 +110,45 @@ def store(tmp_path, monkeypatch):
|
||||
return s
|
||||
|
||||
|
||||
class TestMemoryFileLockPermissions:
|
||||
def test_new_lock_file_is_owner_only_under_permissive_umask(self, tmp_path):
|
||||
memory_path = tmp_path / "MEMORY.md"
|
||||
previous_umask = os.umask(0o002)
|
||||
try:
|
||||
with MemoryStore._file_lock(memory_path):
|
||||
pass
|
||||
finally:
|
||||
os.umask(previous_umask)
|
||||
|
||||
lock_path = tmp_path / "MEMORY.md.lock"
|
||||
assert stat.S_IMODE(lock_path.stat().st_mode) == 0o600
|
||||
|
||||
def test_existing_loose_lock_file_is_tightened(self, tmp_path):
|
||||
memory_path = tmp_path / "MEMORY.md"
|
||||
lock_path = tmp_path / "MEMORY.md.lock"
|
||||
lock_path.write_text("", encoding="utf-8")
|
||||
lock_path.chmod(0o664)
|
||||
|
||||
with MemoryStore._file_lock(memory_path):
|
||||
pass
|
||||
|
||||
assert stat.S_IMODE(lock_path.stat().st_mode) == 0o600
|
||||
|
||||
@pytest.mark.skipif(not hasattr(os, "O_NOFOLLOW"), reason="O_NOFOLLOW unavailable")
|
||||
def test_lock_file_symlink_is_refused(self, tmp_path):
|
||||
memory_path = tmp_path / "MEMORY.md"
|
||||
outside = tmp_path / "outside"
|
||||
outside.write_text("do not touch", encoding="utf-8")
|
||||
lock_path = tmp_path / "MEMORY.md.lock"
|
||||
lock_path.symlink_to(outside)
|
||||
|
||||
with pytest.raises(OSError):
|
||||
with MemoryStore._file_lock(memory_path):
|
||||
pass
|
||||
|
||||
assert outside.read_text(encoding="utf-8") == "do not touch"
|
||||
|
||||
|
||||
class TestMemoryStoreAdd:
|
||||
def test_add_entry(self, store):
|
||||
result = store.add("memory", "Python 3.12 project")
|
||||
|
||||
@@ -4,6 +4,7 @@ Module state that tests monkeypatch (``get_memory_dir``, ``fcntl``/``msvcrt``) s
|
||||
in ``tools.memory_tool`` and is read lazily."""
|
||||
|
||||
import logging
|
||||
import os
|
||||
import time
|
||||
from contextlib import contextmanager, suppress
|
||||
from pathlib import Path
|
||||
@@ -151,7 +152,22 @@ class MemoryStore:
|
||||
if fcntl is None and msvcrt is None:
|
||||
yield
|
||||
return
|
||||
with open(lock_path, "a+", encoding="utf-8") as fd:
|
||||
flags = os.O_RDWR | os.O_CREAT
|
||||
if hasattr(os, "O_NOFOLLOW"):
|
||||
flags |= os.O_NOFOLLOW
|
||||
raw_fd = os.open(lock_path, flags, 0o600)
|
||||
try:
|
||||
# The creation mode is filtered through the process umask and does
|
||||
# not repair a lock left loose by an older Hermes process. Tighten
|
||||
# the opened inode before acquiring the lock so both cases are
|
||||
# owner-only. Operating on the fd avoids a path-swap window.
|
||||
if hasattr(os, "fchmod"):
|
||||
os.fchmod(raw_fd, 0o600)
|
||||
fd = os.fdopen(raw_fd, "r+", encoding="utf-8")
|
||||
except Exception:
|
||||
os.close(raw_fd)
|
||||
raise
|
||||
with fd:
|
||||
def _flock(unlock: bool):
|
||||
if fcntl:
|
||||
fcntl.flock(fd, fcntl.LOCK_UN if unlock else fcntl.LOCK_EX)
|
||||
|
||||
Reference in New Issue
Block a user