Implement automated Fleet Hub upstream client package caching and background update distribution

src/hub/Server.py

@@ -2205,6 +2205,22 @@ def health_check():
     }
 
 
+async def update_checker_loop(interval_hours: int = 1):
+    """
+    Background worker that periodically queries the upstream repository
+    and automatically downloads/caches client binary update packages locally on the Hub.
+    """
+    # Brief initial pause after socket startup
+    await asyncio.sleep(5)
+    while True:
+        try:
+            mgr: ServerUpdateManager = SERVER_STATE.get("update_manager") or UPDATE_MANAGER
+            mgr.check_upstream_updates(force=False, auto_download_clients=True)
+        except Exception as e:
+            print(f"[!] Warning in update checker background loop: {e}")
+        await asyncio.sleep(interval_hours * 3600)
+
+
 async def run_server():
     """Runs the mTLS TCP socket listener and the Hermes REST API concurrently."""
     config = SERVER_STATE["config"]
@@ -2226,6 +2242,7 @@ async def run_server():
     watchdog_task = asyncio.create_task(cert_validity_watchdog())
     collector_task = asyncio.create_task(local_collector_loop(interval_seconds=30))
     buffer_task = asyncio.create_task(ingest_buffer_worker_loop())
+    update_task = asyncio.create_task(update_checker_loop(interval_hours=1))
 
     try:
         await asyncio.gather(
@@ -2233,13 +2250,15 @@ async def run_server():
             uv_server.serve(),
             watchdog_task,
             collector_task,
-            buffer_task
+            buffer_task,
+            update_task
         )
     finally:
         watchdog_task.cancel()
         collector_task.cancel()
         buffer_task.cancel()
-        for t in [watchdog_task, collector_task, buffer_task]:
+        update_task.cancel()
+        for t in [watchdog_task, collector_task, buffer_task, update_task]:
             try:
                 await t
             except asyncio.CancelledError:

src/hub/update_manager.py

@@ -292,10 +292,89 @@ class ServerUpdateManager:
             "mandatory": False
         }
 
-    def check_upstream_updates(self, force: bool = False, timeout: int = 8) -> Dict[str, Any]:
+    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:
+                continue
+
+            target_os = None
+            if ("win" in name or name.endswith(".exe") or "setup" in name) and not results["windows"]:
+                target_os = "windows"
+            elif ("linux" in name or name.endswith((".bin", ".sh")) or "client_linux" in name) and not results["linux"]:
+                target_os = "linux"
+
+            if not target_os:
+                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 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.
+        to check if a newer server or client release is available, and automatically caches client packages.
         """
         if not self.check_enabled and not force:
             return self.get_status()
@@ -337,6 +416,15 @@ class ServerUpdateManager:
                     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 auto_download_clients and assets:
+                        # Check if hosted packages need updating
+                        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:

tests/test_hub_update_auth_deep.py

@@ -73,8 +73,31 @@ class TestHubUpdateAuthDeep(unittest.TestCase):
         self.assertTrue(decision.get("update_available"))
         self.assertEqual(decision.get("latest_version"), "2.5.0")
 
-        decision_uptodate = manager.check_client_update(client_id="node-win-01", current_version="2.5.0", os_type="windows")
+        # Test download and caching from simulated upstream assets
-        self.assertFalse(decision_uptodate.get("update_available"))
+        mock_assets = [
+            {"name": "LOGAR_Client_Setup.exe", "download_url": "http://mock.upstream/LOGAR_Client_Setup.exe"},
+            {"name": "Linux_Client.bin", "download_url": "http://mock.upstream/Linux_Client.bin"}
+        ]
+        class MockHttpResp:
+            status = 200
+            def __init__(self):
+                self._read_done = False
+            def __enter__(self):
+                return self
+            def __exit__(self, *args):
+                pass
+            def read(self, size=65536):
+                if not self._read_done:
+                    self._read_done = True
+                    return b"MOCK_BINARY_DATA_PAYLOAD_2026"
+                return b""
+
+        with patch("urllib.request.urlopen", side_effect=lambda req, timeout=60: MockHttpResp()):
+            cache_res = manager.download_and_cache_client_packages(assets=mock_assets, target_version="2.6.0")
+            self.assertTrue(cache_res["windows"])
+            self.assertTrue(cache_res["linux"])
+            self.assertEqual(len(cache_res["downloaded"]), 2)
+            self.assertEqual(manager.hosted_client_version, "2.6.0")
 
     def test_auth_manager_components(self):
         """Tests PasswordManager, TOTPManager, SessionManager, and RateLimiter."""