fix(pipeline): prevent event loop blocking in local collector and increase CI pipeline timeouts

src/clients/client_updater.py

@@ -144,7 +144,7 @@ class ClientUpdater:
             # If running as a systemd service, restart service
             if shutil.which("systemctl"):
                 try:
-                    subprocess.run(["systemctl", "restart", "logar-client"], check=False)
+                    subprocess.run(["systemctl", "restart", "logar-client"], check=False, timeout=10)
                     print("[+] Signaled systemd to restart logar-client service.")
                 except Exception:
                     pass

src/clients/log_collector.py

@@ -565,23 +565,23 @@ class LinuxLogCollector(BaseLogCollector):
                 last_timestamp_us = 0.0
             sent_cursors = set(self.state.get("sent_cursors", []))
 
-        cmd = ["journalctl", "--since", f"{hours} hours ago", "-p", "info", "--output=json"]
+        cmd = ["journalctl", "--no-pager", "--since", f"{hours} hours ago", "-p", "info", "--output=json"]
         result = None
 
         if last_cursor:
-            cmd_with_cursor = ["journalctl", "--since", f"{hours} hours ago", "--after-cursor", str(last_cursor), "-p", "info", "--output=json"]
+            cmd_with_cursor = ["journalctl", "--no-pager", "--since", f"{hours} hours ago", "--after-cursor", str(last_cursor), "-p", "info", "--output=json"]
             try:
-                res = subprocess.run(cmd_with_cursor, capture_output=True, text=True, check=False)
+                res = subprocess.run(cmd_with_cursor, capture_output=True, text=True, check=False, timeout=8)
                 if res.returncode == 0:
                     result = res
-            except Exception:
+            except (subprocess.TimeoutExpired, Exception):
                 pass
 
         if result is None:
             try:
-                result = subprocess.run(cmd, capture_output=True, text=True, check=False)
+                result = subprocess.run(cmd, capture_output=True, text=True, check=False, timeout=8)
-            except FileNotFoundError:
+            except (subprocess.TimeoutExpired, FileNotFoundError):
-                # journalctl not present, attempt fallback to /var/log/syslog or messages
+                # journalctl not present or timed out, attempt fallback to /var/log/syslog or messages
                 return self._collect_syslog_fallback(hours=hours)
             except Exception as e:
                 print(f"[!] Error running journalctl: {e}")

src/hub/Server.py

@@ -431,8 +431,10 @@ def load_or_init_config(config_path: str = CONFIG_FILE_NAME) -> Dict[str, Any]:
         print(f"[*] Loading server configuration from: {os.path.abspath(actual_config_path)}")
         with open(actual_config_path, "r", encoding="utf-8") as f:
             config = json.load(f)
+        config_changed = False
         if "enrollment_secret" not in config:
             config["enrollment_secret"] = secrets.token_hex(24)
+            config_changed = True
         if "max_seats" not in config:
             config["max_seats"] = 0
         if "cert_dir" not in config:
@@ -469,8 +471,13 @@ def load_or_init_config(config_path: str = CONFIG_FILE_NAME) -> Dict[str, Any]:
             config["encrypt_at_rest"] = False
         if "auth_token" not in config:
             config["auth_token"] = secrets.token_hex(24)
-        if "enrollment_secret" not in config:
+            config_changed = True
-            config["enrollment_secret"] = secrets.token_hex(24)
+        if config_changed:
+            try:
+                with open(actual_config_path, "w", encoding="utf-8") as f:
+                    json.dump(config, f, indent=2)
+            except Exception:
+                pass
         if "auto_update_enabled" not in config:
             config["auto_update_enabled"] = True
         if "update_mode" not in config:
@@ -1063,7 +1070,7 @@ def run_local_log_collection_cycle(db_path: str, window_hours: int = 12, min_run
     now_sql = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
 
     try:
-        conn = sqlite3.connect(db_path)
+        conn = sqlite3.connect(db_path, timeout=10.0)
         c = conn.cursor()
         c.execute("SELECT client_id FROM clients WHERE client_id = ?", (machine_id,))
         if c.fetchone():
@@ -1130,8 +1137,9 @@ async def local_collector_loop(interval_seconds: int = 30):
         except Exception:
             state = {}
 
+    loop = asyncio.get_running_loop()
     try:
-        run_local_log_collection_cycle(db_path, window_hours=window_hours, min_runs=min_runs, state=state)
+        await loop.run_in_executor(None, lambda: run_local_log_collection_cycle(db_path, window_hours=window_hours, min_runs=min_runs, state=state))
         with open(state_file, "w", encoding="utf-8") as f:
             json.dump(state, f)
     except Exception as e:
@@ -1140,7 +1148,7 @@ async def local_collector_loop(interval_seconds: int = 30):
     while True:
         try:
             await asyncio.sleep(interval_seconds)
-            run_local_log_collection_cycle(db_path, window_hours=window_hours, min_runs=min_runs, state=state)
+            await loop.run_in_executor(None, lambda: run_local_log_collection_cycle(db_path, window_hours=window_hours, min_runs=min_runs, state=state))
             with open(state_file, "w", encoding="utf-8") as f:
                 json.dump(state, f)
         except asyncio.CancelledError:
@@ -1154,11 +1162,12 @@ async def cert_validity_watchdog(interval_seconds: int = 43200, threshold_days:
     config = SERVER_STATE.get("config", {})
     cert_dir = config.get("cert_dir", "certs")
     hostnames = [config.get("tcp_host", "0.0.0.0"), "127.0.0.1", "localhost"]
+    loop = asyncio.get_running_loop()
     while True:
         try:
             await asyncio.sleep(interval_seconds)
-            check_and_rotate_server_certs(cert_dir=cert_dir, hostnames=hostnames, threshold_days=threshold_days)
+            await loop.run_in_executor(None, lambda: check_and_rotate_server_certs(cert_dir=cert_dir, hostnames=hostnames, threshold_days=threshold_days))
-            report_seat_quota_to_authority_sync()
+            await loop.run_in_executor(None, report_seat_quota_to_authority_sync)
         except asyncio.CancelledError:
             break
         except Exception as e:
@@ -2246,7 +2255,7 @@ def import_signed_license(req: LicenseImportRequest, request: Request = None, au
 
     db_path = config.get("db_path", DEFAULT_DB_FILE)
     try:
-        conn = sqlite3.connect(db_path)
+        conn = sqlite3.connect(db_path, timeout=10.0)
         conn.execute("""
             UPDATE license_config 
             SET max_seats = ?, license_id = ?, license_server_url = ?, issued_at = ?, expires_at = ?, license_status = 'active', features = ?

src/hub/server_enrollment.py

@@ -47,7 +47,8 @@ def generate_ca_if_needed(
     cert_dir: str = "certs",
     common_name: str = "LOGAR-Root-CA",
     force_renew: bool = False,
-    threshold_days: int = 30
+    threshold_days: int = 30,
+    key_size: int = 4096
 ) -> Tuple[x509.Certificate, rsa.RSAPrivateKey, str, str]:
     """
     Loads an existing Root CA or generates a self-signed Root CA certificate and private key.
@@ -81,8 +82,8 @@ def generate_ca_if_needed(
         except Exception:
             pass
 
-    # Generate RSA 4096 private key for Root CA
+    # Generate RSA private key for Root CA
-    ca_key = rsa.generate_private_key(public_exponent=65537, key_size=4096)
+    ca_key = rsa.generate_private_key(public_exponent=65537, key_size=key_size)
     subject = issuer = x509.Name([
         x509.NameAttribute(NameOID.COUNTRY_NAME, "AT"),
         x509.NameAttribute(NameOID.ORGANIZATION_NAME, "LOGAR"),

tests/test_pipeline.py

@@ -28,6 +28,7 @@ TCP_HOST = "127.0.0.1"
 TCP_PORT = 9443
 HERMES_HOST = "127.0.0.1"
 HERMES_PORT = 8443
+HTTP_TIMEOUT = 30
 
 
 def run_tests():
@@ -69,7 +70,7 @@ def run_tests():
     
     # 1. Check current license status
     req_status_init = urllib.request.Request(f"http://{HERMES_HOST}:{HERMES_PORT}/api/license/status")
-    with urllib.request.urlopen(req_status_init, timeout=15) as resp:
+    with urllib.request.urlopen(req_status_init, timeout=HTTP_TIMEOUT) as resp:
         init_lic_stat = json.loads(resp.read().decode("utf-8"))
     
     if init_lic_stat.get("status") == "UNLICENSED":
@@ -85,7 +86,7 @@ def run_tests():
             headers={"Content-Type": "application/json"}
         )
         try:
-            with urllib.request.urlopen(req_unlicensed, timeout=15):
+            with urllib.request.urlopen(req_unlicensed, timeout=HTTP_TIMEOUT):
                 assert False, "Expected HTTP 403 on unlicensed server enrollment"
         except urllib.error.HTTPError as e:
             assert e.code == 403, f"Expected HTTP 403 on unlicensed server, got {e.code}"
@@ -96,7 +97,7 @@ def run_tests():
         f"http://{HERMES_HOST}:{HERMES_PORT}/api/license/csr",
         headers={"Authorization": f"Bearer {auth_token}"}
     )
-    with urllib.request.urlopen(req_csr, timeout=15) as resp:
+    with urllib.request.urlopen(req_csr, timeout=HTTP_TIMEOUT) as resp:
         assert resp.status == 200
         csr_data = json.loads(resp.read().decode("utf-8"))
         csr_pem = csr_data["csr"]
@@ -104,7 +105,7 @@ def run_tests():
 
     # 3. Generate test master CA and sign subordinate CA with 10 seats
     master_ca_dir = os.path.join(ROOT_DIR, "test_pipeline_master_ca")
-    master_cert, master_key, master_pem, master_key_pem = server_enrollment.generate_ca_if_needed(cert_dir=master_ca_dir, common_name="LOGAR-Master-Root-CA")
+    master_cert, master_key, master_pem, master_key_pem = server_enrollment.generate_ca_if_needed(cert_dir=master_ca_dir, common_name="LOGAR-Master-Root-CA", key_size=2048)
     sub_pem = server_enrollment.sign_ca_csr_with_seats(
         csr_pem=csr_pem,
         master_ca_cert_pem=master_pem,
@@ -121,7 +122,7 @@ def run_tests():
         data=json.dumps(import_payload).encode("utf-8"),
         headers={"Content-Type": "application/json", "Authorization": f"Bearer {auth_token}"}
     )
-    with urllib.request.urlopen(req_import, timeout=15) as resp:
+    with urllib.request.urlopen(req_import, timeout=HTTP_TIMEOUT) as resp:
         assert resp.status == 200
         import_res = json.loads(resp.read().decode("utf-8"))
         assert import_res["status"] == "success"
@@ -130,7 +131,7 @@ def run_tests():
 
     # 5. Verify license status
     req_status = urllib.request.Request(f"http://{HERMES_HOST}:{HERMES_PORT}/api/license/status")
-    with urllib.request.urlopen(req_status, timeout=15) as resp:
+    with urllib.request.urlopen(req_status, timeout=HTTP_TIMEOUT) as resp:
         assert resp.status == 200
         lic_stat = json.loads(resp.read().decode("utf-8"))
         assert lic_stat["status"] == "ACTIVE"
@@ -153,7 +154,7 @@ def run_tests():
         headers={"Content-Type": "application/json"}
     )
     try:
-        with urllib.request.urlopen(req_bad, timeout=15):
+        with urllib.request.urlopen(req_bad, timeout=HTTP_TIMEOUT):
             assert False, "Expected HTTP 403 on invalid secret"
     except urllib.error.HTTPError as e:
         assert e.code == 403, f"Expected HTTP 403, got {e.code}"
@@ -171,7 +172,7 @@ def run_tests():
         data=json.dumps(valid_enroll_payload).encode("utf-8"),
         headers={"Content-Type": "application/json"}
     )
-    with urllib.request.urlopen(req_valid, timeout=15) as resp:
+    with urllib.request.urlopen(req_valid, timeout=HTTP_TIMEOUT) as resp:
         assert resp.status == 200, f"Expected 200, got {resp.status}"
         enroll_data = json.loads(resp.read().decode("utf-8"))
         assert "ca_cert" in enroll_data
@@ -194,7 +195,7 @@ def run_tests():
     auth_tok = server_conf.get("auth_token") or client_conf.get("auth_token")
     client_headers = {"Authorization": f"Bearer {auth_tok}"} if auth_tok else {}
     clients_req = urllib.request.Request(f"http://{HERMES_HOST}:{HERMES_PORT}/api/clients", headers=client_headers)
-    with urllib.request.urlopen(clients_req, timeout=15) as resp:
+    with urllib.request.urlopen(clients_req, timeout=HTTP_TIMEOUT) as resp:
         clients_data = json.loads(resp.read().decode("utf-8"))
         assert clients_data["active_seats"] >= 1
         found_c = any(c["client_id"] == client_id for c in clients_data["clients"])
@@ -294,7 +295,7 @@ def run_tests():
     token = server_conf.get("auth_token") or client_conf.get("auth_token")
     headers = {"Authorization": f"Bearer {token}"} if token else {}
     req = urllib.request.Request(f"http://{HERMES_HOST}:{HERMES_PORT}/api/hermes/report", headers=headers)
-    with urllib.request.urlopen(req, timeout=15) as response:
+    with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as response:
         assert response.status == 200, f"Expected 200, got {response.status}"
         hermes_data = json.loads(response.read().decode("utf-8"))