diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index a039212..bb3c29f 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -43,7 +43,7 @@ jobs: - name: Run tests run: | - pytest tests/test_commands/ -v --tb=short + pytest tests/test_commands/ tests/test_network.py -v --tb=short - name: Determine version id: version diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index b8515ad..fabbc6a 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -28,4 +28,4 @@ jobs: - name: Run tests run: | - pytest tests/test_commands/ -v --tb=short + pytest tests/test_commands/ tests/test_network.py -v --tb=short diff --git a/README.md b/README.md index 6a193d5..78d3c70 100644 --- a/README.md +++ b/README.md @@ -96,6 +96,7 @@ bash = Bash( env={...}, # Environment variables cwd="/home/user", # Working directory network=NetworkConfig(...), # Network configuration (for curl) + fetch=custom_fetch, # Custom async fetch function (for curl) unescape_html=True, # Auto-fix HTML entities in LLM output (default: True) ) ``` @@ -271,6 +272,92 @@ bash = Bash(unescape_html=False) - **Filesystem isolation** - Virtual filesystem keeps host system safe - **SQLite sandboxed** - Only in-memory databases allowed +### Network Access + +Network access is disabled by default. Configure it explicitly to enable `curl`: + +```python +from just_bash import Bash, NetworkConfig + +bash = Bash( + network=NetworkConfig( + allowed_url_prefixes=["https://api.example.com/v1/"], + ) +) + +result = await bash.exec("curl -s https://api.example.com/v1/status") +``` + +`curl` is registered only when `network` or a custom `fetch` function is provided. +Without network configuration, `curl` returns "command not found". + +Allow additional HTTP methods when needed: + +```python +bash = Bash( + network=NetworkConfig( + allowed_url_prefixes=["https://api.example.com/"], + allowed_methods=["GET", "HEAD", "POST"], + ) +) +``` + +Inject trusted headers at the network boundary so secrets never enter the sandbox: + +```python +from just_bash import AllowedUrl, Bash, NetworkConfig, RequestTransform + +bash = Bash( + network=NetworkConfig( + allowed_url_prefixes=[ + AllowedUrl( + url="https://api.example.com/", + transform=[ + RequestTransform(headers={"Authorization": "Bearer secret"}) + ], + ) + ], + ) +) +``` + +Allow all URLs and methods only when the caller already trusts the sandboxed command: + +```python +bash = Bash( + network=NetworkConfig(dangerously_allow_full_internet_access=True) +) +``` + +Pass `fetch=custom_fetch` to provide your own async `fetch(url, options)` +implementation. The default fetch implementation uses `aiohttp` and enforces URL +prefixes, HTTP methods, redirects, timeouts, response-size limits, and optional +private-range blocking. + +#### Allow-List Security + +The allow-list enforces: + +- **Origin matching** - URLs must match the exact scheme, host, and port +- **Path prefix** - Only paths starting with the configured prefix are allowed +- **HTTP method restrictions** - Only GET and HEAD are allowed by default +- **Redirect protection** - Redirect targets are checked before following them +- **Header transforms** - Boundary-injected headers override sandbox-supplied headers with the same name + +#### Using curl + +```bash +# Fetch and process data +curl -s https://api.example.com/data | grep pattern + +# Download into the virtual filesystem +curl -fsSL -o response.json https://api.example.com/data + +# POST JSON data +curl -X POST -H "Content-Type: application/json" \ + -d '{"key":"value"}' https://api.example.com/endpoint +``` + ## Supported Features ### Shell Syntax @@ -395,7 +482,7 @@ bash sh ## Test Results -Test suite history per commit (spec_tests excluded). Each `█` ≈ 57 tests. +Test suite history per commit (spec_tests excluded). Each `█` ≈ 58 tests. ``` Commit Date Passed Failed Skipped Graph @@ -410,6 +497,8 @@ a7e64a4 2026-02-11 2825 0 2 ███████████ e736ca4 2026-02-17 2831 0 2 ███████████████████████████████████████████████████░ 4dddca8 2026-02-18 2849 0 2 █████████████████████████████████████████████████░ 7c83ff3 2026-02-18 2870 0 3 █████████████████████████████████████████████████░ +bbb2f27 2026-05-19 2884 0 3 █████████████████████████████████████████████████░ +ad7fcdb 2026-05-19 2903 0 3 █████████████████████████████████████████████████░ ``` `█` passed · `▒` failed · `░` skipped diff --git a/src/just_bash/__init__.py b/src/just_bash/__init__.py index 8bbadcc..c63d602 100644 --- a/src/just_bash/__init__.py +++ b/src/just_bash/__init__.py @@ -13,6 +13,7 @@ from .bash import Bash from .types import ( + AllowedUrl, ExecResult, BashExecResult, ExecutionLimits, @@ -23,6 +24,8 @@ Command, OK, FAIL, + RequestTransform, + SecureFetch, ) from .fs import InMemoryFs from .parser import Parser, parse, ParseException @@ -37,6 +40,9 @@ "BashExecResult", "ExecutionLimits", "NetworkConfig", + "AllowedUrl", + "RequestTransform", + "SecureFetch", "IFileSystem", "FsStat", "CommandContext", diff --git a/src/just_bash/bash.py b/src/just_bash/bash.py index a97f0b5..cffcb90 100644 --- a/src/just_bash/bash.py +++ b/src/just_bash/bash.py @@ -27,8 +27,10 @@ import nest_asyncio # type: ignore[import-untyped] from .commands import create_command_registry +from .commands.registry import create_network_lazy_commands from .fs import InMemoryFs from .interpreter import ExitError, Interpreter, InterpreterState, ShellOptions, VariableStore +from .network import make_default_fetch from .parser import parse, unescape_html_entities from .parser.parser import ParseException from .types import ( @@ -37,6 +39,7 @@ ExecutionLimits, IFileSystem, NetworkConfig, + SecureFetch, ) @@ -56,6 +59,7 @@ def __init__( env: Optional[dict[str, str]] = None, limits: Optional[ExecutionLimits] = None, network: Optional[NetworkConfig] = None, + fetch: Optional[SecureFetch] = None, commands: Optional[dict[str, Command]] = None, errexit: bool = False, pipefail: bool = False, @@ -71,6 +75,7 @@ def __init__( env: Additional environment variables. limits: Execution limits for security. network: Network configuration (for curl command). + fetch: Custom secure fetch function (for curl command). commands: Custom command registry. If not provided, uses built-in commands. errexit: Enable errexit (set -e) mode. pipefail: Enable pipefail mode. @@ -88,11 +93,18 @@ def __init__( # Set up limits self._limits = limits or ExecutionLimits() - # Set up commands - self._commands = commands or create_command_registry() - # Set up network config self._network = network + self._fetch = fetch or (make_default_fetch(network) if network is not None else None) + + # Set up commands + if commands is None: + self._commands = create_command_registry(include_network=self._fetch is not None) + else: + self._commands = dict(commands) + if self._fetch is not None and "curl" not in self._commands: + for cmd in create_network_lazy_commands(): + self._commands[cmd.name] = cmd # Set up HTML unescaping self._unescape_html = unescape_html @@ -135,6 +147,7 @@ def __init__( commands=self._commands, limits=self._limits, state=self._initial_state, + fetch=self._fetch, ) @property @@ -244,6 +257,7 @@ def reset(self) -> None: fs=self._fs, commands=self._commands, limits=self._limits, + fetch=self._fetch, state=InterpreterState( env=self._initial_state.env.copy() if isinstance(self._initial_state.env, VariableStore) else VariableStore(self._initial_state.env), cwd=self._initial_state.cwd, diff --git a/src/just_bash/commands/curl/curl.py b/src/just_bash/commands/curl/curl.py index 661aa94..e1ae269 100644 --- a/src/just_bash/commands/curl/curl.py +++ b/src/just_bash/commands/curl/curl.py @@ -129,6 +129,13 @@ def format_headers(headers: dict[str, str]) -> str: return "\r\n".join(f"{name}: {value}" for name, value in headers.items()) +def body_to_stdout(body: str | bytes) -> str: + """Convert response bytes to stdout's string representation.""" + if isinstance(body, bytes): + return body.decode("latin-1") + return body + + def extract_filename(url: str) -> str: """Extract filename from URL for -O option.""" try: @@ -776,7 +783,7 @@ def _build_output( # Add body (unless head-only mode) if not options.head_only: - output += body + output += body_to_stdout(body) elif options.include_headers or options.verbose: # For HEAD, we already showed headers pass diff --git a/src/just_bash/interpreter/interpreter.py b/src/just_bash/interpreter/interpreter.py index 5b4ad7f..b25ae0c 100644 --- a/src/just_bash/interpreter/interpreter.py +++ b/src/just_bash/interpreter/interpreter.py @@ -30,7 +30,7 @@ ConditionalCommandNode, ArithmeticCommandNode, ) -from ..types import Command, ExecResult, ExecutionLimits, IFileSystem +from ..types import Command, ExecResult, ExecutionLimits, IFileSystem, SecureFetch from .errors import ( BadSubstitutionError, BreakError, @@ -129,6 +129,7 @@ def __init__( commands: dict[str, Command], limits: ExecutionLimits, state: Optional[InterpreterState] = None, + fetch: Optional[SecureFetch] = None, ): """Initialize the interpreter. @@ -137,10 +138,12 @@ def __init__( commands: Command registry limits: Execution limits state: Optional initial state (creates default if not provided) + fetch: Optional secure fetch function for network-enabled commands """ self._fs = fs self._commands = commands self._limits = limits + self._fetch = fetch self._state = state or InterpreterState( env=VariableStore({ "PATH": "/usr/local/bin:/usr/bin:/bin", @@ -216,6 +219,7 @@ async def _exec_fn( commands=self._commands, limits=self._limits, state=new_state, + fetch=self._fetch, ) try: return await sub_interpreter.execute_script(ast) @@ -570,6 +574,7 @@ async def _execute_subshell(self, node: SubshellNode, stdin: str) -> ExecResult: commands=self._commands, limits=self._limits, state=new_state, + fetch=self._fetch, ) # Execute statements in subshell @@ -1158,6 +1163,7 @@ async def _execute_simple_command( script, opts.get("env"), opts["cwd"] ), get_registered_commands=lambda: list(self._commands.keys()), + fetch=self._fetch, fd_contents=fd_contents, ) result = await cmd.execute(args, ctx) diff --git a/src/just_bash/network/__init__.py b/src/just_bash/network/__init__.py index 1abf7e9..9d61164 100644 --- a/src/just_bash/network/__init__.py +++ b/src/just_bash/network/__init__.py @@ -1 +1,513 @@ """Network module for just-bash.""" + +from __future__ import annotations + +import asyncio +import ipaddress +import socket +from collections.abc import Sequence +from typing import Any +from urllib.parse import SplitResult, urljoin, urlsplit + +import aiohttp +from aiohttp.abc import AbstractResolver, ResolveResult + +from ..types import AllowedUrl, NetworkConfig, RequestTransform + +_ALLOWED_SCHEMES = {"http", "https"} +_BODY_CHUNK_SIZE = 64 * 1024 +_INVALID_ALLOW_LIST_ENTRY = ( + 'Invalid allow-list entry: must be a string URL or an object with a "url" string property' +) + + +class NetworkAccessDeniedError(Exception): + """Raised when a URL is outside the configured network policy.""" + + def __init__(self, url: str, reason: str = "URL not in allow-list") -> None: + super().__init__(f"Network access denied: {reason}: {url}") + + +class MethodNotAllowedError(Exception): + """Raised when an HTTP method is outside the configured network policy.""" + + def __init__(self, method: str, allowed_methods: list[str]) -> None: + super().__init__( + f"HTTP method '{method}' not allowed. Allowed methods: {', '.join(allowed_methods)}" + ) + + +class RedirectNotAllowedError(Exception): + """Raised when a redirect target is outside the configured network policy.""" + + def __init__(self, url: str) -> None: + super().__init__(f"Redirect target not in allow-list: {url}") + + +class TooManyRedirectsError(Exception): + """Raised when a request exceeds the configured redirect limit.""" + + def __init__(self, max_redirects: int) -> None: + super().__init__(f"Too many redirects (max: {max_redirects})") + + +class ResponseTooLargeError(Exception): + """Raised when a response exceeds the configured size limit.""" + + def __init__(self, max_size: int) -> None: + super().__init__(f"Response body too large (max: {max_size} bytes)") + + +def _entry_url(entry: str | AllowedUrl | dict[str, Any]) -> str: + if isinstance(entry, str): + return entry + if isinstance(entry, dict): + return str(entry.get("url", "")) + return entry.url + + +def _default_port(scheme: str) -> int | None: + if scheme == "http": + return 80 + if scheme == "https": + return 443 + return None + + +def _parse_http_url(url: str) -> SplitResult | None: + try: + parsed = urlsplit(url) + # Accessing .port validates malformed ports. + _ = parsed.port + except ValueError: + return None + if parsed.scheme.lower() not in _ALLOWED_SCHEMES or not parsed.hostname: + return None + return parsed + + +def _normalized_origin(parsed: SplitResult) -> tuple[str, str, int]: + scheme = parsed.scheme.lower() + port = parsed.port or _default_port(scheme) + if port is None: + # _parse_http_url guarantees this is unreachable for callers. + raise ValueError(f"unsupported URL scheme: {parsed.scheme}") + return scheme, (parsed.hostname or "").lower(), port + + +def _has_ambiguous_path_separators(path: str) -> bool: + normalized = path.lower() + return "\\" in path or "%2f" in normalized or "%5c" in normalized + + +def _path_matches(path: str, prefix: str) -> bool: + if prefix in ("", "/"): + return True + if _has_ambiguous_path_separators(path): + return False + if prefix.endswith("/"): + return path.startswith(prefix) + return path == prefix or path.startswith(f"{prefix}/") + + +def _matches_allow_entry(url: str, allowed_entry: str) -> bool: + parsed_url = _parse_http_url(url) + parsed_allowed = _parse_http_url(allowed_entry) + if parsed_url is None or parsed_allowed is None: + return False + if _normalized_origin(parsed_url) != _normalized_origin(parsed_allowed): + return False + return _path_matches(parsed_url.path or "/", parsed_allowed.path or "/") + + +def _validate_allow_list(entries: list[str | AllowedUrl | dict[str, Any]]) -> list[str]: + errors: list[str] = [] + for raw_entry in entries: + if isinstance(raw_entry, dict): + entry = raw_entry.get("url") + elif isinstance(raw_entry, str): + entry = raw_entry + elif isinstance(raw_entry, AllowedUrl): + entry = raw_entry.url + else: + errors.append(_INVALID_ALLOW_LIST_ENTRY) + continue + + if not isinstance(entry, str): + errors.append(_INVALID_ALLOW_LIST_ENTRY) + continue + + parsed = _parse_http_url(entry) + if parsed is None: + errors.append( + f'Invalid URL in allow-list: "{entry}" - ' + "must be an http(s) URL with scheme and host" + ) + continue + + if parsed.query or parsed.fragment: + errors.append( + f'Query strings and fragments are ignored in allow-list entries: "{entry}"' + ) + continue + + path = parsed.path or "/" + if path not in ("", "/") and _has_ambiguous_path_separators(path): + errors.append(f'Allow-list entry contains ambiguous path separators: "{entry}"') + return errors + + +def _parse_ipv4_component(part: str) -> int | None: + if not part: + return None + base = 10 + digits = part + if digits.startswith(("0x", "0X")): + base = 16 + digits = digits[2:] + elif len(digits) > 1 and digits.startswith("0"): + base = 8 + try: + value = int(digits, base) + except ValueError: + return None + if value < 0: + return None + return value + + +def _parse_ipv4(host: str) -> ipaddress.IPv4Address | None: + parts = host.split(".") + if not parts or len(parts) > 4: + return None + nums = [_parse_ipv4_component(part) for part in parts] + if any(num is None for num in nums): + return None + + values = [num for num in nums if num is not None] + if len(values) == 1: + number = values[0] + if number > 0xFFFFFFFF: + return None + elif len(values) == 2: + first, second = values + if first > 0xFF or second > 0xFFFFFF: + return None + number = (first << 24) | second + elif len(values) == 3: + first, second, third = values + if first > 0xFF or second > 0xFF or third > 0xFFFF: + return None + number = (first << 24) | (second << 16) | third + else: + first, second, third, fourth = values + if first > 0xFF or second > 0xFF or third > 0xFF or fourth > 0xFF: + return None + number = (first << 24) | (second << 16) | (third << 8) | fourth + + try: + return ipaddress.IPv4Address(number) + except ipaddress.AddressValueError: + return None + + +# Precomputed once at import — these are on the hot path for +# deny_private_ranges=True (checked for the hostname and each resolved address). +_PRIVATE_IPV4_NETWORKS = ( + ipaddress.IPv4Network("0.0.0.0/8"), + ipaddress.IPv4Network("10.0.0.0/8"), + ipaddress.IPv4Network("100.64.0.0/10"), + ipaddress.IPv4Network("127.0.0.0/8"), + ipaddress.IPv4Network("169.254.0.0/16"), + ipaddress.IPv4Network("172.16.0.0/12"), + ipaddress.IPv4Network("192.0.0.0/24"), + ipaddress.IPv4Network("192.0.2.0/24"), + ipaddress.IPv4Network("192.168.0.0/16"), + ipaddress.IPv4Network("198.18.0.0/15"), + ipaddress.IPv4Network("198.51.100.0/24"), + ipaddress.IPv4Network("203.0.113.0/24"), + ipaddress.IPv4Network("224.0.0.0/4"), + ipaddress.IPv4Network("240.0.0.0/4"), +) +_PRIVATE_IPV6_NETWORKS = ( + ipaddress.IPv6Network("::/128"), + ipaddress.IPv6Network("::1/128"), + ipaddress.IPv6Network("fe80::/10"), + ipaddress.IPv6Network("fc00::/7"), + ipaddress.IPv6Network("2001:db8::/32"), + ipaddress.IPv6Network("64:ff9b::/96"), + ipaddress.IPv6Network("64:ff9b:1::/48"), +) +_SIXTOFOUR_NETWORK = ipaddress.IPv6Network("2002::/16") + + +def _is_private_ipv4(ip: ipaddress.IPv4Address) -> bool: + return any(ip in network for network in _PRIVATE_IPV4_NETWORKS) + + +def _is_private_ipv6(ip: ipaddress.IPv6Address) -> bool: + if ip.ipv4_mapped is not None: + return _is_private_ipv4(ip.ipv4_mapped) + if any(ip in network for network in _PRIVATE_IPV6_NETWORKS): + return True + if ip in _SIXTOFOUR_NETWORK: + embedded = int(ip) >> 80 & 0xFFFFFFFF + return _is_private_ipv4(ipaddress.IPv4Address(embedded)) + return False + + +def _is_private_hostname(hostname: str) -> bool: + host = hostname.strip().lower() + if host.startswith("[") and host.endswith("]"): + host = host[1:-1] + if host == "localhost" or host.endswith(".localhost"): + return True + parsed_ipv4 = _parse_ipv4(host) + if parsed_ipv4 is not None: + return _is_private_ipv4(parsed_ipv4) + try: + ip = ipaddress.ip_address(host) + except ValueError: + return False + if isinstance(ip, ipaddress.IPv4Address): + return _is_private_ipv4(ip) + return _is_private_ipv6(ip) + + +async def _resolve_host(hostname: str, port: int) -> list[ResolveResult]: + loop = asyncio.get_running_loop() + infos = await loop.getaddrinfo(hostname, port, type=socket.SOCK_STREAM) + results: list[ResolveResult] = [] + seen: set[tuple[str, int]] = set() + for family, _, proto, _, sockaddr in infos: + address = str(sockaddr[0]) + key = (address, family) + if key in seen: + continue + seen.add(key) + results.append( + { + "hostname": hostname, + "host": address, + "port": port, + "family": family, + "proto": proto, + "flags": socket.AI_NUMERICHOST, + } + ) + return results + + +class _PinnedResolver(AbstractResolver): + def __init__(self, hostname: str, records: list[ResolveResult]) -> None: + self._hostname = hostname + self._records = records + + async def resolve( + self, + host: str, + port: int = 0, + family: socket.AddressFamily = socket.AF_INET, + ) -> list[ResolveResult]: + if host == self._hostname: + return [{**record, "port": port} for record in self._records] + return await _resolve_host(host, port) + + async def close(self) -> None: + return None + + +def _merge_headers( + user_headers: dict[str, str] | None, + firewall_headers: dict[str, str], +) -> dict[str, str]: + merged = dict(user_headers or {}) + for key, value in firewall_headers.items(): + existing = next((k for k in merged if k.lower() == key.lower()), None) + if existing is not None: + del merged[existing] + merged[key] = value + return merged + + +def make_default_fetch(config: NetworkConfig): + """Create an aiohttp-backed secure fetch function for curl.""" + + entries = config.allowed_url_prefixes + if not config.dangerously_allow_full_internet_access: + errors = _validate_allow_list(entries) + if errors: + raise ValueError("Invalid network allow-list:\n" + "\n".join(errors)) + + allowed_methods = ( + ["GET", "HEAD", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"] + if config.dangerously_allow_full_internet_access + else [method.upper() for method in config.allowed_methods] + ) + + async def check_allowed(url: str) -> list[ResolveResult] | None: + parsed = _parse_http_url(url) + if parsed is None: + raise NetworkAccessDeniedError(url, "invalid URL") + + if not config.dangerously_allow_full_internet_access and not any( + _matches_allow_entry(url, _entry_url(entry)) for entry in entries + ): + raise NetworkAccessDeniedError(url) + + if not config.deny_private_ranges: + return None + + hostname = parsed.hostname or "" + if _is_private_hostname(hostname): + raise NetworkAccessDeniedError(url, "private/loopback IP address blocked") + + port = parsed.port or _default_port(parsed.scheme.lower()) or 80 + records = await _resolve_host(hostname, port) + for record in records: + if _is_private_hostname(record["host"]): + raise NetworkAccessDeniedError( + url, "hostname resolves to private/loopback IP address" + ) + return records + + def check_method_allowed(method: str) -> None: + if config.dangerously_allow_full_internet_access: + return + if method.upper() not in allowed_methods: + raise MethodNotAllowedError(method.upper(), allowed_methods) + + def firewall_headers(url: str) -> dict[str, str]: + merged: dict[str, str] = {} + for entry in entries: + entry_url = _entry_url(entry) + if isinstance(entry, str) or not _matches_allow_entry(url, entry_url): + continue + transforms: Sequence[RequestTransform | dict[str, Any]] + if isinstance(entry, dict): + transforms = entry.get("transform", []) + else: + transforms = entry.transform + for transform in transforms: + headers = ( + transform.get("headers", {}) + if isinstance(transform, dict) + else transform.headers + ) + merged.update(headers) + return merged + + async def fetch(url: str, options: dict[str, Any] | None = None) -> dict[str, Any]: + options = options or {} + method = (options.get("method") or "GET").upper() + check_method_allowed(method) + + current_url = url + redirect_count = 0 + follow_redirects = options.get("followRedirects", True) + max_redirects = int(options.get("maxRedirects", config.max_redirects)) + timeout_ms = min( + int(options.get("timeoutMs") or config.timeout_ms), + config.timeout_ms, + ) + body = options.get("body") + if body is not None and method in {"GET", "HEAD", "OPTIONS"}: + body = None + + while True: + pinned_records = await check_allowed(current_url) + timeout = aiohttp.ClientTimeout(total=timeout_ms / 1000) + connector = ( + aiohttp.TCPConnector( + resolver=_PinnedResolver(urlsplit(current_url).hostname or "", pinned_records) + ) + if pinned_records + else None + ) + try: + async with aiohttp.ClientSession(timeout=timeout, connector=connector) as session: + headers = _merge_headers( + options.get("headers") or {}, + firewall_headers(current_url), + ) + async with session.request( + method, + current_url, + headers=headers, + data=body, + allow_redirects=False, + auto_decompress=False, + # Don't let aiohttp auto-advertise `Accept-Encoding: + # gzip, deflate`. The curl layer only opts into + # compression under --compressed (and then decompresses + # itself); with auto_decompress=False an auto-injected + # header makes the server return raw gzip bytes that + # plain `curl` never asked for and won't decode. Honor + # only an Accept-Encoding the caller set explicitly. + skip_auto_headers=["Accept-Encoding"], + ) as resp: + if resp.status in {301, 302, 303, 307, 308} and follow_redirects: + location = resp.headers.get("location") + if not location: + response_body = await _read_limited_body( + resp, config.max_response_size + ) + return { + "status": resp.status, + "statusText": resp.reason or "", + "headers": {k.lower(): v for k, v in resp.headers.items()}, + "body": response_body, + "url": current_url, + "redirectCount": redirect_count, + } + + redirect_url = urljoin(current_url, location) + try: + await check_allowed(redirect_url) + except NetworkAccessDeniedError as exc: + raise RedirectNotAllowedError(redirect_url) from exc + + redirect_count += 1 + if redirect_count > max_redirects: + raise TooManyRedirectsError(max_redirects) + + current_url = redirect_url + continue + + response_body = await _read_limited_body( + resp, config.max_response_size + ) + return { + "status": resp.status, + "statusText": resp.reason or "", + "headers": {k.lower(): v for k, v in resp.headers.items()}, + "body": response_body, + "url": str(resp.url), + "redirectCount": redirect_count, + } + except TimeoutError as exc: + raise TimeoutError("operation timeout") from exc + + return fetch + + +async def _read_limited_body(resp: aiohttp.ClientResponse, max_size: int) -> bytes: + chunks: list[bytes] = [] + total = 0 + + if max_size > 0: + content_length = resp.headers.get("content-length") + if content_length: + try: + size = int(content_length) + except ValueError: + size = None + if size is not None and size > max_size: + raise ResponseTooLargeError(max_size) + + async for chunk in resp.content.iter_chunked(_BODY_CHUNK_SIZE): + total += len(chunk) + if max_size > 0 and total > max_size: + raise ResponseTooLargeError(max_size) + chunks.append(chunk) + return b"".join(chunks) diff --git a/src/just_bash/types.py b/src/just_bash/types.py index 1722978..6f37751 100644 --- a/src/just_bash/types.py +++ b/src/just_bash/types.py @@ -37,17 +37,37 @@ class ExecutionLimits: max_sed_iterations: int = 10_000 +@dataclass +class RequestTransform: + """Headers to inject at the network boundary for an allowed URL.""" + + headers: dict[str, str] + + +@dataclass +class AllowedUrl: + """Allowed URL prefix with optional request transforms.""" + + url: str + transform: list[RequestTransform] = field(default_factory=list) + + @dataclass class NetworkConfig: """Network access configuration.""" - allowed_url_prefixes: list[str] = field(default_factory=list) - allowed_methods: list[str] = field(default_factory=lambda: ["GET", "POST", "PUT", "DELETE"]) - max_redirects: int = 10 + allowed_url_prefixes: list[str | AllowedUrl | dict[str, Any]] = field(default_factory=list) + allowed_methods: list[str] = field(default_factory=lambda: ["GET", "HEAD"]) + max_redirects: int = 20 timeout_ms: int = 30_000 + max_response_size: int = 10_485_760 + deny_private_ranges: bool = False dangerously_allow_full_internet_access: bool = False +SecureFetch = Callable[[str, dict[str, Any]], Awaitable[dict[str, Any]]] + + class IFileSystem(Protocol): """Abstract filesystem interface.""" diff --git a/tests/test_network.py b/tests/test_network.py new file mode 100644 index 0000000..fb5ca29 --- /dev/null +++ b/tests/test_network.py @@ -0,0 +1,438 @@ +import asyncio +import gzip + +import pytest +from aiohttp import web + +from just_bash import AllowedUrl, Bash, NetworkConfig, RequestTransform +from just_bash.network import _is_private_hostname, _matches_allow_entry, _read_limited_body + + +async def make_server(routes): + app = web.Application() + for method, path, handler in routes: + app.router.add_route(method, path, handler) + runner = web.AppRunner(app) + await runner.setup() + site = web.TCPSite(runner, "127.0.0.1", 0) + await site.start() + assert site._server is not None + port = site._server.sockets[0].getsockname()[1] + return runner, f"http://127.0.0.1:{port}" + + +@pytest.mark.parametrize( + "entry", + [ + "https://api.example.com/v1?token=secret", + "https://api.example.com/v1#section", + "ftp://api.example.com/v1", + "https:///v1", + "/v1", + "https://api.example.com/v1%2fadmin", + {"not_url": "https://api.example.com"}, + ], +) +def test_invalid_allow_list_entries_fail_fast(entry): + with pytest.raises(ValueError, match="Invalid network allow-list"): + Bash(network=NetworkConfig(allowed_url_prefixes=[entry])) + + +def test_allow_list_matching_normalizes_origins_and_preserves_path_boundaries(): + assert _matches_allow_entry("https://example.com:443/v1/users", "https://EXAMPLE.com/v1") + assert _matches_allow_entry("http://example.com:80/v1/users", "http://example.com/v1") + assert _matches_allow_entry("https://example.com:8443/v1/users", "https://example.com:8443/v1") + assert not _matches_allow_entry("https://example.com:8443/v1/users", "https://example.com/v1") + assert not _matches_allow_entry("https://example.com/v10", "https://example.com/v1") + assert not _matches_allow_entry("https://example.com/v1-admin", "https://example.com/v1") + + +@pytest.mark.parametrize( + "hostname", + [ + "100.64.0.1", + "2130706433", + "0x7f.0.0.1", + "::1", + "::ffff:127.0.0.1", + ], +) +def test_private_range_detection_matches_upstream_cases(hostname): + assert _is_private_hostname(hostname) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "url", + [ + "http://100.64.0.1/", + "http://2130706433/", + "http://0x7f.0.0.1/", + "http://[::1]/", + "http://[::ffff:127.0.0.1]/", + ], +) +async def test_private_ranges_blocked_even_with_full_access(url): + bash = Bash( + network=NetworkConfig( + dangerously_allow_full_internet_access=True, + deny_private_ranges=True, + ) + ) + + result = await bash.exec(f"curl -sS {url}") + + assert result.exit_code == 7 + assert "private/loopback" in result.stderr + + +@pytest.mark.asyncio +async def test_malformed_content_length_falls_back_to_streamed_size_check(): + class FakeContent: + async def iter_chunked(self, _size): + yield b"ok" + + class FakeResponse: + headers = {"content-length": "nope"} + content = FakeContent() + + assert await _read_limited_body(FakeResponse(), 3) == b"ok" + + +@pytest.mark.asyncio +async def test_curl_disabled_by_default_returns_command_not_found(): + bash = Bash() + result = await bash.exec("curl -s https://example.com") + + assert result.exit_code == 127 + assert "command not found" in result.stderr + + +@pytest.mark.asyncio +async def test_curl_loads_when_network_configured(): + async def ok(_request): + return web.Response(text="ok") + + runner, base_url = await make_server([("GET", "/ok", ok)]) + try: + bash = Bash(network=NetworkConfig(allowed_url_prefixes=[base_url])) + result = await bash.exec(f"curl -s {base_url}/ok") + finally: + await runner.cleanup() + + assert result.exit_code == 0 + assert result.stdout == "ok" + + +@pytest.mark.asyncio +async def test_curl_blocked_when_url_not_in_prefixes(): + async def ok(_request): + return web.Response(text="ok") + + runner, base_url = await make_server([("GET", "/ok", ok)]) + try: + bash = Bash(network=NetworkConfig(allowed_url_prefixes=[f"{base_url}/allowed"])) + result = await bash.exec(f"curl -sS {base_url}/ok") + finally: + await runner.cleanup() + + assert result.exit_code == 7 + assert "Network access denied" in result.stderr + + +@pytest.mark.asyncio +async def test_curl_blocked_when_method_not_allowed(): + async def ok(_request): + return web.Response(text="ok") + + runner, base_url = await make_server([("POST", "/ok", ok)]) + try: + bash = Bash(network=NetworkConfig(allowed_url_prefixes=[base_url])) + result = await bash.exec(f"curl -sS -X POST {base_url}/ok") + finally: + await runner.cleanup() + + assert result.exit_code == 3 + assert "HTTP method" in result.stderr + assert "not allowed" in result.stderr + + +@pytest.mark.asyncio +async def test_curl_head_allowed_by_default(): + async def ok(_request): + return web.Response(text="ok", headers={"x-test": "1"}) + + runner, base_url = await make_server([("HEAD", "/ok", ok)]) + try: + bash = Bash(network=NetworkConfig(allowed_url_prefixes=[base_url])) + result = await bash.exec(f"curl -I {base_url}/ok") + finally: + await runner.cleanup() + + assert result.exit_code == 0 + assert "HTTP/1.1 200" in result.stdout + assert "x-test: 1" in result.stdout.lower() + + +@pytest.mark.asyncio +async def test_dangerously_allow_full_internet_access_bypasses_prefixes_and_methods(): + async def ok(_request): + return web.Response(text="posted") + + runner, base_url = await make_server([("POST", "/ok", ok)]) + try: + bash = Bash( + network=NetworkConfig( + allowed_url_prefixes=[], + dangerously_allow_full_internet_access=True, + ) + ) + result = await bash.exec(f"curl -s -X POST {base_url}/ok") + finally: + await runner.cleanup() + + assert result.exit_code == 0 + assert result.stdout == "posted" + + +@pytest.mark.asyncio +async def test_redirect_target_checked_against_allow_list(): + async def redirect(_request): + raise web.HTTPFound("/final") + + async def final(_request): + return web.Response(text="final") + + runner, base_url = await make_server( + [("GET", "/redirect", redirect), ("GET", "/final", final)] + ) + try: + bash = Bash(network=NetworkConfig(allowed_url_prefixes=[f"{base_url}/redirect"])) + result = await bash.exec(f"curl -sS {base_url}/redirect") + finally: + await runner.cleanup() + + assert result.exit_code == 47 + assert "Redirect target not in allow-list" in result.stderr + + +@pytest.mark.asyncio +async def test_max_redirects_honored(): + async def one(_request): + raise web.HTTPFound("/two") + + async def two(_request): + raise web.HTTPFound("/final") + + async def final(_request): + return web.Response(text="final") + + runner, base_url = await make_server( + [("GET", "/one", one), ("GET", "/two", two), ("GET", "/final", final)] + ) + try: + bash = Bash( + network=NetworkConfig( + allowed_url_prefixes=[base_url], + max_redirects=1, + ) + ) + result = await bash.exec(f"curl -sS {base_url}/one") + finally: + await runner.cleanup() + + assert result.exit_code == 47 + assert "Too many redirects" in result.stderr + + +@pytest.mark.asyncio +async def test_timeout_ms_honored(): + async def slow(_request): + await asyncio.sleep(0.2) + return web.Response(text="slow") + + runner, base_url = await make_server([("GET", "/slow", slow)]) + try: + bash = Bash( + network=NetworkConfig( + allowed_url_prefixes=[base_url], + timeout_ms=10, + ) + ) + result = await bash.exec(f"curl -sS {base_url}/slow") + finally: + await runner.cleanup() + + assert result.exit_code == 28 + + +@pytest.mark.asyncio +async def test_max_response_size_honored(): + async def large(_request): + return web.Response(body=b"too large") + + runner, base_url = await make_server([("GET", "/large", large)]) + try: + bash = Bash( + network=NetworkConfig( + allowed_url_prefixes=[base_url], + max_response_size=3, + ) + ) + result = await bash.exec(f"curl -sS {base_url}/large") + finally: + await runner.cleanup() + + assert result.exit_code == 1 + assert "Response body too large" in result.stderr + + +@pytest.mark.asyncio +async def test_deny_private_ranges_blocks_loopback_even_with_full_access(): + async def ok(_request): + return web.Response(text="ok") + + runner, base_url = await make_server([("GET", "/ok", ok)]) + try: + bash = Bash( + network=NetworkConfig( + dangerously_allow_full_internet_access=True, + deny_private_ranges=True, + ) + ) + result = await bash.exec(f"curl -sS {base_url}/ok") + finally: + await runner.cleanup() + + assert result.exit_code == 7 + assert "private/loopback" in result.stderr + + +@pytest.mark.asyncio +async def test_header_transform_overrides_user_header(): + async def ok(request): + return web.Response(text=request.headers.get("Authorization", "")) + + runner, base_url = await make_server([("GET", "/ok", ok)]) + try: + bash = Bash( + network=NetworkConfig( + allowed_url_prefixes=[ + AllowedUrl( + url=base_url, + transform=[ + RequestTransform(headers={"Authorization": "Bearer secret"}) + ], + ) + ], + ) + ) + result = await bash.exec( + f"curl -s -H 'Authorization: Bearer user' {base_url}/ok" + ) + finally: + await runner.cleanup() + + assert result.exit_code == 0 + assert result.stdout == "Bearer secret" + + +@pytest.mark.asyncio +async def test_curl_writes_bytes_to_sandbox_fs_with_custom_fetch(): + async def fetch(url, _options): + return { + "status": 200, + "statusText": "OK", + "headers": {"content-type": "application/octet-stream"}, + "body": b"\x00\xffdata", + "url": url, + } + + bash = Bash(fetch=fetch) + result = await bash.exec("curl -s -o out.bin https://example.com/blob") + + assert result.exit_code == 0 + assert await bash.fs.read_file_bytes("/home/user/out.bin") == b"\x00\xffdata" + + +@pytest.mark.asyncio +async def test_curl_write_out_format_string(): + async def ok(_request): + return web.Response(text="data", content_type="text/plain") + + runner, base_url = await make_server([("GET", "/ok", ok)]) + try: + bash = Bash(network=NetworkConfig(allowed_url_prefixes=[base_url])) + result = await bash.exec(f"curl -s -w '%{{content_type}}|%{{size_download}}' {base_url}/ok") + finally: + await runner.cleanup() + + assert result.exit_code == 0 + assert result.stdout.endswith("text/plain; charset=utf-8|4") + + +def _make_compressing_server_routes(payload, seen): + """A handler that gzips its response iff the client advertises gzip. + + Mirrors real-world servers (e.g. whitehouse.gov) that compress only when + the request carries `Accept-Encoding: gzip`. `seen` captures the header the + server actually received so tests can assert what curl advertised. + """ + + async def handler(request): + accept_encoding = request.headers.get("Accept-Encoding", "") + seen["accept_encoding"] = accept_encoding + if "gzip" in accept_encoding: + return web.Response( + body=gzip.compress(payload.encode()), + headers={"Content-Encoding": "gzip"}, + content_type="text/plain", + ) + return web.Response(text=payload, content_type="text/plain") + + return [("GET", "/page", handler)] + + +@pytest.mark.asyncio +async def test_plain_curl_does_not_advertise_compression(): + """Plain `curl` must not request gzip and must return decoded text. + + Regression: aiohttp auto-injects `Accept-Encoding: gzip, deflate` on every + request. Combined with `auto_decompress=False`, plain `curl` (no + --compressed) received raw gzip bytes the curl layer correctly refused to + decompress, surfacing as binary garbage. Real curl sends no Accept-Encoding + without --compressed, so the server returns identity and the body is clean. + """ + payload = "hello-world-not-binary-garbage" + seen = {} + runner, base_url = await make_server(_make_compressing_server_routes(payload, seen)) + try: + bash = Bash(network=NetworkConfig(allowed_url_prefixes=[base_url])) + result = await bash.exec(f"curl -s {base_url}/page") + finally: + await runner.cleanup() + + assert result.exit_code == 0 + # curl did not opt into compression, so the server saw no gzip request... + assert "gzip" not in seen["accept_encoding"] + # ...and the body is clean text, not raw gzip bytes. + assert result.stdout == payload + + +@pytest.mark.asyncio +async def test_compressed_flag_still_negotiates_and_decompresses(): + """`curl --compressed` must advertise gzip and transparently decompress.""" + payload = "hello-from-a-gzipped-response" + seen = {} + runner, base_url = await make_server(_make_compressing_server_routes(payload, seen)) + try: + bash = Bash(network=NetworkConfig(allowed_url_prefixes=[base_url])) + result = await bash.exec(f"curl -s --compressed {base_url}/page") + finally: + await runner.cleanup() + + assert result.exit_code == 0 + # --compressed opted in, so the server compressed the response... + assert "gzip" in seen["accept_encoding"] + # ...and curl decompressed it back to clean text. + assert result.stdout == payload