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

"""Fetch latest agent release versions from git repositories."""

from __future__ import annotations

import hashlib
import re
import shutil
import subprocess
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlparse

from sqlalchemy.orm import Session

from app.config import get_settings
from app.models.ui_settings import UI_SETTINGS_ROW_ID, UiSettings
from app.services.agent_update_settings import PRODUCT_RDP, PRODUCT_SSH, AgentUpdateConfig
from app.services.agent_version import parse_agent_version

_SCRIPT_VERSION_RE = re.compile(r'\$ScriptVersion\s*=\s*["\']([^"\']+)["\']', re.IGNORECASE)
_SSH_VERSION_RE = re.compile(r'^SSH_MONITOR_VERSION=["\']([^"\']+)["\']', re.MULTILINE)

_MEMORY_CACHE: dict[str, tuple[dict[str, str], float, str]] = {}


@dataclass(frozen=True)
class GitReleaseVersions:
    versions: dict[str, str]
    fetched_at: datetime | None
    from_cache: bool
    errors: dict[str, str] = field(default_factory=dict)


def normalize_git_repo_url(raw: str) -> str:
    text = (raw or "").strip()
    if not text:
        return ""
    if not text.startswith(("http://", "https://", "git@")):
        text = f"https://{text.lstrip('/')}"
    if text.startswith("git@"):
        return text
    if not text.endswith(".git"):
        text = f"{text.rstrip('/')}.git"
    return text


def git_repo_url_with_auth(repo_url: str, token: str) -> str:
    normalized = normalize_git_repo_url(repo_url)
    if not token or not normalized.startswith("https://"):
        return normalized
    parsed = urlparse(normalized)
    host = parsed.netloc
    path = parsed.path or ""
    return f"https://oauth2:{token}@{host}{path}"


def _cache_key(cfg: AgentUpdateConfig) -> str:
    parts = [
        cfg.rdp_git_repo_url,
        cfg.ssh_git_repo_url,
        cfg.git_branch,
    ]
    return hashlib.sha256("|".join(parts).encode()).hexdigest()


def _read_text(path: Path) -> str | None:
    if not path.is_file():
        return None
    return path.read_text(encoding="utf-8-sig")


def parse_rdp_version_from_files(version_txt: str | None, login_monitor_ps1: str | None) -> str | None:
    if version_txt:
        first_line = version_txt.strip().splitlines()[0].strip() if version_txt.strip() else ""
        if parse_agent_version(first_line):
            return first_line
    if login_monitor_ps1:
        match = _SCRIPT_VERSION_RE.search(login_monitor_ps1)
        if match:
            candidate = match.group(1).strip()
            if parse_agent_version(candidate):
                return candidate
    return None


def parse_ssh_version_from_files(ssh_monitor: str | None, version_txt: str | None) -> str | None:
    if ssh_monitor:
        match = _SSH_VERSION_RE.search(ssh_monitor)
        if match:
            candidate = match.group(1).strip()
            if parse_agent_version(candidate):
                return candidate
    if version_txt:
        first_line = version_txt.strip().splitlines()[0].strip() if version_txt.strip() else ""
        if parse_agent_version(first_line):
            return first_line
    return None


def parse_product_version_from_dir(repo_dir: Path, product: str) -> str | None:
    if product == PRODUCT_RDP:
        return parse_rdp_version_from_files(
            _read_text(repo_dir / "version.txt"),
            _read_text(repo_dir / "Login_Monitor.ps1"),
        )
    if product == PRODUCT_SSH:
        return parse_ssh_version_from_files(
            _read_text(repo_dir / "ssh-monitor"),
            _read_text(repo_dir / "version.txt"),
        )
    return None


def _run_git(args: list[str], *, cwd: Path | None = None) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["git", *args],
        cwd=cwd,
        capture_output=True,
        text=True,
        timeout=120,
        check=False,
    )


def _repo_slug(repo_url: str) -> str:
    normalized = normalize_git_repo_url(repo_url)
    name = normalized.rstrip("/").split("/")[-1]
    if name.endswith(".git"):
        name = name[:-4]
    digest = hashlib.sha256(normalized.encode()).hexdigest()[:12]
    return f"{name}-{digest}"


def clone_or_update_repo(repo_url: str, branch: str, cache_dir: Path, token: str = "") -> Path:
    clone_url = git_repo_url_with_auth(repo_url, token)
    normalized = normalize_git_repo_url(repo_url)
    if not normalized:
        raise ValueError("empty repo url")

    branch_name = (branch or "main").strip() or "main"
    target = cache_dir / _repo_slug(normalized)
    cache_dir.mkdir(parents=True, exist_ok=True)

    if target.is_dir() and (target / ".git").is_dir():
        fetch = _run_git(["fetch", "--depth", "1", "origin", branch_name], cwd=target)
        if fetch.returncode != 0:
            shutil.rmtree(target, ignore_errors=True)
        else:
            checkout = _run_git(["checkout", "FETCH_HEAD"], cwd=target)
            if checkout.returncode == 0:
                return target
            shutil.rmtree(target, ignore_errors=True)

    clone = _run_git(
        ["clone", "--depth", "1", "--branch", branch_name, clone_url, str(target)],
    )
    if clone.returncode != 0:
        err = (clone.stderr or clone.stdout or "git clone failed").strip()
        raise RuntimeError(err)
    return target


def fetch_product_version(repo_url: str, branch: str, product: str, cache_dir: Path, token: str = "") -> str:
    repo_dir = clone_or_update_repo(repo_url, branch, cache_dir, token=token)
    version = parse_product_version_from_dir(repo_dir, product)
    if not version:
        raise RuntimeError(f"version not found in {repo_url} ({product})")
    return version


def recommended_from_git(cfg: AgentUpdateConfig, git_versions: dict[str, str], product: str) -> str:
    git_version = (git_versions.get(product) or "").strip()
    if git_version:
        return git_version
    manual = cfg.recommended_for_product(product)
    return manual


def _ttl_seconds() -> int:
    settings = get_settings()
    return max(60, int(settings.sac_agent_git_cache_ttl_minutes) * 60)


def _cache_dir() -> Path:
    settings = get_settings()
    return Path(settings.sac_agent_git_cache_dir or "/tmp/sac-agent-repos")


def _db_cache_fresh(row: UiSettings | None, cfg: AgentUpdateConfig) -> bool:
    if row is None or row.agent_git_fetched_at is None:
        return False
    fetched_at = row.agent_git_fetched_at
    if fetched_at.tzinfo is None:
        fetched_at = fetched_at.replace(tzinfo=timezone.utc)
    age = (datetime.now(timezone.utc) - fetched_at).total_seconds()
    if age > _ttl_seconds():
        return False
    if (row.agent_rdp_git_repo_url or "").strip() != cfg.rdp_git_repo_url:
        return False
    if (row.agent_ssh_git_repo_url or "").strip() != cfg.ssh_git_repo_url:
        return False
    if (row.agent_git_branch or "main").strip() != cfg.git_branch:
        return False
    return True


def _versions_from_db_row(row: UiSettings) -> dict[str, str]:
    out: dict[str, str] = {}
    rdp = (row.agent_git_rdp_version or "").strip()
    ssh = (row.agent_git_ssh_version or "").strip()
    if rdp:
        out[PRODUCT_RDP] = rdp
    if ssh:
        out[PRODUCT_SSH] = ssh
    return out


def _persist_db_cache(db: Session, cfg: AgentUpdateConfig, versions: dict[str, str]) -> datetime:
    now = datetime.now(timezone.utc)
    row = db.get(UiSettings, UI_SETTINGS_ROW_ID)
    if row is None:
        row = UiSettings(id=UI_SETTINGS_ROW_ID, show_sidebar_system_stats=True)
        db.add(row)
    row.agent_rdp_git_repo_url = cfg.rdp_git_repo_url or None
    row.agent_ssh_git_repo_url = cfg.ssh_git_repo_url or None
    row.agent_git_branch = cfg.git_branch or "main"
    row.agent_git_rdp_version = versions.get(PRODUCT_RDP) or None
    row.agent_git_ssh_version = versions.get(PRODUCT_SSH) or None
    row.agent_git_fetched_at = now
    db.commit()
    return now


def get_git_release_versions(
    cfg: AgentUpdateConfig,
    *,
    db: Session | None = None,
    force_refresh: bool = False,
) -> GitReleaseVersions:
    key = _cache_key(cfg)
    now_ts = time.time()
    ttl = _ttl_seconds()

    if not force_refresh:
        cached = _MEMORY_CACHE.get(key)
        if cached and now_ts - cached[1] <= ttl:
            fetched_at = datetime.fromtimestamp(cached[1], tz=timezone.utc)
            return GitReleaseVersions(versions=dict(cached[0]), fetched_at=fetched_at, from_cache=True)

        if db is not None:
            row = db.get(UiSettings, UI_SETTINGS_ROW_ID)
            if _db_cache_fresh(row, cfg):
                versions = _versions_from_db_row(row)
                if versions:
                    _MEMORY_CACHE[key] = (versions, row.agent_git_fetched_at.timestamp(), key)
                    return GitReleaseVersions(
                        versions=versions,
                        fetched_at=row.agent_git_fetched_at,
                        from_cache=True,
                    )

    settings = get_settings()
    cache_dir = _cache_dir()
    token = (settings.sac_agent_git_token or "").strip()
    branch = cfg.git_branch or "main"
    versions: dict[str, str] = {}
    errors: dict[str, str] = {}

    product_repo = {
        PRODUCT_RDP: cfg.rdp_git_repo_url,
        PRODUCT_SSH: cfg.ssh_git_repo_url,
    }
    for product, repo_url in product_repo.items():
        if not repo_url:
            errors[product] = "repo url not configured"
            continue
        try:
            versions[product] = fetch_product_version(repo_url, branch, product, cache_dir, token=token)
        except (OSError, RuntimeError, ValueError, subprocess.SubprocessError) as exc:
            errors[product] = str(exc).strip() or "fetch failed"

    fetched_at: datetime | None = None
    if versions:
        fetched_at = datetime.now(timezone.utc)
        _MEMORY_CACHE[key] = (dict(versions), fetched_at.timestamp(), key)
        if db is not None:
            fetched_at = _persist_db_cache(db, cfg, versions)

    if not versions and db is not None:
        row = db.get(UiSettings, UI_SETTINGS_ROW_ID)
        if row is not None:
            stale = _versions_from_db_row(row)
            if stale:
                return GitReleaseVersions(
                    versions=stale,
                    fetched_at=row.agent_git_fetched_at,
                    from_cache=True,
                    errors=errors,
                )

    return GitReleaseVersions(
        versions=versions,
        fetched_at=fetched_at,
        from_cache=False,
        errors=errors,
    )