fix(file-ops): verify the byte-exact read against the file's own byte count

The fence drops backend output around the read, not output printed while
the transport runs: a BASH_ENV DEBUG hook firing for base64 alone puts
text inside the payload that still decodes (TERM is b"LDL"), and the edit
paths wrote it back. Emit wc -c in its own fenced segment and hand bytes to
a writer only when their length matches; the od fallback is checked the
same way.
This commit is contained in:
John Paul Soliva
2026-09-25 22:35:19 +09:00
committed by Austin Pickett
parent 7c686fe7a7
commit 48a4a360a0
3 changed files with 70 additions and 27 deletions

View File

@@ -424,13 +424,15 @@ class TestShellFileOpsWriteDenied:
def _fenced_base64_reply(command: str, payload: bytes, rc: int = 0) -> str:
"""The reply shape ``_read_exact_bytes`` asks for: its per-call sentinel around the base64
payload, then the read's exit status. Mirrors what the real shell emits, so a double stays
honest about the fence the transport relies on."""
payload (and the file's ``wc -c`` when the command asks for it), then the read's exit status.
Mirrors what the real shell emits, so a double stays honest about the fence the transport
relies on."""
import base64 as _b64
import re as _re
sentinel = _re.search(r"__HERMES_RB_[0-9a-f]+__", command).group(0)
body = _b64.b64encode(payload).decode() if rc == 0 else ""
return f"{sentinel}\n{body}\n{sentinel}\n{rc}\n"
size = f"{len(payload)}\n{sentinel}\n" if "wc -c <" in command else ""
return f"{sentinel}\n{body}\n{sentinel}\n{size}{rc}\n"
class TestPatchReplacePostWriteVerification:
"""Tests for the post-write verification added in patch_replace.

View File

@@ -524,6 +524,29 @@ class TestBomHandling:
assert dst.read_bytes() == b"PRECIOUS DESTINATION\n"
assert not (tmp_path / "new-src.txt").exists()
@pytest.mark.parametrize("transport,noise", [("base64", "TERM"), ("od", "4c 44")])
def test_output_inside_the_byte_exact_read_never_reaches_a_write(
self, tmp_path: Path, monkeypatch, transport, noise):
# The fence drops noise around the read, not noise printed WHILE it runs: a BASH_ENV DEBUG
# hook firing for the transport command alone puts text inside the payload that still
# decodes ("TERM" is b"LDL"; "4c 44" is hex). Such a read must fail, and no edit may write.
from tools.file_operations import ShellFileOperations
monkeypatch.setenv("HERMES_NATIVE_FILE_READ", "0")
hook = tmp_path / "hook.sh"
hook.write_text(f"trap '[[ $BASH_COMMAND == {transport}* ]] && echo \"{noise}\"' DEBUG\n")
target = tmp_path / "conf.txt"
original = b"HEADER\nVERSION=1\n"
target.write_bytes(original)
missing = ("base64",) if transport == "od" else ()
env = self._env_without(*missing)(cwd=str(tmp_path), env={"BASH_ENV": str(hook)})
ops = ShellFileOperations(env, cwd=str(tmp_path))
assert ops._read_exact_bytes(str(target))[0] is None
assert not ops.patch_replace(str(target), "VERSION=1", "VERSION=2").success
assert not ops.patch_v4a(
f"*** Begin Patch\n*** Update File: {target}\n@@\n-VERSION=1\n+VERSION=2\n*** End Patch").success
assert target.read_bytes() == original
@pytest.mark.skipif(not hasattr(os, "mkfifo"), reason="POSIX only: needs os.mkfifo and SIGALRM")
def test_native_byte_exact_read_never_opens_a_non_regular_file(self, tmp_path: Path, monkeypatch):
# The native fast path bypasses the backend timeout, so a blocking open there hangs the

View File

@@ -249,28 +249,39 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
result = self._exec(f"python -c {self._escape_shell_arg(snippet)}")
return result
def _fenced_read(self, body: str) -> "tuple[Optional[str], Optional[int], ExecuteResult]":
"""Run BODY between two sentinel lines; return (payload segment, BODY's exit status, reply).
def _fenced_read(self, body: str, *more: str) -> "tuple[Optional[list[str]], Optional[int], ExecuteResult]":
"""Run BODY, then each of MORE, each in its own sentinel-delimited segment; return (those
segments, BODY's exit status, reply).
The transport merges the backend's own stdout with the command's, and every caller here
decodes the payload into file bytes, so the payload has to be delimited rather than taken
to be the whole reply: a remote shell announcing ``TERM`` is four base64 characters that
would otherwise join the payload and decode to ``b"LDL"`` at the head of it. Noise outside
the fence is dropped; noise inside fails base64/hex validation, because ``_new_sentinel``'s
underscores are outside both alphabets. The status rides in its own trailing segment so a
failed BODY is still told apart from an empty file. ``(None, None, reply)`` when no fenced
reply came back at all — the command never ran as written.
decodes a segment into file bytes, so the payload has to be delimited rather than taken to
be the whole reply: a remote shell announcing ``TERM`` is four base64 characters that would
otherwise join the payload and decode to ``b"LDL"`` at the head of it. The fence drops noise
OUTSIDE it only; output emitted while BODY runs (a ``BASH_ENV`` DEBUG hook) lands inside the
payload, so a caller that writes the bytes back must verify them independently (MORE). The
status rides in its own trailing segment so a failed BODY is still told apart from an empty
file. ``(None, None, reply)`` when no fenced reply came back — the command never ran as
written.
"""
sentinel = _new_sentinel(_BYTES_SENTINEL_PREFIX)
mark = f"echo {sentinel}"
result = self._exec(f"{mark}; {body}; __hb=$?; {mark}; echo $__hb")
rest = "".join(f"{mark}; {cmd}; " for cmd in more)
result = self._exec(f"{mark}; {body}; __hb=$?; {rest}{mark}; echo $__hb")
segments = _split_segments(result.stdout or "", sentinel)
if len(segments) != 3:
if len(segments) != len(more) + 3:
return None, None, result
try:
return segments[1], int(_strip_terminal_fence_leaks(segments[2]).split()[0]), result
return segments[1:-1], int(_strip_terminal_fence_leaks(segments[-1]).split()[0]), result
except (IndexError, ValueError):
return segments[1], None, result
return segments[1:-1], None, result
@staticmethod
def _matches_size(data: bytes, size_segment: str) -> bool:
"""Whether DATA is exactly as long as the file's own ``wc -c``. Noise inside the payload
only ever ADDS text, and any addition that still decodes adds bytes, so equal length is
the check; noise in the size segment breaks its single-integer shape instead."""
tokens = _strip_terminal_fence_leaks(size_segment).split()
return len(tokens) == 1 and tokens[0].isdigit() and int(tokens[0]) == len(data)
def _sample_file_bytes(self, path: str, length: int = 1000):
"""First ``length`` raw bytes, base64-wrapped so they survive the terminal
@@ -284,11 +295,11 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
byte-exact read below: this sample is the binary-admission gate in FRONT of that read, so
backend noise decoded into it decides whether a file is editable at all.
"""
payload, read_rc, _ = self._fenced_read(
segments, read_rc, _ = self._fenced_read(
f"head -c {length} {self._escape_shell_arg(path)} 2>/dev/null | base64")
if payload is None or read_rc != 0:
if segments is None or read_rc != 0:
return None
return self._decode_base64_sample(payload)
return self._decode_base64_sample(segments[0])
def _read_exact_bytes(self, path: str) -> "tuple[Optional[bytes], Optional[ExecuteResult]]":
"""The file's bytes exactly, for the edit paths that write back every line they did not touch.
@@ -319,10 +330,12 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
# Fenced like the compound read probe, and for the same reason: a backend whose merged
# stdout carries login-shell noise (a remote shell announcing TERM, a banner) would
# otherwise have it whitespace-joined onto the payload and decoded INTO the file's bytes,
# which the edit paths then write back.
payload, read_rc, result = self._fenced_read(f"base64 < {self._escape_shell_arg(path)}")
# which the edit paths then write back. The file's own byte count travels beside it: output
# INSIDE the fence decodes too, so only a read that matches it is ever handed to a writer.
arg = self._escape_shell_arg(path)
segments, read_rc, result = self._fenced_read(f"base64 < {arg}", f"wc -c < {arg}")
garbled = ExecuteResult(stdout=f"{path}: the backend returned a garbled byte-exact read", exit_code=1)
if payload is None:
if segments is None:
# No fenced reply: the command never ran as written (a wrapper ``cd`` failed, the backend
# refused it). Hand the backend's own text back so the caller reports what it said.
return None, result if result.exit_code != 0 else garbled
@@ -330,12 +343,13 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
return None, garbled
if read_rc == 127: # no base64 on this backend (busybox, distroless): try the hex transport
return self._read_exact_bytes_hex(path)
payload, size = segments
if read_rc != 0:
# stderr is merged into the fenced segment, so that segment holds base64's own diagnostic
# ("No such file or directory", "Permission denied"): keep it for the caller's message.
return None, self._failed_read(path, payload, read_rc)
data = self._decode_base64_sample(payload)
if data is None: # stray output in the payload: refuse rather than guess (never echo it back)
if data is None or not self._matches_size(data, size): # stray output: refuse, never echo it back
return None, garbled
return data, None
@@ -353,22 +367,26 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
file" — that clobbers the file the Add was refusing to overwrite. ``od`` is POSIX and
present in busybox; when it is missing too the caller gets a transport error, never a
not-found."""
payload, read_rc, result = self._fenced_read(
f"od -An -v -tx1 < {self._escape_shell_arg(path)}")
arg = self._escape_shell_arg(path)
segments, read_rc, result = self._fenced_read(f"od -An -v -tx1 < {arg}", f"wc -c < {arg}")
unavailable = ExecuteResult(
stdout=f"{path}: this backend has neither base64 nor od, so a byte-exact read is unavailable",
exit_code=1)
if payload is None:
if segments is None:
return None, result if result.exit_code != 0 else unavailable
if read_rc is None or read_rc == 127:
return None, unavailable
payload, size = segments
if read_rc != 0:
return None, self._failed_read(path, payload, read_rc)
try:
return bytes.fromhex("".join(_strip_terminal_fence_leaks(payload).split())), None
data = bytes.fromhex("".join(_strip_terminal_fence_leaks(payload).split()))
except ValueError:
data = None
if data is None or not self._matches_size(data, size):
return None, ExecuteResult(stdout=f"{path}: the backend returned a garbled byte-exact read",
exit_code=1)
return data, None
@staticmethod
def _decode_base64_sample(text: str) -> Optional[bytes]: