Python: harden Hyperlight output capture against symlinks

python/packages/hyperlight/agent_framework_hyperlight/_execute_code_tool.py

@@ -4,7 +4,9 @@
 
 import asyncio
 import mimetypes
+import os
 import shutil
+import stat
 import threading
 import time
 from collections.abc import Callable, Iterator, Sequence
@@ -594,10 +596,41 @@ def _populate_input_dir(*, config: _RunConfig, input_root: Path) -> None:
         _copy_path(mount.host_path, input_root / mount.mount_path)
 
 
+def _read_output_file_bytes(file_path: Path) -> bytes:
+    """Read ``file_path`` without following a symlink, even under a TOCTOU swap.
+
+    ``Path.read_bytes`` follows symlinks, so a sandbox payload that replaces an
+    output file with ``/output/leak.txt -> /host/secret`` between validation and
+    read could still exfiltrate a host file. Two layers defend against this:
+
+    * ``os.O_NOFOLLOW`` makes the kernel reject a final-component symlink with
+      ``ELOOP``. The flag is absent on some platforms (notably Windows), where
+      it degrades to ``0``, so it cannot be the only defense.
+    * The file is ``lstat``-ed before opening and ``fstat``-ed after; if the
+      ``(st_dev, st_ino)`` identity changed, or the pre-open entry is a symlink,
+      the read is refused. This closes the swap window on every platform.
+    """
+    pre_stat = file_path.lstat()
+    if stat.S_ISLNK(pre_stat.st_mode):
+        raise OSError(f"refusing to read symlinked output file: {file_path}")
+
+    fd = os.open(file_path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
+    try:
+        opened_stat = os.fstat(fd)
+        if (opened_stat.st_dev, opened_stat.st_ino) != (pre_stat.st_dev, pre_stat.st_ino):
+            raise OSError(f"output file changed between validation and read: {file_path}")
+    except BaseException:
+        os.close(fd)
+        raise
+
+    with os.fdopen(fd, "rb") as handle:
+        return handle.read()
+
+
 def _create_file_content(file_path: Path, *, relative_path: str) -> Content:
     media_type = mimetypes.guess_type(file_path.name)[0] or "application/octet-stream"
     return Content.from_data(
-        data=file_path.read_bytes(),
+        data=_read_output_file_bytes(file_path),
         media_type=media_type,
         additional_properties={"path": f"/output/{relative_path}"},
     )
@@ -621,6 +654,50 @@ def _normalize_output_relative_path(*, output_file: object, root: Path) -> str |
     return "/".join(parts)
 
 
+def _is_safe_output_file(*, root: Path, host_path: Path) -> bool:
+    """Return True only if ``host_path`` is a real regular file safely under ``root``.
+
+    The ``/output`` directory is sandbox-controlled, so a payload can plant a
+    final-component symlink (``/output/leak.txt -> /host/secret``) or an
+    intermediate directory symlink to escape ``root`` and read host files.
+    ``Path.is_file`` follows symlinks, so this validator instead walks each path
+    component from ``root`` to ``host_path`` with ``lstat`` and rejects the path
+    if any component is a symlink, requiring the final entry to be a regular
+    file. ``..``/``.`` components are rejected up front because
+    ``Path.relative_to`` is purely lexical and would otherwise allow a listing
+    such as ``root / ".." / "secret.txt"`` to escape ``root`` without any
+    symlink. This mirrors the symlink-hardening already applied to the input
+    staging path (``_copy_path`` / ``_iter_real_entries``).
+    """
+    try:
+        relative = host_path.relative_to(root)
+    except ValueError:
+        return False
+
+    if not relative.parts or any(part in {"..", "."} for part in relative.parts):
+        return False
+
+    *parent_parts, final_part = relative.parts
+    current = root
+    for part in parent_parts:
+        current = current / part
+        try:
+            parent_stat = current.lstat()
+        except OSError:
+            return False
+        if stat.S_ISLNK(parent_stat.st_mode):
+            return False
+
+    current = current / final_part
+    try:
+        final_stat = current.lstat()
+    except OSError:
+        return False
+    if stat.S_ISLNK(final_stat.st_mode):
+        return False
+    return stat.S_ISREG(final_stat.st_mode)
+
+
 def _collect_output_relative_paths(*, sandbox: Any, root: Path) -> set[str]:
     relative_paths: set[str] = set()
 
@@ -634,7 +711,11 @@ def _collect_output_relative_paths(*, sandbox: Any, root: Path) -> set[str]:
             if (relative_path := _normalize_output_relative_path(output_file=output_file, root=root)) is not None:
                 relative_paths.add(relative_path)
 
-    for host_path in root.rglob("*"):
+    # ``Path.rglob`` follows directory symlinks and ``Path.is_file`` follows
+    # symlinks, both of which would surface paths outside the sandbox-controlled
+    # output tree. ``_iter_real_entries`` skips symlinks and never descends
+    # through a symlinked directory, yielding only real entries under ``root``.
+    for host_path in _iter_real_entries(root):
         if host_path.is_file():
             relative_paths.add(host_path.relative_to(root).as_posix())
 
@@ -659,12 +740,12 @@ def _parse_output_files(
 
         for relative_path in sorted(relative_paths):
             host_path = root.joinpath(*PurePosixPath(relative_path).parts)
-            if not host_path.is_file():
+            if not _is_safe_output_file(root=root, host_path=host_path):
                 missing_files = True
                 continue
             try:
                 contents.append(_create_file_content(host_path, relative_path=relative_path))
-            except PermissionError:
+            except (PermissionError, OSError):
                 missing_files = True
 
         if not missing_files or attempt == OUTPUT_FILE_RETRY_ATTEMPTS - 1:

python/packages/hyperlight/tests/hyperlight/test_hyperlight_codeact.py

@@ -594,6 +594,141 @@ def test_path_tree_signature_walks_through_symlinked_root(tmp_path: Path) -> Non
     assert signature_v1 != signature_v2, "signature should change when symlinked target contents change"
 
 
+class _OutputDirShim:
+    """Minimal stand-in for ``TemporaryDirectory`` exposing only ``.name``."""
+
+    def __init__(self, path: Path) -> None:
+        self.name = str(path)
+
+
+class _SandboxWithListing:
+    def __init__(self, output_files: list[str]) -> None:
+        self._output_files = output_files
+
+    def get_output_files(self) -> list[str]:
+        return self._output_files
+
+
+def _decode_content_bytes(item: Content) -> bytes:
+    import base64
+
+    assert item.uri is not None
+    _, _, encoded = item.uri.partition("base64,")
+    return base64.b64decode(encoded)
+
+
+def test_collect_output_relative_paths_skips_symlinked_file(tmp_path: Path) -> None:
+    """A final-component symlink planted in /output must not be surfaced."""
+    if not _symlinks_supported(tmp_path):
+        pytest.skip("Symlinks not supported on this platform/environment")
+    output_root = tmp_path / "output"
+    output_root.mkdir()
+    (output_root / "report.txt").write_text("real-report", encoding="utf-8")
+    secret = tmp_path / "host_secret.txt"
+    secret.write_text("HOST_SECRET", encoding="utf-8")
+    (output_root / "leak.txt").symlink_to(secret)
+
+    relative_paths = execute_code_module._collect_output_relative_paths(sandbox=object(), root=output_root)
+
+    assert "report.txt" in relative_paths
+    assert "leak.txt" not in relative_paths
+
+
+def test_collect_output_relative_paths_skips_symlinked_directory(tmp_path: Path) -> None:
+    """A symlinked directory in /output must not be descended into."""
+    if not _symlinks_supported(tmp_path):
+        pytest.skip("Symlinks not supported on this platform/environment")
+    output_root = tmp_path / "output"
+    output_root.mkdir()
+    outside_dir = tmp_path / "outside_dir"
+    outside_dir.mkdir()
+    (outside_dir / "deep.txt").write_text("deep-secret", encoding="utf-8")
+    (output_root / "linked_dir").symlink_to(outside_dir, target_is_directory=True)
+
+    relative_paths = execute_code_module._collect_output_relative_paths(sandbox=object(), root=output_root)
+
+    assert relative_paths == set()
+
+
+def test_parse_output_files_skips_symlink_to_host_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """End-to-end: a /output symlink to a host file is never returned as Content."""
+    if not _symlinks_supported(tmp_path):
+        pytest.skip("Symlinks not supported on this platform/environment")
+    monkeypatch.setattr(execute_code_module, "OUTPUT_FILE_RETRY_ATTEMPTS", 1)
+    output_root = tmp_path / "output"
+    output_root.mkdir()
+    (output_root / "report.txt").write_text("real-report", encoding="utf-8")
+    secret = tmp_path / "host_secret.txt"
+    secret.write_text("HOST_SECRET", encoding="utf-8")
+    (output_root / "leak.txt").symlink_to(secret)
+
+    contents = execute_code_module._parse_output_files(
+        sandbox=object(),
+        output_dir=_OutputDirShim(output_root),
+        expect_output_files=False,
+    )
+
+    paths = {item.additional_properties["path"] for item in contents if item.type == "data"}
+    assert "/output/report.txt" in paths
+    assert "/output/leak.txt" not in paths
+    assert all(b"HOST_SECRET" not in _decode_content_bytes(item) for item in contents if item.type == "data")
+
+
+def test_parse_output_files_rejects_intermediate_dir_symlink_from_listing(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A backend-listed path traversing an intermediate dir symlink must be rejected."""
+    if not _symlinks_supported(tmp_path):
+        pytest.skip("Symlinks not supported on this platform/environment")
+    monkeypatch.setattr(execute_code_module, "OUTPUT_FILE_RETRY_ATTEMPTS", 1)
+    output_root = tmp_path / "output"
+    output_root.mkdir()
+    outside_dir = tmp_path / "outside_dir"
+    outside_dir.mkdir()
+    (outside_dir / "leak.txt").write_text("HOST_SECRET", encoding="utf-8")
+    (output_root / "sub").symlink_to(outside_dir, target_is_directory=True)
+
+    contents = execute_code_module._parse_output_files(
+        sandbox=_SandboxWithListing(["output/sub/leak.txt"]),
+        output_dir=_OutputDirShim(output_root),
+        expect_output_files=False,
+    )
+
+    assert all(b"HOST_SECRET" not in _decode_content_bytes(item) for item in contents if item.type == "data")
+    assert all(item.additional_properties.get("path") != "/output/sub/leak.txt" for item in contents)
+
+
+def test_is_safe_output_file_rejects_parent_traversal(tmp_path: Path) -> None:
+    """A lexical ``..`` component must be rejected even without any symlink."""
+    output_root = tmp_path / "output"
+    output_root.mkdir()
+    secret = tmp_path / "secret.txt"
+    secret.write_text("HOST_SECRET", encoding="utf-8")
+
+    assert (
+        execute_code_module._is_safe_output_file(root=output_root, host_path=output_root / ".." / "secret.txt") is False
+    )
+    assert execute_code_module._is_safe_output_file(root=output_root, host_path=secret) is False
+
+
+def test_parse_output_files_collects_real_output_file(tmp_path: Path) -> None:
+    """Regression: a genuine /output file is still collected and returned."""
+    output_root = tmp_path / "output"
+    output_root.mkdir()
+    (output_root / "report.txt").write_text("artifact", encoding="utf-8")
+
+    contents = execute_code_module._parse_output_files(
+        sandbox=object(),
+        output_dir=_OutputDirShim(output_root),
+        expect_output_files=True,
+    )
+
+    data_items = [item for item in contents if item.type == "data"]
+    assert len(data_items) == 1
+    assert data_items[0].additional_properties["path"] == "/output/report.txt"
+    assert _decode_content_bytes(data_items[0]) == b"artifact"
+
+
 def test_execute_code_tool_allowed_domains_use_structured_entries_and_replace_by_target() -> None:
     execute_code = HyperlightExecuteCodeTool(_registry=_FakeRuntime())