vignette/scripts/provision-nas-preview-env.py

#!/usr/bin/env python3
"""NAS preview 전용 Compose 환경 파일을 비밀값 노출 없이 생성한다."""

from __future__ import annotations

import argparse
import json
import os
import secrets
import tempfile
from pathlib import Path
from urllib.parse import urlsplit


MIN_SECRET_LENGTH = 32
INTERNAL_TOKEN_KEYS = (
    "VIGNETTE_RUPTURE_INTERNAL_TOKEN",
    "VIGNETTE_PRACTICE_INTERNAL_TOKEN",
    "VIGNETTE_CALIBRATION_TRANSFER_INTERNAL_TOKEN",
    "VIGNETTE_SUPERVISION_RESEARCH_INTERNAL_TOKEN",
    "VIGNETTE_MULTIMODAL_ALLIANCE_INTERNAL_TOKEN",
    "VIGNETTE_CONTINUOUS_IMPROVEMENT_INTERNAL_TOKEN",
)
SOURCE_SECRET_KEYS = (
    "OPENAI_API_KEY",
    "ANTHROPIC_API_KEY",
)
PREVIEW_ALLOWED_EMAIL_DOMAINS = ("hs.ac.kr", "twentyoz.kr")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--source-env", type=Path, required=True)
    parser.add_argument("--out", type=Path, required=True)
    parser.add_argument("--tailnet-base-url", required=True)
    parser.add_argument("--lan-base-url", required=True)
    parser.add_argument("--engine-url", required=True)
    parser.add_argument("--http-port", type=int, default=8088)
    parser.add_argument("--https-port", type=int, default=8444)
    return parser.parse_args()


def parse_env(path: Path) -> dict[str, str]:
    values: dict[str, str] = {}
    for raw_line in path.read_text(encoding="utf-8-sig").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        values[key.strip()] = value.strip().strip('"').strip("'")
    return values


def validate_base_url(value: str, name: str) -> str:
    normalized = value.rstrip("/")
    parsed = urlsplit(normalized)
    if parsed.scheme != "http" or not parsed.hostname or parsed.path not in ("", "/"):
        raise ValueError(f"{name} must be an origin-style http URL")
    return normalized


def random_secret() -> str:
    value = secrets.token_urlsafe(48)
    if len(value) < MIN_SECRET_LENGTH:  # pragma: no cover - defensive invariant
        raise RuntimeError("generated secret was unexpectedly short")
    return value


def render_env(
    *,
    source_values: dict[str, str],
    tailnet_base_url: str,
    lan_base_url: str,
    engine_url: str,
    http_port: int,
    https_port: int,
) -> str:
    openai_key = source_values.get("OPENAI_API_KEY", "").strip()
    if not openai_key:
        raise ValueError("OPENAI_API_KEY is required in --source-env")

    tailnet_origin = validate_base_url(tailnet_base_url, "tailnet base URL")
    lan_origin = validate_base_url(lan_base_url, "LAN base URL")
    engine_origin = validate_base_url(engine_url, "engine URL")
    if not (1 <= http_port <= 65535 and 1 <= https_port <= 65535):
        raise ValueError("preview ports must be between 1 and 65535")
    if http_port == https_port:
        raise ValueError("HTTP and HTTPS ports must differ")

    origins_json = json.dumps(
        [tailnet_origin, lan_origin],
        ensure_ascii=True,
        separators=(",", ":"),
    )
    allowed_domains_json = json.dumps(
        PREVIEW_ALLOWED_EMAIL_DOMAINS,
        ensure_ascii=True,
        separators=(",", ":"),
    )
    rows: list[tuple[str, str]] = [
        ("ENVIRONMENT", "dev"),
        ("POSTGRES_USER", "vignette_owner"),
        ("POSTGRES_PASSWORD", random_secret()),
        ("POSTGRES_DB", "vignette_preview"),
        ("APP_DB_USER", "vignette_app"),
        ("APP_DB_PASSWORD", random_secret()),
        ("INSTALL_RAG", "false"),
        ("USER_UPLOAD_DIR", "/app/uploads"),
        ("SESSION_SECRET", random_secret()),
    ]
    rows.extend((key, random_secret()) for key in INTERNAL_TOKEN_KEYS)
    rows.extend(
        (
            ("AUTH_ALLOWED_EMAIL_DOMAINS", allowed_domains_json),
            ("AUTH_TEACHER_EMAILS", "[]"),
            ("AUTH_ADMIN_EMAILS", "[]"),
            ("AUTH_DEV_LOGIN_ENABLED", "true"),
            ("AUTH_DEV_LOGIN_EXTRA_ORIGINS", origins_json),
            ("DEFAULT_AFFILIATION", "nas-preview"),
            ("AUTO_SEED_PERSONAS", "true"),
            ("ALLOW_SEED_PERSONA_FALLBACK", "true"),
            ("FRONTEND_BASE_URL", tailnet_origin),
            ("FRONTEND_ORIGIN_MAP", "{}"),
            ("CORS_ORIGINS", origins_json),
            ("OAUTH_GOOGLE_CLIENT_ID", "nas-preview-dev-login"),
            ("OAUTH_GOOGLE_CLIENT_SECRET", random_secret()),
            ("OAUTH_REDIRECT_URI", f"{tailnet_origin}/api/auth/callback"),
            ("OPENAI_API_KEY", openai_key),
            ("OPENAI_BASE_URL", "https://api.openai.com/v1"),
            ("ANTHROPIC_API_KEY", source_values.get("ANTHROPIC_API_KEY", "")),
            ("ENGINE_MODE", "claude_cli"),
            ("ENGINE_URL", engine_origin),
            ("ENGINE_GATEWAY_SHARED_SECRET", random_secret()),
            ("PUBLIC_API_BASE", "/api"),
            ("HTTP_PORT", str(http_port)),
            ("HTTPS_PORT", str(https_port)),
            ("SITE_ADDRESS", ":80"),
            ("VIGNETTE_VOICE_STT_PROVIDER", "openai"),
            ("VIGNETTE_VOICE_TTS_PROVIDER", "openai"),
            ("VIGNETTE_CONTINUOUS_IMPROVEMENT_PRODUCER_ENABLED", "false"),
            ("VIGNETTE_CONTINUOUS_IMPROVEMENT_DRIFT_TRIGGER_ENABLED", "false"),
        )
    )
    return "\n".join(f"{key}={value}" for key, value in rows) + "\n"


def atomic_write(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temp_name = tempfile.mkstemp(
        dir=path.parent,
        prefix=f".{path.name}.",
        suffix=".tmp",
        text=True,
    )
    temp_path = Path(temp_name)
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as stream:
            stream.write(text)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temp_path, path)
    finally:
        if temp_path.exists():
            temp_path.unlink()


def main() -> int:
    args = parse_args()
    source_path = args.source_env.resolve()
    out_path = args.out.resolve()
    rendered = render_env(
        source_values=parse_env(source_path),
        tailnet_base_url=args.tailnet_base_url,
        lan_base_url=args.lan_base_url,
        engine_url=args.engine_url,
        http_port=args.http_port,
        https_port=args.https_port,
    )
    atomic_write(out_path, rendered)
    report = {
        "ok": True,
        "out": str(out_path),
        "managed_keys": len(rendered.splitlines()),
        "secret_values_emitted": False,
    }
    print(json.dumps(report, ensure_ascii=False, sort_keys=True))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())