site/apps/api/app/core/install_secrets.py

from __future__ import annotations

import os
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
from urllib.parse import ParseResult, urlparse, urlunparse
from uuid import uuid4

from app.core.crypto import generate_install_bundle

_API_ROOT = Path(__file__).resolve().parents[2]
INSTALL_SECRETS_DIR = _API_ROOT / "data" / "secrets"
INSTALL_SECRETS_FILE = INSTALL_SECRETS_DIR / "install.env"
INSTALL_SECRETS_META_FILE = INSTALL_SECRETS_DIR / "install.meta.json"
REQUIRED_KEYS = (
    "POSTGRES_USER",
    "POSTGRES_PASSWORD",
    "POSTGRES_DB",
    "DATABASE_URL",
    "JWT_ACCESS_SECRET",
    "JWT_REFRESH_PEPPER",
    "S3_ACCESS_KEY",
    "S3_SECRET_KEY",
    "MINIO_ROOT_USER",
    "MINIO_ROOT_PASSWORD",
)


@dataclass
class InstallSecretsStatus:
    initialized: bool
    locked: bool
    path: str
    created: bool


def _parse_env_text(raw: str) -> dict[str, str]:
    values: dict[str, str] = {}
    for line in raw.splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        values[key.strip()] = value.strip()
    return values


def _render_env(values: dict[str, str]) -> str:
    ordered = [f"{key}={values[key]}" for key in sorted(values.keys())]
    return "\n".join(ordered) + "\n"


def read_install_secrets() -> dict[str, str]:
    if not INSTALL_SECRETS_FILE.exists():
        return {}
    return _parse_env_text(INSTALL_SECRETS_FILE.read_text(encoding="utf-8"))


def _write_install_secrets(values: dict[str, str]) -> None:
    INSTALL_SECRETS_DIR.mkdir(parents=True, exist_ok=True)
    INSTALL_SECRETS_FILE.write_text(_render_env(values), encoding="utf-8")


def _write_meta() -> None:
    payload = (
        "{\n"
        f'  "install_id": "{uuid4()}",\n'
        f'  "locked_at": "{datetime.now(UTC).isoformat()}"\n'
        "}\n"
    )
    INSTALL_SECRETS_META_FILE.write_text(payload, encoding="utf-8")


def _adopt_from_environment() -> dict[str, str]:
    env_values = {key: os.getenv(key, "") for key in REQUIRED_KEYS}
    db_url = os.getenv("DATABASE_URL", "")
    if db_url:
        parsed = urlparse(db_url)
        if parsed.username:
            env_values["POSTGRES_USER"] = parsed.username
        if parsed.password:
            env_values["POSTGRES_PASSWORD"] = parsed.password
        if parsed.path and parsed.path != "/":
            env_values["POSTGRES_DB"] = parsed.path.lstrip("/")
    return {key: value for key, value in env_values.items() if value}


def _sync_minio_s3_secrets(values: dict[str, str]) -> dict[str, str]:
    """MinIO root credentials are the S3 access key pair — keep them aligned."""
    if (
        values.get("S3_ACCESS_KEY") == values.get("MINIO_ROOT_USER")
        and values.get("MINIO_ROOT_PASSWORD")
        and values.get("S3_SECRET_KEY") != values["MINIO_ROOT_PASSWORD"]
    ):
        values = dict(values)
        values["S3_SECRET_KEY"] = values["MINIO_ROOT_PASSWORD"]
    return values


def load_install_secrets_to_env() -> None:
    values = _sync_minio_s3_secrets(read_install_secrets())
    for key, value in values.items():
        os.environ[key] = value


def ensure_install_secrets() -> InstallSecretsStatus:
    existing = read_install_secrets()
    if existing:
        synced = _sync_minio_s3_secrets(existing)
        if synced != existing:
            synced["SECRETS_LOCKED"] = existing.get("SECRETS_LOCKED", "true")
            _write_install_secrets(synced)
            existing = synced
        load_install_secrets_to_env()
        return InstallSecretsStatus(True, existing.get("SECRETS_LOCKED", "false") == "true", str(INSTALL_SECRETS_FILE), False)

    adopted = _adopt_from_environment()
    generated = generate_install_bundle()
    values = generated | adopted
    values["SECRETS_LOCKED"] = "true"
    _write_install_secrets(values)
    _write_meta()
    load_install_secrets_to_env()
    return InstallSecretsStatus(True, True, str(INSTALL_SECRETS_FILE), True)


def masked_database_url(database_url: str) -> str:
    parsed = urlparse(database_url)
    if not parsed.username:
        return database_url
    password = "***" if parsed.password else ""
    credentials = f"{parsed.username}:{password}" if password else parsed.username
    host = parsed.hostname or ""
    if parsed.port:
        host = f"{host}:{parsed.port}"
    netloc = f"{credentials}@{host}"
    sanitized = ParseResult(
        scheme=parsed.scheme,
        netloc=netloc,
        path=parsed.path,
        params=parsed.params,
        query=parsed.query,
        fragment=parsed.fragment,
    )
    return urlunparse(sanitized)


def install_secrets_payload() -> dict:
    values = read_install_secrets()
    db = urlparse(values.get("DATABASE_URL", ""))
    return {
        "initialized": bool(values),
        "locked": values.get("SECRETS_LOCKED") == "true",
        "secrets_path": str(INSTALL_SECRETS_FILE),
        "database": {
            "host": db.hostname,
            "port": db.port,
            "database": db.path.lstrip("/") if db.path else "",
            "user": db.username,
            "password_configured": bool(values.get("POSTGRES_PASSWORD")),
        },
        "connection_string_masked": masked_database_url(values.get("DATABASE_URL", "")),
        "secrets_status": {
            "jwt_access_secret": "configured" if bool(values.get("JWT_ACCESS_SECRET")) else "missing",
            "jwt_refresh_pepper": "configured" if bool(values.get("JWT_REFRESH_PEPPER")) else "missing",
            "postgres_password": "configured" if bool(values.get("POSTGRES_PASSWORD")) else "missing",
            "s3_secret_key": "configured" if bool(values.get("S3_SECRET_KEY")) else "missing",
            "password_bcrypt_salt": "per_user_in_db",
        },
    }


def reveal_install_secret(key: str) -> str:
    mapping = {
        "database_password": "POSTGRES_PASSWORD",
        "jwt_access_secret": "JWT_ACCESS_SECRET",
        "jwt_refresh_pepper": "JWT_REFRESH_PEPPER",
        "s3_secret_key": "S3_SECRET_KEY",
    }
    env_key = mapping.get(key)
    if not env_key:
        raise ValueError("UNSUPPORTED_SECRET_KEY")
    values = read_install_secrets()
    if env_key in values:
        return values[env_key]
    if env_key == "POSTGRES_PASSWORD":
        database_url = values.get("DATABASE_URL") or os.getenv("DATABASE_URL", "")
        parsed = urlparse(database_url)
        return parsed.password or ""
    return os.getenv(env_key, "")