wehub-resource-sync
858 lines
28 KiB
Python
858 lines
28 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import os
|
|
import tarfile
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from agents.sandbox import SandboxArchiveLimits
|
|
from agents.sandbox.entries import GCSMount, InContainerMountStrategy, MountpointMountPattern
|
|
from agents.sandbox.errors import (
|
|
InvalidCompressionSchemeError,
|
|
InvalidManifestPathError,
|
|
WorkspaceArchiveWriteError,
|
|
)
|
|
from agents.sandbox.files import EntryKind, FileEntry
|
|
from agents.sandbox.manifest import Manifest
|
|
from agents.sandbox.sandboxes.unix_local import (
|
|
UnixLocalSandboxSession,
|
|
UnixLocalSandboxSessionState,
|
|
)
|
|
from agents.sandbox.session.archive_extraction import zipfile_compatible_stream
|
|
from agents.sandbox.session.archive_ops import extract_archive
|
|
from agents.sandbox.session.base_sandbox_session import BaseSandboxSession
|
|
from agents.sandbox.snapshot import NoopSnapshot
|
|
from agents.sandbox.types import ExecResult, Permissions
|
|
|
|
|
|
def _build_session(tmp_path: Path) -> UnixLocalSandboxSession:
|
|
state = UnixLocalSandboxSessionState(
|
|
manifest=Manifest(root=str(tmp_path / "workspace")),
|
|
snapshot=NoopSnapshot(id="noop"),
|
|
)
|
|
return UnixLocalSandboxSession.from_state(state)
|
|
|
|
|
|
class _CountingExtractSession(BaseSandboxSession):
|
|
def __init__(self, workspace_root: Path) -> None:
|
|
self.state = UnixLocalSandboxSessionState(
|
|
manifest=Manifest(root=str(workspace_root)),
|
|
snapshot=NoopSnapshot(id="noop"),
|
|
)
|
|
self.ls_calls: list[Path] = []
|
|
|
|
async def _exec_internal(
|
|
self,
|
|
*command: str | Path,
|
|
timeout: float | None = None,
|
|
) -> ExecResult:
|
|
_ = (command, timeout)
|
|
raise AssertionError("exec() should not be called in this test")
|
|
|
|
async def read(self, path: Path, *, user: object = None) -> io.IOBase:
|
|
_ = user
|
|
return self.normalize_path(path).open("rb")
|
|
|
|
async def write(self, path: Path, data: io.IOBase, *, user: object = None) -> None:
|
|
_ = user
|
|
workspace_path = self.normalize_path(path)
|
|
workspace_path.parent.mkdir(parents=True, exist_ok=True)
|
|
payload = data.read()
|
|
if isinstance(payload, str):
|
|
payload = payload.encode("utf-8")
|
|
workspace_path.write_bytes(payload)
|
|
|
|
async def running(self) -> bool:
|
|
return True
|
|
|
|
async def persist_workspace(self) -> io.IOBase:
|
|
return io.BytesIO()
|
|
|
|
async def hydrate_workspace(self, data: io.IOBase) -> None:
|
|
_ = data
|
|
|
|
async def shutdown(self) -> None:
|
|
return
|
|
|
|
async def mkdir(
|
|
self,
|
|
path: Path | str,
|
|
*,
|
|
parents: bool = False,
|
|
user: object = None,
|
|
) -> None:
|
|
_ = user
|
|
self.normalize_path(path).mkdir(parents=parents, exist_ok=True)
|
|
|
|
async def ls(
|
|
self,
|
|
path: Path | str,
|
|
*,
|
|
user: object = None,
|
|
) -> list[FileEntry]:
|
|
_ = user
|
|
directory = self.normalize_path(path)
|
|
self.ls_calls.append(directory)
|
|
if not directory.exists():
|
|
raise AssertionError(f"ls() called for missing directory: {directory}")
|
|
|
|
entries: list[FileEntry] = []
|
|
for child in directory.iterdir():
|
|
if child.is_symlink():
|
|
kind = EntryKind.SYMLINK
|
|
elif child.is_dir():
|
|
kind = EntryKind.DIRECTORY
|
|
else:
|
|
kind = EntryKind.FILE
|
|
entries.append(
|
|
FileEntry(
|
|
path=str(child),
|
|
permissions=Permissions(),
|
|
owner="root",
|
|
group="root",
|
|
size=0,
|
|
kind=kind,
|
|
)
|
|
)
|
|
return entries
|
|
|
|
|
|
def _tar_bytes(*, members: dict[str, bytes]) -> io.BytesIO:
|
|
buf = io.BytesIO()
|
|
with tarfile.open(fileobj=buf, mode="w") as archive:
|
|
for name, payload in members.items():
|
|
info = tarfile.TarInfo(name=name)
|
|
info.size = len(payload)
|
|
archive.addfile(info, io.BytesIO(payload))
|
|
buf.seek(0)
|
|
return buf
|
|
|
|
|
|
def _zip_bytes(*, members: dict[str, bytes]) -> io.BytesIO:
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, mode="w") as archive:
|
|
for name, payload in members.items():
|
|
archive.writestr(name, payload)
|
|
buf.seek(0)
|
|
return buf
|
|
|
|
|
|
async def _assert_extract_rejects_member(
|
|
tmp_path: Path,
|
|
archive_name: str,
|
|
data: io.IOBase,
|
|
*,
|
|
expected_member: str,
|
|
expected_reason: str,
|
|
) -> Path:
|
|
session = _build_session(tmp_path)
|
|
await session.start()
|
|
try:
|
|
workspace = Path(session.state.manifest.root)
|
|
with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
|
|
await session.extract(archive_name, data)
|
|
|
|
assert exc_info.value.context["member"] == expected_member
|
|
assert exc_info.value.context["reason"] == expected_reason
|
|
return workspace
|
|
finally:
|
|
await session.shutdown()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_tar_writes_archive_and_unpacks_contents(tmp_path: Path) -> None:
|
|
session = _build_session(tmp_path)
|
|
await session.start()
|
|
try:
|
|
await session.extract(
|
|
"bundle.tar",
|
|
_tar_bytes(members={"nested/hello.txt": b"hello from tar"}),
|
|
)
|
|
finally:
|
|
await session.shutdown()
|
|
|
|
workspace = Path(session.state.manifest.root)
|
|
assert (workspace / "bundle.tar").is_file()
|
|
assert (workspace / "nested" / "hello.txt").read_text(encoding="utf-8") == "hello from tar"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_zip_writes_archive_and_unpacks_contents(tmp_path: Path) -> None:
|
|
session = _build_session(tmp_path)
|
|
await session.start()
|
|
try:
|
|
await session.extract(
|
|
"bundle.zip",
|
|
_zip_bytes(members={"nested/hello.txt": b"hello from zip"}),
|
|
)
|
|
finally:
|
|
await session.shutdown()
|
|
|
|
workspace = Path(session.state.manifest.root)
|
|
assert (workspace / "bundle.zip").is_file()
|
|
assert (workspace / "nested" / "hello.txt").read_text(encoding="utf-8") == "hello from zip"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_default_archive_limits_none_preserves_no_resource_limit_behavior(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
session = _build_session(tmp_path)
|
|
await session.start()
|
|
try:
|
|
await session.extract(
|
|
"bundle.tar",
|
|
_tar_bytes(members={"one.txt": b"1", "two.txt": b"2"}),
|
|
)
|
|
finally:
|
|
await session.shutdown()
|
|
|
|
workspace = Path(session.state.manifest.root)
|
|
assert (workspace / "one.txt").read_text(encoding="utf-8") == "1"
|
|
assert (workspace / "two.txt").read_text(encoding="utf-8") == "2"
|
|
|
|
|
|
def test_sandbox_archive_limits_defaults_enable_sdk_thresholds() -> None:
|
|
limits = SandboxArchiveLimits()
|
|
|
|
assert limits.max_input_bytes == 1024 * 1024 * 1024
|
|
assert limits.max_extracted_bytes == 4 * 1024 * 1024 * 1024
|
|
assert limits.max_members == 100_000
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_archive_rejects_missing_compression_scheme(tmp_path: Path) -> None:
|
|
session = _CountingExtractSession(tmp_path / "workspace")
|
|
|
|
with pytest.raises(InvalidCompressionSchemeError) as exc_info:
|
|
await extract_archive(session, "bundle", io.BytesIO(b"not an archive"))
|
|
|
|
assert exc_info.value.context["path"] == "bundle"
|
|
assert exc_info.value.context["scheme"] is None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("kwargs", "message"),
|
|
[
|
|
({"max_input_bytes": 0}, "archive_limits.max_input_bytes must be at least 1"),
|
|
({"max_extracted_bytes": 0}, "archive_limits.max_extracted_bytes must be at least 1"),
|
|
({"max_members": 0}, "archive_limits.max_members must be at least 1"),
|
|
],
|
|
)
|
|
def test_sandbox_archive_limits_rejects_non_positive_values(
|
|
kwargs: dict[str, int],
|
|
message: str,
|
|
) -> None:
|
|
with pytest.raises(ValueError) as exc_info:
|
|
SandboxArchiveLimits(**kwargs)
|
|
|
|
assert str(exc_info.value) == message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_rejects_archive_input_over_limit(tmp_path: Path) -> None:
|
|
session = _build_session(tmp_path)
|
|
await session.start()
|
|
try:
|
|
with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
|
|
await session.extract(
|
|
"bundle.zip",
|
|
_ChunkedBinaryStream([b"123", b"45"]),
|
|
archive_limits=SandboxArchiveLimits(
|
|
max_input_bytes=4,
|
|
max_extracted_bytes=None,
|
|
max_members=None,
|
|
),
|
|
)
|
|
|
|
assert exc_info.value.context["reason"] == "archive input size exceeds limit"
|
|
assert exc_info.value.context["limit"] == 4
|
|
assert exc_info.value.context["actual"] == 5
|
|
finally:
|
|
await session.shutdown()
|
|
|
|
workspace = Path(session.state.manifest.root)
|
|
assert not (workspace / "bundle.zip").exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_tar_rejects_extracted_bytes_over_limit(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
session = _build_session(tmp_path)
|
|
await session.start()
|
|
try:
|
|
with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
|
|
await session.extract(
|
|
"bundle.tar",
|
|
_tar_bytes(members={"large.txt": b"12345"}),
|
|
archive_limits=SandboxArchiveLimits(
|
|
max_input_bytes=None,
|
|
max_extracted_bytes=4,
|
|
max_members=None,
|
|
),
|
|
)
|
|
|
|
assert exc_info.value.context["member"] == "large.txt"
|
|
assert exc_info.value.context["reason"] == "archive extracted size exceeds limit"
|
|
assert exc_info.value.context["limit"] == 4
|
|
assert exc_info.value.context["actual"] == 5
|
|
finally:
|
|
await session.shutdown()
|
|
|
|
workspace = Path(session.state.manifest.root)
|
|
assert not (workspace / "large.txt").exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_zip_rejects_extracted_bytes_over_limit(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
session = _build_session(tmp_path)
|
|
await session.start()
|
|
try:
|
|
with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
|
|
await session.extract(
|
|
"bundle.zip",
|
|
_zip_bytes(members={"large.txt": b"12345"}),
|
|
archive_limits=SandboxArchiveLimits(
|
|
max_input_bytes=None,
|
|
max_extracted_bytes=4,
|
|
max_members=None,
|
|
),
|
|
)
|
|
|
|
assert exc_info.value.context["member"] == "large.txt"
|
|
assert exc_info.value.context["reason"] == "archive extracted size exceeds limit"
|
|
assert exc_info.value.context["limit"] == 4
|
|
assert exc_info.value.context["actual"] == 5
|
|
finally:
|
|
await session.shutdown()
|
|
|
|
workspace = Path(session.state.manifest.root)
|
|
assert not (workspace / "large.txt").exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_tar_rejects_member_count_over_limit(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
data = _tar_bytes(members={"one.txt": b"1", "two.txt": b"2"})
|
|
|
|
def fail_getmembers(_self: tarfile.TarFile) -> list[tarfile.TarInfo]:
|
|
raise AssertionError("tar extraction should not materialize all members")
|
|
|
|
monkeypatch.setattr(tarfile.TarFile, "getmembers", fail_getmembers)
|
|
|
|
session = _build_session(tmp_path)
|
|
await session.start()
|
|
try:
|
|
with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
|
|
await session.extract(
|
|
"bundle.tar",
|
|
data,
|
|
archive_limits=SandboxArchiveLimits(
|
|
max_input_bytes=None,
|
|
max_extracted_bytes=None,
|
|
max_members=1,
|
|
),
|
|
)
|
|
|
|
assert exc_info.value.context["member"] == "two.txt"
|
|
assert exc_info.value.context["reason"] == "archive member count exceeds limit"
|
|
assert exc_info.value.context["limit"] == 1
|
|
assert exc_info.value.context["actual"] == 2
|
|
finally:
|
|
await session.shutdown()
|
|
|
|
workspace = Path(session.state.manifest.root)
|
|
assert not (workspace / "one.txt").exists()
|
|
assert not (workspace / "two.txt").exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_zip_rejects_member_count_over_limit(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
session = _build_session(tmp_path)
|
|
await session.start()
|
|
try:
|
|
with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
|
|
await session.extract(
|
|
"bundle.zip",
|
|
_zip_bytes(members={"one.txt": b"1", "two.txt": b"2"}),
|
|
archive_limits=SandboxArchiveLimits(
|
|
max_input_bytes=None,
|
|
max_extracted_bytes=None,
|
|
max_members=1,
|
|
),
|
|
)
|
|
|
|
assert exc_info.value.context["member"] == "two.txt"
|
|
assert exc_info.value.context["reason"] == "archive member count exceeds limit"
|
|
assert exc_info.value.context["limit"] == 1
|
|
assert exc_info.value.context["actual"] == 2
|
|
finally:
|
|
await session.shutdown()
|
|
|
|
workspace = Path(session.state.manifest.root)
|
|
assert not (workspace / "one.txt").exists()
|
|
assert not (workspace / "two.txt").exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_archive_limits_none_disables_only_selected_limits(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
session = _build_session(tmp_path)
|
|
await session.start()
|
|
try:
|
|
await session.extract(
|
|
"bundle.tar",
|
|
_tar_bytes(members={"large.txt": b"12345"}),
|
|
archive_limits=SandboxArchiveLimits(
|
|
max_input_bytes=None,
|
|
max_extracted_bytes=None,
|
|
max_members=1,
|
|
),
|
|
)
|
|
finally:
|
|
await session.shutdown()
|
|
|
|
workspace = Path(session.state.manifest.root)
|
|
assert (workspace / "large.txt").read_text(encoding="utf-8") == "12345"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_archive_limits_per_call_override_session_default(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
session = _build_session(tmp_path)
|
|
session._set_archive_limits(
|
|
SandboxArchiveLimits(max_input_bytes=None, max_extracted_bytes=None, max_members=1)
|
|
)
|
|
await session.start()
|
|
try:
|
|
await session.extract(
|
|
"bundle.tar",
|
|
_tar_bytes(members={"one.txt": b"1", "two.txt": b"2"}),
|
|
archive_limits=SandboxArchiveLimits(
|
|
max_input_bytes=None,
|
|
max_extracted_bytes=None,
|
|
max_members=2,
|
|
),
|
|
)
|
|
finally:
|
|
await session.shutdown()
|
|
|
|
workspace = Path(session.state.manifest.root)
|
|
assert (workspace / "one.txt").read_text(encoding="utf-8") == "1"
|
|
assert (workspace / "two.txt").read_text(encoding="utf-8") == "2"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_uses_session_default_archive_limits(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
session = _build_session(tmp_path)
|
|
session._set_archive_limits(
|
|
SandboxArchiveLimits(max_input_bytes=None, max_extracted_bytes=None, max_members=1)
|
|
)
|
|
await session.start()
|
|
try:
|
|
with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
|
|
await session.extract(
|
|
"bundle.tar",
|
|
_tar_bytes(members={"one.txt": b"1", "two.txt": b"2"}),
|
|
)
|
|
|
|
assert exc_info.value.context["member"] == "two.txt"
|
|
assert exc_info.value.context["reason"] == "archive member count exceeds limit"
|
|
assert exc_info.value.context["limit"] == 1
|
|
assert exc_info.value.context["actual"] == 2
|
|
finally:
|
|
await session.shutdown()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_archive_limits_object_with_all_none_overrides_session_default(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
session = _build_session(tmp_path)
|
|
session._set_archive_limits(
|
|
SandboxArchiveLimits(max_input_bytes=None, max_extracted_bytes=None, max_members=1)
|
|
)
|
|
await session.start()
|
|
try:
|
|
await session.extract(
|
|
"bundle.tar",
|
|
_tar_bytes(members={"one.txt": b"1", "two.txt": b"2"}),
|
|
archive_limits=SandboxArchiveLimits(
|
|
max_input_bytes=None,
|
|
max_extracted_bytes=None,
|
|
max_members=None,
|
|
),
|
|
)
|
|
finally:
|
|
await session.shutdown()
|
|
|
|
workspace = Path(session.state.manifest.root)
|
|
assert (workspace / "one.txt").read_text(encoding="utf-8") == "1"
|
|
assert (workspace / "two.txt").read_text(encoding="utf-8") == "2"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_rejects_invalid_per_call_archive_limits(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
limits = SandboxArchiveLimits(max_input_bytes=1)
|
|
limits.max_input_bytes = 0
|
|
session = _build_session(tmp_path)
|
|
await session.start()
|
|
try:
|
|
with pytest.raises(ValueError) as exc_info:
|
|
await session.extract(
|
|
"bundle.tar",
|
|
_tar_bytes(members={"one.txt": b"1"}),
|
|
archive_limits=limits,
|
|
)
|
|
|
|
assert str(exc_info.value) == "archive_limits.max_input_bytes must be at least 1"
|
|
finally:
|
|
await session.shutdown()
|
|
|
|
|
|
class _NoSeekableZipStream(io.IOBase):
|
|
def __init__(self, payload: bytes) -> None:
|
|
self._buffer = io.BytesIO(payload)
|
|
|
|
def tell(self) -> int:
|
|
return self._buffer.tell()
|
|
|
|
def seek(self, offset: int, whence: int = io.SEEK_SET) -> int:
|
|
return self._buffer.seek(offset, whence)
|
|
|
|
def read(self, size: int = -1) -> bytes:
|
|
return self._buffer.read(size)
|
|
|
|
|
|
class _ChunkedBinaryStream(io.IOBase):
|
|
def __init__(self, chunks: list[bytes]) -> None:
|
|
self._chunks = list(chunks)
|
|
self.headers = {"Content-Length": str(sum(len(chunk) for chunk in chunks))}
|
|
|
|
def read(self, size: int = -1) -> bytes:
|
|
if not self._chunks:
|
|
return b""
|
|
if size < 0:
|
|
data = b"".join(self._chunks)
|
|
self._chunks.clear()
|
|
return data
|
|
|
|
remaining = size
|
|
out = bytearray()
|
|
while remaining > 0 and self._chunks:
|
|
chunk = self._chunks[0]
|
|
if len(chunk) <= remaining:
|
|
out.extend(self._chunks.pop(0))
|
|
remaining -= len(chunk)
|
|
continue
|
|
out.extend(chunk[:remaining])
|
|
self._chunks[0] = chunk[remaining:]
|
|
remaining = 0
|
|
return bytes(out)
|
|
|
|
|
|
class _SeekableFalseZipStream(io.IOBase):
|
|
def __init__(self, payload: bytes) -> None:
|
|
self._buffer = io.BytesIO(payload)
|
|
|
|
def seekable(self) -> bool:
|
|
return False
|
|
|
|
def read(self, size: int = -1) -> bytes:
|
|
return self._buffer.read(size)
|
|
|
|
|
|
def test_zipfile_compatible_stream_supports_streams_without_seekable() -> None:
|
|
raw_stream = _NoSeekableZipStream(_zip_bytes(members={"file.txt": b"hello"}).getvalue())
|
|
|
|
with zipfile_compatible_stream(raw_stream) as compatible:
|
|
assert compatible.seekable() is True
|
|
with zipfile.ZipFile(compatible) as archive:
|
|
assert archive.read("file.txt") == b"hello"
|
|
|
|
|
|
def test_zipfile_compatible_stream_buffers_streams_with_seekable_false() -> None:
|
|
raw_stream = _SeekableFalseZipStream(_zip_bytes(members={"file.txt": b"hello"}).getvalue())
|
|
|
|
with zipfile_compatible_stream(raw_stream) as compatible:
|
|
assert compatible.seekable() is True
|
|
with zipfile.ZipFile(compatible) as archive:
|
|
assert archive.read("file.txt") == b"hello"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unix_local_write_accepts_chunked_non_seekable_binary_stream(tmp_path: Path) -> None:
|
|
session = _build_session(tmp_path)
|
|
await session.start()
|
|
try:
|
|
await session.write(
|
|
Path("streamed.bin"),
|
|
_ChunkedBinaryStream([b"hello ", b"from ", b"stream"]),
|
|
)
|
|
finally:
|
|
await session.shutdown()
|
|
|
|
workspace = Path(session.state.manifest.root)
|
|
assert (workspace / "streamed.bin").read_bytes() == b"hello from stream"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_tar_rejects_symlinked_parent_paths(tmp_path: Path) -> None:
|
|
session = _build_session(tmp_path)
|
|
await session.start()
|
|
try:
|
|
workspace = Path(session.state.manifest.root)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
os.symlink(outside, workspace / "link", target_is_directory=True)
|
|
|
|
with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
|
|
await session.extract(
|
|
"bundle.tar",
|
|
_tar_bytes(members={"link/hello.txt": b"hello from tar"}),
|
|
)
|
|
|
|
assert exc_info.value.context["member"] == "link/hello.txt"
|
|
assert exc_info.value.context["reason"] == "symlink in parent path: link"
|
|
assert not (outside / "hello.txt").exists()
|
|
finally:
|
|
await session.shutdown()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_zip_rejects_symlinked_parent_paths(tmp_path: Path) -> None:
|
|
session = _build_session(tmp_path)
|
|
await session.start()
|
|
try:
|
|
workspace = Path(session.state.manifest.root)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
os.symlink(outside, workspace / "link", target_is_directory=True)
|
|
|
|
with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
|
|
await session.extract(
|
|
"bundle.zip",
|
|
_zip_bytes(members={"link/hello.txt": b"hello from zip"}),
|
|
)
|
|
|
|
assert exc_info.value.context["member"] == "link/hello.txt"
|
|
assert exc_info.value.context["reason"] == "symlink in parent path: link"
|
|
assert not (outside / "hello.txt").exists()
|
|
finally:
|
|
await session.shutdown()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unix_local_hydrate_workspace_rejects_external_symlink_targets(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
session = _build_session(tmp_path)
|
|
await session.start()
|
|
try:
|
|
archive = io.BytesIO()
|
|
with tarfile.open(fileobj=archive, mode="w") as tar:
|
|
info = tarfile.TarInfo(name="leak")
|
|
info.type = tarfile.SYMTYPE
|
|
info.linkname = "/etc/passwd"
|
|
tar.addfile(info)
|
|
archive.seek(0)
|
|
|
|
with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
|
|
await session.hydrate_workspace(archive)
|
|
|
|
assert exc_info.value.context["member"] == "leak"
|
|
assert (
|
|
exc_info.value.context["reason"] == "absolute symlink target not allowed: /etc/passwd"
|
|
)
|
|
assert not (Path(session.state.manifest.root) / "leak").exists()
|
|
finally:
|
|
await session.shutdown()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_tar_rejects_windows_drive_member_paths(tmp_path: Path) -> None:
|
|
await _assert_extract_rejects_member(
|
|
tmp_path,
|
|
"bundle.tar",
|
|
_tar_bytes(members={"C:/tmp/evil.txt": b"evil"}),
|
|
expected_member="C:/tmp/evil.txt",
|
|
expected_reason="windows drive path",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_zip_rejects_windows_drive_member_paths(tmp_path: Path) -> None:
|
|
await _assert_extract_rejects_member(
|
|
tmp_path,
|
|
"bundle.zip",
|
|
_zip_bytes(members={r"C:\tmp\evil.txt": b"evil"}),
|
|
expected_member=r"C:\tmp\evil.txt",
|
|
expected_reason="windows drive path",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_tar_rejects_windows_separator_member_paths(tmp_path: Path) -> None:
|
|
await _assert_extract_rejects_member(
|
|
tmp_path,
|
|
"bundle.tar",
|
|
_tar_bytes(members={r"..\evil.txt": b"evil"}),
|
|
expected_member=r"..\evil.txt",
|
|
expected_reason="windows path separator",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_zip_rejects_windows_separator_member_paths(tmp_path: Path) -> None:
|
|
await _assert_extract_rejects_member(
|
|
tmp_path,
|
|
"bundle.zip",
|
|
_zip_bytes(members={r"\evil.txt": b"evil"}),
|
|
expected_member=r"\evil.txt",
|
|
expected_reason="windows path separator",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_tar_rejects_member_under_non_directory_member(tmp_path: Path) -> None:
|
|
workspace = await _assert_extract_rejects_member(
|
|
tmp_path,
|
|
"bundle.tar",
|
|
_tar_bytes(
|
|
members={
|
|
"nested/hello.txt": b"hello from tar",
|
|
"nested": b"not a directory",
|
|
}
|
|
),
|
|
expected_member="nested/hello.txt",
|
|
expected_reason="archive path descends through non-directory: nested",
|
|
)
|
|
|
|
assert not (workspace / "nested").exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_zip_rejects_member_under_non_directory_member(tmp_path: Path) -> None:
|
|
workspace = await _assert_extract_rejects_member(
|
|
tmp_path,
|
|
"bundle.zip",
|
|
_zip_bytes(
|
|
members={
|
|
"nested/hello.txt": b"hello from zip",
|
|
"nested": b"not a directory",
|
|
}
|
|
),
|
|
expected_member="nested/hello.txt",
|
|
expected_reason="archive path descends through non-directory: nested",
|
|
)
|
|
|
|
assert not (workspace / "nested").exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unix_local_persist_workspace_excludes_resolved_mount_path(tmp_path: Path) -> None:
|
|
workspace_root = tmp_path / "workspace"
|
|
actual_mount_path = workspace_root / "actual"
|
|
actual_mount_path.mkdir(parents=True)
|
|
(actual_mount_path / "remote.txt").write_text("remote", encoding="utf-8")
|
|
(workspace_root / "keep.txt").write_text("keep", encoding="utf-8")
|
|
|
|
state = UnixLocalSandboxSessionState(
|
|
manifest=Manifest(
|
|
root=str(workspace_root),
|
|
entries={
|
|
"logical": GCSMount(
|
|
bucket="bucket",
|
|
mount_path=Path("actual"),
|
|
mount_strategy=InContainerMountStrategy(pattern=MountpointMountPattern()),
|
|
)
|
|
},
|
|
),
|
|
snapshot=NoopSnapshot(id="noop"),
|
|
)
|
|
session = UnixLocalSandboxSession.from_state(state)
|
|
|
|
archive = await session.persist_workspace()
|
|
|
|
with tarfile.open(fileobj=archive, mode="r:*") as tar:
|
|
names = set(tar.getnames())
|
|
|
|
assert "./keep.txt" in names
|
|
assert "./actual" not in names
|
|
assert "./actual/remote.txt" not in names
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_tar_reuses_directory_listings_during_symlink_checks(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir()
|
|
session = _CountingExtractSession(workspace)
|
|
|
|
await session.extract(
|
|
"bundle.tar",
|
|
_tar_bytes(
|
|
members={
|
|
"nested/one.txt": b"one",
|
|
"nested/two.txt": b"two",
|
|
}
|
|
),
|
|
)
|
|
|
|
assert (workspace / "nested" / "one.txt").read_text(encoding="utf-8") == "one"
|
|
assert (workspace / "nested" / "two.txt").read_text(encoding="utf-8") == "two"
|
|
assert session.ls_calls == [
|
|
workspace,
|
|
workspace / "nested",
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unix_local_helpers_reject_paths_outside_workspace_root(tmp_path: Path) -> None:
|
|
session = _build_session(tmp_path)
|
|
await session.start()
|
|
try:
|
|
with pytest.raises(InvalidManifestPathError, match="must not escape root"):
|
|
await session.ls("../outside")
|
|
with pytest.raises(InvalidManifestPathError, match="must not escape root"):
|
|
await session.mkdir("../outside", parents=True)
|
|
with pytest.raises(InvalidManifestPathError, match="must not escape root"):
|
|
await session.rm("../outside")
|
|
with pytest.raises(InvalidManifestPathError, match="must be relative"):
|
|
await session.extract("/tmp/bundle.tar", _tar_bytes(members={"a.txt": b"a"}))
|
|
finally:
|
|
await session.shutdown()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unix_local_helpers_reject_symlink_escape_paths(tmp_path: Path) -> None:
|
|
session = _build_session(tmp_path)
|
|
await session.start()
|
|
try:
|
|
workspace = Path(session.state.manifest.root)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
os.symlink(outside, workspace / "link", target_is_directory=True)
|
|
|
|
with pytest.raises(InvalidManifestPathError, match="must not escape root"):
|
|
await session.mkdir("link/nested", parents=True)
|
|
with pytest.raises(InvalidManifestPathError, match="must not escape root"):
|
|
await session.ls("link")
|
|
finally:
|
|
await session.shutdown()
|