agentic-mobile-control/packages/cloud-platform/cloud/control_config.py

from __future__ import annotations

import os
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Literal


EnvironmentName = Literal["local", "test", "production"]
SUPPORTED_DATABASE_PREFIXES = (
    "sqlite:///",
    "postgresql://",
    "postgresql+psycopg://",
)


class CloudConfigurationError(ValueError):
    """Raised when control-plane configuration is unsafe or invalid."""


@dataclass(frozen=True)
class CloudControlConfig:
    environment: EnvironmentName = "local"
    database_url: str = "sqlite:///cloud/cloud.sqlite3"
    scheduler_interval_seconds: float = 1.0
    lease_reaper_interval_seconds: float = 5.0
    lease_duration_seconds: float = 60.0
    max_task_attempts: int = 3
    allow_insecure_anonymous: bool = False
    cors_allowed_origins: tuple[str, ...] = ()
    console_static_dir: str | None = None
    user_session_idle_seconds: int = 28_800
    user_session_absolute_seconds: int = 604_800
    login_failure_limit: int = 5
    login_failure_window_seconds: int = 900
    login_block_seconds: int = 900
    session_cookie_secure: bool = False
    trust_proxy_headers: bool = False


def load_control_config(
    env: Mapping[str, str] | None = None,
) -> CloudControlConfig:
    values = os.environ if env is None else env
    environment = values.get("CLOUD_ENVIRONMENT", "local").strip().lower()
    if environment not in {"local", "test", "production"}:
        raise CloudConfigurationError(
            "CLOUD_ENVIRONMENT must be local, test, or production"
        )

    database_url = values.get(
        "CLOUD_DATABASE_URL",
        "sqlite:///cloud/cloud.sqlite3",
    ).strip()
    if not database_url.startswith(SUPPORTED_DATABASE_PREFIXES):
        raise CloudConfigurationError(
            "CLOUD_DATABASE_URL must use sqlite or postgresql"
        )

    config = CloudControlConfig(
        environment=environment,  # type: ignore[arg-type]
        database_url=database_url,
        scheduler_interval_seconds=_positive_float(
            values,
            "CLOUD_SCHEDULER_INTERVAL_SECONDS",
            1.0,
        ),
        lease_reaper_interval_seconds=_positive_float(
            values,
            "CLOUD_LEASE_REAPER_INTERVAL_SECONDS",
            5.0,
        ),
        lease_duration_seconds=_positive_float(
            values,
            "CLOUD_LEASE_DURATION_SECONDS",
            60.0,
        ),
        max_task_attempts=_positive_int(
            values,
            "CLOUD_MAX_TASK_ATTEMPTS",
            3,
        ),
        allow_insecure_anonymous=_parse_bool(
            values.get("CLOUD_ALLOW_INSECURE_ANONYMOUS"),
            default=False,
        ),
        cors_allowed_origins=_parse_cors_origins(
            values.get("CLOUD_CONSOLE_CORS_ORIGINS")
        ),
        console_static_dir=_parse_optional_string(
            values.get("CLOUD_CONSOLE_STATIC_DIR")
        ),
        user_session_idle_seconds=_positive_int(
            values,
            "CLOUD_USER_SESSION_IDLE_SECONDS",
            28_800,
        ),
        user_session_absolute_seconds=_positive_int(
            values,
            "CLOUD_USER_SESSION_ABSOLUTE_SECONDS",
            604_800,
        ),
        login_failure_limit=_positive_int(values, "CLOUD_LOGIN_FAILURE_LIMIT", 5),
        login_failure_window_seconds=_positive_int(
            values,
            "CLOUD_LOGIN_FAILURE_WINDOW_SECONDS",
            900,
        ),
        login_block_seconds=_positive_int(values, "CLOUD_LOGIN_BLOCK_SECONDS", 900),
        session_cookie_secure=_parse_bool(
            values.get("CLOUD_SESSION_COOKIE_SECURE"),
            default=environment == "production",
        ),
        trust_proxy_headers=_parse_bool(
            values.get("CLOUD_TRUST_PROXY_HEADERS"),
            default=False,
        ),
    )
    validate_control_config(config)
    return config


def validate_control_config(config: CloudControlConfig) -> None:
    if config.environment == "production" and config.allow_insecure_anonymous:
        raise CloudConfigurationError(
            "anonymous access cannot be enabled in production"
        )
    if config.user_session_absolute_seconds < config.user_session_idle_seconds:
        raise CloudConfigurationError(
            "CLOUD_USER_SESSION_ABSOLUTE_SECONDS must be at least the idle TTL"
        )
    if config.environment == "production" and not config.session_cookie_secure:
        raise CloudConfigurationError(
            "production requires secure user session cookies"
        )


def _positive_float(
    values: Mapping[str, str],
    name: str,
    default: float,
) -> float:
    raw_value = values.get(name)
    if raw_value is None:
        return default
    try:
        value = float(raw_value)
    except ValueError as exc:
        raise CloudConfigurationError(f"{name} must be a number") from exc
    if value <= 0:
        raise CloudConfigurationError(f"{name} must be greater than zero")
    return value


def _positive_int(
    values: Mapping[str, str],
    name: str,
    default: int,
) -> int:
    raw_value = values.get(name)
    if raw_value is None:
        return default
    try:
        value = int(raw_value)
    except ValueError as exc:
        raise CloudConfigurationError(f"{name} must be an integer") from exc
    if value <= 0:
        raise CloudConfigurationError(f"{name} must be greater than zero")
    return value


def _parse_bool(value: str | None, *, default: bool) -> bool:
    if value is None:
        return default
    normalized = value.strip().lower()
    if normalized in {"1", "true", "yes", "on", "enabled"}:
        return True
    if normalized in {"0", "false", "no", "off", "disabled", ""}:
        return False
    raise CloudConfigurationError("boolean configuration value is invalid")


def _parse_cors_origins(raw_value: str | None) -> tuple[str, ...]:
    if raw_value is None or not raw_value.strip():
        return ()
    return tuple(origin.strip() for origin in raw_value.split(",") if origin.strip())


def _parse_optional_string(raw_value: str | None) -> str | None:
    if raw_value is None:
        return None
    stripped = raw_value.strip()
    return stripped or None