fix(identity): spawn-ledger write survives surrogate-escaped argv
`_append_entry` moved to `utils.atomic_json_write`, which dumps with `ensure_ascii=False` through a utf-8 text handle. An argv token holding surrogate-escaped bytes (a non-UTF-8 project path via os.fsdecode) makes json.dump raise UnicodeEncodeError — a ValueError, so the `except OSError` does not catch it and callers silently lose their registration. Expose `ensure_ascii` on `atomic_json_write` (default unchanged) and pass True at the ledger call site, restoring the previous json.dumps behaviour while keeping mode=0o600. One round-trip test, red on base. Follow-up to #109156.
This commit is contained in:
@@ -267,7 +267,9 @@ def _append_entry(entry: LedgerEntry) -> bool:
|
||||
pruned.append(asdict(entry))
|
||||
try:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
atomic_json_write(path, pruned, mode=0o600)
|
||||
# argv may carry surrogate-escaped bytes (non-UTF-8 paths); ensure_ascii keeps the
|
||||
# utf-8 text handle from raising UnicodeEncodeError (a ValueError, not an OSError).
|
||||
atomic_json_write(path, pruned, mode=0o600, ensure_ascii=True)
|
||||
return True
|
||||
except OSError:
|
||||
logger.debug("spawn ledger write failed", exc_info=True)
|
||||
|
||||
@@ -134,6 +134,19 @@ def test_register_self_writes_and_prunes_dead(tmp_path):
|
||||
assert me["create_time"] == pytest.approx(50.0, abs=0.01)
|
||||
|
||||
|
||||
def test_register_self_survives_non_utf8_argv(tmp_path):
|
||||
ledger = tmp_path / "spawn-ledger.json"
|
||||
fake = _fake_psutil({999: 50.0})
|
||||
bad_argv = ["hermes", "serve", os.fsdecode(b"/tmp/project-\xff")] # surrogate-escaped path
|
||||
with patch.dict(sys.modules, {"psutil": fake}), \
|
||||
patch.object(pi, "_ledger_path", return_value=ledger), \
|
||||
patch.object(pi.os, "getpid", return_value=999), \
|
||||
patch.object(sys, "argv", bad_argv):
|
||||
assert pi.register_self("serve", project_root=Path("/x/install")) is True
|
||||
me = next(e for e in json.loads(ledger.read_text(encoding="utf-8")) if e["pid"] == 999)
|
||||
assert me["argv"] == " ".join(bad_argv)
|
||||
|
||||
|
||||
@pytest.mark.skipif(os.name == "nt", reason="POSIX mode bits are platform-specific")
|
||||
def test_register_self_writes_ledger_with_0600(tmp_path):
|
||||
ledger = tmp_path / "spawn-ledger.json"
|
||||
|
||||
@@ -217,10 +217,17 @@ def atomic_write_text(path: Union[str, Path], content: str, *, encoding: str = "
|
||||
mode=_mode_for_write(path, create_mode, preserve=preserve_mode), preserve_owner=preserve_mode)
|
||||
|
||||
|
||||
def atomic_json_write(path: Union[str, Path], data: Any, *, indent: int = 2, mode: int | None = None, **dump_kwargs: Any) -> None:
|
||||
"""Write JSON to *path* atomically (temp file + fsync + replace)."""
|
||||
def atomic_json_write(
|
||||
path: Union[str, Path], data: Any, *, indent: int = 2, mode: int | None = None,
|
||||
ensure_ascii: bool = False, **dump_kwargs: Any,
|
||||
) -> None:
|
||||
"""Write JSON to *path* atomically (temp file + fsync + replace).
|
||||
|
||||
``ensure_ascii=True`` lets callers persist surrogate-escaped strings (non-UTF-8 argv/paths)
|
||||
that a utf-8 text handle would otherwise reject with ``UnicodeEncodeError``.
|
||||
"""
|
||||
path = Path(path)
|
||||
_atomic_write(path, lambda f: json.dump(data, f, indent=indent, ensure_ascii=False, **dump_kwargs),
|
||||
_atomic_write(path, lambda f: json.dump(data, f, indent=indent, ensure_ascii=ensure_ascii, **dump_kwargs),
|
||||
prefix=f".{path.stem}_", mode=mode if mode is not None else _preserve_file_mode(path))
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user