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

# 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.

"""
Secure runtime resolver for translating secure runtime configuration
to backend-specific parameters (Docker --runtime, Kubernetes RuntimeClass).

This module provides:
- SecureRuntimeResolver: Translates AppConfig to runtime parameters
- validate_secure_runtime_on_startup: Validates runtime availability at server startup
"""

from __future__ import annotations

import asyncio
import logging
from typing import TYPE_CHECKING, Optional

from kubernetes.client.exceptions import ApiException

logger = logging.getLogger(__name__)

if TYPE_CHECKING:
    from docker import DockerClient
    from opensandbox_server.config import AppConfig, SecureRuntimeConfig
    from opensandbox_server.services.k8s.client import K8sClient


class SecureRuntimeResolver:
    """
    Resolver for secure container runtime configuration.

    Translates server-level secure_runtime configuration into
    backend-specific parameters:
    - Docker: OCI runtime name (e.g., "runsc", "kata-runtime")
    - Kubernetes: RuntimeClass name (e.g., "gvisor", "kata-qemu")
    """

    # Default runtime mappings
    DEFAULT_DOCKER_RUNTIMES = {
        "gvisor": "runsc",
        "kata": "kata-runtime",
    }

    DEFAULT_K8S_RUNTIME_CLASSES = {
        "gvisor": "gvisor",
        "kata": "kata-qemu",
        "firecracker": "kata-fc",
    }

    def __init__(self, config: AppConfig):
        """
        Initialize the resolver with application configuration.

        Args:
            config: Application configuration containing secure_runtime settings
        """
        self.secure_runtime: Optional[SecureRuntimeConfig] = getattr(
            config, "secure_runtime", None
        )
        self.runtime_mode = config.runtime.type  # "docker" or "kubernetes"

    def is_enabled(self) -> bool:
        """Check if secure runtime is configured and enabled."""
        return (
            self.secure_runtime is not None
            and self.secure_runtime.type != ""
        )

    def get_docker_runtime(self) -> Optional[str]:
        """
        Get the Docker OCI runtime name for secure containers.

        Returns the configured docker_runtime if set, otherwise uses
        the default mapping for the secure runtime type.

        Returns:
            OCI runtime name (e.g., "runsc", "kata-runtime") or None
        """
        if not self.is_enabled():
            return None

        if self.secure_runtime is None:
            return None

        # Use explicit docker_runtime if configured
        if self.secure_runtime.docker_runtime:
            return self.secure_runtime.docker_runtime

        # Fall back to default mapping
        runtime_type = self.secure_runtime.type
        return self.DEFAULT_DOCKER_RUNTIMES.get(runtime_type)

    def get_k8s_runtime_class(self) -> Optional[str]:
        """
        Get the Kubernetes RuntimeClass name for secure containers.

        Returns the configured k8s_runtime_class if set, otherwise uses
        the default mapping for the secure runtime type.

        Returns:
            RuntimeClass name (e.g., "gvisor", "kata-qemu") or None
        """
        if not self.is_enabled():
            return None

        if self.secure_runtime is None:
            return None

        # Use explicit k8s_runtime_class if configured
        if self.secure_runtime.k8s_runtime_class:
            return self.secure_runtime.k8s_runtime_class

        # Fall back to default mapping
        runtime_type = self.secure_runtime.type
        return self.DEFAULT_K8S_RUNTIME_CLASSES.get(runtime_type)


async def validate_secure_runtime_on_startup(
    config: AppConfig,
    docker_client: Optional["DockerClient"] = None,
    k8s_client: Optional["K8sClient"] = None,
) -> None:
    """
    Validate that configured secure runtimes are available at startup.

    This function performs fail-fast validation to ensure the server
    starts with a valid secure runtime configuration. It checks:
    - Docker runtimes: Verifies the runtime exists in Docker daemon
    - Kubernetes RuntimeClasses: Verifies the RuntimeClass exists in cluster

    Args:
        config: Application configuration
        docker_client: Optional Docker client for runtime validation
        k8s_client: Optional K8s client wrapper for RuntimeClass validation

    Raises:
        ValueError: If a configured secure runtime is not available
        Exception: For other validation errors
    """
    resolver = SecureRuntimeResolver(config)

    if not resolver.is_enabled():
        logger.info("Secure runtime is not configured.")
        return

    if config.runtime.type == "docker":
        await _validate_docker_runtime(resolver, docker_client, config)
    elif config.runtime.type == "kubernetes":
        await _validate_k8s_runtime_class(resolver, k8s_client, config)
    else:
        logger.warning(
            "Secure runtime validation skipped for unknown runtime type: %s",
            config.runtime.type,
        )


async def _validate_docker_runtime(
    resolver: SecureRuntimeResolver,
    docker_client: Optional["DockerClient"],
    config: "AppConfig",
) -> None:
    """Validate that the Docker OCI runtime exists."""
    runtime_name = resolver.get_docker_runtime()

    if not runtime_name:
        logger.info("No Docker runtime configured for secure containers.")
        return

    logger.info("Validating Docker OCI runtime: %s", runtime_name)

    if docker_client is None:
        logger.warning(
            "Docker client not available; skipping runtime validation. "
            "Runtime '%s' will be used but not validated.",
            runtime_name,
        )
        return

    try:
        # Get list of available runtimes from Docker daemon
        # Docker stores runtimes in daemon configuration
        info = docker_client.info()
        runtimes = info.get("Runtimes", {})

        if runtime_name not in runtimes:
            available = ", ".join(runtimes.keys()) if runtimes else "none"
            raise ValueError(
                f"Configured Docker runtime '{runtime_name}' is not available. "
                f"Available runtimes: {available}. "
                f"Please install and configure the runtime before starting the server."
            )

        logger.info(
            "Docker OCI runtime '%s' is available: %s",
            runtime_name,
            runtimes.get(runtime_name, {}),
        )
    except Exception as exc:
        logger.error("Failed to validate Docker runtime: %s", exc)
        raise

    _warn_gvisor_egress_incompatibility(config)


async def _validate_k8s_runtime_class(
    resolver: SecureRuntimeResolver,
    k8s_client: Optional["K8sClient"],
    config: AppConfig,
) -> None:
    """Validate that the Kubernetes RuntimeClass exists."""
    runtime_class_name = resolver.get_k8s_runtime_class()

    if not runtime_class_name:
        logger.info("No Kubernetes RuntimeClass configured for secure containers.")
        return

    logger.info("Validating Kubernetes RuntimeClass: %s", runtime_class_name)

    if k8s_client is None:
        logger.warning(
            "Kubernetes client not available; skipping RuntimeClass validation. "
            "RuntimeClass '%s' will be used but not validated.",
            runtime_class_name,
        )
        return

    try:
        loop = asyncio.get_event_loop()
        await loop.run_in_executor(None, k8s_client.read_runtime_class, runtime_class_name)
        logger.info("Kubernetes RuntimeClass '%s' is available.", runtime_class_name)
    except ApiException as exc:
        if exc.status == 404:
            raise ValueError(
                f"Configured Kubernetes RuntimeClass '{runtime_class_name}' does not exist. "
                f"Please create the RuntimeClass before starting the server."
            ) from exc
        logger.error("Failed to validate RuntimeClass: %s", exc)
        raise
    except Exception as exc:
        logger.error("Failed to validate RuntimeClass: %s", exc)
        raise

    _warn_gvisor_egress_incompatibility(config)


def _warn_gvisor_egress_incompatibility(config: "AppConfig") -> None:
    """Log a warning when gVisor is configured alongside an egress sidecar image."""
    egress_image = config.egress.image if getattr(config, "egress", None) else None
    if config.secure_runtime and config.secure_runtime.type == "gvisor" and egress_image:
        logger.warning(
            "gVisor runtime is configured with egress sidecar image. "
            "The egress sidecar's iptables nat-based DNS redirect is incompatible with gVisor. "
            "Sandboxes created with network_policy will be rejected at creation time."
        )


__all__ = [
    "SecureRuntimeResolver",
    "validate_secure_runtime_on_startup",
]