"""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()