# Copyright 2025 Alibaba Group Holding Ltd. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """ Shared helpers for container-based sandbox services. These utilities centralize common parsing, filtering, and transformation logic so multiple container runtimes (docker, kubernetes, etc.) can reuse them. """ from __future__ import annotations import logging import re from datetime import datetime, timezone from typing import Dict, Optional from opensandbox_server.api.schema import Endpoint, Sandbox, SandboxFilter from opensandbox_server.services.constants import ( ALLOWED_EGRESS_ENV_VARS, EGRESS_ENV_PREFIX, OPENSANDBOX_EGRESS_MITMPROXY_TRANSPARENT, OPEN_SANDBOX_INGRESS_HEADER, ) from opensandbox_server.config import ( GATEWAY_ROUTE_MODE_HEADER, GATEWAY_ROUTE_MODE_URI, GATEWAY_ROUTE_MODE_WILDCARD, INGRESS_MODE_GATEWAY, IngressConfig, ) logger = logging.getLogger(__name__) MEMORY_PATTERN = re.compile(r"^\s*(\d+)([kmgti]i?|[kmgti]?b)?\s*$", re.IGNORECASE) MEMORY_MULTIPLIERS: Dict[str, int] = { "": 1, "b": 1, "k": 1_000, "kb": 1_000, "ki": 1024, "m": 1_000_000, "mb": 1_000_000, "mi": 1024**2, "g": 1_000_000_000, "gb": 1_000_000_000, "gi": 1024**3, "t": 1_000_000_000_000, "tb": 1_000_000_000_000, "ti": 1024**4, } def parse_memory_limit(value: Optional[str]) -> Optional[int]: """Convert memory string (e.g., 512Mi) to bytes.""" if not value: return None match = MEMORY_PATTERN.match(value) if not match: logger.warning("Invalid memory limit format '%s'; ignoring.", value) return None amount = int(match.group(1)) unit = (match.group(2) or "").lower() multiplier = MEMORY_MULTIPLIERS.get(unit) if not multiplier: logger.warning("Unsupported memory unit '%s'; ignoring.", unit) return None return amount * multiplier def parse_nano_cpus(value: Optional[str]) -> Optional[int]: """Convert CPU string (e.g., 500m, 2) to nano_cpus.""" if not value: return None cpu_str = value.strip().lower() try: if cpu_str.endswith("m"): cpus = float(cpu_str[:-1]) / 1000 else: cpus = float(cpu_str) except ValueError: logger.warning("Invalid CPU limit format '%s'; ignoring.", value) return None if cpus <= 0: logger.warning("CPU limit must be positive. Got '%s'. Ignoring.", value) return None return int(cpus * 1_000_000_000) def parse_gpu_request(value: Optional[str]) -> Optional[int]: """Convert GPU limit string to a device count. Accepts a positive integer string (e.g. ``"1"``, ``"2"``) or the literal ``"all"`` to request every available GPU. Returns ``-1`` for ``"all"`` (the sentinel the Docker Engine uses for unbounded device requests), a positive int for a numeric count, and ``None`` when unset or the value cannot be parsed. """ if not value: return None gpu_str = value.strip().lower() if gpu_str == "all": return -1 try: count = int(gpu_str) except ValueError: logger.warning("Invalid GPU limit format '%s'; ignoring.", value) return None if count <= 0: logger.warning("GPU limit must be positive. Got '%s'. Ignoring.", value) return None return count def parse_timestamp(timestamp: Optional[str]) -> datetime: """ Parse RFC3339 timestamp into timezone-aware datetime. Fallback to now. Docker often returns RFC3339Nano (up to 9 fractional digits). Python's datetime.fromisoformat only supports microseconds (6 digits), so we truncate the fractional part to 6 digits before parsing. """ if not timestamp or timestamp == "0001-01-01T00:00:00Z": return datetime.now(timezone.utc) normalized = timestamp if normalized.endswith("Z"): normalized = normalized[:-1] + "+00:00" if "." in normalized: main, rest = normalized.split(".", 1) tz_sep = None for sep in ("+", "-"): pos = rest.find(sep) if pos != -1: tz_sep = pos break if tz_sep is None: frac = rest tz = "" else: frac = rest[:tz_sep] tz = rest[tz_sep:] frac = frac[:6] # truncate to microseconds precision normalized = f"{main}.{frac}{tz}" if frac else f"{main}{tz}" try: return datetime.fromisoformat(normalized) except ValueError: logger.warning("Invalid timestamp '%s'; defaulting to current time.", timestamp) return datetime.now(timezone.utc) def normalize_external_endpoint_url(endpoint: str, default_scheme: str = "https") -> str: """Normalize host or URL to a full URL with an explicit scheme.""" endpoint = endpoint.strip() if endpoint.startswith("http://") or endpoint.startswith("https://"): return endpoint return f"{default_scheme}://{endpoint}" def matches_filter(sandbox: Sandbox, filter_: SandboxFilter) -> bool: """Apply state/metadata filters to a sandbox instance.""" if not filter_: return True if filter_.state: desired = {state.lower() for state in filter_.state} current_state = (sandbox.status.state or "").lower() if current_state not in desired: return False if filter_.metadata: metadata = sandbox.metadata or {} for key, value in filter_.metadata.items(): if metadata.get(key) != value: return False return True # ============================================================================ # Ingress helpers # ============================================================================ def format_ingress_endpoint( ingress_config: Optional[IngressConfig], sandbox_id: str, port: int, expires_b36: Optional[str] = None, signature: Optional[str] = None, ) -> Optional[Endpoint]: """ Build an ingress-based endpoint string for a sandbox. When *expires_b36* and *signature* are provided, the endpoint embeds a signed route token (OSEP-0011). Otherwise a plain ingress endpoint is returned. Returns None when ingress is not in gateway mode. """ if not ingress_config or ingress_config.mode != INGRESS_MODE_GATEWAY: return None gateway_cfg = ingress_config.gateway if gateway_cfg is None: return None address = gateway_cfg.address route_mode = gateway_cfg.route.mode is_signed = expires_b36 is not None and signature is not None if route_mode == GATEWAY_ROUTE_MODE_WILDCARD: base = address[2:] if address.startswith("*.") else address if is_signed: return Endpoint(endpoint=f"{sandbox_id}-{port}-{expires_b36}-{signature}.{base}") return Endpoint(endpoint=f"{sandbox_id}-{port}.{base}") if route_mode == GATEWAY_ROUTE_MODE_URI: if is_signed: return Endpoint(endpoint=f"{address}/{sandbox_id}/{port}/{expires_b36}/{signature}") return Endpoint(endpoint=f"{address}/{sandbox_id}/{port}") if route_mode == GATEWAY_ROUTE_MODE_HEADER: if is_signed: header_value = f"{sandbox_id}-{port}-{expires_b36}-{signature}" else: header_value = f"{sandbox_id}-{port}" return Endpoint( endpoint=address, headers={OPEN_SANDBOX_INGRESS_HEADER: header_value}, ) raise RuntimeError(f"Unsupported route mode: {route_mode}") def split_egress_env( env: Optional[Dict[str, Optional[str]]], ) -> tuple[Dict[str, Optional[str]], Dict[str, Optional[str]]]: """Split request env into (sandbox_env, egress_env) by OPENSANDBOX_EGRESS_ prefix. Only env vars listed in ALLOWED_EGRESS_ENV_VARS are forwarded to the egress sidecar. Any other OPENSANDBOX_EGRESS_ key raises ValueError. """ if not env: return {}, {} sandbox_env: Dict[str, Optional[str]] = {} egress_env: Dict[str, Optional[str]] = {} for key, value in env.items(): if key.startswith(EGRESS_ENV_PREFIX): if key not in ALLOWED_EGRESS_ENV_VARS: raise ValueError( f"Environment variable '{key}' is not allowed; " f"permitted OPENSANDBOX_EGRESS_ keys: {sorted(ALLOWED_EGRESS_ENV_VARS)}" ) egress_env[key] = value if key == OPENSANDBOX_EGRESS_MITMPROXY_TRANSPARENT: sandbox_env[key] = value else: sandbox_env[key] = value return sandbox_env, egress_env __all__ = [ "parse_memory_limit", "parse_nano_cpus", "parse_gpu_request", "parse_timestamp", "normalize_external_endpoint_url", "format_ingress_endpoint", "matches_filter", "split_egress_env", ]