security-alert-center/backend/app/services/agent_update.py

"""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_settings import (
    PRODUCT_RDP,
    PRODUCT_SSH,
    AgentUpdateConfig,
    get_effective_agent_update_config,
)
from app.services.agent_version import (
    is_agent_version_outdated,
    latest_agent_versions_by_product,
    parse_agent_version,
)
from app.services.linux_admin_settings import get_effective_linux_admin_config
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_config
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) -> str:
    recommended = cfg.recommended_for_product(host.product or "")
    if recommended:
        return recommended
    latest_map = latest_agent_versions_by_product(db)
    return latest_map.get(host.product or "", "") or (host.product_version or "")


def host_version_status(
    host: Host,
    *,
    cfg: AgentUpdateConfig,
    latest_map: dict[str, str],
) -> str:
    current = host.product_version
    if not current or parse_agent_version(current) is None:
        return "unknown"
    recommended = cfg.recommended_for_product(host.product or "")
    reference = recommended or latest_map.get(host.product or "")
    if not reference:
        return "unknown"
    if is_agent_version_outdated(current, reference):
        return "outdated"
    return "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) -> tuple[bool, str, str | None]:
    if not is_linux_host(host):
        return False, "Host is not Linux", None
    if not cfg_linux.configured:
        return False, "Linux SSH admin is not configured", None

    last_message = "SSH fallback failed"
    version: str | None = None
    try:
        targets = iter_ssh_targets(host)
    except (HostNotLinuxError, HostTargetMissingError) as exc:
        return False, str(exc), None

    for target in targets:
        result = run_ssh_monitor_update(target=target, user=cfg_linux.user, password=cfg_linux.password)
        last_message = result.message
        if result.ok:
            version = result.agent_version
            return True, result.message, version
    return False, last_message, version


def run_windows_agent_update_fallback(host: Host, cfg_win, script_path: str) -> tuple[bool, str, str | None]:
    if not is_windows_host(host):
        return False, "Host is not Windows", None
    if not cfg_win.configured:
        return False, "Windows domain admin is not configured", None

    try:
        targets = iter_winrm_targets(host)
    except (HostNotWindowsError, WinRmHostTargetMissingError) as exc:
        return False, str(exc), 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,
        ),
    )
    if result is None:
        return False, "WinRM fallback failed", 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 result.ok, message, version


def execute_agent_update_fallback(db: Session, host: Host) -> tuple[bool, str, str | None]:
    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_config(db)
        ok, message, version = run_linux_agent_update_fallback(host, linux_cfg)
    elif product == PRODUCT_RDP or is_windows_host(host):
        win_cfg = get_effective_win_admin_config(db)
        ok, message, version = run_windows_agent_update_fallback(host, win_cfg, cfg.win_agent_update_script)
    else:
        return False, f"Unsupported product for fallback update: {product}", None

    now = datetime.now(timezone.utc)
    host.agent_update_last_at = now
    if ok:
        host.agent_update_state = "success"
        host.agent_update_last_error = None
        _clear_pending(host)
        if version:
            host.product_version = 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 = message[:2000]

    db.flush()
    return ok, message, version


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:
        ok, message, version = execute_agent_update_fallback(db, host)
        results.append(
            {
                "host_id": host.id,
                "hostname": host.hostname,
                "ok": ok,
                "message": message,
                "product_version": version,
            }
        )
    return results