"""Request agent self-update, ingest lifecycle, SSH/WinRM fallback.""" from __future__ import annotations from datetime import datetime, timedelta, timezone from sqlalchemy import select from sqlalchemy.orm import Session from app.models import Host from app.services.agent_update_types import AgentUpdateFallbackResult from app.services.agent_update_settings import ( PRODUCT_RDP, PRODUCT_SSH, AgentUpdateConfig, get_effective_agent_update_config, ) from app.services.agent_git_release import get_git_release_versions, recommended_from_git from app.services.agent_version import ( host_version_outdated, latest_agent_versions_by_product, ) from app.services.linux_admin_settings import get_effective_linux_admin_for_host from app.services.ssh_connect import ( HostNotLinuxError, HostTargetMissingError, is_linux_host, iter_ssh_targets, run_ssh_monitor_update, ) from app.services.win_admin_settings import get_effective_win_admin_for_host from app.services.winrm_connect import ( HostNotWindowsError, HostTargetMissingError as WinRmHostTargetMissingError, is_windows_host, iter_winrm_targets, run_winrm_on_host_targets, run_winrm_rdp_monitor_update, ) AGENT_UPDATE_STARTED = "agent.update.started" AGENT_UPDATE_SUCCESS = "agent.update.success" AGENT_UPDATE_FAILED = "agent.update.failed" AGENT_UPDATE_TYPES = frozenset( {AGENT_UPDATE_STARTED, AGENT_UPDATE_SUCCESS, AGENT_UPDATE_FAILED} ) class AgentUpdateNotAllowedError(Exception): pass def resolve_target_version( db: Session, host: Host, cfg: AgentUpdateConfig, *, git_versions: dict[str, str] | None = None, ) -> str: product = host.product or "" if git_versions is None: git_versions = get_git_release_versions(cfg, db=db).versions git_target = recommended_from_git(cfg, git_versions, product) if git_target: return git_target latest_map = latest_agent_versions_by_product(db) return latest_map.get(product, "") or (host.product_version or "") def host_version_status( host: Host, *, cfg: AgentUpdateConfig, latest_map: dict[str, str], git_versions: dict[str, str] | None = None, ) -> str: product = host.product or "" if git_versions is None: git_versions = get_git_release_versions(cfg).versions outdated = host_version_outdated( host.product_version, recommended=recommended_from_git(cfg, git_versions, product), fleet_latest=latest_map.get(product, ""), min_version=cfg.min_for_product(product), ) if outdated is None: return "unknown" return "outdated" if outdated else "ok" def request_agent_update(db: Session, host: Host) -> Host: cfg = get_effective_agent_update_config(db) if not cfg.sac_managed: raise AgentUpdateNotAllowedError( "Agent updates from SAC are disabled (mode=gpo). Use GPO/manual update or switch to mode=sac." ) if not host.agent_instance_id: raise AgentUpdateNotAllowedError("Host has no agent_instance_id (agent never connected via SAC)") target = resolve_target_version(db, host, cfg) now = datetime.now(timezone.utc) host.pending_agent_update = True host.pending_update_requested_at = now host.pending_update_target_version = target or None host.agent_update_state = "requested" host.agent_update_last_error = None db.flush() return host def _clear_pending(host: Host) -> None: host.pending_agent_update = False host.pending_update_requested_at = None host.pending_update_target_version = None def process_agent_update_ingest(db: Session, host: Host, event_type: str, details: dict | None) -> None: if event_type not in AGENT_UPDATE_TYPES: return now = datetime.now(timezone.utc) host.agent_update_last_at = now details = details if isinstance(details, dict) else {} if event_type == AGENT_UPDATE_STARTED: host.agent_update_state = "started" return if event_type == AGENT_UPDATE_SUCCESS: host.agent_update_state = "success" host.agent_update_last_error = None _clear_pending(host) version = details.get("product_version") or details.get("version") if isinstance(version, str) and version.strip(): host.product_version = version.strip() return if event_type == AGENT_UPDATE_FAILED: host.agent_update_state = "failed" err = details.get("error") or details.get("message") host.agent_update_last_error = str(err)[:2000] if err else "agent.update.failed" _clear_pending(host) def run_linux_agent_update_fallback( host: Host, cfg_linux, *, repo_url: str | None = None, git_branch: str | None = None, ) -> AgentUpdateFallbackResult: if not is_linux_host(host): return AgentUpdateFallbackResult(ok=False, message="Host is not Linux", product_version=None) if not cfg_linux.configured: return AgentUpdateFallbackResult( ok=False, message="Linux SSH admin is not configured", product_version=None, ) last = AgentUpdateFallbackResult(ok=False, message="SSH fallback failed", product_version=None) try: targets = iter_ssh_targets(host) except (HostNotLinuxError, HostTargetMissingError) as exc: return AgentUpdateFallbackResult(ok=False, message=str(exc), product_version=None) effective_repo = (repo_url or "").strip() or None effective_branch = (git_branch or "main").strip() or "main" for target in targets: result = run_ssh_monitor_update( target=target, user=cfg_linux.user, password=cfg_linux.password, repo_url=effective_repo, git_branch=effective_branch, ) last = AgentUpdateFallbackResult( ok=result.ok, message=result.message, product_version=result.agent_version, target=result.target, stdout=result.stdout, stderr=result.stderr, exit_code=result.exit_code, ) if result.ok: return last return last def run_windows_agent_update_fallback( host: Host, cfg_win, script_path: str, *, repo_url: str = "", git_branch: str = "main", ) -> AgentUpdateFallbackResult: if not is_windows_host(host): return AgentUpdateFallbackResult(ok=False, message="Host is not Windows", product_version=None) if not cfg_win.configured: return AgentUpdateFallbackResult( ok=False, message="Windows domain admin is not configured", product_version=None, ) try: targets = iter_winrm_targets(host) except (HostNotWindowsError, WinRmHostTargetMissingError) as exc: return AgentUpdateFallbackResult(ok=False, message=str(exc), product_version=None) result, attempts = run_winrm_on_host_targets( host, user=cfg_win.user, password=cfg_win.password, action=lambda target: run_winrm_rdp_monitor_update( target=target, user=cfg_win.user, password=cfg_win.password, script_path=script_path, repo_url=repo_url, git_branch=git_branch, ), ) if result is None: return AgentUpdateFallbackResult(ok=False, message="WinRM fallback failed", product_version=None) message = result.message if attempts: message = f"{message} (пробовали: {', '.join(attempts)})" version = None if result.ok and result.stdout: from app.services.ssh_connect import parse_ssh_monitor_version_text version = parse_ssh_monitor_version_text(result.stdout) return AgentUpdateFallbackResult( ok=result.ok, message=message, product_version=version, target=result.target, stdout=result.stdout, stderr=result.stderr, exit_code=result.exit_code, ) def execute_agent_update_fallback(db: Session, host: Host) -> AgentUpdateFallbackResult: cfg = get_effective_agent_update_config(db) product = (host.product or "").strip() if product == PRODUCT_SSH or is_linux_host(host): linux_cfg = get_effective_linux_admin_for_host(db, host) agent_cfg = get_effective_agent_update_config(db) repo_url = (agent_cfg.ssh_git_repo_url or "").strip() or None branch = (agent_cfg.git_branch or "main").strip() or "main" result = run_linux_agent_update_fallback( host, linux_cfg, repo_url=repo_url, git_branch=branch, ) elif product == PRODUCT_RDP or is_windows_host(host): win_cfg = get_effective_win_admin_for_host(db, host) result = run_windows_agent_update_fallback( host, win_cfg, cfg.win_agent_update_script, repo_url=cfg.rdp_git_repo_url, git_branch=cfg.git_branch, ) else: return AgentUpdateFallbackResult( ok=False, message=f"Unsupported product for fallback update: {product}", product_version=None, ) now = datetime.now(timezone.utc) host.agent_update_last_at = now if result.ok: host.agent_update_state = "success" host.agent_update_last_error = None _clear_pending(host) if result.product_version: host.product_version = result.product_version if is_linux_host(host): host.ssh_admin_ok = True host.ssh_admin_checked_at = now else: host.agent_update_state = "failed" host.agent_update_last_error = result.message[:2000] db.flush() return result def process_agent_update_fallbacks(db: Session) -> list[dict]: """Find stale pending updates and run SSH/WinRM fallback when enabled.""" cfg = get_effective_agent_update_config(db) if not cfg.sac_managed or not cfg.fallback_enabled: return [] cutoff = datetime.now(timezone.utc) - timedelta(minutes=cfg.fallback_after_minutes) rows = db.scalars( select(Host).where( Host.pending_agent_update.is_(True), Host.pending_update_requested_at.isnot(None), Host.pending_update_requested_at <= cutoff, Host.agent_update_state.in_(("requested", "started")), ) ).all() results: list[dict] = [] for host in rows: result = execute_agent_update_fallback(db, host) results.append( { "host_id": host.id, "hostname": host.hostname, "ok": result.ok, "message": result.message, "product_version": result.product_version, } ) return results