"""Default device credentials and ports from environment variables. Naming convention (catalog_key → env prefix): proxmox → DEVICE_PROXMOX_* (API token + SSH fallback) docker_vm → DEVICE_DOCKER_VM_* pfsense → DEVICE_PFSENSE_* Per slot: DEVICE__PORT DEVICE__USERNAME DEVICE__PASSWORD (SSH / login) DEVICE__API_KEY (REST API devices) Proxmox API (proxmox-mcp — tried before SSH): DEVICE_PROXMOX_API_URL (optional override; leave blank to use https://:8006) DEVICE_PROXMOX_API_PORT (optional; default 8006 when API URL is auto-built) DEVICE_PROXMOX_TOKEN_ID (e.g. root@pam!agentic) DEVICE_PROXMOX_TOKEN_SECRET DEVICE_PROXMOX_VERIFY_SSL (true/false, default false) Global fallbacks for SSH slots: DEVICE_SSH_USERNAME DEVICE_SSH_PASSWORD Global fallback for API-key slots: DEVICE_API_KEY """ from __future__ import annotations import os from dataclasses import dataclass from typing import TYPE_CHECKING if TYPE_CHECKING: from app.inventory_catalog import CatalogEntry # catalog_key → how secrets are stored / resolved SLOT_AUTH: dict[str, dict] = { "proxmox": {"auth": "ssh", "secret_field": "PASSWORD"}, "docker_vm": {"auth": "ssh", "secret_field": "PASSWORD"}, "asterisk": {"auth": "ssh", "secret_field": "PASSWORD"}, "asterisk_geneseasx": {"auth": "ssh", "secret_field": "PASSWORD"}, "asterisk_satbox": {"auth": "ssh", "secret_field": "PASSWORD"}, "debian_host": {"auth": "ssh", "secret_field": "PASSWORD"}, "tplink": {"auth": "ssh", "secret_field": "PASSWORD"}, "pfsense": {"auth": "api_key", "secret_field": "API_KEY"}, "fortigate": {"auth": "api_key", "secret_field": "API_KEY"}, "fortiswitch": {"auth": "password", "secret_field": "PASSWORD"}, } @dataclass class SlotDefaults: port: int | None = None username: str | None = None secret: str | None = None secret_kind: str = "password" # password | api_key from_env: bool = False def _env(name: str) -> str | None: val = os.environ.get(name, "").strip() return val or None def _prefix(catalog_key: str) -> str: return catalog_key.upper() def resolve_slot_defaults(catalog_key: str, entry: CatalogEntry) -> SlotDefaults: """Load defaults for a catalog slot from .env (never exposed via API as plaintext).""" prefix = _prefix(catalog_key) meta = SLOT_AUTH.get(catalog_key, {"auth": "ssh", "secret_field": "PASSWORD"}) secret_kind = "api_key" if meta["auth"] == "api_key" else "password" port_raw = _env(f"DEVICE_{prefix}_PORT") port = int(port_raw) if port_raw and port_raw.isdigit() else None username = _env(f"DEVICE_{prefix}_USERNAME") secret: str | None = None if meta["secret_field"] == "API_KEY": secret = _env(f"DEVICE_{prefix}_API_KEY") or _env("DEVICE_API_KEY") else: secret = _env(f"DEVICE_{prefix}_PASSWORD") if meta["auth"] == "ssh": username = username or _env("DEVICE_SSH_USERNAME") secret = secret or _env("DEVICE_SSH_PASSWORD") if meta["auth"] == "api_key" and not username: username = _env(f"DEVICE_{prefix}_USERNAME") from_env = any( [ port is not None, bool(username), bool(secret), ] ) return SlotDefaults( port=port, username=username, secret=secret, secret_kind=secret_kind, from_env=from_env, ) def _truthy_env(name: str) -> bool: val = os.environ.get(name, "").strip().lower() return val in ("1", "true", "yes", "on") PROXMOX_DEFAULT_API_PORT = 8006 def build_proxmox_api_url(public_ip: str) -> str | None: """Build Proxmox API URL from vessel/device public IP unless DEVICE_PROXMOX_API_URL is set.""" override = _env("DEVICE_PROXMOX_API_URL") if override: return override if not public_ip: return None port_raw = _env("DEVICE_PROXMOX_API_PORT") port = int(port_raw) if port_raw and port_raw.isdigit() else PROXMOX_DEFAULT_API_PORT host = public_ip.strip().removeprefix("https://").removeprefix("http://").split("/")[0] if ":" in host and not host.startswith("["): # Already host:port — use as-is with scheme only. return f"https://{host}" return f"https://{host}:{port}" def resolve_proxmox_api_defaults(public_ip: str) -> dict: """Proxmox VE API token settings from env (used by proxmox-mcp connect). ``public_ip`` is the vessel public IP stored on the proxmox device row (``device.address``). When DEVICE_PROXMOX_API_URL is blank, the URL becomes ``https://:8006``. """ return { "proxmox_api_url": build_proxmox_api_url(public_ip), "proxmox_token_id": _env("DEVICE_PROXMOX_TOKEN_ID"), "proxmox_token_secret": _env("DEVICE_PROXMOX_TOKEN_SECRET"), "proxmox_verify_ssl": _truthy_env("DEVICE_PROXMOX_VERIFY_SSL"), } def proxmox_api_configured(ctx: dict) -> bool: return bool(ctx.get("proxmox_token_id") and ctx.get("proxmox_token_secret")) def merge_selection_with_defaults( catalog_key: str, entry: CatalogEntry, *, port: int | None, username: str | None, secret: str | None, existing_secret: str | None = None, ) -> tuple[int | None, str | None, str | None]: """Apply env defaults where the UI/API did not supply a value.""" defaults = resolve_slot_defaults(catalog_key, entry) resolved_port = port if port is not None else defaults.port if defaults.port is not None else entry.default_port resolved_username = username if username is not None else defaults.username or entry.default_username if secret: resolved_secret = secret elif existing_secret: resolved_secret = existing_secret else: resolved_secret = defaults.secret return resolved_port, resolved_username, resolved_secret