Nexus/server/infrastructure/btpanel/credentials.py

"""Per-server Baota panel URL and encrypted API key in ``servers.extra_attrs``."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any

from server.domain.models import Server
from server.infrastructure.database.crypto import decrypt_value, encrypt_value

BT_PANEL_ATTR = "bt_panel"


@dataclass(frozen=True)
class BtPanelCredentials:
    base_url: str
    api_key: str
    verify_ssl: bool = False


def _attrs(server: Server) -> dict[str, Any]:
    raw = server.extra_attrs
    return raw if isinstance(raw, dict) else {}


def get_bt_panel_block(server: Server) -> dict[str, Any]:
    block = _attrs(server).get(BT_PANEL_ATTR)
    return block if isinstance(block, dict) else {}


def bt_panel_configured(server: Server) -> bool:
    block = get_bt_panel_block(server)
    return bool(str(block.get("base_url") or "").strip() and block.get("api_key_enc"))


def get_bt_panel_base_url(server: Server) -> str | None:
    url = str(get_bt_panel_block(server).get("base_url") or "").strip().rstrip("/")
    return url or None


def read_bt_panel_credentials(server: Server) -> BtPanelCredentials | None:
    block = get_bt_panel_block(server)
    base_url = str(block.get("base_url") or "").strip().rstrip("/")
    enc = block.get("api_key_enc")
    if not base_url or not enc:
        return None
    api_key = decrypt_value(str(enc))
    if not api_key:
        return None
    verify_ssl = bool(block.get("verify_ssl"))
    return BtPanelCredentials(base_url=base_url, api_key=api_key, verify_ssl=verify_ssl)


def merge_bt_panel_extra(
    extra_attrs: dict[str, Any] | None,
    *,
    base_url: str | None = None,
    api_key: str | None = None,
    verify_ssl: bool | None = None,
    clear_api_key: bool = False,
) -> dict[str, Any]:
    out = dict(extra_attrs or {})
    block = dict(out.get(BT_PANEL_ATTR) or {}) if isinstance(out.get(BT_PANEL_ATTR), dict) else {}

    if base_url is not None:
        block["base_url"] = base_url.strip().rstrip("/")
    if api_key is not None and api_key.strip():
        block["api_key_enc"] = encrypt_value(api_key.strip())
    elif clear_api_key:
        block.pop("api_key_enc", None)
    if verify_ssl is not None:
        block["verify_ssl"] = verify_ssl

    if block:
        out[BT_PANEL_ATTR] = block
    else:
        out.pop(BT_PANEL_ATTR, None)
    return out


def public_bt_panel_status(server: Server) -> dict[str, Any]:
    from server.infrastructure.btpanel.bootstrap_state import public_bootstrap_fields

    block = get_bt_panel_block(server)
    return {
        "base_url": str(block.get("base_url") or "").strip() or None,
        "api_key_set": bool(block.get("api_key_enc")),
        "verify_ssl": bool(block.get("verify_ssl")),
        "configured": bt_panel_configured(server),
        **public_bootstrap_fields(server),
    }