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

"""Сравнение версий агентов (ssh-monitor / rdp-login-monitor, формат X.Y.Z-SAC)."""

from __future__ import annotations

import re
from dataclasses import dataclass

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models import Host

_VERSION_RE = re.compile(r"^(\d+)\.(\d+)\.(\d+)(?:-SAC)?$", re.IGNORECASE)
_DEFAULT_LAG_THRESHOLD = 3


@dataclass(frozen=True, order=True)
class ParsedAgentVersion:
    major: int
    minor: int
    patch: int


def parse_agent_version(raw: str | None) -> ParsedAgentVersion | None:
    if not raw:
        return None
    text = raw.strip()
    match = _VERSION_RE.match(text)
    if not match:
        return None
    return ParsedAgentVersion(
        major=int(match.group(1)),
        minor=int(match.group(2)),
        patch=int(match.group(3)),
    )


def agent_version_lag(host: ParsedAgentVersion, latest: ParsedAgentVersion) -> int | None:
    """Отставание от latest: patch при том же major.minor, иначе None (любое отставание)."""
    if host > latest:
        return 0
    if host.major != latest.major or host.minor != latest.minor:
        return None
    return latest.patch - host.patch


def is_agent_version_outdated(
    host_version: str | None,
    latest_version: str | None,
    *,
    lag_threshold: int = _DEFAULT_LAG_THRESHOLD,
) -> bool:
    host = parse_agent_version(host_version)
    latest = parse_agent_version(latest_version)
    if host is None or latest is None or host >= latest:
        return False
    lag = agent_version_lag(host, latest)
    if lag is None:
        return True
    return lag >= lag_threshold


def effective_reference_version(
    *,
    recommended: str = "",
    fleet_latest: str = "",
    min_version: str = "",
) -> str | None:
    """Единый эталон: max(recommended, min_version, max версия в fleet по продукту)."""
    best: ParsedAgentVersion | None = None
    best_raw: str | None = None
    for raw in (recommended, min_version, fleet_latest):
        text = (raw or "").strip()
        parsed = parse_agent_version(text)
        if parsed is None:
            continue
        if best is None or parsed > best:
            best = parsed
            best_raw = text
    return best_raw


def reference_agent_versions_by_product(
    cfg,
    latest_map: dict[str, str],
    *,
    git_versions: dict[str, str] | None = None,
    products: tuple[str, ...] = ("rdp-login-monitor", "ssh-monitor"),
) -> dict[str, str]:
    from app.services.agent_git_release import recommended_from_git
    from app.services.agent_update_settings import PRODUCT_RDP, PRODUCT_SSH

    git_versions = git_versions or {}
    product_cfg = {
        PRODUCT_RDP: (recommended_from_git(cfg, git_versions, PRODUCT_RDP), cfg.min_for_product(PRODUCT_RDP)),
        PRODUCT_SSH: (recommended_from_git(cfg, git_versions, PRODUCT_SSH), cfg.min_for_product(PRODUCT_SSH)),
    }
    out: dict[str, str] = {}
    for product in products:
        recommended, min_version = product_cfg.get(product, ("", ""))
        ref = effective_reference_version(
            recommended=recommended,
            fleet_latest=latest_map.get(product, ""),
            min_version=min_version,
        )
        if ref:
            out[product] = ref
    return out


def host_version_outdated(
    host_version: str | None,
    *,
    recommended: str = "",
    fleet_latest: str = "",
    min_version: str = "",
    lag_threshold: int = _DEFAULT_LAG_THRESHOLD,
) -> bool | None:
    """True=outdated, False=ok, None=unknown (нет эталона или непарсится версия хоста)."""
    if not host_version or parse_agent_version(host_version) is None:
        return None
    reference = effective_reference_version(
        recommended=recommended,
        fleet_latest=fleet_latest,
        min_version=min_version,
    )
    if not reference:
        return None
    return is_agent_version_outdated(host_version, reference, lag_threshold=lag_threshold)


def latest_agent_versions_by_product(db: Session) -> dict[str, str]:
    rows = db.execute(
        select(Host.product, Host.product_version).where(
            Host.product_version.isnot(None),
            Host.product_version != "",
        )
    ).all()
    best: dict[str, ParsedAgentVersion] = {}
    best_raw: dict[str, str] = {}
    for product, version in rows:
        parsed = parse_agent_version(version)
        if parsed is None:
            continue
        current = best.get(product)
        if current is None or parsed > current:
            best[product] = parsed
            best_raw[product] = version.strip()
    return best_raw