From 0e1ec4d21978188d90f7334917b0531a0e49ba3b Mon Sep 17 00:00:00 2001
From: liramon2 <liramon@amazon.com>
Date: Wed, 30 Sep 2026 20:45:07 -0400
Subject: [PATCH 1/4] feat(sandbox): add sandboxed fetch to Python SDK

---
 strands-py/src/strands/sandbox/__init__.py    |   5 +-
 strands-py/src/strands/sandbox/base.py        |  27 ++-
 strands-py/src/strands/sandbox/errors.py      |   4 +
 .../not_a_sandbox_local_environment.py        |  51 +++++-
 strands-py/src/strands/sandbox/posix_shell.py | 120 +++++++++++-
 strands-py/src/strands/sandbox/types.py       |  15 ++
 .../test_not_a_sandbox_local_environment.py   |  88 ++++++++-
 .../tests/strands/sandbox/test_posix_shell.py | 171 +++++++++++++++++-
 8 files changed, 473 insertions(+), 8 deletions(-)

diff --git a/strands-py/src/strands/sandbox/__init__.py b/strands-py/src/strands/sandbox/__init__.py
index 127a695620..4da87a472b 100644
--- a/strands-py/src/strands/sandbox/__init__.py
+++ b/strands-py/src/strands/sandbox/__init__.py
@@ -32,16 +32,19 @@ async def execute_streaming(self, command, *, timeout=None, cwd=None, env=None,
 
 from .base import Sandbox
 from .constants import LANGUAGE_PATTERN
+from .errors import SandboxFetchError
 from .posix_shell import PosixShellSandbox
-from .types import ExecutionResult, FileInfo, OutputFile, StreamChunk, StreamType
+from .types import ExecutionResult, FetchResult, FileInfo, OutputFile, StreamChunk, StreamType
 
 __all__ = [
     "ExecutionResult",
+    "FetchResult",
     "FileInfo",
     "LANGUAGE_PATTERN",
     "OutputFile",
     "PosixShellSandbox",
     "Sandbox",
+    "SandboxFetchError",
     "StreamChunk",
     "StreamType",
 ]
diff --git a/strands-py/src/strands/sandbox/base.py b/strands-py/src/strands/sandbox/base.py
index b538d13069..20eddb8515 100644
--- a/strands-py/src/strands/sandbox/base.py
+++ b/strands-py/src/strands/sandbox/base.py
@@ -28,7 +28,7 @@
 from typing import Any
 
 from ..types.tools import AgentTool
-from .types import ExecutionResult, FileInfo, StreamChunk
+from .types import ExecutionResult, FetchResult, FileInfo, StreamChunk
 
 logger = logging.getLogger(__name__)
 
@@ -200,6 +200,31 @@ async def list_files(self, path: str, **kwargs: Any) -> list[FileInfo]:
         """
         ...
 
+    async def fetch(
+        self,
+        url: str,
+        *,
+        max_bytes: int | None = None,
+        timeout: float | None = None,
+        headers: dict[str, str] | None = None,
+        **kwargs: Any,
+    ) -> FetchResult:
+        """Fetch a URL via HTTP GET inside the sandbox.
+
+        Optional because not every sandbox needs HTTP fetching.
+
+        Args:
+            url: The http(s) URL to fetch.
+            max_bytes: Cap the response body to this many bytes. ``None`` means no limit.
+            timeout: Maximum execution time in seconds. ``None`` means no timeout.
+            headers: Request headers.
+            **kwargs: Additional keyword arguments for forward compatibility.
+
+        Raises:
+            SandboxFetchError: On network or transport errors.
+        """
+        raise NotImplementedError(f"{type(self).__name__} does not implement fetch()")
+
     # ---- Tool vending ----
 
     def get_tools(self) -> list[AgentTool]:
diff --git a/strands-py/src/strands/sandbox/errors.py b/strands-py/src/strands/sandbox/errors.py
index 1c1106bc4d..88588746c2 100644
--- a/strands-py/src/strands/sandbox/errors.py
+++ b/strands-py/src/strands/sandbox/errors.py
@@ -25,6 +25,10 @@ def __init__(self, seconds: float | None, stdout: str = "", stderr: str = "") ->
         self.stderr = stderr
 
 
+class SandboxFetchError(RuntimeError):
+    """Raised when :meth:`~strands.sandbox.base.Sandbox.fetch` fails."""
+
+
 class SandboxPathNotFoundError(FileNotFoundError):
     """Raised by :meth:`~strands.sandbox.base.Sandbox.list_files` when the path does not exist.
 
diff --git a/strands-py/src/strands/sandbox/not_a_sandbox_local_environment.py b/strands-py/src/strands/sandbox/not_a_sandbox_local_environment.py
index 8bf288cf9a..37e581583a 100644
--- a/strands-py/src/strands/sandbox/not_a_sandbox_local_environment.py
+++ b/strands-py/src/strands/sandbox/not_a_sandbox_local_environment.py
@@ -19,13 +19,16 @@
 from collections.abc import AsyncGenerator
 from pathlib import Path
 from typing import Any
+from urllib.parse import urlparse
+
+import httpx
 
 from .base import Sandbox
 from .constants import LANGUAGE_PATTERN
-from .errors import SandboxPathNotFoundError
+from .errors import SandboxFetchError, SandboxPathNotFoundError
 from .posix_shell import build_shell_env_prefix
 from .stream_process import _stream_process
-from .types import ExecutionResult, FileInfo, StreamChunk
+from .types import ExecutionResult, FetchResult, FileInfo, StreamChunk
 
 
 class NotASandboxLocalEnvironment(Sandbox):
@@ -205,3 +208,47 @@ async def list_files(self, path: str, **kwargs: Any) -> list[FileInfo]:
                 except OSError:
                     results.append(FileInfo(name=entry.name))
         return results
+
+    async def fetch(
+        self,
+        url: str,
+        *,
+        max_bytes: int | None = None,
+        timeout: float | None = None,
+        headers: dict[str, str] | None = None,
+        **kwargs: Any,
+    ) -> FetchResult:
+        """Fetch a URL with httpx on the host.
+
+        ``max_bytes`` is enforced by streaming: reading stops as soon as
+        the cap is reached and the connection is closed.
+        """
+        url = url.strip()
+        parts = urlparse(url)
+        if parts.scheme not in ("http", "https"):
+            raise SandboxFetchError(f"fetch only supports http(s) URLs, got {url!r}.")
+        if not parts.hostname:
+            raise SandboxFetchError(f"fetch URL has no host: {url!r}.")
+
+        try:
+            async with httpx.AsyncClient(follow_redirects=True, timeout=timeout) as client:
+                async with client.stream("GET", url, headers=headers or {}) as response:
+                    response.raise_for_status()
+                    chunks: list[bytes] = []
+                    total = 0
+                    async for chunk in response.aiter_bytes():
+                        total += len(chunk)
+                        chunks.append(chunk)
+                        if max_bytes is not None and total >= max_bytes:
+                            break
+                    body = b"".join(chunks)
+                    if max_bytes is not None:
+                        body = body[:max_bytes]
+                    result_url = str(response.url)
+                    result_headers = {k: v for k, v in response.headers.items()}
+        except httpx.HTTPStatusError as exc:
+            raise SandboxFetchError(f"HTTP {exc.response.status_code} for {url}") from exc
+        except (httpx.TimeoutException, httpx.RequestError) as exc:
+            raise SandboxFetchError(str(exc)) from exc
+
+        return FetchResult(url=result_url, headers=result_headers, body=body)
diff --git a/strands-py/src/strands/sandbox/posix_shell.py b/strands-py/src/strands/sandbox/posix_shell.py
index 485573276d..3aae9ccff1 100644
--- a/strands-py/src/strands/sandbox/posix_shell.py
+++ b/strands-py/src/strands/sandbox/posix_shell.py
@@ -9,20 +9,28 @@
 """
 
 import base64
+import contextlib
 import logging
+import re
 import shlex
 import uuid
 from abc import ABC
 from collections.abc import AsyncGenerator
 from typing import Any
+from urllib.parse import urlparse
 
 from .base import Sandbox
 from .constants import ENV_KEY_PATTERN, LANGUAGE_PATTERN
-from .errors import SandboxPathNotFoundError
-from .types import ExecutionResult, FileInfo, StreamChunk
+from .errors import SandboxFetchError, SandboxPathNotFoundError, SandboxTimeoutError
+from .types import ExecutionResult, FetchResult, FileInfo, StreamChunk
 
 logger = logging.getLogger(__name__)
 
+# The characters RFC 3986 allows anywhere in a URL.  Anything outside this
+# set (whitespace, quotes, control characters, non-ASCII) is rejected before
+# the URL reaches a shell command.  Security-critical for the curl transport.
+_URL_CHARS = re.compile(r"[A-Za-z0-9\-._~:/?#\[\]@!$&'()*+,;=%]+")
+
 
 def validate_env_keys(env: dict[str, str]) -> None:
     """Validate environment variable names against :data:`ENV_KEY_PATTERN`.
@@ -74,6 +82,35 @@ def _eof_marker() -> str:
     return f"STRANDS_EOF_{uuid.uuid4().hex[:16]}"
 
 
+def _parse_header_dump(raw: str) -> dict[str, str]:
+    r"""Parse a curl ``-D`` header dump into ``{lowered_name: value}``.
+
+    Curl dumps headers for every hop in a redirect chain, separated by blank
+    lines. Only the final response is parsed so that stale headers from
+    intermediate redirects don't leak through.
+    """
+    # Split on blank lines, take the last non-empty block.
+    blocks = re.split(r"\n\s*\n", raw.replace("\r\n", "\n"))
+    last_block = ""
+    for block in reversed(blocks):
+        if block.strip():
+            last_block = block
+            break
+
+    result: dict[str, str] = {}
+    for line in last_block.split("\n"):
+        line = line.strip()
+        if not line or line.upper().startswith("HTTP/"):
+            continue
+        colon = line.find(":")
+        if colon < 1:
+            continue
+        name = line[:colon].strip().lower()
+        value = line[colon + 1 :].strip()
+        result[name] = value
+    return result
+
+
 class PosixShellSandbox(Sandbox, ABC):
     """Abstract sandbox that provides shell-based defaults for file and code operations.
 
@@ -239,3 +276,82 @@ async def list_files(self, path: str, **kwargs: Any) -> list[FileInfo]:
             if name:
                 entries.append(FileInfo(name=name, is_dir=is_dir))
         return entries
+
+    async def fetch(
+        self,
+        url: str,
+        *,
+        max_bytes: int | None = None,
+        timeout: float | None = None,
+        headers: dict[str, str] | None = None,
+        **kwargs: Any,
+    ) -> FetchResult:
+        """Fetch a URL with ``curl`` inside the sandbox.
+
+        ``max_bytes`` is rejected early when ``Content-Length`` is known and
+        truncated for chunked responses.
+        """
+        url = url.strip()
+        if not _URL_CHARS.fullmatch(url):
+            raise SandboxFetchError(
+                "fetch URLs may only contain the characters RFC 3986 allows; "
+                "percent-encode spaces and non-ASCII characters (and punycode the host) and retry."
+            )
+        parts = urlparse(url)
+        if parts.scheme not in ("http", "https"):
+            raise SandboxFetchError(f"fetch only supports http(s) URLs, got {url!r}.")
+        if not parts.hostname:
+            raise SandboxFetchError(f"fetch URL has no host: {url!r}.")
+
+        tag = uuid.uuid4().hex
+        body_file = f"/tmp/strands-fetch-{tag}"
+        header_file = f"/tmp/strands-fetch-{tag}.headers"
+        body_quoted = shlex.quote(body_file)
+        header_quoted = shlex.quote(header_file)
+        part_quoted = shlex.quote(body_file + ".part")
+
+        # Build a hardened curl command: protocol restricted to http(s),
+        # body → file, response headers → file via -D, final URL on stdout.
+        opts = ""
+        if timeout is not None:
+            opts += f" --max-time {timeout}"
+        if max_bytes is not None:
+            opts += f" --max-filesize {max_bytes}"
+        if headers:
+            for n, v in headers.items():
+                opts += f" -H {shlex.quote(f'{n}: {v}')}"
+        cmd = (
+            f"curl -sSL -g --fail"
+            f" --proto '=http,https' --proto-redir '=http,https'"
+            f"{opts}"
+            f" -D {header_quoted} -o {body_quoted}"
+            f" -w '%{{url_effective}}'"
+            f" -- {shlex.quote(url)}"
+        )
+        if max_bytes is not None:
+            cmd += f" && head -c {max_bytes} {body_quoted} > {part_quoted} && mv -f {part_quoted} {body_quoted}"
+
+        # Give curl a few extra seconds beyond its own --max-time so the
+        # sandbox kills it only if curl itself hangs.
+        exec_timeout = (timeout + 5) if timeout is not None else None
+
+        try:
+            try:
+                result = await self.execute(cmd, timeout=exec_timeout, **kwargs)
+            except SandboxTimeoutError as exc:
+                raise SandboxFetchError(f"fetch timed out after {timeout}s") from exc
+            if result.exit_code != 0:
+                raise SandboxFetchError(result.stderr.strip() or f"curl exited with code {result.exit_code}")
+
+            resolved_url = result.stdout.strip() or url
+            raw_headers = (await self.read_file(header_file)).decode("utf-8", errors="replace")
+            response_headers = _parse_header_dump(raw_headers)
+            body = await self.read_file(body_file)
+        finally:
+            with contextlib.suppress(Exception):
+                await self.execute(
+                    f"rm -f {body_quoted} {part_quoted} {header_quoted}",
+                    timeout=10,
+                )
+
+        return FetchResult(url=resolved_url, headers=response_headers, body=body)
diff --git a/strands-py/src/strands/sandbox/types.py b/strands-py/src/strands/sandbox/types.py
index 3e8dd316e9..3d36a81999 100644
--- a/strands-py/src/strands/sandbox/types.py
+++ b/strands-py/src/strands/sandbox/types.py
@@ -85,3 +85,18 @@ class ExecutionResult:
     stdout: str
     stderr: str
     output_files: list[OutputFile] = field(default_factory=list)
+
+
+@dataclass
+class FetchResult:
+    """Result of an HTTP fetch inside a sandbox.
+
+    Attributes:
+        url: Final URL after redirects.
+        headers: Response headers with lowercased keys.
+        body: Raw response body bytes.
+    """
+
+    url: str
+    headers: dict[str, str]
+    body: bytes
diff --git a/strands-py/tests/strands/sandbox/test_not_a_sandbox_local_environment.py b/strands-py/tests/strands/sandbox/test_not_a_sandbox_local_environment.py
index 8e29ac3b0a..f37dde44e5 100644
--- a/strands-py/tests/strands/sandbox/test_not_a_sandbox_local_environment.py
+++ b/strands-py/tests/strands/sandbox/test_not_a_sandbox_local_environment.py
@@ -7,11 +7,13 @@
 
 import os
 import sys
+from unittest.mock import patch
 
+import httpx
 import pytest
 
 from strands.sandbox import FileInfo
-from strands.sandbox.errors import SandboxPathNotFoundError
+from strands.sandbox.errors import SandboxFetchError, SandboxPathNotFoundError
 from strands.sandbox.not_a_sandbox_local_environment import NotASandboxLocalEnvironment
 
 pytestmark = pytest.mark.skipif(sys.platform == "win32", reason="POSIX shell required")
@@ -178,3 +180,87 @@ async def test_absolute_path_written_as_is(sandbox, tmp_path):
     file = str(tmp_path / "abs.txt")
     await sandbox.write_text(file, "absolute")
     assert (tmp_path / "abs.txt").read_text() == "absolute"
+
+
+# ---- fetch (httpx on host) ----
+
+_HTTPX_CLIENT_PATH = "strands.sandbox.not_a_sandbox_local_environment.httpx.AsyncClient"
+
+
+def _mock_client(handler):
+    """Return a factory that ignores kwargs and returns a mock-transport client."""
+    real_client = httpx.AsyncClient(transport=httpx.MockTransport(handler), follow_redirects=True)
+
+    def factory(**kwargs):
+        return real_client
+
+    return factory
+
+
+class TestFetch:
+    @pytest.mark.asyncio
+    async def test_successful_fetch(self, sandbox):
+        def handler(request):
+            return httpx.Response(200, headers={"content-type": "text/html"}, text="<h1>Hello</h1>")
+
+        with patch(_HTTPX_CLIENT_PATH, _mock_client(handler)):
+            result = await sandbox.fetch("https://example.com/")
+
+        assert result.body == b"<h1>Hello</h1>"
+        assert result.headers["content-type"] == "text/html"
+
+    @pytest.mark.asyncio
+    async def test_max_bytes_truncates(self, sandbox):
+        def handler(request):
+            return httpx.Response(200, text="x" * 1000)
+
+        with patch(_HTTPX_CLIENT_PATH, _mock_client(handler)):
+            result = await sandbox.fetch("https://example.com/big", max_bytes=100)
+
+        assert len(result.body) == 100
+
+    @pytest.mark.asyncio
+    async def test_http_error_raises(self, sandbox):
+        def handler(request):
+            return httpx.Response(404, text="Not Found")
+
+        with patch(_HTTPX_CLIENT_PATH, _mock_client(handler)):
+            with pytest.raises(SandboxFetchError, match="404"):
+                await sandbox.fetch("https://example.com/missing")
+
+    @pytest.mark.asyncio
+    async def test_timeout_raises(self, sandbox):
+        def handler(request):
+            raise httpx.ReadTimeout("timed out")
+
+        with patch(_HTTPX_CLIENT_PATH, _mock_client(handler)):
+            with pytest.raises(SandboxFetchError, match="timed out"):
+                await sandbox.fetch("https://example.com/slow")
+
+    @pytest.mark.asyncio
+    async def test_connection_error_raises(self, sandbox):
+        def handler(request):
+            raise httpx.ConnectError("refused")
+
+        with patch(_HTTPX_CLIENT_PATH, _mock_client(handler)):
+            with pytest.raises(SandboxFetchError, match="refused"):
+                await sandbox.fetch("https://example.com/down")
+
+    @pytest.mark.asyncio
+    async def test_rejects_non_http_scheme(self, sandbox):
+        with pytest.raises(SandboxFetchError, match="http\\(s\\)"):
+            await sandbox.fetch("ftp://example.com")
+
+    @pytest.mark.asyncio
+    async def test_rejects_no_host(self, sandbox):
+        with pytest.raises(SandboxFetchError, match="no host"):
+            await sandbox.fetch("http://")
+
+    @pytest.mark.asyncio
+    async def test_strips_whitespace(self, sandbox):
+        def handler(request):
+            return httpx.Response(200, text="ok")
+
+        with patch(_HTTPX_CLIENT_PATH, _mock_client(handler)):
+            result = await sandbox.fetch("  https://example.com  ")
+        assert result.body == b"ok"
diff --git a/strands-py/tests/strands/sandbox/test_posix_shell.py b/strands-py/tests/strands/sandbox/test_posix_shell.py
index 7bc8e75404..94263a89d8 100644
--- a/strands-py/tests/strands/sandbox/test_posix_shell.py
+++ b/strands-py/tests/strands/sandbox/test_posix_shell.py
@@ -22,12 +22,14 @@
 
 from strands.sandbox import (
     ExecutionResult,
+    FetchResult,
     FileInfo,
     PosixShellSandbox,
     StreamChunk,
 )
-from strands.sandbox.errors import SandboxPathNotFoundError
+from strands.sandbox.errors import SandboxFetchError, SandboxPathNotFoundError
 from strands.sandbox.posix_shell import (
+    _parse_header_dump,
     build_shell_env_prefix,
     validate_env_keys,
 )
@@ -607,3 +609,170 @@ def test_file_info_defaults():
     info = FileInfo(name="x")
     assert info.is_dir is None
     assert info.size is None
+
+
+class _MockFetchSandbox:
+    """Mock implementing execute + read_file for fetch() tests."""
+
+    def __init__(self, *, stdout="https://example.com", stderr="", exit_code=0, headers=None, body=None):
+        self._result = ExecutionResult(exit_code=exit_code, stdout=stdout, stderr=stderr)
+        self._headers = headers or b"HTTP/1.1 200 OK\r\nContent-Type: text/html; charset=utf-8\r\n\r\n"
+        self._body = body or b"<html><body>Hello</body></html>"
+        self.commands: list[str] = []
+
+    async def execute(self, command, **kwargs):
+        self.commands.append(command)
+        if command.startswith("rm "):
+            return ExecutionResult(exit_code=0, stdout="", stderr="")
+        return self._result
+
+    async def read_file(self, path, **kwargs):
+        if path.endswith(".headers"):
+            return self._headers
+        if "/tmp/strands-fetch-" in path:
+            return self._body
+        raise FileNotFoundError(path)
+
+
+class TestFetch:
+    @pytest.mark.asyncio
+    async def test_successful_fetch(self):
+        sb = _MockFetchSandbox()
+        result = await PosixShellSandbox.fetch(sb, "https://example.com")
+
+        assert result == FetchResult(
+            url="https://example.com",
+            headers={"content-type": "text/html; charset=utf-8"},
+            body=b"<html><body>Hello</body></html>",
+        )
+
+        # Verify hardened curl flags.
+        cmd = sb.commands[0]
+        assert "curl -sSL -g --fail" in cmd
+        assert "--proto '=http,https'" in cmd
+        assert "-D " in cmd and "-o " in cmd
+
+    @pytest.mark.asyncio
+    async def test_timeout_and_max_bytes_and_headers(self):
+        sb = _MockFetchSandbox()
+        await PosixShellSandbox.fetch(
+            sb,
+            "https://example.com",
+            max_bytes=1024,
+            timeout=15,
+            headers={"X-Key": "val"},
+        )
+        cmd = sb.commands[0]
+        assert "--max-time 15" in cmd
+        assert "head -c 1024" in cmd
+        assert "-H 'X-Key: val'" in cmd
+
+    @pytest.mark.asyncio
+    async def test_no_timeout_no_max_bytes(self):
+        sb = _MockFetchSandbox()
+        await PosixShellSandbox.fetch(sb, "https://example.com")
+        cmd = sb.commands[0]
+        assert "--max-time" not in cmd
+        assert "head -c" not in cmd
+
+    @pytest.mark.asyncio
+    async def test_curl_failure_raises(self):
+        sb = _MockFetchSandbox(exit_code=22, stderr="curl: (22) 404")
+        with pytest.raises(SandboxFetchError, match="404"):
+            await PosixShellSandbox.fetch(sb, "https://example.com")
+
+    @pytest.mark.asyncio
+    async def test_rejects_unsafe_url_chars(self):
+        """RFC 3986 character allowlist rejects shell-unsafe characters."""
+        for url in [
+            "https://example.com/path with spaces",
+            "https://example.com/café",
+            "https://example.com/`id`",
+            'https://example.com/"test"',
+            "https://example.com/path\nnewline",
+            "https://example.com/{path}",
+            "",
+        ]:
+            with pytest.raises(SandboxFetchError):
+                await PosixShellSandbox.fetch(_MockFetchSandbox(), url)
+
+    @pytest.mark.asyncio
+    async def test_rejects_non_http_scheme(self):
+        for url in ["ftp://example.com", "file:///etc/passwd", "javascript:alert(1)"]:
+            with pytest.raises(SandboxFetchError, match="http\\(s\\)"):
+                await PosixShellSandbox.fetch(_MockFetchSandbox(), url)
+
+    @pytest.mark.asyncio
+    async def test_rejects_no_host(self):
+        with pytest.raises(SandboxFetchError, match="no host"):
+            await PosixShellSandbox.fetch(_MockFetchSandbox(), "http://")
+
+    @pytest.mark.asyncio
+    async def test_strips_url_whitespace(self):
+        sb = _MockFetchSandbox()
+        result = await PosixShellSandbox.fetch(sb, "  https://example.com  ")
+        assert result.url == "https://example.com"
+
+    @pytest.mark.asyncio
+    async def test_cleanup_runs(self):
+        # Success
+        sb = _MockFetchSandbox()
+        await PosixShellSandbox.fetch(sb, "https://example.com")
+        assert sb.commands[-1].startswith("rm -f")
+
+        # Failure
+        sb = _MockFetchSandbox(exit_code=1, stderr="fail")
+        with pytest.raises(SandboxFetchError):
+            await PosixShellSandbox.fetch(sb, "https://example.com")
+        assert sb.commands[-1].startswith("rm -f")
+
+    @pytest.mark.asyncio
+    async def test_resolved_url(self):
+        sb = _MockFetchSandbox(stdout="https://example.com/redirected")
+        result = await PosixShellSandbox.fetch(sb, "https://example.com")
+        assert result.url == "https://example.com/redirected"
+
+        sb = _MockFetchSandbox(stdout="")
+        result = await PosixShellSandbox.fetch(sb, "https://example.com")
+        assert result.url == "https://example.com"
+
+
+class TestParseHeaderDump:
+    def test_basic_headers(self):
+        raw = "HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nX-Custom: value\r\n\r\n"
+        assert _parse_header_dump(raw) == {"content-type": "text/html", "x-custom": "value"}
+
+    def test_redirect_chain_uses_final_response_only(self):
+        raw = (
+            "HTTP/1.1 301 Moved\r\nContent-Type: text/html\r\n\r\n"
+            "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\r\n"
+        )
+        assert _parse_header_dump(raw)["content-type"] == "application/json"
+
+    def test_redirect_header_does_not_leak_when_final_omits_it(self):
+        # 302 with Content-Type → 200 without Content-Type.
+        # The final response has no Content-Type, so it must not appear.
+        raw = (
+            "HTTP/1.1 302 Found\r\nContent-Type: text/html\r\nLocation: /new\r\n\r\n"
+            "HTTP/1.1 200 OK\r\nX-Request-Id: abc\r\n\r\n"
+        )
+        result = _parse_header_dump(raw)
+        assert "content-type" not in result
+        assert result["x-request-id"] == "abc"
+
+    def test_lowercases_keys_preserves_values(self):
+        raw = "HTTP/1.1 200 OK\r\nContent-TYPE: Text/HTML; charset=UTF-8\r\n"
+        result = _parse_header_dump(raw)
+        assert "content-type" in result
+        assert result["content-type"] == "Text/HTML; charset=UTF-8"
+
+    def test_colon_in_value(self):
+        raw = "HTTP/1.1 200 OK\r\nLocation: https://example.com:8080/path\r\n"
+        assert _parse_header_dump(raw)["location"] == "https://example.com:8080/path"
+
+    def test_empty_input(self):
+        assert _parse_header_dump("") == {}
+
+    def test_lf_only(self):
+        raw = "HTTP/1.1 200 OK\nContent-Type: text/html\n"
+        assert _parse_header_dump(raw) == {"content-type": "text/html"}

From cdc3da4b49ed8a5b48442f6a25418e34a0db9874 Mon Sep 17 00:00:00 2001
From: liramon2 <liramon@amazon.com>
Date: Thu, 1 Oct 2026 13:08:22 -0400
Subject: [PATCH 2/4] refactor: change fetch to request

---
 strands-py/src/strands/sandbox/__init__.py    |   8 +-
 strands-py/src/strands/sandbox/base.py        |  34 ++-
 strands-py/src/strands/sandbox/errors.py      |   4 +-
 strands-py/src/strands/sandbox/posix_shell.py | 116 +++++++---
 strands-py/src/strands/sandbox/types.py       |  16 +-
 .../tests/strands/sandbox/test_posix_shell.py | 212 +++++++++++++-----
 6 files changed, 279 insertions(+), 111 deletions(-)

diff --git a/strands-py/src/strands/sandbox/__init__.py b/strands-py/src/strands/sandbox/__init__.py
index 4da87a472b..28bd38f27e 100644
--- a/strands-py/src/strands/sandbox/__init__.py
+++ b/strands-py/src/strands/sandbox/__init__.py
@@ -32,19 +32,19 @@ async def execute_streaming(self, command, *, timeout=None, cwd=None, env=None,
 
 from .base import Sandbox
 from .constants import LANGUAGE_PATTERN
-from .errors import SandboxFetchError
+from .errors import SandboxHttpError
 from .posix_shell import PosixShellSandbox
-from .types import ExecutionResult, FetchResult, FileInfo, OutputFile, StreamChunk, StreamType
+from .types import ExecutionResult, FileInfo, HttpResult, OutputFile, StreamChunk, StreamType
 
 __all__ = [
     "ExecutionResult",
-    "FetchResult",
     "FileInfo",
+    "HttpResult",
     "LANGUAGE_PATTERN",
     "OutputFile",
     "PosixShellSandbox",
     "Sandbox",
-    "SandboxFetchError",
+    "SandboxHttpError",
     "StreamChunk",
     "StreamType",
 ]
diff --git a/strands-py/src/strands/sandbox/base.py b/strands-py/src/strands/sandbox/base.py
index 20eddb8515..aa29aafb14 100644
--- a/strands-py/src/strands/sandbox/base.py
+++ b/strands-py/src/strands/sandbox/base.py
@@ -28,7 +28,7 @@
 from typing import Any
 
 from ..types.tools import AgentTool
-from .types import ExecutionResult, FetchResult, FileInfo, StreamChunk
+from .types import ExecutionResult, FileInfo, HttpResult, StreamChunk
 
 logger = logging.getLogger(__name__)
 
@@ -200,30 +200,40 @@ async def list_files(self, path: str, **kwargs: Any) -> list[FileInfo]:
         """
         ...
 
-    async def fetch(
+    async def request(
         self,
+        method: str,
         url: str,
         *,
-        max_bytes: int | None = None,
-        timeout: float | None = None,
         headers: dict[str, str] | None = None,
+        body: bytes | str | None = None,
+        timeout: float | None = None,
+        max_bytes: int | None = None,
         **kwargs: Any,
-    ) -> FetchResult:
-        """Fetch a URL via HTTP GET inside the sandbox.
+    ) -> HttpResult:
+        """Make an HTTP request inside the sandbox.
 
-        Optional because not every sandbox needs HTTP fetching.
+        Works with any HTTP method. Optional because not every sandbox needs
+        HTTP access.
 
         Args:
-            url: The http(s) URL to fetch.
-            max_bytes: Cap the response body to this many bytes. ``None`` means no limit.
-            timeout: Maximum execution time in seconds. ``None`` means no timeout.
+            method: HTTP method (e.g. ``GET``, ``POST``, ``PUT``, ``DELETE``,
+                ``PATCH``, ``HEAD``, ``OPTIONS``).
+            url: The http(s) URL to request.
             headers: Request headers.
+            body: Request body. ``str`` is sent as UTF-8; ``bytes`` is sent as-is.
+                ``None`` sends no body.
+            timeout: Maximum execution time in seconds. ``None`` means no timeout.
+            max_bytes: Cap the response body to this many bytes. ``None`` means no limit.
             **kwargs: Additional keyword arguments for forward compatibility.
 
+        Returns:
+            An :class:`HttpResult`. HTTP error statuses (4xx/5xx) are returned in ``status`` rather than raised.
+
         Raises:
-            SandboxFetchError: On network or transport errors.
+            SandboxHttpError: On network or transport errors.
         """
-        raise NotImplementedError(f"{type(self).__name__} does not implement fetch()")
+        raise NotImplementedError(f"{type(self).__name__} does not implement request()")
 
     # ---- Tool vending ----
 
diff --git a/strands-py/src/strands/sandbox/errors.py b/strands-py/src/strands/sandbox/errors.py
index 88588746c2..f0deda1af2 100644
--- a/strands-py/src/strands/sandbox/errors.py
+++ b/strands-py/src/strands/sandbox/errors.py
@@ -25,8 +25,8 @@ def __init__(self, seconds: float | None, stdout: str = "", stderr: str = "") ->
         self.stderr = stderr
 
 
-class SandboxFetchError(RuntimeError):
-    """Raised when :meth:`~strands.sandbox.base.Sandbox.fetch` fails."""
+class SandboxHttpError(RuntimeError):
+    """Raised when :meth:`~strands.sandbox.base.Sandbox.request` fails."""
 
 
 class SandboxPathNotFoundError(FileNotFoundError):
diff --git a/strands-py/src/strands/sandbox/posix_shell.py b/strands-py/src/strands/sandbox/posix_shell.py
index 3aae9ccff1..96abbcc53a 100644
--- a/strands-py/src/strands/sandbox/posix_shell.py
+++ b/strands-py/src/strands/sandbox/posix_shell.py
@@ -21,8 +21,8 @@
 
 from .base import Sandbox
 from .constants import ENV_KEY_PATTERN, LANGUAGE_PATTERN
-from .errors import SandboxFetchError, SandboxPathNotFoundError, SandboxTimeoutError
-from .types import ExecutionResult, FetchResult, FileInfo, StreamChunk
+from .errors import SandboxHttpError, SandboxPathNotFoundError, SandboxTimeoutError
+from .types import ExecutionResult, FileInfo, HttpResult, StreamChunk
 
 logger = logging.getLogger(__name__)
 
@@ -31,6 +31,11 @@
 # the URL reaches a shell command.  Security-critical for the curl transport.
 _URL_CHARS = re.compile(r"[A-Za-z0-9\-._~:/?#\[\]@!$&'()*+,;=%]+")
 
+# HTTP methods accepted by request(). An explicit allowlist (matching the
+# http_request tool's HttpMethod) keeps the method safe to place in the curl
+# command line and rejects anything unexpected outright.
+_ALLOWED_METHODS = frozenset({"GET", "POST", "PUT", "DELETE", "PATCH", "HEAD", "OPTIONS"})
+
 
 def validate_env_keys(env: dict[str, str]) -> None:
     """Validate environment variable names against :data:`ENV_KEY_PATTERN`.
@@ -82,12 +87,16 @@ def _eof_marker() -> str:
     return f"STRANDS_EOF_{uuid.uuid4().hex[:16]}"
 
 
-def _parse_header_dump(raw: str) -> dict[str, str]:
-    r"""Parse a curl ``-D`` header dump into ``{lowered_name: value}``.
+def _parse_status_and_headers(raw: str) -> tuple[int, str, dict[str, str]]:
+    r"""Parse a curl ``-D`` header dump into ``(status, status_text, headers)``.
 
     Curl dumps headers for every hop in a redirect chain, separated by blank
     lines. Only the final response is parsed so that stale headers from
     intermediate redirects don't leak through.
+
+    Repeated headers (e.g. ``Set-Cookie``) are preserved by joining their values
+    with a newline, matching the ``http_request`` tool's handling. ``status`` is
+    ``0`` and ``status_text`` is ``""`` when the dump has no parseable status line.
     """
     # Split on blank lines, take the last non-empty block.
     blocks = re.split(r"\n\s*\n", raw.replace("\r\n", "\n"))
@@ -97,18 +106,28 @@ def _parse_header_dump(raw: str) -> dict[str, str]:
             last_block = block
             break
 
-    result: dict[str, str] = {}
+    status = 0
+    status_text = ""
+    headers: dict[str, str] = {}
     for line in last_block.split("\n"):
         line = line.strip()
-        if not line or line.upper().startswith("HTTP/"):
+        if not line:
+            continue
+        if line.upper().startswith("HTTP/"):
+            # Status line, e.g. "HTTP/1.1 200 OK" or "HTTP/2 204".
+            parts = line.split(None, 2)
+            if len(parts) >= 2 and parts[1].isdigit():
+                status = int(parts[1])
+                status_text = parts[2].strip() if len(parts) == 3 else ""
             continue
         colon = line.find(":")
         if colon < 1:
             continue
         name = line[:colon].strip().lower()
         value = line[colon + 1 :].strip()
-        result[name] = value
-    return result
+        existing = headers.get(name)
+        headers[name] = value if existing is None else f"{existing}\n{value}"
+    return status, status_text, headers
 
 
 class PosixShellSandbox(Sandbox, ABC):
@@ -277,41 +296,58 @@ async def list_files(self, path: str, **kwargs: Any) -> list[FileInfo]:
                 entries.append(FileInfo(name=name, is_dir=is_dir))
         return entries
 
-    async def fetch(
+    async def request(
         self,
+        method: str,
         url: str,
         *,
-        max_bytes: int | None = None,
-        timeout: float | None = None,
         headers: dict[str, str] | None = None,
+        body: bytes | str | None = None,
+        timeout: float | None = None,
+        max_bytes: int | None = None,
         **kwargs: Any,
-    ) -> FetchResult:
-        """Fetch a URL with ``curl`` inside the sandbox.
-
-        ``max_bytes`` is rejected early when ``Content-Length`` is known and
-        truncated for chunked responses.
+    ) -> HttpResult:
+        """Make an HTTP request with ``curl`` inside the sandbox.
+
+        Works with any HTTP method. HTTP error statuses (4xx/5xx) are returned
+        in :attr:`HttpResult.status` rather than raised; only transport-level
+        failures (DNS, connection, timeout, size cap) raise
+        :class:`SandboxHttpError`. ``max_bytes`` is rejected early when
+        ``Content-Length`` is known and truncated for chunked responses.
         """
+        method = method.strip().upper()
+        if method not in _ALLOWED_METHODS:
+            raise SandboxHttpError(f"Unsupported HTTP method {method!r}; expected one of {sorted(_ALLOWED_METHODS)}.")
         url = url.strip()
         if not _URL_CHARS.fullmatch(url):
-            raise SandboxFetchError(
-                "fetch URLs may only contain the characters RFC 3986 allows; "
+            raise SandboxHttpError(
+                "request URLs may only contain the characters RFC 3986 allows; "
                 "percent-encode spaces and non-ASCII characters (and punycode the host) and retry."
             )
         parts = urlparse(url)
         if parts.scheme not in ("http", "https"):
-            raise SandboxFetchError(f"fetch only supports http(s) URLs, got {url!r}.")
+            raise SandboxHttpError(f"request only supports http(s) URLs, got {url!r}.")
         if not parts.hostname:
-            raise SandboxFetchError(f"fetch URL has no host: {url!r}.")
+            raise SandboxHttpError(f"request URL has no host: {url!r}.")
 
         tag = uuid.uuid4().hex
-        body_file = f"/tmp/strands-fetch-{tag}"
-        header_file = f"/tmp/strands-fetch-{tag}.headers"
+        body_file = f"/tmp/strands-http-{tag}"
+        header_file = f"/tmp/strands-http-{tag}.headers"
+        data_file = f"/tmp/strands-http-{tag}.data"
         body_quoted = shlex.quote(body_file)
         header_quoted = shlex.quote(header_file)
+        data_quoted = shlex.quote(data_file)
         part_quoted = shlex.quote(body_file + ".part")
 
-        # Build a hardened curl command: protocol restricted to http(s),
-        # body → file, response headers → file via -D, final URL on stdout.
+        # A HEAD response carries no body; use curl's --head so it doesn't block
+        # waiting for a body the server will never send.
+        is_head = method == "HEAD"
+
+        # Write the request body to a temp file and feed it via --data-binary so
+        # arbitrary bytes never touch the shell command line.
+        if body is not None:
+            await self.write_file(data_file, body.encode("utf-8") if isinstance(body, str) else body)
+
         opts = ""
         if timeout is not None:
             opts += f" --max-time {timeout}"
@@ -320,15 +356,23 @@ async def fetch(
         if headers:
             for n, v in headers.items():
                 opts += f" -H {shlex.quote(f'{n}: {v}')}"
+        if body is not None:
+            opts += f" --data-binary @{data_quoted}"
+
+        method_opt = "--head" if is_head else f"-X {method}"
+        out_target = "/dev/null" if is_head else body_quoted
+
+        # Build a hardened curl command: protocol restricted to http(s), body →
+        # file, response headers + status → file via -D, final URL on stdout.
         cmd = (
-            f"curl -sSL -g --fail"
+            f"curl -sSL -g {method_opt}"
             f" --proto '=http,https' --proto-redir '=http,https'"
             f"{opts}"
-            f" -D {header_quoted} -o {body_quoted}"
+            f" -D {header_quoted} -o {out_target}"
             f" -w '%{{url_effective}}'"
             f" -- {shlex.quote(url)}"
         )
-        if max_bytes is not None:
+        if max_bytes is not None and not is_head:
             cmd += f" && head -c {max_bytes} {body_quoted} > {part_quoted} && mv -f {part_quoted} {body_quoted}"
 
         # Give curl a few extra seconds beyond its own --max-time so the
@@ -339,19 +383,25 @@ async def fetch(
             try:
                 result = await self.execute(cmd, timeout=exec_timeout, **kwargs)
             except SandboxTimeoutError as exc:
-                raise SandboxFetchError(f"fetch timed out after {timeout}s") from exc
+                raise SandboxHttpError(f"request timed out after {timeout}s") from exc
             if result.exit_code != 0:
-                raise SandboxFetchError(result.stderr.strip() or f"curl exited with code {result.exit_code}")
+                raise SandboxHttpError(result.stderr.strip() or f"curl exited with code {result.exit_code}")
 
             resolved_url = result.stdout.strip() or url
             raw_headers = (await self.read_file(header_file)).decode("utf-8", errors="replace")
-            response_headers = _parse_header_dump(raw_headers)
-            body = await self.read_file(body_file)
+            status, status_text, response_headers = _parse_status_and_headers(raw_headers)
+            response_body = b"" if is_head else await self.read_file(body_file)
         finally:
             with contextlib.suppress(Exception):
                 await self.execute(
-                    f"rm -f {body_quoted} {part_quoted} {header_quoted}",
+                    f"rm -f {body_quoted} {part_quoted} {header_quoted} {data_quoted}",
                     timeout=10,
                 )
 
-        return FetchResult(url=resolved_url, headers=response_headers, body=body)
+        return HttpResult(
+            status=status,
+            status_text=status_text,
+            resolved_url=resolved_url,
+            headers=response_headers,
+            body=response_body,
+        )
diff --git a/strands-py/src/strands/sandbox/types.py b/strands-py/src/strands/sandbox/types.py
index 3d36a81999..6d7619b43b 100644
--- a/strands-py/src/strands/sandbox/types.py
+++ b/strands-py/src/strands/sandbox/types.py
@@ -88,15 +88,21 @@ class ExecutionResult:
 
 
 @dataclass
-class FetchResult:
-    """Result of an HTTP fetch inside a sandbox.
+class HttpResult:
+    """Result of an HTTP request inside a sandbox.
 
     Attributes:
-        url: Final URL after redirects.
-        headers: Response headers with lowercased keys.
+        status: HTTP status code of the final response.
+        status_text: HTTP reason phrase of the final response (may be empty when
+            the server omits it).
+        resolved_url: Final URL after redirects.
+        headers: Response headers with lowercased keys. Repeated headers (e.g.
+            ``Set-Cookie``) are preserved, joined with a newline.
         body: Raw response body bytes.
     """
 
-    url: str
+    status: int
+    status_text: str
+    resolved_url: str
     headers: dict[str, str]
     body: bytes
diff --git a/strands-py/tests/strands/sandbox/test_posix_shell.py b/strands-py/tests/strands/sandbox/test_posix_shell.py
index 94263a89d8..bb1b838bef 100644
--- a/strands-py/tests/strands/sandbox/test_posix_shell.py
+++ b/strands-py/tests/strands/sandbox/test_posix_shell.py
@@ -22,14 +22,14 @@
 
 from strands.sandbox import (
     ExecutionResult,
-    FetchResult,
     FileInfo,
+    HttpResult,
     PosixShellSandbox,
     StreamChunk,
 )
-from strands.sandbox.errors import SandboxFetchError, SandboxPathNotFoundError
+from strands.sandbox.errors import SandboxHttpError, SandboxPathNotFoundError
 from strands.sandbox.posix_shell import (
-    _parse_header_dump,
+    _parse_status_and_headers,
     build_shell_env_prefix,
     validate_env_keys,
 )
@@ -611,14 +611,15 @@ def test_file_info_defaults():
     assert info.size is None
 
 
-class _MockFetchSandbox:
-    """Mock implementing execute + read_file for fetch() tests."""
+class _MockHttpSandbox:
+    """Mock implementing execute + read_file + write_file for request() tests."""
 
     def __init__(self, *, stdout="https://example.com", stderr="", exit_code=0, headers=None, body=None):
         self._result = ExecutionResult(exit_code=exit_code, stdout=stdout, stderr=stderr)
         self._headers = headers or b"HTTP/1.1 200 OK\r\nContent-Type: text/html; charset=utf-8\r\n\r\n"
         self._body = body or b"<html><body>Hello</body></html>"
         self.commands: list[str] = []
+        self.written: dict[str, bytes] = {}
 
     async def execute(self, command, **kwargs):
         self.commands.append(command)
@@ -629,34 +630,41 @@ async def execute(self, command, **kwargs):
     async def read_file(self, path, **kwargs):
         if path.endswith(".headers"):
             return self._headers
-        if "/tmp/strands-fetch-" in path:
+        if "/tmp/strands-http-" in path:
             return self._body
         raise FileNotFoundError(path)
 
+    async def write_file(self, path, content, **kwargs):
+        self.written[path] = content
 
-class TestFetch:
-    @pytest.mark.asyncio
-    async def test_successful_fetch(self):
-        sb = _MockFetchSandbox()
-        result = await PosixShellSandbox.fetch(sb, "https://example.com")
 
-        assert result == FetchResult(
-            url="https://example.com",
+class TestRequest:
+    @pytest.mark.asyncio
+    async def test_successful_request(self):
+        sb = _MockHttpSandbox()
+        result = await PosixShellSandbox.request(sb, "GET", "https://example.com")
+
+        assert result == HttpResult(
+            status=200,
+            status_text="OK",
+            resolved_url="https://example.com",
             headers={"content-type": "text/html; charset=utf-8"},
             body=b"<html><body>Hello</body></html>",
         )
 
-        # Verify hardened curl flags.
+        # Verify hardened curl flags. No --fail: error statuses are reported.
         cmd = sb.commands[0]
-        assert "curl -sSL -g --fail" in cmd
+        assert "curl -sSL -g -X GET" in cmd
+        assert "--fail" not in cmd
         assert "--proto '=http,https'" in cmd
         assert "-D " in cmd and "-o " in cmd
 
     @pytest.mark.asyncio
     async def test_timeout_and_max_bytes_and_headers(self):
-        sb = _MockFetchSandbox()
-        await PosixShellSandbox.fetch(
+        sb = _MockHttpSandbox()
+        await PosixShellSandbox.request(
             sb,
+            "GET",
             "https://example.com",
             max_bytes=1024,
             timeout=15,
@@ -669,17 +677,83 @@ async def test_timeout_and_max_bytes_and_headers(self):
 
     @pytest.mark.asyncio
     async def test_no_timeout_no_max_bytes(self):
-        sb = _MockFetchSandbox()
-        await PosixShellSandbox.fetch(sb, "https://example.com")
+        sb = _MockHttpSandbox()
+        await PosixShellSandbox.request(sb, "GET", "https://example.com")
         cmd = sb.commands[0]
         assert "--max-time" not in cmd
         assert "head -c" not in cmd
 
     @pytest.mark.asyncio
-    async def test_curl_failure_raises(self):
-        sb = _MockFetchSandbox(exit_code=22, stderr="curl: (22) 404")
-        with pytest.raises(SandboxFetchError, match="404"):
-            await PosixShellSandbox.fetch(sb, "https://example.com")
+    async def test_post_with_body_writes_data_file(self):
+        sb = _MockHttpSandbox()
+        await PosixShellSandbox.request(sb, "POST", "https://example.com", body='{"a": 1}')
+        # Body is written to a temp file (never placed on the command line)...
+        assert any(p.endswith(".data") for p in sb.written)
+        data_path = next(p for p in sb.written if p.endswith(".data"))
+        assert sb.written[data_path] == b'{"a": 1}'
+        # ...and fed to curl via --data-binary @file with the method set.
+        cmd = sb.commands[0]
+        assert "-X POST" in cmd
+        assert f"--data-binary @{shlex.quote(data_path)}" in cmd
+        assert '{"a": 1}' not in cmd
+
+    @pytest.mark.asyncio
+    async def test_bytes_body_sent_as_is(self):
+        sb = _MockHttpSandbox()
+        await PosixShellSandbox.request(sb, "PUT", "https://example.com", body=b"\x00\x01\x02")
+        data_path = next(p for p in sb.written if p.endswith(".data"))
+        assert sb.written[data_path] == b"\x00\x01\x02"
+
+    @pytest.mark.asyncio
+    async def test_no_body_writes_nothing_and_omits_data_flag(self):
+        sb = _MockHttpSandbox()
+        await PosixShellSandbox.request(sb, "GET", "https://example.com")
+        assert sb.written == {}
+        assert "--data-binary" not in sb.commands[0]
+
+    @pytest.mark.asyncio
+    async def test_head_uses_head_flag_and_empty_body(self):
+        sb = _MockHttpSandbox()
+        result = await PosixShellSandbox.request(sb, "HEAD", "https://example.com")
+        cmd = sb.commands[0]
+        # --head avoids curl hanging on a body the server never sends; the body
+        # goes to /dev/null and the response body is empty.
+        assert "--head" in cmd
+        assert "-X HEAD" not in cmd
+        assert "-o /dev/null" in cmd
+        assert result.status == 200
+        assert result.status_text == "OK"
+        assert result.body == b""
+        assert result.headers["content-type"] == "text/html; charset=utf-8"
+
+    @pytest.mark.asyncio
+    async def test_method_is_normalized_and_validated(self):
+        sb = _MockHttpSandbox()
+        await PosixShellSandbox.request(sb, "  patch  ", "https://example.com")
+        assert "-X PATCH" in sb.commands[0]
+
+        for method in ["GET;rm -rf /", "GET POST", "", "G-E-T", "POST!"]:
+            with pytest.raises(SandboxHttpError, match="method"):
+                await PosixShellSandbox.request(_MockHttpSandbox(), method, "https://example.com")
+
+    @pytest.mark.asyncio
+    async def test_transport_failure_raises(self):
+        # Without --fail, HTTP error statuses no longer fail curl; only genuine
+        # transport errors (DNS, connection, size cap) make it exit non-zero.
+        sb = _MockHttpSandbox(exit_code=7, stderr="curl: (7) Failed to connect")
+        with pytest.raises(SandboxHttpError, match="connect"):
+            await PosixShellSandbox.request(sb, "GET", "https://example.com")
+
+    @pytest.mark.asyncio
+    async def test_error_status_is_returned_not_raised(self):
+        sb = _MockHttpSandbox(
+            headers=b"HTTP/1.1 404 Not Found\r\nContent-Type: text/plain\r\n\r\n",
+            body=b"nope",
+        )
+        result = await PosixShellSandbox.request(sb, "GET", "https://example.com/missing")
+        assert result.status == 404
+        assert result.status_text == "Not Found"
+        assert result.body == b"nope"
 
     @pytest.mark.asyncio
     async def test_rejects_unsafe_url_chars(self):
@@ -693,61 +767,83 @@ async def test_rejects_unsafe_url_chars(self):
             "https://example.com/{path}",
             "",
         ]:
-            with pytest.raises(SandboxFetchError):
-                await PosixShellSandbox.fetch(_MockFetchSandbox(), url)
+            with pytest.raises(SandboxHttpError):
+                await PosixShellSandbox.request(_MockHttpSandbox(), "GET", url)
 
     @pytest.mark.asyncio
     async def test_rejects_non_http_scheme(self):
         for url in ["ftp://example.com", "file:///etc/passwd", "javascript:alert(1)"]:
-            with pytest.raises(SandboxFetchError, match="http\\(s\\)"):
-                await PosixShellSandbox.fetch(_MockFetchSandbox(), url)
+            with pytest.raises(SandboxHttpError, match="http\\(s\\)"):
+                await PosixShellSandbox.request(_MockHttpSandbox(), "GET", url)
 
     @pytest.mark.asyncio
     async def test_rejects_no_host(self):
-        with pytest.raises(SandboxFetchError, match="no host"):
-            await PosixShellSandbox.fetch(_MockFetchSandbox(), "http://")
+        with pytest.raises(SandboxHttpError, match="no host"):
+            await PosixShellSandbox.request(_MockHttpSandbox(), "GET", "http://")
 
     @pytest.mark.asyncio
     async def test_strips_url_whitespace(self):
-        sb = _MockFetchSandbox()
-        result = await PosixShellSandbox.fetch(sb, "  https://example.com  ")
-        assert result.url == "https://example.com"
+        sb = _MockHttpSandbox()
+        result = await PosixShellSandbox.request(sb, "GET", "  https://example.com  ")
+        assert result.resolved_url == "https://example.com"
 
     @pytest.mark.asyncio
     async def test_cleanup_runs(self):
         # Success
-        sb = _MockFetchSandbox()
-        await PosixShellSandbox.fetch(sb, "https://example.com")
+        sb = _MockHttpSandbox()
+        await PosixShellSandbox.request(sb, "GET", "https://example.com")
         assert sb.commands[-1].startswith("rm -f")
 
         # Failure
-        sb = _MockFetchSandbox(exit_code=1, stderr="fail")
-        with pytest.raises(SandboxFetchError):
-            await PosixShellSandbox.fetch(sb, "https://example.com")
+        sb = _MockHttpSandbox(exit_code=1, stderr="fail")
+        with pytest.raises(SandboxHttpError):
+            await PosixShellSandbox.request(sb, "GET", "https://example.com")
         assert sb.commands[-1].startswith("rm -f")
 
     @pytest.mark.asyncio
     async def test_resolved_url(self):
-        sb = _MockFetchSandbox(stdout="https://example.com/redirected")
-        result = await PosixShellSandbox.fetch(sb, "https://example.com")
-        assert result.url == "https://example.com/redirected"
+        sb = _MockHttpSandbox(stdout="https://example.com/redirected")
+        result = await PosixShellSandbox.request(sb, "GET", "https://example.com")
+        assert result.resolved_url == "https://example.com/redirected"
 
-        sb = _MockFetchSandbox(stdout="")
-        result = await PosixShellSandbox.fetch(sb, "https://example.com")
-        assert result.url == "https://example.com"
+        sb = _MockHttpSandbox(stdout="")
+        result = await PosixShellSandbox.request(sb, "GET", "https://example.com")
+        assert result.resolved_url == "https://example.com"
 
 
-class TestParseHeaderDump:
+class TestParseStatusAndHeaders:
     def test_basic_headers(self):
         raw = "HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nX-Custom: value\r\n\r\n"
-        assert _parse_header_dump(raw) == {"content-type": "text/html", "x-custom": "value"}
+        status, status_text, headers = _parse_status_and_headers(raw)
+        assert status == 200
+        assert status_text == "OK"
+        assert headers == {"content-type": "text/html", "x-custom": "value"}
+
+    def test_status_without_reason_phrase(self):
+        # HTTP/2 responses have no reason phrase.
+        status, status_text, headers = _parse_status_and_headers("HTTP/2 204\r\n\r\n")
+        assert status == 204
+        assert status_text == ""
+
+    def test_multi_word_reason_phrase(self):
+        status, status_text, _ = _parse_status_and_headers("HTTP/1.1 404 Not Found\r\n\r\n")
+        assert status == 404
+        assert status_text == "Not Found"
+
+    def test_repeated_headers_are_preserved(self):
+        raw = "HTTP/1.1 200 OK\r\nSet-Cookie: a=1\r\nSet-Cookie: b=2\r\n\r\n"
+        _, _, headers = _parse_status_and_headers(raw)
+        assert headers["set-cookie"] == "a=1\nb=2"
 
     def test_redirect_chain_uses_final_response_only(self):
         raw = (
             "HTTP/1.1 301 Moved\r\nContent-Type: text/html\r\n\r\n"
             "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\r\n"
         )
-        assert _parse_header_dump(raw)["content-type"] == "application/json"
+        status, status_text, headers = _parse_status_and_headers(raw)
+        assert status == 200
+        assert status_text == "OK"
+        assert headers["content-type"] == "application/json"
 
     def test_redirect_header_does_not_leak_when_final_omits_it(self):
         # 302 with Content-Type → 200 without Content-Type.
@@ -756,23 +852,29 @@ def test_redirect_header_does_not_leak_when_final_omits_it(self):
             "HTTP/1.1 302 Found\r\nContent-Type: text/html\r\nLocation: /new\r\n\r\n"
             "HTTP/1.1 200 OK\r\nX-Request-Id: abc\r\n\r\n"
         )
-        result = _parse_header_dump(raw)
-        assert "content-type" not in result
-        assert result["x-request-id"] == "abc"
+        _, _, headers = _parse_status_and_headers(raw)
+        assert "content-type" not in headers
+        assert headers["x-request-id"] == "abc"
 
     def test_lowercases_keys_preserves_values(self):
         raw = "HTTP/1.1 200 OK\r\nContent-TYPE: Text/HTML; charset=UTF-8\r\n"
-        result = _parse_header_dump(raw)
-        assert "content-type" in result
-        assert result["content-type"] == "Text/HTML; charset=UTF-8"
+        _, _, headers = _parse_status_and_headers(raw)
+        assert "content-type" in headers
+        assert headers["content-type"] == "Text/HTML; charset=UTF-8"
 
     def test_colon_in_value(self):
         raw = "HTTP/1.1 200 OK\r\nLocation: https://example.com:8080/path\r\n"
-        assert _parse_header_dump(raw)["location"] == "https://example.com:8080/path"
+        _, _, headers = _parse_status_and_headers(raw)
+        assert headers["location"] == "https://example.com:8080/path"
 
     def test_empty_input(self):
-        assert _parse_header_dump("") == {}
+        status, status_text, headers = _parse_status_and_headers("")
+        assert status == 0
+        assert status_text == ""
+        assert headers == {}
 
     def test_lf_only(self):
         raw = "HTTP/1.1 200 OK\nContent-Type: text/html\n"
-        assert _parse_header_dump(raw) == {"content-type": "text/html"}
+        status, _, headers = _parse_status_and_headers(raw)
+        assert status == 200
+        assert headers == {"content-type": "text/html"}

From 91e50a70181e007fcf13cc83f61c99041bf2d8df Mon Sep 17 00:00:00 2001
From: liramon2 <liramon@amazon.com>
Date: Thu, 1 Oct 2026 13:09:41 -0400
Subject: [PATCH 3/4] refactor: remove fetch from NotASandboxLocalEnvironment

---
 .../not_a_sandbox_local_environment.py        | 51 +----------
 .../test_not_a_sandbox_local_environment.py   | 88 +------------------
 2 files changed, 3 insertions(+), 136 deletions(-)

diff --git a/strands-py/src/strands/sandbox/not_a_sandbox_local_environment.py b/strands-py/src/strands/sandbox/not_a_sandbox_local_environment.py
index 37e581583a..8bf288cf9a 100644
--- a/strands-py/src/strands/sandbox/not_a_sandbox_local_environment.py
+++ b/strands-py/src/strands/sandbox/not_a_sandbox_local_environment.py
@@ -19,16 +19,13 @@
 from collections.abc import AsyncGenerator
 from pathlib import Path
 from typing import Any
-from urllib.parse import urlparse
-
-import httpx
 
 from .base import Sandbox
 from .constants import LANGUAGE_PATTERN
-from .errors import SandboxFetchError, SandboxPathNotFoundError
+from .errors import SandboxPathNotFoundError
 from .posix_shell import build_shell_env_prefix
 from .stream_process import _stream_process
-from .types import ExecutionResult, FetchResult, FileInfo, StreamChunk
+from .types import ExecutionResult, FileInfo, StreamChunk
 
 
 class NotASandboxLocalEnvironment(Sandbox):
@@ -208,47 +205,3 @@ async def list_files(self, path: str, **kwargs: Any) -> list[FileInfo]:
                 except OSError:
                     results.append(FileInfo(name=entry.name))
         return results
-
-    async def fetch(
-        self,
-        url: str,
-        *,
-        max_bytes: int | None = None,
-        timeout: float | None = None,
-        headers: dict[str, str] | None = None,
-        **kwargs: Any,
-    ) -> FetchResult:
-        """Fetch a URL with httpx on the host.
-
-        ``max_bytes`` is enforced by streaming: reading stops as soon as
-        the cap is reached and the connection is closed.
-        """
-        url = url.strip()
-        parts = urlparse(url)
-        if parts.scheme not in ("http", "https"):
-            raise SandboxFetchError(f"fetch only supports http(s) URLs, got {url!r}.")
-        if not parts.hostname:
-            raise SandboxFetchError(f"fetch URL has no host: {url!r}.")
-
-        try:
-            async with httpx.AsyncClient(follow_redirects=True, timeout=timeout) as client:
-                async with client.stream("GET", url, headers=headers or {}) as response:
-                    response.raise_for_status()
-                    chunks: list[bytes] = []
-                    total = 0
-                    async for chunk in response.aiter_bytes():
-                        total += len(chunk)
-                        chunks.append(chunk)
-                        if max_bytes is not None and total >= max_bytes:
-                            break
-                    body = b"".join(chunks)
-                    if max_bytes is not None:
-                        body = body[:max_bytes]
-                    result_url = str(response.url)
-                    result_headers = {k: v for k, v in response.headers.items()}
-        except httpx.HTTPStatusError as exc:
-            raise SandboxFetchError(f"HTTP {exc.response.status_code} for {url}") from exc
-        except (httpx.TimeoutException, httpx.RequestError) as exc:
-            raise SandboxFetchError(str(exc)) from exc
-
-        return FetchResult(url=result_url, headers=result_headers, body=body)
diff --git a/strands-py/tests/strands/sandbox/test_not_a_sandbox_local_environment.py b/strands-py/tests/strands/sandbox/test_not_a_sandbox_local_environment.py
index f37dde44e5..8e29ac3b0a 100644
--- a/strands-py/tests/strands/sandbox/test_not_a_sandbox_local_environment.py
+++ b/strands-py/tests/strands/sandbox/test_not_a_sandbox_local_environment.py
@@ -7,13 +7,11 @@
 
 import os
 import sys
-from unittest.mock import patch
 
-import httpx
 import pytest
 
 from strands.sandbox import FileInfo
-from strands.sandbox.errors import SandboxFetchError, SandboxPathNotFoundError
+from strands.sandbox.errors import SandboxPathNotFoundError
 from strands.sandbox.not_a_sandbox_local_environment import NotASandboxLocalEnvironment
 
 pytestmark = pytest.mark.skipif(sys.platform == "win32", reason="POSIX shell required")
@@ -180,87 +178,3 @@ async def test_absolute_path_written_as_is(sandbox, tmp_path):
     file = str(tmp_path / "abs.txt")
     await sandbox.write_text(file, "absolute")
     assert (tmp_path / "abs.txt").read_text() == "absolute"
-
-
-# ---- fetch (httpx on host) ----
-
-_HTTPX_CLIENT_PATH = "strands.sandbox.not_a_sandbox_local_environment.httpx.AsyncClient"
-
-
-def _mock_client(handler):
-    """Return a factory that ignores kwargs and returns a mock-transport client."""
-    real_client = httpx.AsyncClient(transport=httpx.MockTransport(handler), follow_redirects=True)
-
-    def factory(**kwargs):
-        return real_client
-
-    return factory
-
-
-class TestFetch:
-    @pytest.mark.asyncio
-    async def test_successful_fetch(self, sandbox):
-        def handler(request):
-            return httpx.Response(200, headers={"content-type": "text/html"}, text="<h1>Hello</h1>")
-
-        with patch(_HTTPX_CLIENT_PATH, _mock_client(handler)):
-            result = await sandbox.fetch("https://example.com/")
-
-        assert result.body == b"<h1>Hello</h1>"
-        assert result.headers["content-type"] == "text/html"
-
-    @pytest.mark.asyncio
-    async def test_max_bytes_truncates(self, sandbox):
-        def handler(request):
-            return httpx.Response(200, text="x" * 1000)
-
-        with patch(_HTTPX_CLIENT_PATH, _mock_client(handler)):
-            result = await sandbox.fetch("https://example.com/big", max_bytes=100)
-
-        assert len(result.body) == 100
-
-    @pytest.mark.asyncio
-    async def test_http_error_raises(self, sandbox):
-        def handler(request):
-            return httpx.Response(404, text="Not Found")
-
-        with patch(_HTTPX_CLIENT_PATH, _mock_client(handler)):
-            with pytest.raises(SandboxFetchError, match="404"):
-                await sandbox.fetch("https://example.com/missing")
-
-    @pytest.mark.asyncio
-    async def test_timeout_raises(self, sandbox):
-        def handler(request):
-            raise httpx.ReadTimeout("timed out")
-
-        with patch(_HTTPX_CLIENT_PATH, _mock_client(handler)):
-            with pytest.raises(SandboxFetchError, match="timed out"):
-                await sandbox.fetch("https://example.com/slow")
-
-    @pytest.mark.asyncio
-    async def test_connection_error_raises(self, sandbox):
-        def handler(request):
-            raise httpx.ConnectError("refused")
-
-        with patch(_HTTPX_CLIENT_PATH, _mock_client(handler)):
-            with pytest.raises(SandboxFetchError, match="refused"):
-                await sandbox.fetch("https://example.com/down")
-
-    @pytest.mark.asyncio
-    async def test_rejects_non_http_scheme(self, sandbox):
-        with pytest.raises(SandboxFetchError, match="http\\(s\\)"):
-            await sandbox.fetch("ftp://example.com")
-
-    @pytest.mark.asyncio
-    async def test_rejects_no_host(self, sandbox):
-        with pytest.raises(SandboxFetchError, match="no host"):
-            await sandbox.fetch("http://")
-
-    @pytest.mark.asyncio
-    async def test_strips_whitespace(self, sandbox):
-        def handler(request):
-            return httpx.Response(200, text="ok")
-
-        with patch(_HTTPX_CLIENT_PATH, _mock_client(handler)):
-            result = await sandbox.fetch("  https://example.com  ")
-        assert result.body == b"ok"

From 15946824f979c9c7cd61277a744955e39e1d7204 Mon Sep 17 00:00:00 2001
From: liramon2 <liramon@amazon.com>
Date: Thu, 1 Oct 2026 18:10:06 -0400
Subject: [PATCH 4/4] fix: harden posix request

---
 strands-py/src/strands/sandbox/base.py        |   7 +-
 strands-py/src/strands/sandbox/posix_shell.py | 122 ++++-
 .../tests/strands/sandbox/test_posix_shell.py | 449 +++++++++++++-----
 3 files changed, 423 insertions(+), 155 deletions(-)

diff --git a/strands-py/src/strands/sandbox/base.py b/strands-py/src/strands/sandbox/base.py
index aa29aafb14..c9b87f3427 100644
--- a/strands-py/src/strands/sandbox/base.py
+++ b/strands-py/src/strands/sandbox/base.py
@@ -224,14 +224,17 @@ async def request(
             body: Request body. ``str`` is sent as UTF-8; ``bytes`` is sent as-is.
                 ``None`` sends no body.
             timeout: Maximum execution time in seconds. ``None`` means no timeout.
-            max_bytes: Cap the response body to this many bytes. ``None`` means no limit.
+            max_bytes: Bound the response body to this many bytes. ``None`` means
+                no limit. A response exceeding the cap raises
+                :class:`SandboxHttpError` rather than returning a truncated body.
             **kwargs: Additional keyword arguments for forward compatibility.
 
         Returns:
             An :class:`HttpResult`. HTTP error statuses (4xx/5xx) are returned in ``status`` rather than raised.
 
         Raises:
-            SandboxHttpError: On network or transport errors.
+            SandboxHttpError: On network/transport errors, or when the response
+                body exceeds ``max_bytes``.
         """
         raise NotImplementedError(f"{type(self).__name__} does not implement request()")
 
diff --git a/strands-py/src/strands/sandbox/posix_shell.py b/strands-py/src/strands/sandbox/posix_shell.py
index 96abbcc53a..5d029dd541 100644
--- a/strands-py/src/strands/sandbox/posix_shell.py
+++ b/strands-py/src/strands/sandbox/posix_shell.py
@@ -31,11 +31,26 @@
 # the URL reaches a shell command.  Security-critical for the curl transport.
 _URL_CHARS = re.compile(r"[A-Za-z0-9\-._~:/?#\[\]@!$&'()*+,;=%]+")
 
+# Anything outside this set (or a CR/LF/NUL in a value) could inject additional headers
+# or smuggle a second equest, so header names are validated against this pattern.
+_HEADER_NAME = re.compile(r"[!#$%&'*+\-.^_|~0-9A-Za-z]+")
+
 # HTTP methods accepted by request(). An explicit allowlist (matching the
 # http_request tool's HttpMethod) keeps the method safe to place in the curl
 # command line and rejects anything unexpected outright.
 _ALLOWED_METHODS = frozenset({"GET", "POST", "PUT", "DELETE", "PATCH", "HEAD", "OPTIONS"})
 
+# curl's "Failure writing output" exit code. When a response is capped, the body
+# is piped through ``head -c``; once ``head`` has read the cap it closes the pipe
+# and curl aborts with this code — the signal that the body exceeded ``max_bytes``.
+_CURL_WRITE_ERROR = 23
+
+# Tags curl's final URL on stderr for the capped path, where stdout is reserved
+# for the (binary) response body. Error text never contains this literal, so the
+# resolved URL can be recovered unambiguously.
+_URL_MARKER = "STRANDS_URL:"
+_URL_MARKER_RE = re.compile(re.escape(_URL_MARKER) + r"(\S*)")
+
 
 def validate_env_keys(env: dict[str, str]) -> None:
     """Validate environment variable names against :data:`ENV_KEY_PATTERN`.
@@ -130,6 +145,28 @@ def _parse_status_and_headers(raw: str) -> tuple[int, str, dict[str, str]]:
     return status, status_text, headers
 
 
+def _resolve_capped_outcome(rc_text: str, stderr_text: str, fallback_exit: int, max_bytes: int | None, url: str) -> str:
+    """Interpret a capped request's curl exit code and tagged stderr.
+
+    The pipeline's exit status is ``head``'s, so curl's own code is read from a file
+    (``fallback_exit`` if unreadable); the resolved URL and error text share the tagged stderr.
+
+    Returns the resolved URL on success.
+
+    Raises:
+        SandboxHttpError: When the body exceeded ``max_bytes`` (curl aborted
+            writing to the closed pipe) or the transfer otherwise failed.
+    """
+    curl_rc = int(rc_text) if rc_text.isdigit() else fallback_exit
+    if curl_rc == _CURL_WRITE_ERROR:
+        raise SandboxHttpError(f"response body exceeded max_bytes ({max_bytes})")
+    if curl_rc != 0:
+        detail = _URL_MARKER_RE.sub("", stderr_text).strip()
+        raise SandboxHttpError(detail or f"curl exited with code {curl_rc}")
+    match = _URL_MARKER_RE.search(stderr_text)
+    return (match.group(1) if match else "") or url
+
+
 class PosixShellSandbox(Sandbox, ABC):
     """Abstract sandbox that provides shell-based defaults for file and code operations.
 
@@ -312,8 +349,10 @@ async def request(
         Works with any HTTP method. HTTP error statuses (4xx/5xx) are returned
         in :attr:`HttpResult.status` rather than raised; only transport-level
         failures (DNS, connection, timeout, size cap) raise
-        :class:`SandboxHttpError`. ``max_bytes`` is rejected early when
-        ``Content-Length`` is known and truncated for chunked responses.
+        :class:`SandboxHttpError`. ``max_bytes`` bounds the download: the
+        transfer is stopped at the cap and a response exceeding it raises
+        :class:`SandboxHttpError` rather than returning a silently truncated body
+        (matching the ``web_fetch`` tool).
         """
         method = method.strip().upper()
         if method not in _ALLOWED_METHODS:
@@ -334,10 +373,13 @@ async def request(
         body_file = f"/tmp/strands-http-{tag}"
         header_file = f"/tmp/strands-http-{tag}.headers"
         data_file = f"/tmp/strands-http-{tag}.data"
+        rc_file = f"/tmp/strands-http-{tag}.rc"
+        err_file = f"/tmp/strands-http-{tag}.err"
         body_quoted = shlex.quote(body_file)
         header_quoted = shlex.quote(header_file)
         data_quoted = shlex.quote(data_file)
-        part_quoted = shlex.quote(body_file + ".part")
+        rc_quoted = shlex.quote(rc_file)
+        err_quoted = shlex.quote(err_file)
 
         # A HEAD response carries no body; use curl's --head so it doesn't block
         # waiting for a body the server will never send.
@@ -351,29 +393,51 @@ async def request(
         opts = ""
         if timeout is not None:
             opts += f" --max-time {timeout}"
-        if max_bytes is not None:
-            opts += f" --max-filesize {max_bytes}"
         if headers:
             for n, v in headers.items():
-                opts += f" -H {shlex.quote(f'{n}: {v}')}"
+                if not _HEADER_NAME.fullmatch(n) or re.search(r"[\r\n\0]", v):
+                    raise SandboxHttpError(
+                        f"invalid header {n!r}: name must be an RFC 7230 token and value may not contain CR/LF/NUL"
+                    )
+                # `-H "name;"` sends an empty-valued header rather than omitting it.
+                opts += f" -H {shlex.quote(f'{n};' if v == '' else f'{n}: {v}')}"
         if body is not None:
             opts += f" --data-binary @{data_quoted}"
 
-        method_opt = "--head" if is_head else f"-X {method}"
-        out_target = "/dev/null" if is_head else body_quoted
-
-        # Build a hardened curl command: protocol restricted to http(s), body →
-        # file, response headers + status → file via -D, final URL on stdout.
-        cmd = (
-            f"curl -sSL -g {method_opt}"
-            f" --proto '=http,https' --proto-redir '=http,https'"
-            f"{opts}"
-            f" -D {header_quoted} -o {out_target}"
-            f" -w '%{{url_effective}}'"
-            f" -- {shlex.quote(url)}"
+        # Only force the method where curl can't infer it. ``-X POST`` with
+        # ``-L`` replays a bodiless POST through 302/303 redirects (curl keeps the
+        # method but drops the body); omitting ``-X`` lets curl switch to GET on
+        # those redirects like browsers/httpx do. GET and a body-carrying POST are
+        # curl's defaults, so they need no ``-X``.
+        if is_head:
+            method_opt = "--head"
+        elif method == "GET" or (method == "POST" and body is not None):
+            method_opt = ""
+        else:
+            method_opt = f"-X {method}"
+
+        # Cap the response body (not meaningful for HEAD, which carries none).
+        cap = None if is_head else max_bytes
+
+        base_curl = (
+            f"curl -sSL -g {method_opt} --proto '=http,https' --proto-redir '=http,https'{opts} -D {header_quoted}"
         )
-        if max_bytes is not None and not is_head:
-            cmd += f" && head -c {max_bytes} {body_quoted} > {part_quoted} && mv -f {part_quoted} {body_quoted}"
+        if cap is not None:
+            # Bound the download: pipe the body through ``head -c`` so curl is stopped at the cap (head closes
+            # the pipe, curl exits 23) rather than buffering the whole response; the pipeline's status is head's,
+            # so curl's code goes to a file and the resolved URL is tagged onto stderr to keep stdout pure body.
+            cmd = (
+                f"{{ {base_curl} -o - -w '%{{stderr}}{_URL_MARKER}%{{url_effective}}'"
+                f" -- {shlex.quote(url)}; echo $? > {rc_quoted}; }}"
+                f" 2> {err_quoted} | head -c {cap + 1} > {body_quoted}"
+            )
+        else:
+            out_target = "/dev/null" if is_head else body_quoted
+            cmd = f"{base_curl} -o {out_target} -w '%{{url_effective}}' -- {shlex.quote(url)}"
+            if not is_head:
+                # curl omits the -o file for a bodyless response (e.g. 304 Not Modified);
+                # pre-create it so the read-back always finds a (possibly empty) file.
+                cmd = f": > {body_quoted} && {cmd}"
 
         # Give curl a few extra seconds beyond its own --max-time so the
         # sandbox kills it only if curl itself hangs.
@@ -384,17 +448,27 @@ async def request(
                 result = await self.execute(cmd, timeout=exec_timeout, **kwargs)
             except SandboxTimeoutError as exc:
                 raise SandboxHttpError(f"request timed out after {timeout}s") from exc
-            if result.exit_code != 0:
-                raise SandboxHttpError(result.stderr.strip() or f"curl exited with code {result.exit_code}")
 
-            resolved_url = result.stdout.strip() or url
+            if cap is not None:
+                # The pipeline's status is ``head``'s, so curl's own code is read from its file;
+                # the resolved URL and any error text come from the tagged stderr.
+                rc_text = (await self.read_file(rc_file)).decode("utf-8", errors="replace").strip()
+                err_text = (await self.read_file(err_file)).decode("utf-8", errors="replace")
+                resolved_url = _resolve_capped_outcome(rc_text, err_text, result.exit_code, cap, url)
+            else:
+                if result.exit_code != 0:
+                    raise SandboxHttpError(result.stderr.strip() or f"curl exited with code {result.exit_code}")
+                resolved_url = result.stdout.strip() or url
+
             raw_headers = (await self.read_file(header_file)).decode("utf-8", errors="replace")
             status, status_text, response_headers = _parse_status_and_headers(raw_headers)
             response_body = b"" if is_head else await self.read_file(body_file)
+            if cap is not None and len(response_body) > cap:
+                raise SandboxHttpError(f"response body exceeded max_bytes ({cap})")
         finally:
             with contextlib.suppress(Exception):
                 await self.execute(
-                    f"rm -f {body_quoted} {part_quoted} {header_quoted} {data_quoted}",
+                    f"rm -f {body_quoted} {header_quoted} {data_quoted} {rc_quoted} {err_quoted}",
                     timeout=10,
                 )
 
diff --git a/strands-py/tests/strands/sandbox/test_posix_shell.py b/strands-py/tests/strands/sandbox/test_posix_shell.py
index bb1b838bef..63ec205fc6 100644
--- a/strands-py/tests/strands/sandbox/test_posix_shell.py
+++ b/strands-py/tests/strands/sandbox/test_posix_shell.py
@@ -12,9 +12,15 @@
 """
 
 import asyncio
+import contextlib
+import http.server
 import os
+import re
 import shlex
+import shutil
 import sys
+import threading
+import urllib.parse
 from collections.abc import AsyncGenerator
 from typing import Any
 
@@ -23,12 +29,12 @@
 from strands.sandbox import (
     ExecutionResult,
     FileInfo,
-    HttpResult,
     PosixShellSandbox,
     StreamChunk,
 )
 from strands.sandbox.errors import SandboxHttpError, SandboxPathNotFoundError
 from strands.sandbox.posix_shell import (
+    _URL_MARKER,
     _parse_status_and_headers,
     build_shell_env_prefix,
     validate_env_keys,
@@ -614,10 +620,23 @@ def test_file_info_defaults():
 class _MockHttpSandbox:
     """Mock implementing execute + read_file + write_file for request() tests."""
 
-    def __init__(self, *, stdout="https://example.com", stderr="", exit_code=0, headers=None, body=None):
+    def __init__(
+        self,
+        *,
+        stdout="https://example.com",
+        stderr="",
+        exit_code=0,
+        headers=None,
+        body=None,
+        rc="0",
+        err=f"{_URL_MARKER}https://example.com",
+    ):
         self._result = ExecutionResult(exit_code=exit_code, stdout=stdout, stderr=stderr)
-        self._headers = headers or b"HTTP/1.1 200 OK\r\nContent-Type: text/html; charset=utf-8\r\n\r\n"
-        self._body = body or b"<html><body>Hello</body></html>"
+        default_headers = b"HTTP/1.1 200 OK\r\nContent-Type: text/html; charset=utf-8\r\n\r\n"
+        self._headers = default_headers if headers is None else headers
+        self._body = b"<html><body>Hello</body></html>" if body is None else body
+        self._rc = rc
+        self._err = err
         self.commands: list[str] = []
         self.written: dict[str, bytes] = {}
 
@@ -630,6 +649,10 @@ async def execute(self, command, **kwargs):
     async def read_file(self, path, **kwargs):
         if path.endswith(".headers"):
             return self._headers
+        if path.endswith(".rc"):
+            return self._rc.encode()
+        if path.endswith(".err"):
+            return self._err.encode()
         if "/tmp/strands-http-" in path:
             return self._body
         raise FileNotFoundError(path)
@@ -639,177 +662,146 @@ async def write_file(self, path, content, **kwargs):
 
 
 class TestRequest:
-    @pytest.mark.asyncio
-    async def test_successful_request(self):
-        sb = _MockHttpSandbox()
-        result = await PosixShellSandbox.request(sb, "GET", "https://example.com")
-
-        assert result == HttpResult(
-            status=200,
-            status_text="OK",
-            resolved_url="https://example.com",
-            headers={"content-type": "text/html; charset=utf-8"},
-            body=b"<html><body>Hello</body></html>",
-        )
+    """Command-shape and validation unit tests against a canned mock.
 
-        # Verify hardened curl flags. No --fail: error statuses are reported.
-        cmd = sb.commands[0]
-        assert "curl -sSL -g -X GET" in cmd
-        assert "--fail" not in cmd
-        assert "--proto '=http,https'" in cmd
-        assert "-D " in cmd and "-o " in cmd
+    The behavioral contract (status/headers/body/resolved_url, raise-vs-return)
+    is proven against a real curl + HTTP server in :class:`TestRequestLive`;
+    these only pin the curl invocation and the input validation that happens
+    before any process runs.
+    """
 
     @pytest.mark.asyncio
-    async def test_timeout_and_max_bytes_and_headers(self):
+    async def test_non_capped_curl_command_shape(self):
         sb = _MockHttpSandbox()
-        await PosixShellSandbox.request(
-            sb,
-            "GET",
-            "https://example.com",
-            max_bytes=1024,
-            timeout=15,
-            headers={"X-Key": "val"},
-        )
+        await PosixShellSandbox.request(sb, "GET", "https://example.com", headers={"X-Key": "val"})
         cmd = sb.commands[0]
-        assert "--max-time 15" in cmd
-        assert "head -c 1024" in cmd
+        # Hardened flags; no --fail (error statuses are reported, not raised).
+        assert "curl -sSL -g " in cmd
+        assert "--fail" not in cmd
+        assert "--proto '=http,https' --proto-redir '=http,https'" in cmd
+        assert "-D " in cmd and "-o " in cmd
         assert "-H 'X-Key: val'" in cmd
-
-    @pytest.mark.asyncio
-    async def test_no_timeout_no_max_bytes(self):
-        sb = _MockHttpSandbox()
-        await PosixShellSandbox.request(sb, "GET", "https://example.com")
-        cmd = sb.commands[0]
-        assert "--max-time" not in cmd
+        # No cap → no download pipe.
         assert "head -c" not in cmd
+        # curl omits the -o file for a bodyless 304, so it is pre-created.
+        body_file = re.search(r"/tmp/strands-http-[0-9a-f]+\b", cmd).group(0)
+        assert cmd.startswith(f": > {body_file} && ")
+        assert f"-o {body_file}" in cmd
+        assert sb.commands[-1].startswith("rm -f")  # temp files cleaned up
 
     @pytest.mark.asyncio
-    async def test_post_with_body_writes_data_file(self):
+    async def test_capped_curl_command_shape(self):
         sb = _MockHttpSandbox()
-        await PosixShellSandbox.request(sb, "POST", "https://example.com", body='{"a": 1}')
-        # Body is written to a temp file (never placed on the command line)...
-        assert any(p.endswith(".data") for p in sb.written)
-        data_path = next(p for p in sb.written if p.endswith(".data"))
-        assert sb.written[data_path] == b'{"a": 1}'
-        # ...and fed to curl via --data-binary @file with the method set.
+        await PosixShellSandbox.request(sb, "GET", "https://example.com", max_bytes=1024, timeout=15)
         cmd = sb.commands[0]
-        assert "-X POST" in cmd
-        assert f"--data-binary @{shlex.quote(data_path)}" in cmd
-        assert '{"a": 1}' not in cmd
+        assert "--max-time 15" in cmd
+        assert "--max-filesize" not in cmd  # unused: would miss chunked responses
+        # Body piped through `head -c` (bounds the transfer); curl writes to
+        # stdout, its exit code is captured, and the URL is tagged onto stderr.
+        assert "-o -" in cmd
+        assert "| head -c 1025 >" in cmd  # cap + 1 byte, so an overshoot is caught by length
+        assert "echo $? >" in cmd
+        assert f"%{{stderr}}{_URL_MARKER}%{{url_effective}}" in cmd
 
     @pytest.mark.asyncio
-    async def test_bytes_body_sent_as_is(self):
+    async def test_body_written_to_file_not_command_line(self):
         sb = _MockHttpSandbox()
         await PosixShellSandbox.request(sb, "PUT", "https://example.com", body=b"\x00\x01\x02")
         data_path = next(p for p in sb.written if p.endswith(".data"))
-        assert sb.written[data_path] == b"\x00\x01\x02"
-
-    @pytest.mark.asyncio
-    async def test_no_body_writes_nothing_and_omits_data_flag(self):
-        sb = _MockHttpSandbox()
-        await PosixShellSandbox.request(sb, "GET", "https://example.com")
-        assert sb.written == {}
-        assert "--data-binary" not in sb.commands[0]
+        assert sb.written[data_path] == b"\x00\x01\x02"  # bytes sent as-is
+        assert f"--data-binary @{shlex.quote(data_path)}" in sb.commands[0]
+        # No body → no data file, no --data-binary.
+        sb2 = _MockHttpSandbox()
+        await PosixShellSandbox.request(sb2, "GET", "https://example.com")
+        assert sb2.written == {}
+        assert "--data-binary" not in sb2.commands[0]
 
     @pytest.mark.asyncio
-    async def test_head_uses_head_flag_and_empty_body(self):
-        sb = _MockHttpSandbox()
-        result = await PosixShellSandbox.request(sb, "HEAD", "https://example.com")
-        cmd = sb.commands[0]
-        # --head avoids curl hanging on a body the server never sends; the body
-        # goes to /dev/null and the response body is empty.
-        assert "--head" in cmd
-        assert "-X HEAD" not in cmd
-        assert "-o /dev/null" in cmd
-        assert result.status == 200
-        assert result.status_text == "OK"
-        assert result.body == b""
-        assert result.headers["content-type"] == "text/html; charset=utf-8"
+    async def test_method_opt_lets_curl_infer_where_it_can(self):
+        # -X POST + -L replays a bodiless POST through 302/303 redirects (curl
+        # keeps the method but drops the body). Omit -X where curl infers it:
+        # GET (default) and a body-carrying POST. HEAD uses --head, never -X.
+        async def token(method, **kw):
+            sb = _MockHttpSandbox()
+            await PosixShellSandbox.request(sb, method, "https://example.com", **kw)
+            return sb.commands[0]
+
+        assert "-X" not in await token("GET")
+        assert "-X" not in await token("POST", body="payload")
+        assert "-X POST" in await token("POST")  # bodiless POST must stay a POST
+        assert "-X PUT" in await token("PUT", body="payload")
+        assert "-X DELETE" in await token("DELETE")
+        head_cmd = await token("HEAD")
+        assert "--head" in head_cmd and "-X" not in head_cmd and "-o /dev/null" in head_cmd
+        assert "head -c" not in head_cmd  # HEAD is never capped
+        assert not head_cmd.startswith(": >")  # /dev/null, no body file to pre-create
 
     @pytest.mark.asyncio
     async def test_method_is_normalized_and_validated(self):
         sb = _MockHttpSandbox()
         await PosixShellSandbox.request(sb, "  patch  ", "https://example.com")
         assert "-X PATCH" in sb.commands[0]
-
         for method in ["GET;rm -rf /", "GET POST", "", "G-E-T", "POST!"]:
             with pytest.raises(SandboxHttpError, match="method"):
                 await PosixShellSandbox.request(_MockHttpSandbox(), method, "https://example.com")
 
     @pytest.mark.asyncio
-    async def test_transport_failure_raises(self):
-        # Without --fail, HTTP error statuses no longer fail curl; only genuine
-        # transport errors (DNS, connection, size cap) make it exit non-zero.
-        sb = _MockHttpSandbox(exit_code=7, stderr="curl: (7) Failed to connect")
-        with pytest.raises(SandboxHttpError, match="connect"):
-            await PosixShellSandbox.request(sb, "GET", "https://example.com")
-
-    @pytest.mark.asyncio
-    async def test_error_status_is_returned_not_raised(self):
-        sb = _MockHttpSandbox(
-            headers=b"HTTP/1.1 404 Not Found\r\nContent-Type: text/plain\r\n\r\n",
-            body=b"nope",
-        )
-        result = await PosixShellSandbox.request(sb, "GET", "https://example.com/missing")
-        assert result.status == 404
-        assert result.status_text == "Not Found"
-        assert result.body == b"nope"
+    async def test_rejects_header_injection(self):
+        # CRLF/NUL in a value, or a non-token name, must not reach the wire.
+        for headers in [
+            {"X-Evil": "val\r\nX-Injected: 1"},
+            {"X-Evil": "val\x00"},
+            {"Bad Name": "val"},
+            {"X:Colon": "val"},
+            {"": "val"},
+        ]:
+            with pytest.raises(SandboxHttpError, match="invalid header"):
+                await PosixShellSandbox.request(_MockHttpSandbox(), "GET", "https://example.com", headers=headers)
+        # An empty value is valid and uses curl's `name;` form (not `name:`,
+        # which curl would drop).
+        sb = _MockHttpSandbox()
+        await PosixShellSandbox.request(sb, "GET", "https://example.com", headers={"X-Empty": ""})
+        assert "-H 'X-Empty;'" in sb.commands[0]
 
     @pytest.mark.asyncio
-    async def test_rejects_unsafe_url_chars(self):
-        """RFC 3986 character allowlist rejects shell-unsafe characters."""
-        for url in [
-            "https://example.com/path with spaces",
+    async def test_rejects_bad_urls(self):
+        bad = [
+            "https://example.com/path with spaces",  # RFC 3986 allowlist
             "https://example.com/café",
             "https://example.com/`id`",
-            'https://example.com/"test"',
             "https://example.com/path\nnewline",
-            "https://example.com/{path}",
             "",
-        ]:
+            "ftp://example.com",  # non-http(s) scheme
+            "file:///etc/passwd",
+            "http://",  # no host
+        ]
+        for url in bad:
             with pytest.raises(SandboxHttpError):
                 await PosixShellSandbox.request(_MockHttpSandbox(), "GET", url)
 
     @pytest.mark.asyncio
-    async def test_rejects_non_http_scheme(self):
-        for url in ["ftp://example.com", "file:///etc/passwd", "javascript:alert(1)"]:
-            with pytest.raises(SandboxHttpError, match="http\\(s\\)"):
-                await PosixShellSandbox.request(_MockHttpSandbox(), "GET", url)
-
-    @pytest.mark.asyncio
-    async def test_rejects_no_host(self):
-        with pytest.raises(SandboxHttpError, match="no host"):
-            await PosixShellSandbox.request(_MockHttpSandbox(), "GET", "http://")
-
-    @pytest.mark.asyncio
-    async def test_strips_url_whitespace(self):
-        sb = _MockHttpSandbox()
-        result = await PosixShellSandbox.request(sb, "GET", "  https://example.com  ")
-        assert result.resolved_url == "https://example.com"
+    async def test_capped_outcome_from_curl_exit_and_marker(self):
+        # Overflow: head closes the pipe at the cap → curl exit 23.
+        sb = _MockHttpSandbox(rc="23", err=f"{_URL_MARKER}https://example.com")
+        with pytest.raises(SandboxHttpError, match="exceeded max_bytes"):
+            await PosixShellSandbox.request(sb, "GET", "https://example.com", max_bytes=100)
+        # Transport error: surfaced with the URL marker stripped from the message.
+        sb = _MockHttpSandbox(rc="7", err=f"curl: (7) Failed to connect{_URL_MARKER}https://example.com")
+        with pytest.raises(SandboxHttpError, match="Failed to connect") as exc:
+            await PosixShellSandbox.request(sb, "GET", "https://example.com", max_bytes=1024)
+        assert _URL_MARKER not in str(exc.value)
+        # Success: resolved URL recovered from the stderr marker.
+        sb = _MockHttpSandbox(rc="0", err=f"{_URL_MARKER}https://example.com/final")
+        result = await PosixShellSandbox.request(sb, "GET", "https://example.com", max_bytes=1024)
+        assert result.resolved_url == "https://example.com/final"
 
     @pytest.mark.asyncio
-    async def test_cleanup_runs(self):
-        # Success
-        sb = _MockHttpSandbox()
-        await PosixShellSandbox.request(sb, "GET", "https://example.com")
-        assert sb.commands[-1].startswith("rm -f")
-
-        # Failure
+    async def test_cleanup_runs_on_failure(self):
         sb = _MockHttpSandbox(exit_code=1, stderr="fail")
         with pytest.raises(SandboxHttpError):
             await PosixShellSandbox.request(sb, "GET", "https://example.com")
         assert sb.commands[-1].startswith("rm -f")
 
-    @pytest.mark.asyncio
-    async def test_resolved_url(self):
-        sb = _MockHttpSandbox(stdout="https://example.com/redirected")
-        result = await PosixShellSandbox.request(sb, "GET", "https://example.com")
-        assert result.resolved_url == "https://example.com/redirected"
-
-        sb = _MockHttpSandbox(stdout="")
-        result = await PosixShellSandbox.request(sb, "GET", "https://example.com")
-        assert result.resolved_url == "https://example.com"
-
 
 class TestParseStatusAndHeaders:
     def test_basic_headers(self):
@@ -878,3 +870,202 @@ def test_lf_only(self):
         status, _, headers = _parse_status_and_headers(raw)
         assert status == 200
         assert headers == {"content-type": "text/html"}
+
+
+# ---- request() against a real HTTP server (end-to-end through sh + curl) ----
+#
+# The command-shape assertions above pin the curl invocation, but can't prove the
+# contract (status/headers/body/resolved_url, raise-vs-return). These drive
+# request() through _ShellTestSandbox (real `sh -c`) against a stdlib HTTP server,
+# so an actual curl parses an actual response. Skipped if curl is missing.
+
+_CURL_AVAILABLE = shutil.which("curl") is not None
+
+
+class _RequestHandler(http.server.BaseHTTPRequestHandler):
+    """Routes used by the live request() tests. See the ``do_*`` dispatch below."""
+
+    protocol_version = "HTTP/1.1"
+
+    def log_message(self, *args):  # noqa: D102 - silence per-request stderr logging
+        pass
+
+    def _handle(self):
+        parsed = urllib.parse.urlparse(self.path)
+        path = parsed.path
+        length = int(self.headers.get("content-length", 0))
+        body = self.rfile.read(length) if length else b""
+
+        if path == "/hello":
+            payload = b"<html><body>Hello</body></html>"
+            self.send_response(200)
+            self.send_header("content-type", "text/html; charset=utf-8")
+            self.send_header("content-length", str(len(payload)))
+            self.end_headers()
+            self.wfile.write(payload)
+        elif path == "/echo":
+            # Report what the server actually received, so the test can assert
+            # the method and body survived (e.g. across a redirect).
+            payload = f"{self.command}|{body.decode('utf-8', 'replace')}".encode()
+            self.send_response(200)
+            self.send_header("content-type", "text/plain")
+            self.send_header("content-length", str(len(payload)))
+            self.end_headers()
+            self.wfile.write(payload)
+        elif path == "/redirect-303":
+            # A 303 forces curl to decide method handling on redirect.
+            self.send_response(303)
+            self.send_header("location", "/echo")
+            self.send_header("content-length", "0")
+            self.end_headers()
+        elif path == "/set-cookie":
+            self.send_response(200)
+            self.send_header("set-cookie", "a=1")
+            self.send_header("set-cookie", "b=2")
+            self.send_header("content-length", "2")
+            self.end_headers()
+            self.wfile.write(b"ok")
+        elif path == "/not-found":
+            self.send_response(404)
+            self.send_header("content-type", "text/plain")
+            self.send_header("content-length", "3")
+            self.end_headers()
+            self.wfile.write(b"no!")
+        elif path == "/not-modified":
+            # 304 carries no body and no content-length; curl omits the -o file.
+            self.send_response(304)
+            self.end_headers()
+        elif path == "/sized":
+            # Fixed-size body with Content-Length: curl delivers it in one read
+            # and exits 0, so an overshoot can't rely on curl's exit-23 fast path.
+            n = int(urllib.parse.parse_qs(parsed.query).get("n", ["0"])[0])
+            payload = b"A" * n
+            self.send_response(200)
+            self.send_header("content-type", "text/plain")
+            self.send_header("content-length", str(len(payload)))
+            self.end_headers()
+            self.wfile.write(payload)
+        elif path == "/big":
+            # Chunked (no content-length) so --max-filesize can't pre-reject it.
+            # curl closes the socket once head -c hits the cap; stop writing on
+            # the resulting broken pipe instead of blocking the server thread.
+            self.send_response(200)
+            self.send_header("content-type", "application/octet-stream")
+            self.end_headers()
+            with contextlib.suppress(BrokenPipeError, ConnectionResetError):
+                for _ in range(64):
+                    self.wfile.write(b"x" * 1024)
+                    self.wfile.flush()
+        else:
+            self.send_response(404)
+            self.send_header("content-length", "0")
+            self.end_headers()
+
+    do_GET = _handle
+    do_POST = _handle
+    do_PUT = _handle
+    do_DELETE = _handle
+    do_HEAD = _handle
+
+
+@pytest.fixture
+def http_server():
+    server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _RequestHandler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        host, port = server.server_address
+        yield f"http://{host}:{port}"
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+
+
+@pytest.mark.skipif(not _CURL_AVAILABLE, reason="curl not available")
+class TestRequestLive:
+    @pytest.mark.asyncio
+    async def test_get_returns_status_headers_body(self, sandbox, http_server):
+        result = await PosixShellSandbox.request(sandbox, "GET", f"{http_server}/hello")
+        assert result.status == 200
+        assert result.status_text == "OK"
+        assert result.body == b"<html><body>Hello</body></html>"
+        assert result.headers["content-type"] == "text/html; charset=utf-8"
+        assert result.resolved_url == f"{http_server}/hello"
+
+    @pytest.mark.asyncio
+    async def test_post_body_reaches_server(self, sandbox, http_server):
+        result = await PosixShellSandbox.request(sandbox, "POST", f"{http_server}/echo", body="payload")
+        assert result.status == 200
+        assert result.body == b"POST|payload"
+
+    @pytest.mark.asyncio
+    async def test_post_body_switches_to_get_on_303_redirect(self, sandbox, http_server):
+        # Regression: `-X POST -L` replayed a bodiless POST; omitting -X lets curl
+        # switch to GET on a 303 the way browsers/httpx do.
+        result = await PosixShellSandbox.request(sandbox, "POST", f"{http_server}/redirect-303", body="payload")
+        assert result.status == 200
+        assert result.body == b"GET|"
+        assert result.resolved_url == f"{http_server}/echo"
+
+    @pytest.mark.asyncio
+    async def test_error_status_is_returned_not_raised(self, sandbox, http_server):
+        result = await PosixShellSandbox.request(sandbox, "GET", f"{http_server}/not-found")
+        assert result.status == 404
+        assert result.status_text == "Not Found"
+        assert result.body == b"no!"
+
+    @pytest.mark.asyncio
+    async def test_repeated_response_headers_are_preserved(self, sandbox, http_server):
+        result = await PosixShellSandbox.request(sandbox, "GET", f"{http_server}/set-cookie")
+        assert result.headers["set-cookie"] == "a=1\nb=2"
+
+    @pytest.mark.asyncio
+    async def test_304_returns_empty_body_not_filenotfound(self, sandbox, http_server):
+        # Regression: curl omits the -o file for a 304, which used to surface as
+        # a bare FileNotFoundError on read-back.
+        result = await PosixShellSandbox.request(sandbox, "GET", f"{http_server}/not-modified")
+        assert result.status == 304
+        assert result.body == b""
+
+    @pytest.mark.asyncio
+    async def test_max_bytes_bounds_chunked_download(self, sandbox, http_server):
+        # A chunked response (no content-length) exceeding the cap must raise,
+        # not silently truncate or buffer the whole 64 KiB; one within the cap
+        # succeeds.
+        with pytest.raises(SandboxHttpError, match="exceeded max_bytes"):
+            await PosixShellSandbox.request(sandbox, "GET", f"{http_server}/big", max_bytes=100)
+        result = await PosixShellSandbox.request(sandbox, "GET", f"{http_server}/hello", max_bytes=1_000_000)
+        assert result.body == b"<html><body>Hello</body></html>"
+
+    @pytest.mark.asyncio
+    async def test_max_bytes_catches_small_overshoot_and_allows_exact(self, sandbox, http_server):
+        # A body one byte over the cap (with Content-Length) arrives in curl's
+        # first read, so curl exits 0 and the exit-23 fast path never fires — the
+        # post-read length check must still raise. A body exactly at the cap is
+        # not an overshoot and comes back intact.
+        with pytest.raises(SandboxHttpError, match="exceeded max_bytes"):
+            await PosixShellSandbox.request(sandbox, "GET", f"{http_server}/sized?n=101", max_bytes=100)
+        result = await PosixShellSandbox.request(sandbox, "GET", f"{http_server}/sized?n=100", max_bytes=100)
+        assert result.status == 200
+        assert result.body == b"A" * 100
+
+    @pytest.mark.asyncio
+    async def test_crlf_header_value_is_rejected_before_the_wire(self, sandbox, http_server):
+        with pytest.raises(SandboxHttpError, match="invalid header"):
+            await PosixShellSandbox.request(
+                sandbox, "GET", f"{http_server}/hello", headers={"X-Evil": "v\r\nX-Injected: 1"}
+            )
+
+    @pytest.mark.asyncio
+    async def test_head_returns_headers_without_body(self, sandbox, http_server):
+        result = await PosixShellSandbox.request(sandbox, "HEAD", f"{http_server}/hello")
+        assert result.status == 200
+        assert result.body == b""
+        assert result.headers["content-type"] == "text/html; charset=utf-8"
+
+    @pytest.mark.asyncio
+    async def test_connection_refused_raises(self, sandbox):
+        # Nothing is listening on this port → genuine transport failure.
+        with pytest.raises(SandboxHttpError):
+            await PosixShellSandbox.request(sandbox, "GET", "http://127.0.0.1:1/nope", timeout=5)