PTah
81 lines
2.3 KiB
Python
81 lines
2.3 KiB
Python
"""Сравнение версий агентов (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 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
|