diff --git a/examples/21_port_security_policy.py b/examples/21_port_security_policy.py new file mode 100644 index 00000000..b4c6e7e6 --- /dev/null +++ b/examples/21_port_security_policy.py @@ -0,0 +1,134 @@ +#!/usr/bin/env python3 +"""Verify exposed-port security policies: no auth, API key, and Basic Auth. + +Follows the same pattern as example 14 (expose_port) and adds a security +policy on the user-facing route so that unauthenticated requests are rejected. +""" + +import os +import sys +import time +import random +import string + +import httpx + +from koyeb import Sandbox +from koyeb.sandbox import ApiKey, BasicAuth + +API_KEY = "my-secret-api-key" +BA_USER = "admin" +BA_PASS = "s3cr3t" + + +def _suffix() -> str: + return "".join(random.choices(string.ascii_lowercase + string.digits, k=8)) + + +def _setup_server(sandbox: Sandbox) -> str: + """Write a test file, start an HTTP server on 8080, expose it, return the base URL.""" + sandbox.filesystem.write_file("/tmp/index.html", "

ok

") + sandbox.launch_process("python3 -m http.server 8080", cwd="/tmp") + time.sleep(3) + + exposed = sandbox.expose_port(8080) + print(f" Exposed at: {exposed.exposed_at}") + time.sleep(2) + return exposed.exposed_at + + +def demo_no_auth(api_token: str) -> None: + print("\n=== No security policy (public) ===") + sandbox = None + try: + sandbox = Sandbox.create( + image="koyeb/sandbox:slim", + name=f"sec-noauth-{_suffix()}", + api_token=api_token, + ) + base = _setup_server(sandbox) + + resp = httpx.get(f"{base}/index.html", timeout=15) + print(f" GET /index.html → {resp.status_code}") + assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" + print(" ✓ Publicly accessible as expected") + finally: + if sandbox: + sandbox.delete() + + +def demo_api_key(api_token: str) -> None: + print("\n=== API key policy ===") + sandbox = None + try: + sandbox = Sandbox.create( + image="koyeb/sandbox:slim", + name=f"sec-apikey-{_suffix()}", + api_token=api_token, + exposed_port_security_policy=ApiKey(API_KEY), + ) + base = _setup_server(sandbox) + + # Without key → rejected + resp = httpx.get(f"{base}/index.html", timeout=15) + print(f" GET (no key) → {resp.status_code}") + assert resp.status_code in (401, 403), f"Expected 401/403, got {resp.status_code}" + print(" ✓ Rejected without key") + + # With correct key → accepted + resp = httpx.get( + f"{base}/index.html", + headers={"x-api-key": f"{API_KEY}"}, + timeout=15, + ) + print(f" GET (Bearer {API_KEY}) → {resp.status_code}") + assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" + print(" ✓ Accepted with correct API key") + finally: + if sandbox: + sandbox.delete() + + +def demo_basic_auth(api_token: str) -> None: + print("\n=== Basic Auth policy ===") + sandbox = None + try: + sandbox = Sandbox.create( + image="koyeb/sandbox:slim", + name=f"sec-basicauth-{_suffix()}", + api_token=api_token, + exposed_port_security_policy=BasicAuth(username=BA_USER, password=BA_PASS), + ) + base = _setup_server(sandbox) + + # Without credentials → rejected + resp = httpx.get(f"{base}/index.html", timeout=15) + print(f" GET (no creds) → {resp.status_code}") + assert resp.status_code in (401, 403), f"Expected 401/403, got {resp.status_code}" + print(" ✓ Rejected without credentials") + + # With correct credentials → accepted + resp = httpx.get(f"{base}/index.html", auth=(BA_USER, BA_PASS), timeout=15) + print(f" GET ({BA_USER}:{BA_PASS}) → {resp.status_code}") + assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" + print(" ✓ Accepted with correct Basic Auth credentials") + finally: + if sandbox: + sandbox.delete() + + +def main() -> int: + api_token = os.getenv("KOYEB_API_TOKEN") + if not api_token: + print("Error: KOYEB_API_TOKEN not set") + return 1 + + demo_no_auth(api_token) + demo_api_key(api_token) + demo_basic_auth(api_token) + print("\nAll assertions passed.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/examples/21_port_security_policy_async.py b/examples/21_port_security_policy_async.py new file mode 100644 index 00000000..15052fee --- /dev/null +++ b/examples/21_port_security_policy_async.py @@ -0,0 +1,141 @@ +#!/usr/bin/env python3 +"""Verify exposed-port security policies: no auth, API key, and Basic Auth (async variant). + +Follows the same pattern as example 14 (expose_port) and adds a security +policy on the user-facing route so that unauthenticated requests are rejected. +""" + +import asyncio +import os +import sys +import random +import string + +import httpx + +from koyeb import AsyncSandbox +from koyeb.sandbox import ApiKey, BasicAuth + +API_KEY = "my-secret-api-key" +BA_USER = "admin" +BA_PASS = "s3cr3t" + + +def _suffix() -> str: + return "".join(random.choices(string.ascii_lowercase + string.digits, k=8)) + + +async def _setup_server(sandbox: AsyncSandbox) -> str: + """Write a test file, start an HTTP server on 8080, expose it, return the base URL.""" + await sandbox.filesystem.write_file("/tmp/index.html", "

ok

") + await sandbox.launch_process("python3 -m http.server 8080", cwd="/tmp") + await asyncio.sleep(3) + + exposed = await sandbox.expose_port(8080) + print(f" Exposed at: {exposed.exposed_at}") + await asyncio.sleep(2) + return exposed.exposed_at + + +async def demo_no_auth(api_token: str) -> None: + print("\n=== No security policy (public) ===") + sandbox = None + try: + sandbox = await AsyncSandbox.create( + image="koyeb/sandbox:slim", + name=f"sec-noauth-{_suffix()}", + api_token=api_token, + ) + base = await _setup_server(sandbox) + + async with httpx.AsyncClient() as client: + resp = await client.get(f"{base}/index.html", timeout=15) + print(f" GET /index.html → {resp.status_code}") + assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" + print(" ✓ Publicly accessible as expected") + finally: + if sandbox: + await sandbox.delete() + + +async def demo_api_key(api_token: str) -> None: + print("\n=== API key policy ===") + sandbox = None + try: + sandbox = await AsyncSandbox.create( + image="koyeb/sandbox:slim", + name=f"sec-apikey-{_suffix()}", + api_token=api_token, + exposed_port_security_policy=ApiKey(API_KEY), + ) + base = await _setup_server(sandbox) + + async with httpx.AsyncClient() as client: + # Without key → rejected + resp = await client.get(f"{base}/index.html", timeout=15) + print(f" GET (no key) → {resp.status_code}") + assert resp.status_code in (401, 403), f"Expected 401/403, got {resp.status_code}" + print(" ✓ Rejected without key") + + # With correct key → accepted + resp = await client.get( + f"{base}/index.html", + headers={"x-api-key": API_KEY}, + timeout=15, + ) + print(f" GET (x-api-key: {API_KEY}) → {resp.status_code}") + assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" + print(" ✓ Accepted with correct API key") + finally: + if sandbox: + await sandbox.delete() + + +async def demo_basic_auth(api_token: str) -> None: + print("\n=== Basic Auth policy ===") + sandbox = None + try: + sandbox = await AsyncSandbox.create( + image="koyeb/sandbox:slim", + name=f"sec-basicauth-{_suffix()}", + api_token=api_token, + exposed_port_security_policy=BasicAuth(username=BA_USER, password=BA_PASS), + ) + base = await _setup_server(sandbox) + + async with httpx.AsyncClient() as client: + # Without credentials → rejected + resp = await client.get(f"{base}/index.html", timeout=15) + print(f" GET (no creds) → {resp.status_code}") + assert resp.status_code in (401, 403), f"Expected 401/403, got {resp.status_code}" + print(" ✓ Rejected without credentials") + + # With correct credentials → accepted + resp = await client.get( + f"{base}/index.html", + auth=(BA_USER, BA_PASS), + timeout=15, + ) + print(f" GET ({BA_USER}:{BA_PASS}) → {resp.status_code}") + assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" + print(" ✓ Accepted with correct Basic Auth credentials") + finally: + if sandbox: + await sandbox.delete() + + +async def main() -> int: + api_token = os.getenv("KOYEB_API_TOKEN") + if not api_token: + print("Error: KOYEB_API_TOKEN not set") + return 1 + + await demo_no_auth(api_token) + await demo_api_key(api_token) + await demo_basic_auth(api_token) + print("\nAll assertions passed.") + return 0 + + +if __name__ == "__main__": + sys.exit(asyncio.run(main())) diff --git a/koyeb/sandbox/__init__.py b/koyeb/sandbox/__init__.py index f1043801..010be1d5 100644 --- a/koyeb/sandbox/__init__.py +++ b/koyeb/sandbox/__init__.py @@ -21,6 +21,8 @@ from .sandbox import AsyncSandbox, ExposedPort, ProcessInfo, Sandbox from .snapshot import DeclarativeSnapshot, Snapshot, SnapshotStatus, SnapshotType from .utils import ( + ApiKey, + BasicAuth, EgressPolicyError, SandboxDeploymentError, SandboxError, @@ -31,6 +33,8 @@ __all__ = [ "Sandbox", "AsyncSandbox", + "ApiKey", + "BasicAuth", "ConfigFile", "Secret", "SandboxFilesystem", diff --git a/koyeb/sandbox/sandbox.py b/koyeb/sandbox/sandbox.py index b858e260..369fec10 100644 --- a/koyeb/sandbox/sandbox.py +++ b/koyeb/sandbox/sandbox.py @@ -25,6 +25,8 @@ from .executor_client import ConnectionInfo from .utils import ( + ApiKey, + BasicAuth, DEFAULT_INSTANCE_WAIT_TIMEOUT, DEFAULT_POLL_INTERVAL, SandboxDeploymentError, @@ -35,7 +37,6 @@ build_env_vars, create_deployment_definition, create_docker_source, - create_koyeb_sandbox_routes, create_sandbox_client, get_api_clients, logger, @@ -124,6 +125,7 @@ def create( wait_ready: bool = True, instance_type: str = "micro", exposed_port_protocol: Optional[str] = None, + exposed_port_security_policy: Optional[Union[ApiKey, BasicAuth]] = None, env: Optional[Dict[str, Any]] = None, config_files: Optional[Dict[str, Any]] = None, region: Optional[str] = None, @@ -160,6 +162,11 @@ def create( exposed_port_protocol: Protocol to expose ports with ("http" or "http2"). If None, defaults to "http". If provided, must be one of "http" or "http2". + exposed_port_security_policy: Optional access control for the user-facing port (3031). + Pass ``ApiKey("my-key")`` to require an API key via + ``x-api-key: ``, or + ``BasicAuth(username="u", password="p")`` to require HTTP Basic Auth. + If None, the port is publicly accessible (default). env: Environment variables config_files: Config files to create in the sandbox, as a dictionary mapping file paths to file contents. Values can be plain strings (default permissions 0644) @@ -216,15 +223,23 @@ def create( ... image="ghcr.io/myorg/myimage:latest", ... registry_secret="my-ghcr-secret" ... ) - + + >>> # Protect the exposed port with an API key + >>> sandbox = Sandbox.create(exposed_port_security_policy=ApiKey("my-secret-key")) + + >>> # Protect the exposed port with Basic Auth + >>> sandbox = Sandbox.create( + ... exposed_port_security_policy=BasicAuth(username="admin", ******) + ... ) + >>> # Create from a Snapshot object >>> from koyeb.sandbox import Snapshot >>> snapshot = Snapshot.get("my-snapshot-id") >>> sandbox = Sandbox.create(snapshot=snapshot) - + >>> # Create from a snapshot ID string >>> sandbox = Sandbox.create(snapshot="my-snapshot-id") - + >>> # Create from a snapshot with custom parameters >>> sandbox = Sandbox.create( ... snapshot="my-snapshot-id", @@ -283,6 +298,7 @@ def create( image=image, instance_type=instance_type, exposed_port_protocol=exposed_port_protocol, + exposed_port_security_policy=exposed_port_security_policy, env=env, config_files=config_files, region=region, @@ -328,6 +344,7 @@ def _create_sync( image: str = "koyeb/sandbox", instance_type: str = "micro", exposed_port_protocol: Optional[str] = None, + exposed_port_security_policy: Optional[Union[ApiKey, BasicAuth]] = None, env: Optional[Dict[str, Any]] = None, config_files: Optional[Dict[str, Any]] = None, region: Optional[str] = None, @@ -364,9 +381,6 @@ def _create_sync( apps_api = clients.apps services_api = clients.services - # Always create routes (ports are always exposed, default to "http") - routes = create_koyeb_sandbox_routes() - # Generate secure sandbox secret if not provided if sandbox_secret is None: sandbox_secret = secrets.token_urlsafe(32) @@ -403,8 +417,8 @@ def _create_sync( env_vars=env_vars, instance_type=instance_type, exposed_port_protocol=exposed_port_protocol, + exposed_port_security_policy=exposed_port_security_policy, region=region, - routes=routes, idle_timeout=idle_timeout, enable_tcp_proxy=enable_tcp_proxy, _experimental_enable_light_sleep=_experimental_enable_light_sleep, @@ -418,31 +432,7 @@ def _create_sync( delete_after_create=delete_after_delay, delete_after_sleep=delete_after_inactivity_delay, ) - - # Build deployment definition - used for both snapshot and non-snapshot cases - 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, - ) - deployment_definition = create_deployment_definition( - name=name, - docker_source=docker_source, - env_vars=env_vars, - instance_type=instance_type, - exposed_port_protocol=exposed_port_protocol, - region=region, - routes=routes, - idle_timeout=idle_timeout, - enable_tcp_proxy=enable_tcp_proxy, - _experimental_enable_light_sleep=_experimental_enable_light_sleep, - _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, - ) - + # Handle snapshot creation based on snapshot type # For FULL snapshots, don't provide definition (API will infer it) # For FILESYSTEM snapshots, always provide definition with snapshot_id @@ -1634,6 +1624,7 @@ async def create( wait_ready: bool = True, instance_type: str = "micro", exposed_port_protocol: Optional[str] = None, + exposed_port_security_policy: Optional[Union[ApiKey, BasicAuth]] = None, env: Optional[Dict[str, Any]] = None, config_files: Optional[Dict[str, Any]] = None, region: Optional[str] = None, @@ -1670,6 +1661,11 @@ async def create( exposed_port_protocol: Protocol to expose ports with ("http" or "http2"). If None, defaults to "http". If provided, must be one of "http" or "http2". + exposed_port_security_policy: Optional access control for the user-facing port. + Pass ``ApiKey("my-key")`` to require an API key via + ``x-api-key: ``, or + ``BasicAuth(username="u", password="p")`` to require HTTP Basic Auth. + If None, the port is publicly accessible (default). env: Environment variables config_files: Config files to create in the sandbox, as a dictionary mapping file paths to file contents. Values can be plain strings (default permissions 0644) @@ -1776,9 +1772,6 @@ async def create( clients = get_async_api_clients(api_token, host) - # Always create routes - routes = create_koyeb_sandbox_routes() - # Generate secure sandbox secret if not provided if sandbox_secret is None: sandbox_secret = secrets.token_urlsafe(32) @@ -1805,30 +1798,11 @@ async def create( entrypoint=entrypoint, command=command, args=args, ) - deployment_definition = create_deployment_definition( - name=name, - docker_source=docker_source, - env_vars=env_vars, - instance_type=instance_type, - exposed_port_protocol=exposed_port_protocol, - region=region, - routes=routes, - idle_timeout=idle_timeout, - enable_tcp_proxy=enable_tcp_proxy, - _experimental_enable_light_sleep=_experimental_enable_light_sleep, - _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( delete_after_create=delete_after_delay, delete_after_sleep=delete_after_inactivity_delay, ) - # Convert sync DeploymentDefinition to dict so the async Pydantic model - # (which expects koyeb.api_async.models.DeploymentDefinition) can coerce it. - + # Handle snapshot creation based on snapshot type # For FULL snapshots, don't provide definition (API will infer it) # For FILESYSTEM snapshots, always provide definition with snapshot_id @@ -1846,20 +1820,14 @@ async def create( ) else: # For FILESYSTEM snapshots (or unknown), provide definition - 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, - ) deployment_definition = create_deployment_definition( name=name, docker_source=docker_source, env_vars=env_vars, instance_type=instance_type, exposed_port_protocol=exposed_port_protocol, + exposed_port_security_policy=exposed_port_security_policy, region=region, - routes=routes, idle_timeout=idle_timeout, enable_tcp_proxy=enable_tcp_proxy, _experimental_enable_light_sleep=_experimental_enable_light_sleep, @@ -1877,20 +1845,14 @@ async def create( ) else: # No snapshot, create normally with definition - 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, - ) deployment_definition = create_deployment_definition( name=name, docker_source=docker_source, env_vars=env_vars, instance_type=instance_type, exposed_port_protocol=exposed_port_protocol, + exposed_port_security_policy=exposed_port_security_policy, region=region, - routes=routes, idle_timeout=idle_timeout, enable_tcp_proxy=enable_tcp_proxy, _experimental_enable_light_sleep=_experimental_enable_light_sleep, diff --git a/koyeb/sandbox/utils.py b/koyeb/sandbox/utils.py index 525d126f..aaccae50 100644 --- a/koyeb/sandbox/utils.py +++ b/koyeb/sandbox/utils.py @@ -10,7 +10,7 @@ import os import shlex from dataclasses import dataclass -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Dict, List, Optional, Tuple, Union from koyeb.api import ApiClient, Configuration from koyeb.api.api import ( @@ -22,6 +22,7 @@ SecretsApi, ServicesApi, ) +from koyeb.api.models.basic_auth_policy import BasicAuthPolicy from koyeb.api.models.config_file import ConfigFile from koyeb.api.models.deployment_definition import DeploymentDefinition from koyeb.api.models.deployment_definition_type import DeploymentDefinitionType @@ -43,6 +44,7 @@ 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 +from koyeb.api.models.security_policies import SecurityPolicies # Setup logging logger = logging.getLogger(__name__) @@ -67,6 +69,43 @@ VALID_DEPLOYMENT_PORT_PROTOCOLS = ("http", "http2") +@dataclass +class ApiKey: + """An API key used to restrict access to the user-facing sandbox port.""" + + key: str + + +@dataclass +class BasicAuth: + """HTTP Basic Auth credentials used to restrict access to the user-facing sandbox port.""" + + username: str + password: str + + +def _build_security_policy( + policy, +) -> Optional[SecurityPolicies]: + """ + Build a SecurityPolicies object from an ApiKey or BasicAuth instance. + + Raises: + ValueError: If policy is not an ApiKey or BasicAuth. + """ + if policy is None: + return None + if isinstance(policy, ApiKey): + return SecurityPolicies(api_keys=[policy.key]) + if isinstance(policy, BasicAuth): + return SecurityPolicies( + basic_auths=[BasicAuthPolicy(username=policy.username, password=policy.password)] + ) + raise ValueError( + "exposed_port_security_policy must be an ApiKey or BasicAuth instance" + ) + + def _validate_port_protocol(protocol: str) -> str: """ Validate port protocol using API model structure. @@ -388,20 +427,25 @@ def create_koyeb_sandbox_proxy_ports() -> List[DeploymentProxyPort]: ] -def create_koyeb_sandbox_routes() -> List[DeploymentRoute]: +def create_koyeb_sandbox_routes( + security_policy: Optional[SecurityPolicies] = None, +) -> List[DeploymentRoute]: """ Create route configuration for koyeb/sandbox image to make it publicly accessible. Creates two routes: - - Port 3030 accessible at /koyeb-sandbox/ - - Port 3031 accessible at / + - Port 3030 accessible at /koyeb-sandbox/ (control plane, no security policy) + - Port 3031 accessible at / (user-facing, optional security policy) + + Args: + security_policy: Optional security policy to apply to the user-facing port 3031 route. Returns: List of DeploymentRoute objects configured for koyeb/sandbox """ return [ DeploymentRoute(port=3030, path="/koyeb-sandbox/"), - DeploymentRoute(port=3031, path="/"), + DeploymentRoute(port=3031, path="/", security_policies=security_policy), ] @@ -411,6 +455,7 @@ def create_deployment_definition( env_vars: List[DeploymentEnv], instance_type: str, exposed_port_protocol: Optional[str] = None, + exposed_port_security_policy: Optional[Union[ApiKey, BasicAuth]] = None, region: Optional[str] = None, routes: Optional[List[DeploymentRoute]] = None, idle_timeout: int = 300, @@ -432,6 +477,8 @@ def create_deployment_definition( exposed_port_protocol: Protocol to expose ports with ("http" or "http2"). If None, defaults to "http". If provided, must be one of "http" or "http2". + exposed_port_security_policy: Optional security policy for the user-facing port 3031 route. + Pass ApiKey("my-key") or BasicAuth(username="u", password="p"). region: Region to deploy to. Defaults to KOYEB_REGION env var, or "na" if not set. routes: List of routes for public access idle_timeout: Number of seconds to wait before sleeping the instance if it receives no traffic @@ -458,6 +505,10 @@ def create_deployment_definition( protocol = _validate_port_protocol(protocol) ports = create_koyeb_sandbox_ports(protocol) + if routes is None: + security_policy = _build_security_policy(exposed_port_security_policy) + routes = create_koyeb_sandbox_routes(security_policy=security_policy) + # Create TCP proxy ports if enabled proxy_ports = None if enable_tcp_proxy: