From 99b65c6313e3d18a405af609a2cb66134dd25cb6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=81ukasz=20Ole=C5=9B?= Date: Fri, 3 Jul 2026 03:53:32 +0200 Subject: [PATCH 1/2] Support Network Polcices in the Sandboxes --- docs/sandbox.md | 236 ++++++++++++++++++++++++---- examples/21_egress_policy.py | 74 +++++++++ examples/21_egress_policy_async.py | 75 +++++++++ examples/README.md | 6 + koyeb/sandbox/__init__.py | 9 +- koyeb/sandbox/filesystem.py | 4 +- koyeb/sandbox/sandbox.py | 169 +++++++++++++++++++- koyeb/sandbox/test_egress_policy.py | 185 ++++++++++++++++++++++ koyeb/sandbox/test_utils.py | 80 +++++++++- koyeb/sandbox/utils.py | 91 ++++++++++- uv.lock | 10 +- 11 files changed, 895 insertions(+), 44 deletions(-) create mode 100644 examples/21_egress_policy.py create mode 100644 examples/21_egress_policy_async.py create mode 100644 koyeb/sandbox/test_egress_policy.py diff --git a/docs/sandbox.md b/docs/sandbox.md index 0654f3b7..0ee6e5b3 100644 --- a/docs/sandbox.md +++ b/docs/sandbox.md @@ -18,6 +18,50 @@ class TestCreateDockerSource(unittest.TestCase) Tests for create_docker_source entrypoint, command, and args support. + + +## TestBuildEgressPolicy Objects + +```python +class TestBuildEgressPolicy(unittest.TestCase) +``` + +Tests for build_network_policy validation and normalization. + + + +# koyeb/sandbox.test\_egress\_policy + + + +## TestCreateEgressWiring Objects + +```python +class TestCreateEgressWiring(unittest.TestCase) +``` + +Tests that Sandbox.create wires egress kwargs into the deployment definition. + + + +## TestUpdateNetworkPolicy Objects + +```python +class TestUpdateNetworkPolicy(unittest.TestCase) +``` + +Tests for Sandbox.update_network_policy. + + + +## TestAsyncUpdateNetworkPolicy Objects + +```python +class TestAsyncUpdateNetworkPolicy(unittest.TestCase) +``` + +Tests for AsyncSandbox.update_network_policy. + # koyeb/sandbox.exec @@ -958,6 +1002,12 @@ Exit code (if completed) ISO 8601 timestamp when process started + + +#### completed\_at + +ISO 8601 timestamp when process completed (if applicable) + ## ExposedPort Objects @@ -1022,7 +1072,9 @@ def create(cls, entrypoint: Optional[List[str]] = None, command: Optional[str] = None, args: Optional[List[str]] = None, - host: Optional[str] = None) -> Sandbox + host: Optional[str] = None, + block_network: bool = False, + outbound_allowlist: Optional[List[str]] = None) -> Sandbox ``` Create a new sandbox instance. @@ -1065,6 +1117,10 @@ Create a new sandbox instance. - `entrypoint` - Override the default entrypoint of the Docker image (e.g., ["/bin/sh", "-c"]) - `command` - Override the default command of the Docker image (e.g., "python app.py") - `host` - Koyeb API host URL. If not provided, will try to get from KOYEB_API_HOST env var (defaults to https://app.koyeb.com) +- `block_network` - If True, block all outbound network access from the sandbox +- `outbound_allowlist` - List of IPs/CIDRs allowed as outbound destinations; + all other outbound traffic is blocked. Bare IPs are normalized to + /32 (IPv4) or /128 (IPv6). Mutually exclusive with block_network. **Returns**: @@ -1076,6 +1132,8 @@ Create a new sandbox instance. - `ValueError` - If API token is not provided - `SandboxTimeoutError` - If wait_ready is True and sandbox does not become ready within timeout +- `EgressPolicyError` - If both block_network and outbound_allowlist are passed, + or an allowlist entry is not a valid IP address or CIDR **Example**: @@ -1476,6 +1534,47 @@ Update the sandbox's life cycle settings. >>> sandbox.update_life_cycle(delete_after_delay=600, delete_after_inactivity=300) + + +#### update\_network\_policy + +```python +def update_network_policy( + block_network: bool = False, + outbound_allowlist: Optional[List[str]] = None) -> None +``` + +Update the sandbox's egress network policy. + +Warning: applying a new network policy triggers a redeployment of the +sandbox service. The sandbox is restarted and any in-memory or +non-persisted state is lost. This method does not wait for the +redeployment to finish. + +**Arguments**: + +- `block_network` - If True, block all outbound network access from the sandbox +- `outbound_allowlist` - List of IPs/CIDRs allowed as outbound destinations; + all other outbound traffic is blocked. Bare IPs are normalized to + /32 (IPv4) or /128 (IPv6). Mutually exclusive with block_network. + + With both arguments unset, the egress policy is reset to the + platform default (unrestricted outbound access). + + +**Raises**: + +- `EgressPolicyError` - If both block_network and outbound_allowlist are + passed, or an allowlist entry is not a valid IP address or CIDR +- `SandboxError` - If updating the egress policy fails + + +**Example**: + + >>> sandbox.update_network_policy(block_network=True) + >>> sandbox.update_network_policy(outbound_allowlist=["10.0.0.0/8", "1.2.3.4"]) + >>> sandbox.update_network_policy() # reset to unrestricted + #### \_\_enter\_\_ @@ -1544,32 +1643,35 @@ Get a sandbox by service ID asynchronously. ```python @classmethod -async def create(cls, - image: str = "koyeb/sandbox", - name: str = "quick-sandbox", - wait_ready: bool = True, - instance_type: str = "micro", - exposed_port_protocol: Optional[str] = None, - env: Optional[Dict[str, Any]] = None, - config_files: Optional[Dict[str, Any]] = None, - region: Optional[str] = None, - api_token: Optional[str] = None, - timeout: int = 300, - idle_timeout: int = 0, - enable_tcp_proxy: bool = False, - privileged: bool = False, - registry_secret: Optional[str] = None, - _experimental_enable_light_sleep: bool = False, - _experimental_deep_sleep_value: int = 3900, - delete_after_delay: int = 0, - delete_after_inactivity_delay: int = 0, - app_id: Optional[str] = None, - enable_mesh: bool = False, - poll_interval: float = DEFAULT_POLL_INTERVAL, - entrypoint: Optional[List[str]] = None, - command: Optional[str] = None, - args: Optional[List[str]] = None, - host: Optional[str] = None) -> AsyncSandbox +async def create( + cls, + image: str = "koyeb/sandbox", + name: str = "quick-sandbox", + wait_ready: bool = True, + instance_type: str = "micro", + exposed_port_protocol: Optional[str] = None, + env: Optional[Dict[str, Any]] = None, + config_files: Optional[Dict[str, Any]] = None, + region: Optional[str] = None, + api_token: Optional[str] = None, + timeout: int = 300, + idle_timeout: int = 0, + enable_tcp_proxy: bool = False, + privileged: bool = False, + registry_secret: Optional[str] = None, + _experimental_enable_light_sleep: bool = False, + _experimental_deep_sleep_value: int = 3900, + delete_after_delay: int = 0, + delete_after_inactivity_delay: int = 0, + app_id: Optional[str] = None, + enable_mesh: bool = False, + poll_interval: float = DEFAULT_POLL_INTERVAL, + entrypoint: Optional[List[str]] = None, + command: Optional[str] = None, + args: Optional[List[str]] = None, + host: Optional[str] = None, + block_network: bool = False, + outbound_allowlist: Optional[List[str]] = None) -> AsyncSandbox ``` Create a new sandbox instance with async support. @@ -1614,6 +1716,10 @@ Create a new sandbox instance with async support. - `entrypoint` - Override the default entrypoint of the Docker image (e.g., ["/bin/sh", "-c"]) - `command` - Override the default command of the Docker image (e.g., "python app.py") - `host` - Koyeb API host URL. If not provided, will try to get from KOYEB_API_HOST env var (defaults to https://app.koyeb.com) +- `block_network` - If True, block all outbound network access from the sandbox +- `outbound_allowlist` - List of IPs/CIDRs allowed as outbound destinations; + all other outbound traffic is blocked. Bare IPs are normalized to + /32 (IPv4) or /128 (IPv6). Mutually exclusive with block_network. **Returns**: @@ -1625,6 +1731,8 @@ Create a new sandbox instance with async support. - `ValueError` - If API token is not provided - `SandboxTimeoutError` - If wait_ready is True and sandbox does not become ready within timeout +- `EgressPolicyError` - If both block_network and outbound_allowlist are passed, + or an allowlist entry is not a valid IP address or CIDR @@ -1792,6 +1900,31 @@ async def update_lifecycle( Update the sandbox's life cycle settings asynchronously. + + +#### update\_network\_policy + +```python +async def update_network_policy( + block_network: bool = False, + outbound_allowlist: Optional[List[str]] = None) -> None +``` + +Update the sandbox's egress network policy asynchronously. + +Warning: applying a new network policy triggers a redeployment of the +sandbox service. The sandbox is restarted and any in-memory or +non-persisted state is lost. This method does not wait for the +redeployment to finish. + +See Sandbox.update_network_policy for full documentation. + +**Raises**: + +- `EgressPolicyError` - If both block_network and outbound_allowlist are + passed, or an allowlist entry is not a valid IP address or CIDR +- `SandboxError` - If updating the egress policy fails + #### \_\_aenter\_\_ @@ -2071,7 +2204,8 @@ def create_deployment_definition( _experimental_enable_light_sleep: bool = False, _experimental_deep_sleep_value: int = 3900, enable_mesh: bool = None, - config_files: Optional[List[ConfigFile]] = None + config_files: Optional[List[ConfigFile]] = None, + network_policy: Optional[NetworkPolicy] = None ) -> DeploymentDefinition ``` @@ -2095,12 +2229,46 @@ Create deployment definition for a sandbox service. - `_experimental_deep_sleep_value` - Number of seconds for deep sleep when light sleep is enabled (default: 3900). Only used if _experimental_enable_light_sleep is True. Ignored otherwise. - `enable_mesh` - Enable or disable mesh for this sandbox. Disabled by default +- `network_policy` - Optional network policy restricting egress traffic **Returns**: DeploymentDefinition object + + +#### build\_network\_policy + +```python +def build_network_policy( + block_network: bool = False, + outbound_allowlist: Optional[List[str]] = None +) -> Optional[NetworkPolicy] +``` + +Build a NetworkPolicy from sandbox egress arguments. + +**Arguments**: + +- `block_network` - If True, block all outbound network access +- `outbound_allowlist` - List of IPs/CIDRs allowed as outbound + destinations; all other outbound traffic is blocked. Bare IPs + are normalized to /32 (IPv4) or /128 (IPv6). An empty list + blocks all outbound traffic. + + +**Returns**: + + NetworkPolicy, or None when both arguments are unset + (block_network=False and outbound_allowlist=None) + + +**Raises**: + +- `EgressPolicyError` - If both arguments are passed, or an allowlist + entry is not a valid IP address or CIDR + #### escape\_shell\_arg @@ -2165,7 +2333,7 @@ Uses case-insensitive matching against known error patterns. #### create\_sandbox\_client ```python -def create_sandbox_client(conn_info: Optional['ConnectionInfo'], +def create_sandbox_client(conn_info: Optional["ConnectionInfo"], existing_client: Optional[Any] = None) -> Any ``` @@ -2258,6 +2426,16 @@ class SandboxServiceError(SandboxError) Raised when the sandbox executor returns an HTTP 5xx error + + +## EgressPolicyError Objects + +```python +class EgressPolicyError(SandboxError) +``` + +Raised when egress policy arguments are invalid or conflicting + # koyeb/sandbox.executor\_client diff --git a/examples/21_egress_policy.py b/examples/21_egress_policy.py new file mode 100644 index 00000000..416ab9d1 --- /dev/null +++ b/examples/21_egress_policy.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +"""Egress network policy: block all outbound traffic or restrict it to an allowlist""" + +import os +import random +import string +import sys + +from koyeb import Sandbox +from koyeb.sandbox import EgressPolicyError + +# Outbound probe run inside the sandbox; fails when egress is blocked +PROBE = ( + 'python3 -c "import urllib.request; ' + "urllib.request.urlopen('https://example.com', timeout=5)\"" +) + + +def main(): + api_token = os.getenv("KOYEB_API_TOKEN") + if not api_token: + print("Error: KOYEB_API_TOKEN not set") + return 1 + + suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=8)) + + # block_network and outbound_allowlist are mutually exclusive; passing + # both is rejected client-side, before any API call + try: + Sandbox.create( + name=f"egress-{suffix}", + api_token=api_token, + block_network=True, + outbound_allowlist=["1.1.1.1"], + ) + raise AssertionError("Expected EgressPolicyError") + except EgressPolicyError as e: + print(f"Conflicting arguments rejected: {e}") + + sandbox = None + try: + # Create a sandbox with all outbound network access blocked + sandbox = Sandbox.create( + image="koyeb/sandbox", + name=f"egress-{suffix}", + wait_ready=True, + api_token=api_token, + block_network=True, + ) + print(f"Created sandbox with block_network=True: {sandbox.name}") + + # Outbound requests from inside the sandbox fail + result = sandbox.exec(PROBE) + assert result.exit_code != 0, "Expected outbound request to fail" + print(f"Outbound request blocked (exit code {result.exit_code})") + + # Switch to an allowlist: only the listed destinations are reachable. + # Entries are CIDRs or bare IPs (normalized to /32 for IPv4, /128 for + # IPv6). This triggers a redeployment of the sandbox service. + sandbox.update_network_policy(outbound_allowlist=["1.1.1.1", "9.9.0.0/16"]) + print("Egress policy updated to allowlist: 1.1.1.1/32, 9.9.0.0/16") + + # Reset to the platform default (unrestricted outbound access) + sandbox.update_network_policy() + print("Egress policy reset to default") + + return 0 + finally: + if sandbox: + sandbox.delete() + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/examples/21_egress_policy_async.py b/examples/21_egress_policy_async.py new file mode 100644 index 00000000..7e5a58d5 --- /dev/null +++ b/examples/21_egress_policy_async.py @@ -0,0 +1,75 @@ +#!/usr/bin/env python3 +"""Egress network policy: block all outbound traffic or restrict it to an allowlist (async variant)""" + +import asyncio +import os +import random +import string +import sys + +from koyeb import AsyncSandbox +from koyeb.sandbox import EgressPolicyError + +# Outbound probe run inside the sandbox; fails when egress is blocked +PROBE = ( + 'python3 -c "import urllib.request; ' + "urllib.request.urlopen('https://example.com', timeout=5)\"" +) + + +async def main(): + api_token = os.getenv("KOYEB_API_TOKEN") + if not api_token: + print("Error: KOYEB_API_TOKEN not set") + return 1 + + suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=8)) + + # block_network and outbound_allowlist are mutually exclusive; passing + # both is rejected client-side, before any API call + try: + await AsyncSandbox.create( + name=f"egress-{suffix}", + api_token=api_token, + block_network=True, + outbound_allowlist=["1.1.1.1"], + ) + raise AssertionError("Expected EgressPolicyError") + except EgressPolicyError as e: + print(f"Conflicting arguments rejected: {e}") + + sandbox = None + try: + # Create a sandbox with all outbound network access blocked + sandbox = await AsyncSandbox.create( + image="koyeb/sandbox", + name=f"egress-{suffix}", + wait_ready=True, + api_token=api_token, + block_network=True, + ) + print(f"Created sandbox with block_network=True: {sandbox.name}") + + # Outbound requests from inside the sandbox fail + result = await sandbox.exec(PROBE) + assert result.exit_code != 0, "Expected outbound request to fail" + print(f"Outbound request blocked (exit code {result.exit_code})") + + # Switch to an allowlist: only the listed destinations are reachable. + # Entries are CIDRs or bare IPs (normalized to /32 for IPv4, /128 for + # IPv6). This triggers a redeployment of the sandbox service. + await sandbox.update_network_policy(outbound_allowlist=["1.1.1.1", "9.9.0.0/16"]) + print("Egress policy updated to allowlist: 1.1.1.1/32, 9.9.0.0/16") + + # Reset to the platform default (unrestricted outbound access) + await sandbox.update_network_policy() + print("Egress policy reset to default") + + return 0 + finally: + if sandbox: + await sandbox.delete() + + +if __name__ == "__main__": + sys.exit(asyncio.run(main())) diff --git a/examples/README.md b/examples/README.md index c10b4eb7..c3efbd6a 100644 --- a/examples/README.md +++ b/examples/README.md @@ -29,6 +29,12 @@ uv run python examples/01_create_sandbox.py - **13_background_processes.py** - Background process management (launch, list, kill) - **14_expose_port.py** - Port exposure via TCP proxy with HTTP verification - **15_get_sandbox.py** - Create a sandbox and then retrieve it by ID +- **16_create_sandbox_with_auto_delete_simple.py** - Create a sandbox with auto-delete lifecycle setting +- **17_create_sandbox_with_auto_delete.py** - Create sandboxes with auto-delete lifecycle settings +- **18_create_sandbox_with_existing_app.py** - Create a sandbox in an existing app +- **19_entrypoint_and_command.py** - Create sandboxes with custom entrypoint and command +- **20_config_files.py** - Config files with secrets and interpolation +- **21_egress_policy.py** - Block outbound network access or restrict it to an allowlist ## Basic Usage diff --git a/koyeb/sandbox/__init__.py b/koyeb/sandbox/__init__.py index 4688c7c6..cdc9fc17 100644 --- a/koyeb/sandbox/__init__.py +++ b/koyeb/sandbox/__init__.py @@ -19,7 +19,13 @@ ) from .filesystem import FileInfo, SandboxFilesystem from .sandbox import AsyncSandbox, ExposedPort, ProcessInfo, Sandbox -from .utils import SandboxDeploymentError, SandboxError, SandboxServiceError, SandboxTimeoutError +from .utils import ( + EgressPolicyError, + SandboxDeploymentError, + SandboxError, + SandboxServiceError, + SandboxTimeoutError, +) __all__ = [ "Sandbox", @@ -31,6 +37,7 @@ "AsyncSandboxExecutor", "FileInfo", "SandboxStatus", + "EgressPolicyError", "SandboxDeploymentError", "SandboxError", "SandboxServiceError", diff --git a/koyeb/sandbox/filesystem.py b/koyeb/sandbox/filesystem.py index 74806873..1485d91a 100644 --- a/koyeb/sandbox/filesystem.py +++ b/koyeb/sandbox/filesystem.py @@ -346,7 +346,9 @@ def upload_file( with open(local_path, "rb") as f: content_bytes = f.read() - SandboxFilesystem.write_file(self, remote_path, content_bytes, encoding=encoding) + SandboxFilesystem.write_file( + self, remote_path, content_bytes, encoding=encoding + ) def download_file( self, remote_path: str, local_path: str, encoding: str = "utf-8" diff --git a/koyeb/sandbox/sandbox.py b/koyeb/sandbox/sandbox.py index 7c67bbbd..e4c7812e 100644 --- a/koyeb/sandbox/sandbox.py +++ b/koyeb/sandbox/sandbox.py @@ -18,6 +18,9 @@ from koyeb.api.models.create_app import AppLifeCycle, CreateApp from koyeb.api.models.create_service import CreateService, ServiceLifeCycle from koyeb.api.models.deployment_status import DeploymentStatus +from koyeb.api.models.egress_policy import EgressPolicy +from koyeb.api.models.egress_policy_mode import EgressPolicyMode +from koyeb.api.models.network_policy import NetworkPolicy from koyeb.api.models.update_service import UpdateService from .executor_client import ConnectionInfo @@ -28,6 +31,7 @@ SandboxError, SandboxTimeoutError, build_config_files, + build_network_policy, build_env_vars, create_deployment_definition, create_docker_source, @@ -54,9 +58,9 @@ class ProcessInfo: pid: Optional[int] = None # OS process ID (if running) exit_code: Optional[int] = None # Exit code (if completed) started_at: Optional[str] = None # ISO 8601 timestamp when process started - completed_at: Optional[str] = ( - None # ISO 8601 timestamp when process completed (if applicable) - ) + completed_at: Optional[ + str + ] = None # ISO 8601 timestamp when process completed (if applicable) @dataclass @@ -137,6 +141,8 @@ def create( command: Optional[str] = None, args: Optional[List[str]] = None, host: Optional[str] = None, + block_network: bool = False, + outbound_allowlist: Optional[List[str]] = None, ) -> Sandbox: """ Create a new sandbox instance. @@ -178,6 +184,10 @@ def create( entrypoint: Override the default entrypoint of the Docker image (e.g., ["/bin/sh", "-c"]) command: Override the default command of the Docker image (e.g., "python app.py") host: Koyeb API host URL. If not provided, will try to get from KOYEB_API_HOST env var (defaults to https://app.koyeb.com) + block_network: If True, block all outbound network access from the sandbox + outbound_allowlist: List of IPs/CIDRs allowed as outbound destinations; + all other outbound traffic is blocked. Bare IPs are normalized to + /32 (IPv4) or /128 (IPv6). Mutually exclusive with block_network. Returns: Sandbox: A new Sandbox instance @@ -185,6 +195,8 @@ def create( Raises: ValueError: If API token is not provided SandboxTimeoutError: If wait_ready is True and sandbox does not become ready within timeout + EgressPolicyError: If both block_network and outbound_allowlist are passed, + or an allowlist entry is not a valid IP address or CIDR Example: >>> # Public image (default) @@ -228,6 +240,8 @@ def create( command=command, args=args, host=host, + block_network=block_network, + outbound_allowlist=outbound_allowlist, ) if wait_ready: @@ -268,11 +282,14 @@ def _create_sync( command: Optional[str] = None, args: Optional[List[str]] = None, host: Optional[str] = None, + block_network: bool = False, + outbound_allowlist: Optional[List[str]] = None, ) -> Sandbox: """ Synchronous creation method that returns creation parameters. Subclasses can override to return their own type. """ + network_policy = build_network_policy(block_network, outbound_allowlist) clients = get_api_clients(api_token, host) apps_api = clients.apps @@ -302,8 +319,12 @@ def _create_sync( env_vars = build_env_vars(env) config_file_objects = build_config_files(config_files) docker_source = create_docker_source( - image, privileged=privileged, image_registry_secret=registry_secret, - entrypoint=entrypoint, command=command, args=args, + image, + privileged=privileged, + image_registry_secret=registry_secret, + entrypoint=entrypoint, + command=command, + args=args, ) deployment_definition = create_deployment_definition( @@ -320,6 +341,7 @@ def _create_sync( _experimental_deep_sleep_value=_experimental_deep_sleep_value, enable_mesh=enable_mesh, config_files=config_file_objects if config_file_objects else None, + network_policy=network_policy, ) service_life_cycle = ServiceLifeCycle( @@ -1102,6 +1124,68 @@ def update_lifecycle( raise raise SandboxError(f"Failed to update life cycle: {str(e)}") from e + def update_network_policy( + self, + block_network: bool = False, + outbound_allowlist: Optional[List[str]] = None, + ) -> None: + """ + Update the sandbox's egress network policy. + + Warning: applying a new network policy triggers a redeployment of the + sandbox service. The sandbox is restarted and any in-memory or + non-persisted state is lost. This method does not wait for the + redeployment to finish. + + Args: + block_network: If True, block all outbound network access from the sandbox + outbound_allowlist: List of IPs/CIDRs allowed as outbound destinations; + all other outbound traffic is blocked. Bare IPs are normalized to + /32 (IPv4) or /128 (IPv6). Mutually exclusive with block_network. + + With both arguments unset, the egress policy is reset to the + platform default (unrestricted outbound access). + + Raises: + EgressPolicyError: If both block_network and outbound_allowlist are + passed, or an allowlist entry is not a valid IP address or CIDR + SandboxError: If updating the egress policy fails + + Example: + >>> sandbox.update_network_policy(block_network=True) + >>> sandbox.update_network_policy(outbound_allowlist=["10.0.0.0/8", "1.2.3.4"]) + >>> sandbox.update_network_policy() # reset to unrestricted + """ + network_policy = build_network_policy(block_network, outbound_allowlist) + if network_policy is None: + network_policy = NetworkPolicy( + egress=EgressPolicy(mode=EgressPolicyMode.EGRESS_POLICY_MODE_DEFAULT) + ) + try: + clients = get_api_clients(self.api_token) + services_api = clients.services + deployments_api = clients.deployments + service_response = services_api.get_service(self.service_id) + service = service_response.service + + if not service: + raise SandboxError("Sandbox service not found") + + deployment_response = deployments_api.get_deployment( + service.latest_deployment_id + ) + definition = deployment_response.deployment.definition + definition.network_policy = network_policy + + services_api.update_service( + id=self.service_id, + service=UpdateService(definition=definition), + ) + except Exception as e: + if isinstance(e, SandboxError): + raise + raise SandboxError(f"Failed to update egress policy: {str(e)}") + def __enter__(self) -> "Sandbox": """Context manager entry - returns self.""" return self @@ -1265,6 +1349,8 @@ async def create( command: Optional[str] = None, args: Optional[List[str]] = None, host: Optional[str] = None, + block_network: bool = False, + outbound_allowlist: Optional[List[str]] = None, ) -> AsyncSandbox: """ Create a new sandbox instance with async support. @@ -1308,6 +1394,10 @@ async def create( entrypoint: Override the default entrypoint of the Docker image (e.g., ["/bin/sh", "-c"]) command: Override the default command of the Docker image (e.g., "python app.py") host: Koyeb API host URL. If not provided, will try to get from KOYEB_API_HOST env var (defaults to https://app.koyeb.com) + block_network: If True, block all outbound network access from the sandbox + outbound_allowlist: List of IPs/CIDRs allowed as outbound destinations; + all other outbound traffic is blocked. Bare IPs are normalized to + /32 (IPv4) or /128 (IPv6). Mutually exclusive with block_network. Returns: AsyncSandbox: A new AsyncSandbox instance @@ -1315,6 +1405,8 @@ async def create( Raises: ValueError: If API token is not provided SandboxTimeoutError: If wait_ready is True and sandbox does not become ready within timeout + EgressPolicyError: If both block_network and outbound_allowlist are passed, + or an allowlist entry is not a valid IP address or CIDR """ if api_token is None: api_token = os.getenv("KOYEB_API_TOKEN") @@ -1323,12 +1415,16 @@ async def create( "API token is required. Set KOYEB_API_TOKEN environment variable or pass api_token parameter" ) - from .utils import get_async_api_clients + from .utils import get_async_api_clients, build_network_policy from koyeb.api_async.models.create_app import CreateApp as AsyncCreateApp from koyeb.api_async.models.create_app import AppLifeCycle as AsyncAppLifeCycle from koyeb.api_async.models.create_service import CreateService as AsyncCreateService from koyeb.api_async.models.create_service import ServiceLifeCycle as AsyncServiceLifeCycle + # Build network policy from egress policy parameters. Validate before + # any API call so invalid input fails fast without orphaning an app. + network_policy = build_network_policy(block_network, outbound_allowlist) + clients = get_async_api_clients(api_token, host) # Always create routes @@ -1373,6 +1469,7 @@ async def create( _experimental_deep_sleep_value=_experimental_deep_sleep_value, enable_mesh=enable_mesh, config_files=config_file_objects if config_file_objects else None, + network_policy=network_policy, ) service_life_cycle = AsyncServiceLifeCycle( @@ -1724,6 +1821,66 @@ async def update_lifecycle( raise raise SandboxError(f"Failed to update life cycle: {str(e)}") from e + async def update_network_policy( + self, + block_network: bool = False, + outbound_allowlist: Optional[List[str]] = None, + ) -> None: + """Update the sandbox's egress network policy asynchronously. + + Warning: applying a new network policy triggers a redeployment of the + sandbox service. The sandbox is restarted and any in-memory or + non-persisted state is lost. This method does not wait for the + redeployment to finish. + + See Sandbox.update_network_policy for full documentation. + + Raises: + EgressPolicyError: If both block_network and outbound_allowlist are + passed, or an allowlist entry is not a valid IP address or CIDR + SandboxError: If updating the egress policy fails + """ + from .utils import get_async_api_clients, build_network_policy + from koyeb.api_async.models.network_policy import NetworkPolicy as AsyncNetworkPolicy + from koyeb.api_async.models.egress_policy import EgressPolicy as AsyncEgressPolicy + from koyeb.api_async.models.egress_policy_mode import ( + EgressPolicyMode as AsyncEgressPolicyMode, + ) + from koyeb.api_async.models.update_service import UpdateService as AsyncUpdateService + + sync_policy = build_network_policy(block_network, outbound_allowlist) + if sync_policy is None: + network_policy = AsyncNetworkPolicy( + egress=AsyncEgressPolicy( + mode=AsyncEgressPolicyMode.EGRESS_POLICY_MODE_DEFAULT + ) + ) + else: + network_policy = AsyncNetworkPolicy.from_dict(sync_policy.to_dict()) + + try: + clients = get_async_api_clients(self.api_token, self.host) + service_response = await clients.services.get_service(self.service_id) + service = service_response.service + + if not service: + raise SandboxError("Sandbox service not found") + + deployment_response = await clients.deployments.get_deployment( + service.latest_deployment_id + ) + definition = deployment_response.deployment.definition + definition.network_policy = network_policy + + await clients.services.update_service( + id=self.service_id, + service=AsyncUpdateService(definition=definition), + ) + except Exception as e: + if isinstance(e, SandboxError): + raise + raise SandboxError(f"Failed to update egress policy: {str(e)}") from e + async def __aenter__(self) -> "AsyncSandbox": """Async context manager entry - returns self.""" return self diff --git a/koyeb/sandbox/test_egress_policy.py b/koyeb/sandbox/test_egress_policy.py new file mode 100644 index 00000000..4de25a32 --- /dev/null +++ b/koyeb/sandbox/test_egress_policy.py @@ -0,0 +1,185 @@ +import asyncio +import unittest +from unittest.mock import AsyncMock, MagicMock, patch + +from koyeb.api.models.deployment_definition import DeploymentDefinition +from koyeb.api.models.egress_policy_mode import EgressPolicyMode +from koyeb.api_async.models.deployment_definition import ( + DeploymentDefinition as AsyncDeploymentDefinition, +) +from koyeb.sandbox.sandbox import AsyncSandbox, Sandbox +from koyeb.sandbox.utils import EgressPolicyError + + +class TestCreateEgressWiring(unittest.TestCase): + """Tests that Sandbox.create wires egress kwargs into the deployment definition.""" + + def _create(self, mock_get_clients, **kwargs): + clients = MagicMock() + clients.apps.create_app.return_value.app.id = "mock-app-id" + mock_get_clients.return_value = clients + Sandbox.create(name="t", api_token="tok", wait_ready=False, **kwargs) + return clients.services.create_service.call_args.kwargs["service"] + + @patch("koyeb.sandbox.sandbox.get_api_clients") + def test_default_sends_no_network_policy(self, mock_get_clients): + service = self._create(mock_get_clients) + self.assertIsNone(service.definition.network_policy) + + @patch("koyeb.sandbox.sandbox.get_api_clients") + def test_block_network_sends_deny_all(self, mock_get_clients): + service = self._create(mock_get_clients, block_network=True) + egress = service.definition.network_policy.egress + self.assertEqual(egress.mode, EgressPolicyMode.EGRESS_POLICY_MODE_DENY_ALL) + self.assertIsNone(egress.allow_list) + + @patch("koyeb.sandbox.sandbox.get_api_clients") + def test_allowlist_sends_deny_all_with_destinations(self, mock_get_clients): + service = self._create( + mock_get_clients, outbound_allowlist=["1.2.3.4", "10.0.0.0/8"] + ) + egress = service.definition.network_policy.egress + self.assertEqual(egress.mode, EgressPolicyMode.EGRESS_POLICY_MODE_DENY_ALL) + self.assertEqual( + [d.cidr for d in egress.allow_list], ["1.2.3.4/32", "10.0.0.0/8"] + ) + + @patch("koyeb.sandbox.sandbox.get_api_clients") + def test_mutually_exclusive_fails_before_any_api_call(self, mock_get_clients): + with self.assertRaises(EgressPolicyError): + Sandbox.create( + name="t", + api_token="tok", + wait_ready=False, + block_network=True, + outbound_allowlist=["1.2.3.4"], + ) + mock_get_clients.assert_not_called() + + @patch("koyeb.sandbox.utils.get_async_api_clients") + def test_async_create_forwards_egress_kwargs(self, mock_get_clients): + clients = MagicMock() + clients.apps.create_app = AsyncMock() + clients.apps.create_app.return_value.app.id = "mock-app-id" + clients.services.create_service = AsyncMock() + mock_get_clients.return_value = clients + asyncio.run( + AsyncSandbox.create( + name="t", api_token="tok", wait_ready=False, block_network=True + ) + ) + service = clients.services.create_service.call_args.kwargs["service"] + egress = service.definition.network_policy.egress + self.assertEqual(egress.mode, EgressPolicyMode.EGRESS_POLICY_MODE_DENY_ALL) + + @patch("koyeb.sandbox.utils.get_async_api_clients") + def test_async_mutually_exclusive_fails_before_any_api_call(self, mock_get_clients): + with self.assertRaises(EgressPolicyError): + asyncio.run( + AsyncSandbox.create( + name="t", + api_token="tok", + wait_ready=False, + block_network=True, + outbound_allowlist=["1.2.3.4"], + ) + ) + mock_get_clients.assert_not_called() + + +def _make_clients(): + clients = MagicMock() + clients.deployments.get_deployment.return_value.deployment.definition = ( + DeploymentDefinition(name="sb") + ) + return clients + + +def _make_sandbox(cls=Sandbox): + return cls( + sandbox_id="sb", + app_id="app-id", + service_id="svc-id", + name="sb", + api_token="tok", + sandbox_secret="secret", + ) + + +class TestUpdateNetworkPolicy(unittest.TestCase): + """Tests for Sandbox.update_network_policy.""" + + def _update(self, mock_get_clients, **kwargs): + clients = _make_clients() + mock_get_clients.return_value = clients + _make_sandbox().update_network_policy(**kwargs) + call = clients.services.update_service.call_args + self.assertEqual(call.kwargs["id"], "svc-id") + return call.kwargs["service"] + + @patch("koyeb.sandbox.sandbox.get_api_clients") + def test_block_network(self, mock_get_clients): + update = self._update(mock_get_clients, block_network=True) + egress = update.definition.network_policy.egress + self.assertEqual(egress.mode, EgressPolicyMode.EGRESS_POLICY_MODE_DENY_ALL) + self.assertIsNone(egress.allow_list) + + @patch("koyeb.sandbox.sandbox.get_api_clients") + def test_allowlist(self, mock_get_clients): + update = self._update(mock_get_clients, outbound_allowlist=["2001:db8::1"]) + egress = update.definition.network_policy.egress + self.assertEqual(egress.mode, EgressPolicyMode.EGRESS_POLICY_MODE_DENY_ALL) + self.assertEqual([d.cidr for d in egress.allow_list], ["2001:db8::1/128"]) + + @patch("koyeb.sandbox.sandbox.get_api_clients") + def test_no_args_resets_to_default(self, mock_get_clients): + update = self._update(mock_get_clients) + egress = update.definition.network_policy.egress + self.assertEqual(egress.mode, EgressPolicyMode.EGRESS_POLICY_MODE_DEFAULT) + self.assertIsNone(egress.allow_list) + + @patch("koyeb.sandbox.sandbox.get_api_clients") + def test_mutually_exclusive_fails_before_any_api_call(self, mock_get_clients): + with self.assertRaises(EgressPolicyError): + _make_sandbox().update_network_policy( + block_network=True, outbound_allowlist=["1.2.3.4"] + ) + mock_get_clients.assert_not_called() + + @patch("koyeb.sandbox.sandbox.get_api_clients") + def test_api_failure_wrapped_in_sandbox_error(self, mock_get_clients): + from koyeb.sandbox.utils import SandboxError + + clients = _make_clients() + clients.services.update_service.side_effect = RuntimeError("boom") + mock_get_clients.return_value = clients + with self.assertRaises(SandboxError): + _make_sandbox().update_network_policy(block_network=True) + + +def _make_async_clients(): + clients = MagicMock() + clients.services.get_service = AsyncMock() + clients.services.get_service.return_value.service.latest_deployment_id = "dep-id" + clients.deployments.get_deployment = AsyncMock() + clients.deployments.get_deployment.return_value.deployment.definition = ( + AsyncDeploymentDefinition(name="sb") + ) + clients.services.update_service = AsyncMock() + return clients + + +class TestAsyncUpdateNetworkPolicy(unittest.TestCase): + """Tests for AsyncSandbox.update_network_policy.""" + + @patch("koyeb.sandbox.utils.get_async_api_clients") + def test_async_block_network(self, mock_get_clients): + clients = _make_async_clients() + mock_get_clients.return_value = clients + sandbox = _make_sandbox(AsyncSandbox) + asyncio.run(sandbox.update_network_policy(block_network=True)) + update = clients.services.update_service.call_args.kwargs["service"] + self.assertEqual( + update.definition.network_policy.egress.mode, + EgressPolicyMode.EGRESS_POLICY_MODE_DENY_ALL, + ) diff --git a/koyeb/sandbox/test_utils.py b/koyeb/sandbox/test_utils.py index 872b4fba..a076a240 100644 --- a/koyeb/sandbox/test_utils.py +++ b/koyeb/sandbox/test_utils.py @@ -1,6 +1,11 @@ import unittest -from koyeb.sandbox.utils import create_docker_source +from koyeb.api.models.egress_policy_mode import EgressPolicyMode +from koyeb.sandbox.utils import ( + EgressPolicyError, + build_network_policy, + create_docker_source, +) class TestCreateDockerSource(unittest.TestCase): @@ -60,5 +65,78 @@ def test_privileged_and_registry_secret_still_work(self): self.assertEqual(ds.command, "serve") +class TestBuildEgressPolicy(unittest.TestCase): + """Tests for build_network_policy validation and normalization.""" + + def test_both_unset_returns_none(self): + self.assertIsNone(build_network_policy(False, None)) + self.assertIsNone(build_network_policy()) + + def test_both_passed_raises(self): + with self.assertRaises(EgressPolicyError): + build_network_policy(True, ["1.2.3.4"]) + with self.assertRaises(EgressPolicyError): + build_network_policy(True, []) + + def test_block_network_deny_all_without_allowlist(self): + policy = build_network_policy(True, None) + self.assertEqual( + policy.egress.mode, EgressPolicyMode.EGRESS_POLICY_MODE_DENY_ALL + ) + self.assertIsNone(policy.egress.allow_list) + + def test_bare_ipv4_normalized_to_slash_32(self): + policy = build_network_policy(False, ["1.2.3.4"]) + self.assertEqual( + policy.egress.mode, EgressPolicyMode.EGRESS_POLICY_MODE_DENY_ALL + ) + self.assertEqual([d.cidr for d in policy.egress.allow_list], ["1.2.3.4/32"]) + + def test_bare_ipv6_normalized_to_slash_128(self): + policy = build_network_policy(False, ["2001:db8::1"]) + self.assertEqual( + [d.cidr for d in policy.egress.allow_list], ["2001:db8::1/128"] + ) + + def test_cidr_passthrough(self): + policy = build_network_policy(False, ["10.0.0.0/8", "2001:db8::/32"]) + self.assertEqual( + [d.cidr for d in policy.egress.allow_list], + ["10.0.0.0/8", "2001:db8::/32"], + ) + + def test_cidr_host_bits_normalized(self): + policy = build_network_policy(False, ["10.0.0.1/8"]) + self.assertEqual([d.cidr for d in policy.egress.allow_list], ["10.0.0.0/8"]) + + def test_invalid_entries_raise(self): + for bad in [ + "example.com", + "not-an-ip", + "", + " ", + "10.0.0.0/33", + "1.2.3", + "fe80::1%eth0", + ]: + with self.assertRaises(EgressPolicyError, msg=f"entry: {bad!r}"): + build_network_policy(False, [bad]) + + def test_empty_allowlist_means_deny_all(self): + policy = build_network_policy(False, []) + self.assertEqual( + policy.egress.mode, EgressPolicyMode.EGRESS_POLICY_MODE_DENY_ALL + ) + self.assertEqual(policy.egress.allow_list, []) + + def test_error_is_sandbox_error_and_exported(self): + import koyeb.sandbox as sandbox_pkg + from koyeb.sandbox.utils import SandboxError + + self.assertTrue(issubclass(EgressPolicyError, SandboxError)) + self.assertIs(sandbox_pkg.EgressPolicyError, EgressPolicyError) + self.assertIn("EgressPolicyError", sandbox_pkg.__all__) + + if __name__ == "__main__": unittest.main() diff --git a/koyeb/sandbox/utils.py b/koyeb/sandbox/utils.py index 4482386f..4b0ac336 100644 --- a/koyeb/sandbox/utils.py +++ b/koyeb/sandbox/utils.py @@ -4,6 +4,8 @@ Utility functions for Koyeb Sandbox """ +import asyncio +import ipaddress import logging import os import shlex @@ -35,6 +37,10 @@ ) from koyeb.api.models.deployment_mesh import DeploymentMesh from koyeb.api.models.docker_source import DockerSource +from koyeb.api.models.egress_policy import EgressPolicy +from koyeb.api.models.egress_policy_mode import EgressPolicyMode +from koyeb.api.models.network_policy import NetworkPolicy +from koyeb.api.models.network_policy_destination import NetworkPolicyDestination from koyeb.api.models.proxy_port_protocol import ProxyPortProtocol # Setup logging @@ -407,6 +413,7 @@ def create_deployment_definition( _experimental_deep_sleep_value: int = 3900, enable_mesh: bool = None, config_files: Optional[List[ConfigFile]] = None, + network_policy: Optional[NetworkPolicy] = None, ) -> DeploymentDefinition: """ Create deployment definition for a sandbox service. @@ -428,6 +435,7 @@ def create_deployment_definition( _experimental_deep_sleep_value: Number of seconds for deep sleep when light sleep is enabled (default: 3900). Only used if _experimental_enable_light_sleep is True. Ignored otherwise. enable_mesh: Enable or disable mesh for this sandbox. Disabled by default + network_policy: Optional network policy restricting egress traffic Returns: DeploymentDefinition object @@ -500,6 +508,83 @@ def create_deployment_definition( regions=regions_list, mesh=mesh, config_files=config_files if config_files else None, + network_policy=network_policy, + ) + + +def _normalize_destination(entry: str) -> str: + """ + Normalize an outbound allowlist entry to a CIDR string. + + Bare IPv4 addresses become /32, bare IPv6 addresses become /128, + CIDR notation is validated and passed through (host bits are cleared, + e.g. "10.0.0.1/8" becomes "10.0.0.0/8"). + + Raises: + EgressPolicyError: If the entry is not a valid IP address or CIDR + """ + value = entry.strip() if isinstance(entry, str) else "" + if not value: + raise EgressPolicyError(f"Invalid outbound_allowlist entry: {entry!r}") + if "%" in value: + raise EgressPolicyError( + f"Invalid outbound_allowlist entry {entry!r}: " + "expected an IP address or CIDR (scoped/zone-ID addresses are not allowed)" + ) + try: + if "/" in value: + return str(ipaddress.ip_network(value, strict=False)) + address = ipaddress.ip_address(value) + except ValueError as e: + raise EgressPolicyError( + f"Invalid outbound_allowlist entry {entry!r}: " + "expected an IP address or CIDR" + ) from e + prefix = 32 if address.version == 4 else 128 + return f"{address}/{prefix}" + + +def build_network_policy( + block_network: bool = False, + outbound_allowlist: Optional[List[str]] = None, +) -> Optional[NetworkPolicy]: + """ + Build a NetworkPolicy from sandbox egress arguments. + + Args: + block_network: If True, block all outbound network access + outbound_allowlist: List of IPs/CIDRs allowed as outbound + destinations; all other outbound traffic is blocked. Bare IPs + are normalized to /32 (IPv4) or /128 (IPv6). An empty list + blocks all outbound traffic. + + Returns: + NetworkPolicy, or None when both arguments are unset + (block_network=False and outbound_allowlist=None) + + Raises: + EgressPolicyError: If both arguments are passed, or an allowlist + entry is not a valid IP address or CIDR + """ + if block_network and outbound_allowlist is not None: + raise EgressPolicyError( + "block_network and outbound_allowlist are mutually exclusive; " + "pass at most one" + ) + if not block_network and outbound_allowlist is None: + return None + + allow_list = None + if outbound_allowlist is not None: + allow_list = [ + NetworkPolicyDestination(cidr=_normalize_destination(entry)) + for entry in outbound_allowlist + ] + return NetworkPolicy( + egress=EgressPolicy( + mode=EgressPolicyMode.EGRESS_POLICY_MODE_DENY_ALL, + allow_list=allow_list, + ) ) @@ -553,7 +638,7 @@ def check_error_message(error_msg: str, error_type: str) -> bool: def create_sandbox_client( - conn_info: Optional['ConnectionInfo'], + conn_info: Optional["ConnectionInfo"], existing_client: Optional[Any] = None, ) -> Any: """ @@ -636,3 +721,7 @@ class SandboxServiceError(SandboxError): def __init__(self, status_code: int, message: str): self.status_code = status_code super().__init__(f"Sandbox service error ({status_code}): {message}") + + +class EgressPolicyError(SandboxError): + """Raised when egress policy arguments are invalid or conflicting""" diff --git a/uv.lock b/uv.lock index f28089cf..38bb34be 100644 --- a/uv.lock +++ b/uv.lock @@ -7,7 +7,7 @@ resolution-markers = [ ] [options] -exclude-newer = "2026-07-04T09:58:40.887231Z" +exclude-newer = "2026-07-05T19:43:52.954684Z" exclude-newer-span = "P2D" [[package]] @@ -38,7 +38,7 @@ wheels = [ [[package]] name = "anyio" -version = "4.13.0" +version = "4.14.1" source = { registry = "https://pypi.org/simple" } resolution-markers = [ "python_full_version >= '3.10'", @@ -48,9 +48,9 @@ dependencies = [ { name = "idna", marker = "python_full_version >= '3.10'" }, { name = "typing-extensions", marker = "python_full_version >= '3.10' and python_full_version < '3.13'" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/19/14/2c5dd9f512b66549ae92767a9c7b330ae88e1932ca57876909410251fe13/anyio-4.13.0.tar.gz", hash = "sha256:334b70e641fd2221c1505b3890c69882fe4a2df910cba14d97019b90b24439dc", size = 231622, upload-time = "2026-03-24T12:59:09.671Z" } +sdist = { url = "https://files.pythonhosted.org/packages/3b/72/5562aabb8dd7181e8e860622a38bea08d17842b99ecd4c91f84ac95251b0/anyio-4.14.1.tar.gz", hash = "sha256:8d648a3544c1a700e3ff78615cd679e4c5c3f149904287e73687b2596963629e", size = 254831, upload-time = "2026-06-24T20:56:06.017Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/da/42/e921fccf5015463e32a3cf6ee7f980a6ed0f395ceeaa45060b61d86486c2/anyio-4.13.0-py3-none-any.whl", hash = "sha256:08b310f9e24a9594186fd75b4f73f4a4152069e3853f1ed8bfbf58369f4ad708", size = 114353, upload-time = "2026-03-24T12:59:08.246Z" }, + { url = "https://files.pythonhosted.org/packages/b0/7b/90df4a0a816d98d6ea26f559d87836d494a2cf1fcf063be67df50a7bcc30/anyio-4.14.1-py3-none-any.whl", hash = "sha256:4e5533c5b8ff0a24f5d7a176cbe6877129cd183893f66b537f8f227d10527d72", size = 124875, upload-time = "2026-06-24T20:56:04.413Z" }, ] [[package]] @@ -411,7 +411,7 @@ version = "0.28.1" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "anyio", version = "4.12.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.10'" }, - { name = "anyio", version = "4.13.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, + { name = "anyio", version = "4.14.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, { name = "certifi" }, { name = "httpcore" }, { name = "idna" }, From 8661e1ea4f3577642a211285a617d93c16566661 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=81ukasz=20Ole=C5=9B?= Date: Wed, 8 Jul 2026 15:03:35 +0200 Subject: [PATCH 2/2] Rename egress wording to network policy in comments and docs Generated by Mistral Vibe. Co-Authored-By: Mistral Vibe --- docs/sandbox.md | 12 ++++++------ koyeb/sandbox/sandbox.py | 16 ++++++++-------- koyeb/sandbox/utils.py | 2 +- 3 files changed, 15 insertions(+), 15 deletions(-) diff --git a/docs/sandbox.md b/docs/sandbox.md index 0ee6e5b3..d5d1a9bc 100644 --- a/docs/sandbox.md +++ b/docs/sandbox.md @@ -1544,7 +1544,7 @@ def update_network_policy( outbound_allowlist: Optional[List[str]] = None) -> None ``` -Update the sandbox's egress network policy. +Update the sandbox's network policy. Warning: applying a new network policy triggers a redeployment of the sandbox service. The sandbox is restarted and any in-memory or @@ -1558,7 +1558,7 @@ redeployment to finish. all other outbound traffic is blocked. Bare IPs are normalized to /32 (IPv4) or /128 (IPv6). Mutually exclusive with block_network. - With both arguments unset, the egress policy is reset to the + With both arguments unset, the network policy is reset to the platform default (unrestricted outbound access). @@ -1566,7 +1566,7 @@ redeployment to finish. - `EgressPolicyError` - If both block_network and outbound_allowlist are passed, or an allowlist entry is not a valid IP address or CIDR -- `SandboxError` - If updating the egress policy fails +- `SandboxError` - If updating the network policy fails **Example**: @@ -1910,7 +1910,7 @@ async def update_network_policy( outbound_allowlist: Optional[List[str]] = None) -> None ``` -Update the sandbox's egress network policy asynchronously. +Update the sandbox's network policy asynchronously. Warning: applying a new network policy triggers a redeployment of the sandbox service. The sandbox is restarted and any in-memory or @@ -1923,7 +1923,7 @@ See Sandbox.update_network_policy for full documentation. - `EgressPolicyError` - If both block_network and outbound_allowlist are passed, or an allowlist entry is not a valid IP address or CIDR -- `SandboxError` - If updating the egress policy fails +- `SandboxError` - If updating the network policy fails @@ -2247,7 +2247,7 @@ def build_network_policy( ) -> Optional[NetworkPolicy] ``` -Build a NetworkPolicy from sandbox egress arguments. +Build a NetworkPolicy from sandbox network policy arguments. **Arguments**: diff --git a/koyeb/sandbox/sandbox.py b/koyeb/sandbox/sandbox.py index e4c7812e..d3b9c6f9 100644 --- a/koyeb/sandbox/sandbox.py +++ b/koyeb/sandbox/sandbox.py @@ -1130,7 +1130,7 @@ def update_network_policy( outbound_allowlist: Optional[List[str]] = None, ) -> None: """ - Update the sandbox's egress network policy. + Update the sandbox's network policy. Warning: applying a new network policy triggers a redeployment of the sandbox service. The sandbox is restarted and any in-memory or @@ -1143,13 +1143,13 @@ def update_network_policy( all other outbound traffic is blocked. Bare IPs are normalized to /32 (IPv4) or /128 (IPv6). Mutually exclusive with block_network. - With both arguments unset, the egress policy is reset to the + With both arguments unset, the network policy is reset to the platform default (unrestricted outbound access). Raises: EgressPolicyError: If both block_network and outbound_allowlist are passed, or an allowlist entry is not a valid IP address or CIDR - SandboxError: If updating the egress policy fails + SandboxError: If updating the network policy fails Example: >>> sandbox.update_network_policy(block_network=True) @@ -1184,7 +1184,7 @@ def update_network_policy( except Exception as e: if isinstance(e, SandboxError): raise - raise SandboxError(f"Failed to update egress policy: {str(e)}") + raise SandboxError(f"Failed to update network policy: {str(e)}") def __enter__(self) -> "Sandbox": """Context manager entry - returns self.""" @@ -1421,7 +1421,7 @@ async def create( from koyeb.api_async.models.create_service import CreateService as AsyncCreateService from koyeb.api_async.models.create_service import ServiceLifeCycle as AsyncServiceLifeCycle - # Build network policy from egress policy parameters. Validate before + # Build the network policy from the provided parameters. Validate before # any API call so invalid input fails fast without orphaning an app. network_policy = build_network_policy(block_network, outbound_allowlist) @@ -1826,7 +1826,7 @@ async def update_network_policy( block_network: bool = False, outbound_allowlist: Optional[List[str]] = None, ) -> None: - """Update the sandbox's egress network policy asynchronously. + """Update the sandbox's network policy asynchronously. Warning: applying a new network policy triggers a redeployment of the sandbox service. The sandbox is restarted and any in-memory or @@ -1838,7 +1838,7 @@ async def update_network_policy( Raises: EgressPolicyError: If both block_network and outbound_allowlist are passed, or an allowlist entry is not a valid IP address or CIDR - SandboxError: If updating the egress policy fails + SandboxError: If updating the network policy fails """ from .utils import get_async_api_clients, build_network_policy from koyeb.api_async.models.network_policy import NetworkPolicy as AsyncNetworkPolicy @@ -1879,7 +1879,7 @@ async def update_network_policy( except Exception as e: if isinstance(e, SandboxError): raise - raise SandboxError(f"Failed to update egress policy: {str(e)}") from e + raise SandboxError(f"Failed to update network policy: {str(e)}") from e async def __aenter__(self) -> "AsyncSandbox": """Async context manager entry - returns self.""" diff --git a/koyeb/sandbox/utils.py b/koyeb/sandbox/utils.py index 4b0ac336..b8483989 100644 --- a/koyeb/sandbox/utils.py +++ b/koyeb/sandbox/utils.py @@ -549,7 +549,7 @@ def build_network_policy( outbound_allowlist: Optional[List[str]] = None, ) -> Optional[NetworkPolicy]: """ - Build a NetworkPolicy from sandbox egress arguments. + Build a NetworkPolicy from sandbox network policy arguments. Args: block_network: If True, block all outbound network access