Skip to content

fix(windows): preserve binary bytes in anchored file reads #7077

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
allenliang2022 wants to merge 1 commit into from
Closed

fix(windows): preserve binary bytes in anchored file reads #7077

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
13 changes: 11 additions & 2 deletions api/workspace.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1016,6 +1016,7 @@ def safe_resolve_ws(root: Path, requested: str) -> Path:
_DIR_FD_OK = os.open in getattr(os, "supports_dir_fd", set())
_O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0)
_O_DIRECTORY = getattr(os, "O_DIRECTORY", 0)
_O_BINARY = getattr(os, "O_BINARY", 0)


def open_anchored_fd(workspace: Path, target: Path, *, want_dir: bool) -> int:
Expand All @@ -1037,7 +1038,11 @@ def open_anchored_fd(workspace: Path, target: Path, *, want_dir: bool) -> int:
# Windows / no openat: fall back to a plain pathname open. No new race
# protection, but no regression vs the prior path-based behaviour, and
# symlink creation needs admin on Windows anyway.
flags = os.O_RDONLY | (_O_DIRECTORY if want_dir else 0) | _O_NOFOLLOW
flags = (
os.O_RDONLY
| (_O_DIRECTORY if want_dir else _O_BINARY)
| _O_NOFOLLOW
)
try:
return os.open(str(target), flags)
except OSError:
Expand All @@ -1052,7 +1057,11 @@ def open_anchored_fd(workspace: Path, target: Path, *, want_dir: bool) -> int:
for i, part in enumerate(rel_parts):
is_last = i == len(rel_parts) - 1
want_directory = (not is_last) or want_dir
flags = os.O_RDONLY | _O_NOFOLLOW | (_O_DIRECTORY if want_directory else 0)
flags = (
os.O_RDONLY
| _O_NOFOLLOW
| (_O_DIRECTORY if want_directory else _O_BINARY)
)
try:
nfd = os.open(part, flags, dir_fd=fd)
except OSError:
Expand Down
66 changes: 66 additions & 0 deletions tests/test_media_message_snapshots.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,72 @@ def test_capture_snapshot_stores_content_addressed_bytes(snap_dir, tmp_path):
assert stored.read_bytes() == b"<html>v1</html>"


def test_anchored_snapshot_read_preserves_windows_ctrl_z(tmp_path):
"""Anchored binary reads must not treat DOS Ctrl-Z as text EOF on Windows."""
from api.routes import _etag_and_snapshot, _open_file_read_fd

payload = b"a" * 568 + b"\x1a" + b"b" * 4096
source = tmp_path / "clip.mp4"
source.write_bytes(payload)

fd = _open_file_read_fd(source, tmp_path)
try:
_etag, snapshot, actual_size = _etag_and_snapshot(fd, file_size=len(payload))
finally:
os.close(fd)

assert actual_size == len(payload)
assert snapshot == payload


def test_anchored_file_leaf_uses_binary_open_flag(monkeypatch, tmp_path):
"""The platform-independent flag contract keeps Windows binary-safe in CI."""
from api import workspace

source = tmp_path / "clip.mp4"
source.write_bytes(b"payload")
seen = {}

def fake_open(path, flags, *args, **kwargs):
seen["path"] = path
seen["flags"] = flags
return 123

monkeypatch.setattr(workspace, "_DIR_FD_OK", False)
monkeypatch.setattr(workspace, "_O_BINARY", 0x40000000)
monkeypatch.setattr(workspace.os, "open", fake_open)

fd = workspace.open_anchored_fd(tmp_path, source, want_dir=False)

assert fd == 123
assert seen["path"] == str(source)
assert seen["flags"] & workspace._O_BINARY


def test_anchored_directory_does_not_use_binary_open_flag(monkeypatch, tmp_path):
"""Directory opens keep their directory-only flag contract."""
from api import workspace

target = tmp_path / "folder"
target.mkdir()
seen = {}

def fake_open(path, flags, *args, **kwargs):
seen["flags"] = flags
return 124

monkeypatch.setattr(workspace, "_DIR_FD_OK", False)
monkeypatch.setattr(workspace, "_O_BINARY", 0x40000000)
monkeypatch.setattr(workspace, "_O_DIRECTORY", 0x20000000)
monkeypatch.setattr(workspace.os, "open", fake_open)

fd = workspace.open_anchored_fd(tmp_path, target, want_dir=True)

assert fd == 124
assert seen["flags"] & workspace._O_DIRECTORY
assert not seen["flags"] & workspace._O_BINARY


def test_capture_snapshot_dedupes_identical_content(snap_dir, tmp_path):
from api.media_snapshots import capture_snapshot

Expand Down
Loading