cx-ui/backend/app/config.py

"""Environment-driven configuration.

Everything deployment-specific comes from the environment so the same image runs
locally under compose and in Kubernetes with only a ConfigMap/Secret difference.
"""
from __future__ import annotations

import os
from functools import lru_cache


def _bool(name: str, default: bool = False) -> bool:
    return str(os.environ.get(name, str(default))).strip().lower() in {"1", "true", "yes", "on"}


def _int(name: str, default: int) -> int:
    try:
        return int(os.environ.get(name, default))
    except (TypeError, ValueError):
        return default


class Settings:
    """Read once per instance, at construction time.

    These were class attributes, which meant the environment was read when the
    module was first imported and never again - so a fresh Settings() silently
    returned stale values. Reading in __init__ makes construction mean what it
    looks like it means.
    """

    def __init__(self) -> None:
        # --- app ---------------------------------------------------------------
        self.app_name = os.environ.get("CX_APP_NAME", "CX Triage")
        self.base_url = os.environ.get("CX_BASE_URL", "http://localhost:8080")
        self.secret_key = os.environ.get("CX_SECRET_KEY", "dev-only-change-me")
        self.session_hours = _int("CX_SESSION_HOURS", 12)
        self.static_dir = os.environ.get("CX_STATIC_DIR", "/app/static")

        # --- database ----------------------------------------------------------
        # sqlite for local/compose, postgres in the cluster.
        self.database_url = os.environ.get("CX_DATABASE_URL", "sqlite:////data/cx-triage.db")

        # --- data sources ------------------------------------------------------
        self.prometheus_base = os.environ.get("CX_PROMETHEUS_BASE", "http://10.11.254.250:9090")
        self.prometheus_relay = os.environ.get("CX_PROMETHEUS_RELAY", "")
        self.cx_tools_path = os.environ.get("CX_TOOLS_PATH", "")

        # --- auth --------------------------------------------------------------
        # Local accounts are for development and for a cluster without SSO yet.
        # When CX_OIDC_ENABLED is on, Authentik becomes the source of truth.
        self.auth_local_enabled = _bool("CX_AUTH_LOCAL_ENABLED", True)
        self.bootstrap_admin_email = os.environ.get("CX_BOOTSTRAP_ADMIN_EMAIL", "admin@localhost")
        self.bootstrap_admin_password = os.environ.get("CX_BOOTSTRAP_ADMIN_PASSWORD", "")

        self.oidc_enabled = _bool("CX_OIDC_ENABLED", False)
        self.oidc_issuer = os.environ.get("CX_OIDC_ISSUER", "")            # e.g. https://sso/application/o/cx-triage/
        self.oidc_client_id = os.environ.get("CX_OIDC_CLIENT_ID", "")
        self.oidc_client_secret = os.environ.get("CX_OIDC_CLIENT_SECRET", "")
        self.oidc_scopes = os.environ.get("CX_OIDC_SCOPES", "openid email profile")
        self.oidc_admin_group = os.environ.get("CX_OIDC_ADMIN_GROUP", "cx-triage-admins")
        self.oidc_groups_claim = os.environ.get("CX_OIDC_GROUPS_CLAIM", "groups")

        # --- feature flags -----------------------------------------------------
        # Sending must be switched on deliberately; a demo instance cannot email.
        self.feature_send_enabled = _bool("CX_FEATURE_SEND_ENABLED", False)
        self.feature_zendesk = _bool("CX_FEATURE_ZENDESK", False)
        self.feature_jira = _bool("CX_FEATURE_JIRA", False)
        self.feature_linkage_scan = _bool("CX_FEATURE_LINKAGE_SCAN", True)
        self.send_daily_cap = _int("CX_SEND_DAILY_CAP", 25)

        # --- integrations ------------------------------------------------------
        self.zendesk_subdomain = os.environ.get("CX_ZENDESK_SUBDOMAIN", "")
        self.zendesk_email = os.environ.get("CX_ZENDESK_EMAIL", "")
        self.zendesk_token = os.environ.get("CX_ZENDESK_TOKEN", "")
        self.zendesk_default_public = _bool("CX_ZENDESK_PUBLIC_REPLY", True)

        # Jira, instance 1: the Infrahub/OpenStack side (OIE).
        self.jira_base = os.environ.get("CX_JIRA_BASE", "")
        self.jira_email = os.environ.get("CX_JIRA_EMAIL", "")
        self.jira_token = os.environ.get("CX_JIRA_TOKEN", "")
        self.jira_project = os.environ.get("CX_JIRA_PROJECT", "OIE")
        self.jira_issue_type = os.environ.get("CX_JIRA_ISSUE_TYPE", "Task")

        # Jira, instance 2: RunPod hosts and RMAs. These may be a different
        # Atlassian site entirely, so they get their own credentials; anything left
        # blank falls back to the values above rather than failing.
        self.runpod_jira_base = os.environ.get("CX_RUNPOD_JIRA_BASE", "")
        self.runpod_jira_email = os.environ.get("CX_RUNPOD_JIRA_EMAIL", "")
        self.runpod_jira_token = os.environ.get("CX_RUNPOD_JIRA_TOKEN", "")
        self.runpod_jira_project = os.environ.get("CX_RUNPOD_JIRA_PROJECT", "RMA")
        self.runpod_jira_issue_type = os.environ.get("CX_RUNPOD_JIRA_ISSUE_TYPE", "Task")

        # --- RunPod ------------------------------------------------------------
        # The API key is the supported path. Email/password only reaches a console
        # login that now requires a second factor, so it cannot run unattended -
        # see app/runpod/client.py.
        self.runpod_api_key = os.environ.get("CX_RUNPOD_API_KEY", "")
        self.runpod_email = os.environ.get("CX_RUNPOD_EMAIL", "")
        self.runpod_password = os.environ.get("CX_RUNPOD_PASSWORD", "")
        self.runpod_team_id = os.environ.get("CX_RUNPOD_TEAM_ID", "")
        self.runpod_totp_secret = os.environ.get("CX_RUNPOD_TOTP_SECRET", "")
        self.feature_runpod = _bool("CX_FEATURE_RUNPOD", True)
        self.feature_handover = _bool("CX_FEATURE_HANDOVER", True)
        # Unlisting a machine is destructive to earnings; gated separately.
        self.feature_runpod_write = _bool("CX_FEATURE_RUNPOD_WRITE", False)

    def jira_for(self, scope: str) -> dict:
        """Credentials for a Jira scope: 'runpod' or anything else (default)."""
        if scope == "runpod":
            return {
                "base": self.runpod_jira_base or self.jira_base,
                "email": self.runpod_jira_email or self.jira_email,
                "token": self.runpod_jira_token or self.jira_token,
                "project": self.runpod_jira_project,
                "issue_type": self.runpod_jira_issue_type,
            }
        return {
            "base": self.jira_base, "email": self.jira_email, "token": self.jira_token,
            "project": self.jira_project, "issue_type": self.jira_issue_type,
        }

    @property
    def runpod_ready(self) -> bool:
        return bool(self.feature_runpod and (self.runpod_api_key or
                                             (self.runpod_email and self.runpod_password)))

    @property
    def runpod_jira_ready(self) -> bool:
        creds = self.jira_for("runpod")
        return bool(self.feature_jira and creds["base"] and creds["email"] and creds["token"])

    @property
    def zendesk_ready(self) -> bool:
        return bool(self.feature_zendesk and self.zendesk_subdomain
                    and self.zendesk_email and self.zendesk_token)

    @property
    def jira_ready(self) -> bool:
        return bool(self.feature_jira and self.jira_base and self.jira_email
                    and self.jira_token and self.jira_project)

    def public_flags(self) -> dict:
        """What the frontend is allowed to know - never secrets."""
        return {
            "app_name": self.app_name,
            "oidc_enabled": self.oidc_enabled,
            "local_login": self.auth_local_enabled,
            "zendesk_ready": self.zendesk_ready,
            "jira_ready": self.jira_ready,
            "send_enabled": self.feature_send_enabled,
            "linkage_scan": self.feature_linkage_scan,
            "jira_project": self.jira_project if self.jira_ready else "",
            "runpod_ready": self.runpod_ready,
            "runpod_write": self.feature_runpod_write,
            "runpod_jira_ready": self.runpod_jira_ready,
            "runpod_jira_project": self.runpod_jira_project if self.runpod_jira_ready else "",
            "handover": self.feature_handover,
        }


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