[BUG] Sandbox + `httpProxyPort` enabled, but Claude cannot access HTTP endpoint (`Connection refused`)
Resolved 💬 3 comments Opened Mar 11, 2026 by momenta-jack Closed Apr 8, 2026
Preflight Checklist
- [x] I have searched existing issues and this hasn't been reported yet
- [x] This is a single bug report (please file separate reports for different bugs)
- [x] I am using the latest version of Claude Code
What's Wrong?
Description
Even with sandbox and HTTP proxy enabled, Claude still cannot access HTTP endpoints.
In my Ubuntu shell, the same command works normally:
curl http://127.0.0.1:5555 -x http://127.0.0.1:58088But when Claude executes curl, it always fails with Connection refused on port 5555.
Environment
- OS: Ubuntu 20.04
- Claude version: 2.1.72
Expected behavior
Claude should be able to access the same HTTP endpoint through the configured proxy, just like native shell execution.
Actual behavior
Claude fails with Connection refused on port 5555.
What Should Happen?
Connect http successfully via proxy
Error Messages/Logs
❯ run curl http://127.0.0.1:5555
● Bash(curl http://127.0.0.1:5555)
⎿ % Total % Received % Xferd Average Speed Time Time Time Current
Dload Upload Total Spent Left Speed
0 0 0 0 0 0 0 0 --:--:-- --:--:-- --:--:-- 0 0 0 0 0 0 0 0 0 --:--:-- --:
… +2 lines (ctrl+o to expand)
● Connection refused — nothing is listening on 127.0.0.1:5555. Steps to Reproduce
Steps to reproduce
- start proxy.py(58088)
- Enable sandbox with
httpProxyPortin .calude/settings.json
"sandbox": {
"enabled": true,
"autoAllowBashIfSandboxed": true,
"excludedCommands": [
"curl",
"wget",
"python"
],
"network": {
"httpProxyPort": 58088,
"allowedDomains": [
"*"
]
}
}- Verify in native Linux shell(works):
``bash``
curl http://127.0.0.1:5555 -x http://127.0.0.1:58088
- Ask Claude to run the same
curlcommand.(failed)
❯ run curl http://127.0.0.1:5555
● Bash(curl http://127.0.0.1:5555)
⎿ % Total % Received % Xferd Average Speed Time Time Time Current
Dload Upload Total Spent Left Speed
0 0 0 0 0 0 0 0 --:--:-- --:--:-- --:--:-- 0 0 0 0 0 0 0 0 0 --:--:-- --:
… +2 lines (ctrl+o to expand)
● Connection refused — nothing is listening on 127.0.0.1:5555. Observe failure: Connection refused on port 5555.
files
proxy.py
# python3 proxy.py
"""HTTP proxy server for sandboxed Claude Code execution."""
import asyncio
import errno
import logging
import re
from typing import Optional
logger = logging.getLogger(__name__)
class ProxyServer:
"""Lightweight asyncio-based HTTP/HTTPS proxy server."""
def __init__(
self,
host: str = "0.0.0.0",
port: int = 58088,
allowed_domains: list[str] | None = None,
timeout: int = 30,
buffer_size: int = 8192,
):
"""Initialize proxy server.
Args:
host: Host to bind to
port: Port to listen on
allowed_domains: List of allowed domains (supports wildcards like *.example.com)
Use ["*"] to allow all domains
timeout: Connection timeout in seconds
buffer_size: Buffer size for data transfer
"""
self.host = host
self.port = port
self.allowed_domains = allowed_domains or ["*"]
self.timeout = timeout
self.buffer_size = buffer_size
self.server: Optional[asyncio.Server] = None
self._active_connections: set[asyncio.Task] = set()
async def start(self) -> None:
"""Start the proxy server - multi-process safe."""
try:
self.server = await asyncio.start_server(
self._handle_client, self.host, self.port
)
logger.info(
f"Proxy server started on {self.host}:{self.port} "
f"(allowed_domains: {self.allowed_domains})"
)
except OSError as e:
if e.errno == errno.EADDRINUSE:
# Port already in use by another worker, skip silently
logger.info(
f"Proxy port {self.port} already in use (another worker), skipping"
)
self.server = None
else:
raise
except Exception as e:
logger.error(f"Failed to start proxy server: {e}")
raise
async def stop(self) -> None:
"""Stop the proxy server and wait for active connections - multi-process safe."""
if self.server is None:
# This worker never started the server, nothing to stop
return
self.server.close()
await self.server.wait_closed()
# Cancel and wait for active connections
if self._active_connections:
logger.info(
f"Waiting for {len(self._active_connections)} active connections to close"
)
for task in self._active_connections:
task.cancel()
await asyncio.gather(*self._active_connections, return_exceptions=True)
logger.info("Proxy server stopped")
async def _handle_client(
self, client_reader: asyncio.StreamReader, client_writer: asyncio.StreamWriter
) -> None:
"""Handle incoming client connection."""
task = asyncio.current_task()
if task:
self._active_connections.add(task)
try:
# Read first line to determine request type
request_line = await asyncio.wait_for(
client_reader.readline(), timeout=self.timeout
)
request_line = request_line.decode("utf-8", errors="ignore").strip()
if not request_line:
return
parts = request_line.split()
if len(parts) < 2:
await self._send_error(client_writer, 400, "Bad Request")
return
method, url = parts[0], parts[1]
# Extract domain from URL
domain = self._extract_domain(method, url)
if not domain:
await self._send_error(client_writer, 400, "Bad Request")
return
# Check domain whitelist
if not self._is_domain_allowed(domain):
logger.warning(f"Blocked request to {domain}")
await self._send_error(client_writer, 403, "Forbidden")
return
# Route based on method
if method == "CONNECT":
await self._handle_connect(
client_reader, client_writer, url, request_line
)
else:
await self._handle_http(
client_reader, client_writer, method, url, request_line
)
except asyncio.TimeoutError:
logger.warning("Client connection timeout")
await self._send_error(client_writer, 504, "Gateway Timeout")
except Exception as e:
logger.error(f"Error handling client: {e}")
await self._send_error(client_writer, 500, "Internal Server Error")
finally:
try:
client_writer.close()
await client_writer.wait_closed()
except Exception:
pass
if task:
self._active_connections.discard(task)
async def _handle_connect(
self,
client_reader: asyncio.StreamReader,
client_writer: asyncio.StreamWriter,
target: str,
_request_line: str,
) -> None:
"""Handle HTTPS CONNECT tunneling."""
# Parse target (host:port)
if ":" in target:
host, port_str = target.rsplit(":", 1)
port = int(port_str)
else:
host = target
port = 443
logger.info(f"CONNECT {host}:{port}")
try:
# Connect to target server
target_reader, target_writer = await asyncio.wait_for(
asyncio.open_connection(host, port), timeout=self.timeout
)
# Send success response to client
client_writer.write(b"HTTP/1.1 200 Connection Established\r\n\r\n")
await client_writer.drain()
# Consume remaining headers from client
while True:
line = await client_reader.readline()
if line == b"\r\n" or not line:
break
# Bidirectional forwarding
await asyncio.gather(
self._forward_data(client_reader, target_writer, "client->target"),
self._forward_data(target_reader, client_writer, "target->client"),
return_exceptions=True,
)
except asyncio.TimeoutError:
logger.warning(f"Timeout connecting to {host}:{port}")
await self._send_error(client_writer, 504, "Gateway Timeout")
except Exception as e:
logger.error(f"Error in CONNECT to {host}:{port}: {e}")
await self._send_error(client_writer, 502, "Bad Gateway")
async def _handle_http(
self,
client_reader: asyncio.StreamReader,
client_writer: asyncio.StreamWriter,
method: str,
url: str,
_request_line: str,
) -> None:
"""Handle plain HTTP requests."""
# Parse URL
if url.startswith("http://"):
url = url[7:] # Remove http://
elif url.startswith("https://"):
# Redirect to CONNECT for HTTPS
await self._send_error(
client_writer, 400, "Use CONNECT method for HTTPS"
)
return
if "/" in url:
host_port, path = url.split("/", 1)
path = "/" + path
else:
host_port = url
path = "/"
if ":" in host_port:
host, port_str = host_port.rsplit(":", 1)
port = int(port_str)
else:
host = host_port
port = 80
logger.info(f"{method} {host}:{port}{path}")
try:
# Connect to target server
target_reader, target_writer = await asyncio.wait_for(
asyncio.open_connection(host, port), timeout=self.timeout
)
# Read all client headers
headers = []
while True:
line = await client_reader.readline()
if line == b"\r\n" or not line:
break
headers.append(line)
# Build and send request to target
request = f"{method} {path} HTTP/1.1\r\n".encode()
target_writer.write(request)
# Forward headers (filter out proxy-specific ones)
for header in headers:
header_str = header.decode("utf-8", errors="ignore").lower()
if not header_str.startswith("proxy-"):
target_writer.write(header)
# Ensure Host header is present
has_host = any(
h.decode("utf-8", errors="ignore").lower().startswith("host:")
for h in headers
)
if not has_host:
target_writer.write(f"Host: {host}\r\n".encode())
target_writer.write(b"\r\n")
await target_writer.drain()
# Forward request body (for POST/PUT/PATCH)
content_length = 0
is_chunked = False
for header in headers:
h = header.decode("utf-8", errors="ignore").lower()
if h.startswith("content-length:"):
try:
content_length = int(h.split(":", 1)[1].strip())
except (ValueError, IndexError):
logger.warning(f"Invalid Content-Length header: {h}")
content_length = 0
elif h.startswith("transfer-encoding:") and "chunked" in h:
is_chunked = True
if content_length > 0:
remaining = content_length
while remaining > 0:
chunk = await asyncio.wait_for(
client_reader.read(min(self.buffer_size, remaining)),
timeout=self.timeout,
)
if not chunk:
break
target_writer.write(chunk)
remaining -= len(chunk)
await target_writer.drain()
elif is_chunked:
while True:
line = await asyncio.wait_for(
client_reader.readline(), timeout=self.timeout
)
target_writer.write(line)
try:
size = int(line.strip(), 16)
except (ValueError, UnicodeDecodeError):
logger.error(f"Invalid chunk size: {line}")
break
if size == 0:
target_writer.write(b"\r\n")
break
data = await asyncio.wait_for(
client_reader.readexactly(size + 2), timeout=self.timeout
)
target_writer.write(data)
await target_writer.drain()
# Forward response back to client
while True:
data = await asyncio.wait_for(
target_reader.read(self.buffer_size), timeout=self.timeout
)
if not data:
break
client_writer.write(data)
await client_writer.drain()
except asyncio.TimeoutError:
logger.warning(f"Timeout connecting to {host}:{port}")
await self._send_error(client_writer, 504, "Gateway Timeout")
except Exception as e:
logger.error(f"Error in HTTP request to {host}:{port}: {e}")
await self._send_error(client_writer, 502, "Bad Gateway")
async def _forward_data(
self,
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
direction: str,
) -> None:
"""Forward data from reader to writer."""
try:
while True:
data = await reader.read(self.buffer_size)
if not data:
break
writer.write(data)
await writer.drain()
except Exception as e:
logger.debug(f"Forward error ({direction}): {e}")
finally:
try:
writer.close()
await writer.wait_closed()
except Exception:
pass
def _extract_domain(self, method: str, url: str) -> Optional[str]:
"""Extract domain from request URL."""
if method == "CONNECT":
# CONNECT format: host:port
if ":" in url:
return url.split(":")[0]
return url
# HTTP format: http://host:port/path or just /path
if url.startswith("http://") or url.startswith("https://"):
# Remove protocol
url = url.split("://", 1)[1]
# Extract host part
if "/" in url:
url = url.split("/")[0]
if ":" in url:
url = url.split(":")[0]
return url if url else None
def _is_domain_allowed(self, domain: str) -> bool:
"""Check if domain is in whitelist."""
if "*" in self.allowed_domains:
return True
domain = domain.lower()
for allowed in self.allowed_domains:
allowed = allowed.lower()
# Exact match
if domain == allowed:
return True
# Wildcard match (e.g., *.example.com)
if allowed.startswith("*."):
pattern = allowed[2:] # Remove *.
if domain.endswith(pattern) or domain == pattern:
return True
# Support glob-style patterns
pattern = re.escape(allowed).replace(r"\*", ".*")
if re.match(f"^{pattern}$", domain):
return True
return False
async def _send_error(
self, writer: asyncio.StreamWriter, code: int, message: str
) -> None:
"""Send HTTP error response to client."""
try:
response = (
f"HTTP/1.1 {code} {message}\r\n"
f"Content-Type: text/plain\r\n"
f"Connection: close\r\n"
f"\r\n"
f"{code} {message}\r\n"
)
writer.write(response.encode())
await writer.drain()
except Exception as e:
logger.debug(f"Error sending error response: {e}")
if __name__ == "__main__":
async def _main() -> None:
proxy_server = ProxyServer(port=58088)
await proxy_server.start()
try:
await asyncio.sleep(120) # Keep running for testing purposes
finally:
await proxy_server.stop()
asyncio.run(_main())Claude Model
Sonnet (default)
Is this a regression?
I don't know
Last Working Version
_No response_
Claude Code Version
2.1.72
Platform
Anthropic API
Operating System
Ubuntu/Debian Linux
Terminal/Shell
VS Code integrated terminal
Additional Information
_No response_
This issue has 3 comments on GitHub. Read the full discussion on GitHub ↗