fix: guard yaml.safe_load, flock unlock, TOCTOU races, and atomic writes

1. trajectory_compressor.py: yaml.safe_load() returns None on empty
   files, crashing with TypeError on `if 'tokenizer' in data`. Fix by
   adding `or {}` fallback. (HIGH — blocks startup with empty config)

2. 6 files with fcntl.flock(LOCK_UN) in finally blocks without
   try/except: cron/scheduler.py, hermes_cli/auth.py,
   agent/shell_hooks.py, tools/skill_usage.py,
   tools/environments/file_sync.py, tools/memory_tool.py. If unlock
   raises OSError, fd.close() is skipped and the lock is held forever.
   The msvcrt branches already had try/except; the fcntl branches did
   not. Fix by wrapping in try/except (OSError, IOError): pass.

3. agent/copilot_acp_client.py line 639: TOCTOU race — path.exists()
   followed by path.read_text() with no try/except. If file is deleted
   between the check and the read, FileNotFoundError propagates. Fix
   by using try/except FileNotFoundError.

4. gateway/sticker_cache.py: non-atomic write via Path.write_text()
   can leave truncated JSON on crash, causing JSONDecodeError on next
   load. Fix by writing to tempfile + fsync + os.replace (atomic).
This commit is contained in:
vanthinh6886
2026-05-19 00:12:34 -07:00
committed by Teknium
parent d759a67c0f
commit 62573f44cf
9 changed files with 48 additions and 14 deletions

View File

@@ -9,6 +9,8 @@ Cache location: ~/.hermes/sticker_cache.json
"""
import json
import os
import tempfile
import time
from typing import Optional
@@ -35,12 +37,23 @@ def _load_cache() -> dict:
def _save_cache(cache: dict) -> None:
"""Save the sticker cache to disk."""
"""Save the sticker cache to disk atomically."""
CACHE_PATH.parent.mkdir(parents=True, exist_ok=True)
CACHE_PATH.write_text(
json.dumps(cache, indent=2, ensure_ascii=False),
encoding="utf-8",
fd, tmp_path = tempfile.mkstemp(
dir=str(CACHE_PATH.parent), suffix=".tmp"
)
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(cache, f, indent=2, ensure_ascii=False)
f.flush()
os.fsync(f.fileno())
os.replace(tmp_path, str(CACHE_PATH))
except BaseException:
try:
os.unlink(tmp_path)
except OSError:
pass
raise
def get_cached_description(file_unique_id: str) -> Optional[dict]: