Files
LOGAR/src/hub/update_manager.py
T
me0nline 5456d50774
CI Test Suite / Run Component Tests & Pipeline Verification (push) Successful in 1m54s
feat(hub): add three image acquisition sources (official repo, custom repo, manual provided)
2026-10-04 01:31:38 +02:00

496 lines
22 KiB
Python

"""
LOGAR - Central Server Update & Fleet Distribution Manager
Handles upstream release checking, update alerting, local client package hosting,
and client auto-update check/download endpoints.
"""
import os
import sys
import json
import hashlib
import urllib.request
import urllib.error
from datetime import datetime, timezone
from typing import Dict, Any, List, Optional, Tuple
try:
from src.version import LOGAR_VERSION, is_newer_version, compare_versions
except ImportError:
from version import LOGAR_VERSION, is_newer_version, compare_versions
class ServerUpdateManager:
DEFAULT_GITEA_RELEASE_URL = "https://gitea.eibl.tech/api/v1/repos/me0nline/LOGAR/releases/latest"
def __init__(
self,
update_dir: Optional[str] = None,
upstream_url: Optional[str] = None,
check_enabled: bool = True,
auto_update_enabled: bool = True,
update_mode: str = "auto"
):
self.update_dir = update_dir or os.path.join(os.getcwd(), "updates")
self.upstream_url = upstream_url or self.DEFAULT_GITEA_RELEASE_URL
self.check_enabled = check_enabled
self.auto_update_enabled = auto_update_enabled
raw_mode = (update_mode or "official").lower()
if raw_mode in ("official", "custom", "manual"):
self.update_mode = raw_mode
elif raw_mode == "auto":
self.update_mode = "official" if (not upstream_url or upstream_url == self.DEFAULT_GITEA_RELEASE_URL) else "custom"
else:
self.update_mode = "official"
# Cached upstream release state
self.last_checked_utc: Optional[str] = None
self.latest_upstream_version: Optional[str] = None
self.server_update_available: bool = False
self.release_title: Optional[str] = None
self.release_notes: Optional[str] = None
self.release_url: Optional[str] = None
self.release_assets: List[Dict[str, Any]] = []
self.last_error: Optional[str] = None
# Hosted client packages cache
self.hosted_client_version: str = LOGAR_VERSION
self._sha256_cache: Dict[str, Tuple[float, int, str]] = {}
self._package_cache: Dict[str, Dict[str, Any]] = {}
self._package_cache_mtimes: Dict[str, float] = {}
self._ensure_package_structure()
def _ensure_package_structure(self):
"""Initializes default subdirectories for OS-specific client packages."""
for os_name in ["windows", "linux"]:
os.makedirs(os.path.join(self.update_dir, os_name), exist_ok=True)
@classmethod
def calculate_sha256(cls, filepath: str) -> str:
"""Computes the SHA-256 digest of a local file with stat-based memoization."""
if not os.path.exists(filepath):
return ""
try:
stat = os.stat(filepath)
mtime = stat.st_mtime
size = stat.st_size
if hasattr(cls, "_global_sha256_cache"):
cached = cls._global_sha256_cache.get(filepath)
if cached and cached[0] == mtime and cached[1] == size:
return cached[2]
except Exception:
mtime = 0.0
size = 0
h = hashlib.sha256()
with open(filepath, "rb") as f:
while chunk := f.read(65536):
h.update(chunk)
digest = h.hexdigest().lower()
if not hasattr(cls, "_global_sha256_cache"):
cls._global_sha256_cache = {}
cls._global_sha256_cache[filepath] = (mtime, size, digest)
return digest
def set_hosted_client_version(self, version_str: str):
"""Sets the version tag for currently hosted client packages."""
if version_str:
self.hosted_client_version = version_str.strip().lstrip("v")
self._package_cache.clear()
def invalidate_package_cache(self, os_type: Optional[str] = None):
"""Clears in-memory package cache."""
if os_type:
self._package_cache.pop(os_type.lower(), None)
self._package_cache_mtimes.pop(os_type.lower(), None)
else:
self._package_cache.clear()
self._package_cache_mtimes.clear()
def register_client_package(
self,
os_type: str,
file_path: str,
version: Optional[str] = None,
filename: Optional[str] = None
) -> Dict[str, Any]:
"""
Saves/copies a client binary package into the local update repository.
Computes SHA-256 checksum and updates metadata.
"""
norm_os = os_type.lower()
if norm_os not in ["windows", "linux"]:
raise ValueError(f"Unsupported OS type: {os_type}")
if not os.path.exists(file_path):
raise FileNotFoundError(f"Package file not found: {file_path}")
target_dir = os.path.join(self.update_dir, norm_os)
os.makedirs(target_dir, exist_ok=True)
raw_name = filename or os.path.basename(file_path)
target_name = os.path.basename(raw_name)
dest_path = os.path.join(target_dir, target_name)
if os.path.abspath(file_path) != os.path.abspath(dest_path):
with open(file_path, "rb") as src_f, open(dest_path, "wb") as dst_f:
while chunk := src_f.read(65536):
dst_f.write(chunk)
sha256 = self.calculate_sha256(dest_path)
pkg_version = version or self.hosted_client_version or LOGAR_VERSION
self.set_hosted_client_version(pkg_version)
# Write manifest file next to package
manifest = {
"os": norm_os,
"filename": target_name,
"version": pkg_version,
"sha256": sha256,
"size_bytes": os.path.getsize(dest_path),
"updated_at": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
}
manifest_path = os.path.join(target_dir, "manifest.json")
with open(manifest_path, "w", encoding="utf-8") as mf:
json.dump(manifest, mf, indent=2)
# Update cache directly
cached_entry = dict(manifest)
cached_entry["file_path"] = dest_path
cached_entry["exists"] = True
self._package_cache[norm_os] = cached_entry
self._package_cache_mtimes[norm_os] = os.path.getmtime(dest_path)
return manifest
def get_hosted_packages(self, force_refresh: bool = False) -> Dict[str, Dict[str, Any]]:
"""
Returns info on currently hosted client packages for Windows and Linux.
Uses in-memory cache validated against file modification times to minimize disk I/O.
"""
result: Dict[str, Dict[str, Any]] = {}
for os_name in ["windows", "linux"]:
manifest_path = os.path.join(self.update_dir, os_name, "manifest.json")
target_file = None
if not force_refresh and os_name in self._package_cache:
cached = self._package_cache[os_name]
tf = cached.get("file_path")
if tf and os.path.exists(tf):
try:
cur_mtime = os.path.getmtime(tf)
if cur_mtime == self._package_cache_mtimes.get(os_name):
result[os_name] = cached
continue
except Exception:
pass
if os.path.exists(manifest_path):
try:
with open(manifest_path, "r", encoding="utf-8") as mf:
info = json.load(mf)
target_file = os.path.join(self.update_dir, os_name, info.get("filename", ""))
if os.path.exists(target_file):
info["exists"] = True
info["file_path"] = target_file
result[os_name] = info
self._package_cache[os_name] = info
self._package_cache_mtimes[os_name] = os.path.getmtime(target_file)
continue
except Exception:
pass
# Fallback scan in update_dir/{os_name}
os_dir = os.path.join(self.update_dir, os_name)
found_pkg = None
if os.path.exists(os_dir):
# Search by priority: .bin/.exe first, then archives, never .sh
valid_exts = (".exe", ".msi", ".zip") if os_name == "windows" else (".bin", ".tar.gz", ".tgz")
for f in sorted(os.listdir(os_dir), key=lambda x: (not x.endswith((".exe", ".bin")), x)):
if f.endswith(valid_exts) and not f.startswith(".") and not f.endswith(".sh"):
found_pkg = os.path.join(os_dir, f)
break
# Secondary fallback: search project dist/
if not found_pkg:
dist_dir = os.path.join(os.path.dirname(self.update_dir), "dist")
if os.path.exists(dist_dir):
patterns = [
"LOGAR_Client_Setup.exe", "LOGAR-Client-Setup.exe", "Win_Client.exe", "LOGAR_Client.exe", "LOGAR-Windows-Client.zip"
] if os_name == "windows" else [
"Linux_Client.bin", "Linux_Client", "LOGAR_Client_Linux", "LOGAR-Linux-Client.tar.gz"
]
for pat in patterns:
candidate = os.path.join(dist_dir, pat)
if os.path.exists(candidate):
found_pkg = candidate
break
if found_pkg and os.path.exists(found_pkg):
sha256 = self.calculate_sha256(found_pkg)
pkg_data = {
"os": os_name,
"filename": os.path.basename(found_pkg),
"version": self.hosted_client_version or LOGAR_VERSION,
"sha256": sha256,
"size_bytes": os.path.getsize(found_pkg),
"file_path": found_pkg,
"exists": True,
"updated_at": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
}
result[os_name] = pkg_data
self._package_cache[os_name] = pkg_data
self._package_cache_mtimes[os_name] = os.path.getmtime(found_pkg)
else:
pkg_data = {
"os": os_name,
"filename": None,
"version": self.hosted_client_version or LOGAR_VERSION,
"sha256": "",
"size_bytes": 0,
"file_path": None,
"exists": False
}
result[os_name] = pkg_data
self._package_cache[os_name] = pkg_data
self._package_cache_mtimes[os_name] = 0.0
return result
def get_client_package_path(self, os_type: str) -> Optional[Tuple[str, str, str]]:
"""
Returns (file_path, filename, sha256) for downloading a client update package.
"""
packages = self.get_hosted_packages()
pkg_info = packages.get(os_type.lower())
if pkg_info and pkg_info.get("exists") and pkg_info.get("file_path"):
return pkg_info["file_path"], pkg_info["filename"], pkg_info["sha256"]
return None
def check_client_update(
self,
client_id: str,
os_type: str,
current_version: Optional[str],
arch: str = "x86_64"
) -> Dict[str, Any]:
"""
Evaluates whether a client requires an update based on its reported version
and the server's hosted client package version.
"""
norm_os = (os_type or "windows").lower()
packages = self.get_hosted_packages()
pkg = packages.get(norm_os, {})
target_version = pkg.get("version") or self.hosted_client_version or LOGAR_VERSION
has_pkg = bool(pkg.get("exists"))
client_ver = (current_version or "0.0.0.0").strip().lstrip("v")
# An update is available if auto_update is enabled, package exists, and hosted version is strictly newer
if not self.auto_update_enabled:
return {
"update_available": False,
"client_id": client_id,
"os": norm_os,
"current_version": client_ver,
"latest_version": target_version,
"reason": "Fleet auto-update is disabled on this server."
}
update_available = has_pkg and is_newer_version(target_version, client_ver)
return {
"update_available": update_available,
"client_id": client_id,
"os": norm_os,
"current_version": client_ver,
"latest_version": target_version,
"sha256": pkg.get("sha256", "") if update_available else "",
"download_url": f"/api/client/update/download?os={norm_os}" if update_available else "",
"size_bytes": pkg.get("size_bytes", 0) if update_available else 0,
"filename": pkg.get("filename", "") if update_available else "",
"release_notes": self.release_notes or f"LOGAR {target_version} update for {norm_os.capitalize()}.",
"mandatory": False
}
def download_and_cache_client_packages(
self,
assets: Optional[List[Dict[str, Any]]] = None,
target_version: Optional[str] = None,
timeout: int = 60
) -> Dict[str, Any]:
"""
Downloads OS client binary packages from upstream release assets and caches them locally.
Ensures the Hub acts as a caching distribution point so edge clients don't contact external servers.
"""
release_assets = assets if assets is not None else self.release_assets
version = target_version or self.latest_upstream_version or self.hosted_client_version or LOGAR_VERSION
results = {"windows": False, "linux": False, "downloaded": []}
if not release_assets:
return results
temp_dir = os.path.join(self.update_dir, "temp")
os.makedirs(temp_dir, exist_ok=True)
for asset in release_assets:
name = (asset.get("name") or "").lower()
download_url = asset.get("download_url")
if not download_url or name.endswith(".sh") or "server" in name or "checksum" in name or "sha256" in name:
continue
target_os = None
if ("win" in name or "setup" in name or name.endswith((".exe", ".msi", ".zip"))) and "client" in name and not results["windows"]:
target_os = "windows"
elif ("win" in name or name.endswith(".exe")) and not results["windows"]:
target_os = "windows"
elif ("linux" in name or name.endswith((".bin", ".tar.gz", ".tgz"))) and ("client" in name or "agent" in name) and not results["linux"]:
target_os = "linux"
elif name in ("linux_client.bin", "linux_client", "logar_client_linux") and not results["linux"]:
target_os = "linux"
if not target_os:
continue
# Check if matching version is already cached locally
existing_pkg = self.get_hosted_packages().get(target_os)
if existing_pkg and existing_pkg.get("exists") and existing_pkg.get("version") == version:
results[target_os] = True
continue
filename = os.path.basename(asset.get("name") or f"client_{target_os}")
temp_file = os.path.join(temp_dir, filename)
try:
print(f"[*] Fleet Hub caching upstream client package for {target_os}: {filename} ({download_url})...")
req = urllib.request.Request(
download_url,
headers={"User-Agent": f"LOGAR-Server-Hub/{LOGAR_VERSION}"}
)
with urllib.request.urlopen(req, timeout=timeout) as resp:
if resp.status == 200:
with open(temp_file, "wb") as f:
while chunk := resp.read(65536):
f.write(chunk)
# Register package in Hub repository
manifest = self.register_client_package(
os_type=target_os,
file_path=temp_file,
version=version,
filename=filename
)
results[target_os] = True
results["downloaded"].append({
"os": target_os,
"filename": filename,
"version": version,
"sha256": manifest.get("sha256")
})
print(f"[+] Successfully cached {target_os} client package v{version} (SHA-256: {manifest.get('sha256')[:12]}...).")
except Exception as e:
print(f"[!] Warning: Failed to download and cache upstream client asset '{filename}': {e}")
finally:
if os.path.exists(temp_file):
try:
os.remove(temp_file)
except Exception:
pass
if results["downloaded"]:
self.hosted_client_version = version
return results
def sync_upstream_packages(self, force: bool = True, timeout: int = 60) -> Dict[str, Any]:
"""
Manually or programmatically triggers an immediate upstream release check and downloads
the latest client binary packages to the Hub repository.
"""
self.check_upstream_updates(force=force, auto_download_clients=True, timeout=timeout)
if self.release_assets:
return self.download_and_cache_client_packages(assets=self.release_assets, timeout=timeout)
return {"windows": False, "linux": False, "downloaded": [], "error": self.last_error or "No release assets found."}
def check_upstream_updates(self, force: bool = False, auto_download_clients: bool = True, timeout: int = 15) -> Dict[str, Any]:
"""
Queries the upstream repository release API (Gitea / GitHub / JSON manifest)
to check if a newer server or client release is available, and automatically caches client packages
when in automatic acquisition mode.
"""
if not self.check_enabled and not force:
return self.get_status()
now_utc = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
self.last_checked_utc = now_utc
self.last_error = None
try:
req = urllib.request.Request(
self.upstream_url,
headers={
"User-Agent": f"LOGAR-Server/{LOGAR_VERSION}",
"Accept": "application/json"
}
)
with urllib.request.urlopen(req, timeout=timeout) as resp:
if resp.status == 200:
data = json.loads(resp.read().decode("utf-8"))
raw_tag = data.get("tag_name") or data.get("name") or data.get("version") or ""
upstream_ver = raw_tag.strip().lstrip("v")
self.latest_upstream_version = upstream_ver
self.release_title = data.get("name") or raw_tag
self.release_notes = data.get("body") or data.get("notes") or ""
self.release_url = data.get("html_url") or data.get("url") or ""
# Process assets list
assets = []
for a in data.get("assets", []):
assets.append({
"name": a.get("name"),
"size": a.get("size"),
"download_url": a.get("browser_download_url") or a.get("download_url")
})
self.release_assets = assets
# Compare version against current LOGAR_VERSION
self.server_update_available = is_newer_version(upstream_ver, LOGAR_VERSION)
if self.server_update_available:
print(f"[!] 🔔 LOGAR Server Update Available: Current v{LOGAR_VERSION} -> Upstream v{upstream_ver}")
# Automatically download and cache client update packages on the Hub if in official or custom repo mode
should_auto_download = auto_download_clients and (self.update_mode in ("official", "custom", "auto") or force)
if should_auto_download and assets:
hosted = self.get_hosted_packages()
needs_win = not hosted.get("windows", {}).get("exists") or is_newer_version(upstream_ver, hosted.get("windows", {}).get("version", "0.0.0"))
needs_linux = not hosted.get("linux", {}).get("exists") or is_newer_version(upstream_ver, hosted.get("linux", {}).get("version", "0.0.0"))
if needs_win or needs_linux or force:
self.download_and_cache_client_packages(assets=assets, target_version=upstream_ver)
else:
self.last_error = f"Upstream returned HTTP status {resp.status}"
except Exception as e:
self.last_error = f"Failed to check upstream updates: {e}"
return self.get_status()
def get_status(self) -> Dict[str, Any]:
"""Returns structured update status for WebUI and REST APIs."""
return {
"current_server_version": LOGAR_VERSION,
"latest_upstream_version": self.latest_upstream_version or LOGAR_VERSION,
"server_update_available": self.server_update_available,
"hosted_client_version": self.hosted_client_version or LOGAR_VERSION,
"release_title": self.release_title or "",
"release_notes": self.release_notes or "",
"release_url": self.release_url or "",
"release_assets": self.release_assets,
"last_checked_utc": self.last_checked_utc,
"last_error": self.last_error,
"check_enabled": self.check_enabled,
"auto_update_enabled": self.auto_update_enabled,
"update_mode": self.update_mode,
"upstream_url": self.upstream_url,
"hosted_packages": self.get_hosted_packages()
}