feat: git-based agent version reference and unified outdated badge (0.20.0)
SAC reads latest RDP/ssh-monitor versions from configured git repos for outdated detection, update targets, and settings test-git.
This commit is contained in:
@@ -60,6 +60,75 @@ def is_agent_version_outdated(
|
||||
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(
|
||||
|
||||
Reference in New Issue
Block a user