"""Global agent update policy (DB overrides env).""" from __future__ import annotations from dataclasses import dataclass from sqlalchemy.orm import Session from app.config import get_settings from app.models.ui_settings import UI_SETTINGS_ROW_ID, UiSettings PRODUCT_RDP = "rdp-login-monitor" PRODUCT_SSH = "ssh-monitor" AGENT_UPDATE_MODES = frozenset({"gpo", "sac"}) @dataclass(frozen=True) class AgentUpdateConfig: mode: str fallback_enabled: bool fallback_after_minutes: int recommended_rdp_version: str recommended_ssh_version: str min_rdp_version: str min_ssh_version: str win_agent_update_script: str rdp_git_repo_url: str ssh_git_repo_url: str git_branch: str source: str # env | db @property def sac_managed(self) -> bool: return self.mode == "sac" def recommended_for_product(self, product: str) -> str: if product == PRODUCT_RDP: return self.recommended_rdp_version if product == PRODUCT_SSH: return self.recommended_ssh_version return "" def min_for_product(self, product: str) -> str: if product == PRODUCT_RDP: return self.min_rdp_version if product == PRODUCT_SSH: return self.min_ssh_version return "" def _agent_update_from_env() -> AgentUpdateConfig: settings = get_settings() mode = (settings.sac_agent_update_mode or "gpo").strip().lower() if mode not in AGENT_UPDATE_MODES: mode = "gpo" return AgentUpdateConfig( mode=mode, fallback_enabled=settings.sac_agent_update_fallback_enabled, fallback_after_minutes=max(1, int(settings.sac_agent_update_fallback_minutes)), recommended_rdp_version=(settings.sac_agent_recommended_rdp_version or "").strip(), recommended_ssh_version=(settings.sac_agent_recommended_ssh_version or "").strip(), min_rdp_version=(settings.sac_agent_min_rdp_version or "").strip(), min_ssh_version=(settings.sac_agent_min_ssh_version or "").strip(), win_agent_update_script=(settings.sac_win_agent_update_script or "").strip(), rdp_git_repo_url=(settings.sac_agent_rdp_git_repo_url or "").strip(), ssh_git_repo_url=(settings.sac_agent_ssh_git_repo_url or "").strip(), git_branch=(settings.sac_agent_git_branch or "main").strip() or "main", source="env", ) def _row_has_agent_update_values(row: UiSettings) -> bool: return any( [ (row.agent_update_mode or "").strip() not in ("", "gpo"), row.agent_update_fallback_enabled is not None, row.agent_update_fallback_minutes not in (None, 15), bool((row.agent_recommended_rdp_version or "").strip()), bool((row.agent_recommended_ssh_version or "").strip()), bool((row.agent_min_rdp_version or "").strip()), bool((row.agent_min_ssh_version or "").strip()), bool((row.win_agent_update_script or "").strip()), bool((row.agent_rdp_git_repo_url or "").strip()), bool((row.agent_ssh_git_repo_url or "").strip()), (row.agent_git_branch or "main").strip() not in ("", "main"), ] ) def get_effective_agent_update_config(db: Session | None = None) -> AgentUpdateConfig: if db is None: from app.database import SessionLocal session = SessionLocal() try: return get_effective_agent_update_config(session) finally: session.close() row = db.get(UiSettings, UI_SETTINGS_ROW_ID) env_cfg = _agent_update_from_env() if row is None: return env_cfg mode = (row.agent_update_mode or env_cfg.mode).strip().lower() if mode not in AGENT_UPDATE_MODES: mode = env_cfg.mode return AgentUpdateConfig( mode=mode, fallback_enabled=( row.agent_update_fallback_enabled if row.agent_update_fallback_enabled is not None else env_cfg.fallback_enabled ), fallback_after_minutes=max( 1, int(row.agent_update_fallback_minutes or env_cfg.fallback_after_minutes), ), recommended_rdp_version=(row.agent_recommended_rdp_version or "").strip() or env_cfg.recommended_rdp_version, recommended_ssh_version=(row.agent_recommended_ssh_version or "").strip() or env_cfg.recommended_ssh_version, min_rdp_version=(row.agent_min_rdp_version or "").strip() or env_cfg.min_rdp_version, min_ssh_version=(row.agent_min_ssh_version or "").strip() or env_cfg.min_ssh_version, win_agent_update_script=(row.win_agent_update_script or "").strip() or env_cfg.win_agent_update_script, rdp_git_repo_url=(row.agent_rdp_git_repo_url or "").strip() or env_cfg.rdp_git_repo_url, ssh_git_repo_url=(row.agent_ssh_git_repo_url or "").strip() or env_cfg.ssh_git_repo_url, git_branch=(row.agent_git_branch or env_cfg.git_branch).strip() or env_cfg.git_branch, source="db" if _row_has_agent_update_values(row) else env_cfg.source, ) def upsert_agent_update_settings( db: Session, *, mode: str | None = None, fallback_enabled: bool | None = None, fallback_after_minutes: int | None = None, recommended_rdp_version: str | None = None, recommended_ssh_version: str | None = None, min_rdp_version: str | None = None, min_ssh_version: str | None = None, win_agent_update_script: str | None = None, rdp_git_repo_url: str | None = None, ssh_git_repo_url: str | None = None, git_branch: str | None = None, ) -> AgentUpdateConfig: row = db.get(UiSettings, UI_SETTINGS_ROW_ID) if row is None: row = UiSettings(id=UI_SETTINGS_ROW_ID, show_sidebar_system_stats=True) db.add(row) if mode is not None: normalized = mode.strip().lower() if normalized not in AGENT_UPDATE_MODES: raise ValueError(f"mode must be one of: {sorted(AGENT_UPDATE_MODES)}") row.agent_update_mode = normalized if fallback_enabled is not None: row.agent_update_fallback_enabled = fallback_enabled if fallback_after_minutes is not None: row.agent_update_fallback_minutes = max(1, int(fallback_after_minutes)) if recommended_rdp_version is not None: row.agent_recommended_rdp_version = recommended_rdp_version.strip() or None if recommended_ssh_version is not None: row.agent_recommended_ssh_version = recommended_ssh_version.strip() or None if min_rdp_version is not None: row.agent_min_rdp_version = min_rdp_version.strip() or None if min_ssh_version is not None: row.agent_min_ssh_version = min_ssh_version.strip() or None if win_agent_update_script is not None: row.win_agent_update_script = win_agent_update_script.strip() or None if rdp_git_repo_url is not None: row.agent_rdp_git_repo_url = rdp_git_repo_url.strip() or None row.agent_git_rdp_version = None row.agent_git_fetched_at = None if ssh_git_repo_url is not None: row.agent_ssh_git_repo_url = ssh_git_repo_url.strip() or None row.agent_git_ssh_version = None row.agent_git_fetched_at = None if git_branch is not None: branch = git_branch.strip() or "main" if branch != (row.agent_git_branch or "main"): row.agent_git_rdp_version = None row.agent_git_ssh_version = None row.agent_git_fetched_at = None row.agent_git_branch = branch db.commit() db.refresh(row) return get_effective_agent_update_config(db)