opensandbox-group--opensandbox/server/opensandbox_server/services/windows_common.py

# Copyright 2026 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 Windows-profile helpers used by both Docker and Kubernetes runtimes.
"""

from __future__ import annotations

import math
import re
from typing import Optional, TYPE_CHECKING

from fastapi import HTTPException, status

from opensandbox_server.services.constants import SandboxErrorCodes
from opensandbox_server.services.helpers import parse_nano_cpus

if TYPE_CHECKING:
    from opensandbox_server.api.schema import PlatformSpec

WINDOWS_USER_PORTS_ENV = "USER_PORTS"
WINDOWS_RAM_SIZE_ENV = "RAM_SIZE"
WINDOWS_CPU_CORES_ENV = "CPU_CORES"
WINDOWS_DISK_SIZE_ENV = "DISK_SIZE"
_WINDOWS_SIZE_PATTERN = re.compile(r"^\s*(\d+)\s*([a-zA-Z]*)\s*$")
WINDOWS_MIN_CPU_NANO_CPUS = 2_000_000_000
WINDOWS_MIN_MEMORY_GB = 4
WINDOWS_MIN_DISK_GB = 64


def is_windows_platform(platform: Optional["PlatformSpec"]) -> bool:
    return bool(platform and platform.os == "windows")


def inject_windows_user_ports(environment: list[str], exposed_ports: Optional[list[str]]) -> list[str]:
    """Ensure USER_PORTS includes container ports exposed for windows profile."""
    if not exposed_ports:
        return environment

    resolved_ports: list[str] = []
    for port_spec in exposed_ports:
        token = str(port_spec).split("/", 1)[0].strip()
        if token.isdigit() and token not in resolved_ports:
            resolved_ports.append(token)
    if not resolved_ports:
        return environment

    env_items = list(environment)
    user_ports_index: Optional[int] = None
    existing_ports: list[str] = []

    for idx, item in enumerate(env_items):
        if "=" not in item:
            continue
        key, value = item.split("=", 1)
        if key != WINDOWS_USER_PORTS_ENV:
            continue
        user_ports_index = idx
        existing_ports = [p.strip() for p in value.split(",") if p.strip()]
        break

    merged = list(existing_ports)
    for port in resolved_ports:
        if port not in merged:
            merged.append(port)
    merged_value = ",".join(merged)

    if user_ports_index is None:
        env_items.append(f"{WINDOWS_USER_PORTS_ENV}={merged_value}")
    else:
        env_items[user_ports_index] = f"{WINDOWS_USER_PORTS_ENV}={merged_value}"
    return env_items


def _normalize_windows_size_env(limit_value: Optional[str]) -> Optional[str]:
    if not limit_value:
        return None
    match = _WINDOWS_SIZE_PATTERN.match(limit_value)
    if not match:
        return None

    amount = int(match.group(1))
    unit = (match.group(2) or "").lower()
    if amount <= 0:
        return None
    if unit in {"g", "gi", "gb"}:
        return f"{amount}G"
    if unit in {"m", "mi", "mb"}:
        return f"{max(1, math.ceil(amount / 1024))}G"
    if unit in {"t", "ti", "tb"}:
        return f"{amount * 1024}G"
    return None


def inject_windows_resource_limits_env(
    environment: list[str],
    resource_limits: Optional[dict[str, str]],
) -> list[str]:
    """Convert resourceLimits into dockur/windows resource envs.

    resourceLimits takes precedence over user-supplied env values when present.
    """
    if not resource_limits:
        return environment

    env_items = list(environment)

    def upsert_env(key: str, value: str) -> None:
        for idx, item in enumerate(env_items):
            if "=" not in item:
                continue
            existing_key, _ = item.split("=", 1)
            if existing_key == key:
                env_items[idx] = f"{key}={value}"
                return
        env_items.append(f"{key}={value}")

    cpu_limit = resource_limits.get("cpu")
    nano_cpus = parse_nano_cpus(cpu_limit)
    if nano_cpus:
        cores = max(1, math.ceil(nano_cpus / 1_000_000_000))
        upsert_env(WINDOWS_CPU_CORES_ENV, str(cores))

    memory_limit = resource_limits.get("memory")
    normalized_memory = _normalize_windows_size_env(memory_limit)
    if normalized_memory:
        upsert_env(WINDOWS_RAM_SIZE_ENV, normalized_memory)

    disk_limit = (
        resource_limits.get("disk")
        or resource_limits.get("storage")
        or resource_limits.get("ephemeral-storage")
    )
    normalized_disk = _normalize_windows_size_env(disk_limit)
    if normalized_disk:
        upsert_env(WINDOWS_DISK_SIZE_ENV, normalized_disk)

    return env_items


def validate_windows_resource_limits(resource_limits: Optional[dict[str, str]]) -> None:
    """Validate minimum resource limits for windows profile."""
    if not resource_limits:
        return

    cpu_limit = resource_limits.get("cpu")
    if cpu_limit is not None:
        nano_cpus = parse_nano_cpus(cpu_limit)
        if nano_cpus is None or nano_cpus < WINDOWS_MIN_CPU_NANO_CPUS:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail={
                    "code": SandboxErrorCodes.INVALID_PARAMETER,
                    "message": (
                        "Windows profile requires resourceLimits.cpu >= 2 "
                        f"(received: {cpu_limit})."
                    ),
                },
            )

    memory_limit = resource_limits.get("memory")
    if memory_limit is not None:
        normalized_memory = _normalize_windows_size_env(memory_limit)
        if normalized_memory is None or int(normalized_memory[:-1]) < WINDOWS_MIN_MEMORY_GB:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail={
                    "code": SandboxErrorCodes.INVALID_PARAMETER,
                    "message": (
                        "Windows profile requires resourceLimits.memory >= 4G "
                        f"(received: {memory_limit})."
                    ),
                },
            )

    disk_limit = (
        resource_limits.get("disk")
        or resource_limits.get("storage")
        or resource_limits.get("ephemeral-storage")
    )
    if disk_limit is not None:
        normalized_disk = _normalize_windows_size_env(disk_limit)
        if normalized_disk is None or int(normalized_disk[:-1]) < WINDOWS_MIN_DISK_GB:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail={
                    "code": SandboxErrorCodes.INVALID_PARAMETER,
                    "message": (
                        "Windows profile requires resourceLimits.disk (or storage/ephemeral-storage) >= 64G "
                        f"(received: {disk_limit})."
                    ),
                },
            )