stackpilot/backend/config.py

"""Application settings, loaded from environment variables."""
from __future__ import annotations

import os
import secrets
import stat
from functools import lru_cache
from typing import Annotated

from pydantic import ValidationInfo, field_validator
from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    # Paths
    STACKS_DIR: str = "/opt/stackpilot/stacks"
    DATA_DIR: str = "/opt/stackpilot/data"

    # Security
    # Auto-generated and persisted to ${DATA_DIR}/secret_key when left empty.
    SECRET_KEY: str = ""
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60
    REFRESH_TOKEN_EXPIRE_DAYS: int = 30

    # Update checker
    UPDATE_CHECK_INTERVAL_MINUTES: int = 60

    # Notifications (webhook URLs)
    NOTIFY_WEBHOOKS: Annotated[list[str], NoDecode] = []

    # Docker
    DOCKER_SOCKET: str = "/var/run/docker.sock"
    HOST_PROC_PATH: str = "/host_proc"

    # Throwaway image used to read/write named-volume contents during backup.
    BACKUP_HELPER_IMAGE: str = "alpine:latest"

    # Vulnerability scanner, run as a throwaway container (services/scan_service).
    # Pinning a tag here is the supported way to hold a scanner version still.
    SCANNER_IMAGE: str = "aquasec/trivy:latest"

    # Host browser sandbox roots. Deliberately does NOT contain "/": that entry
    # makes _is_allowed() wave through every path, i.e. it switches the sandbox
    # off. Add it back explicitly if you really want the whole filesystem.
    ALLOWED_BROWSE_ROOTS: Annotated[list[str], NoDecode] = [
        "/mnt", "/media", "/srv", "/opt", "/home",
    ]
    HOST_ROOT_PREFIX: str = ""  # e.g. "/host_root" when host / is bind-mounted

    # CORS
    CORS_ORIGINS: Annotated[list[str], NoDecode] = [
        "http://localhost:5009", "http://localhost:5173",
    ]

    # Server
    PORT: int = 5008

    @field_validator("SECRET_KEY", mode="after")
    @classmethod
    def _ensure_secret(cls, v: str, info: ValidationInfo) -> str:
        """Return the configured key, or a persisted auto-generated one.

        Generating a fresh key per process (the old behaviour) silently
        invalidated every session on each restart, and would now also make the
        encrypted backup-destination credentials undecryptable. So the
        generated key is written next to the database instead, mode 0600, and
        read back on the next start. An explicitly configured SECRET_KEY always
        wins and nothing is written.
        """
        if v:
            return v
        data_dir = info.data.get("DATA_DIR") or "/opt/stackpilot/data"
        key_file = os.path.join(data_dir, "secret_key")
        try:
            with open(key_file, "r", encoding="utf-8") as fh:
                if existing := fh.read().strip():
                    return existing
        except OSError:
            pass
        generated = secrets.token_urlsafe(48)
        try:
            os.makedirs(data_dir, exist_ok=True)
            with open(key_file, "w", encoding="utf-8") as fh:
                fh.write(generated + "\n")
            os.chmod(key_file, stat.S_IRUSR | stat.S_IWUSR)
        except OSError:
            # Read-only data dir: fall back to the old per-process behaviour
            # rather than refusing to boot. Sessions won't survive a restart.
            pass
        return generated

    @field_validator(
        "NOTIFY_WEBHOOKS", "ALLOWED_BROWSE_ROOTS", "CORS_ORIGINS", mode="before"
    )
    @classmethod
    def _split_csv(cls, v):
        if isinstance(v, str):
            v = v.strip()
            if not v:
                return []
            if v.startswith("["):  # tolerate a JSON list too
                import json

                try:
                    return json.loads(v)
                except json.JSONDecodeError:
                    pass
            return [item.strip() for item in v.split(",") if item.strip()]
        return v


@lru_cache
def get_settings() -> Settings:
    return Settings()


settings = get_settings()