from __future__ import annotations from typing import Any from actions.cron import handle_cron_create, handle_cron_delete from actions.sftp import handle_sftp_create, handle_sftp_delete, handle_sftp_reset_password from actions.database import ( handle_database_create, handle_database_drop, handle_database_user_create, handle_database_user_delete, handle_database_user_reset_password, handle_database_user_update_grants, ) from actions.nginx import ( handle_nginx_config_read, handle_nginx_config_write, handle_nginx_service_reload, ) from actions.service import ( handle_service_start, handle_service_stop, handle_service_restart, handle_service_status, ) from actions.ssl import ( handle_ssl_generate, handle_ssl_renew, handle_ssl_delete, ) from actions.pipeline import handle_pipeline_run from actions.site import handle_site_create, handle_site_delete from actions.logs import handle_log_read from actions.file_manager import ( handle_file_list, handle_file_read, handle_file_write, handle_file_delete, handle_file_rename, handle_file_create, ) from actions.backup import ( handle_backup_create, handle_backup_delete, handle_backup_list, handle_backup_restore, ) from actions.agent import handle_agent_update from actions.power import handle_server_reboot from api_client import ApiClient, ApiError from config import Config from identity import load_agent_token from logger import get_logger log = get_logger("runner") HANDLERS = { "cron_job.create": handle_cron_create, "cron_job.delete": handle_cron_delete, "sftp_user.create": handle_sftp_create, "sftp_user.delete": handle_sftp_delete, "sftp_user.reset_password": handle_sftp_reset_password, "database.create": handle_database_create, "database.drop": handle_database_drop, "database_user.create": handle_database_user_create, "database_user.delete": handle_database_user_delete, "database_user.reset_password": handle_database_user_reset_password, "database_user.update_grants": handle_database_user_update_grants, "nginx.config.read": handle_nginx_config_read, "nginx.config.write": handle_nginx_config_write, "nginx.service.reload": handle_nginx_service_reload, "service.start": handle_service_start, "service.stop": handle_service_stop, "service.status": handle_service_status, "service.restart": handle_service_restart, "ssl.generate": handle_ssl_generate, "ssl.renew": handle_ssl_renew, "ssl.delete": handle_ssl_delete, "pipeline.run": handle_pipeline_run, "site.create": handle_site_create, "site.delete": handle_site_delete, "log.read": handle_log_read, "file.list": handle_file_list, "file.read": handle_file_read, "file.write": handle_file_write, "file.delete": handle_file_delete, "file.rename": handle_file_rename, "file.create": handle_file_create, "backup.create": handle_backup_create, "backup.delete": handle_backup_delete, "backup.list": handle_backup_list, "backup.restore": handle_backup_restore, "agent.update": handle_agent_update, "server.reboot": handle_server_reboot, } import queue import threading import os import json import atexit from datetime import datetime from pathlib import Path _state_lock = threading.Lock() _worker_manager: CommandWorkerManager | None = None def load_command_states(var_dir: Path) -> dict[str, Any]: path = var_dir / "command_states.json" if not path.is_file(): return {} try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return {} def save_command_states(var_dir: Path, states: dict[str, Any]) -> None: path = var_dir / "command_states.json" try: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_name(path.name + ".tmp") with open(tmp, "w", encoding="utf-8") as f: json.dump(states, f, indent=2) f.flush() os.fsync(f.fileno()) tmp.replace(path) if os.name != "nt": try: dir_fd = os.open(str(path.parent), os.O_RDONLY) try: os.fsync(dir_fd) finally: os.close(dir_fd) except Exception: pass except Exception as exc: log.warning("failed to persist command states: %s", exc) def update_command_state(var_dir: Path, command_id: str, **changes: Any) -> None: with _state_lock: states = load_command_states(var_dir) if command_id not in states: states[command_id] = {} states[command_id].update(changes) # Keep only the last 100 commands to avoid file growing indefinitely if len(states) > 100: sorted_keys = sorted(states.keys(), key=lambda k: states[k].get("updated_at", "")) for k in sorted_keys[:-100]: states.pop(k, None) save_command_states(var_dir, states) def get_system_boot_time() -> float | None: import time # 1. Read btime from /proc/stat try: if os.path.exists("/proc/stat"): with open("/proc/stat", "r", encoding="utf-8") as f: for line in f: if line.startswith("btime "): parts = line.strip().split() if len(parts) >= 2: return float(parts[1]) except Exception: pass # 2. Fallback to /proc/uptime calculations try: if os.path.exists("/proc/uptime"): with open("/proc/uptime", "r", encoding="utf-8") as f: line = f.readline().strip() if line: uptime_seconds = float(line.split()[0]) return time.time() - uptime_seconds except Exception: pass return None def recover_interrupted_commands(config: Config, api_client: ApiClient) -> None: token = load_agent_token(config) if not token: return try: var_dir = config.var_dir except AttributeError: return states = load_command_states(var_dir) interrupted_ids = [] for cmd_id, info in states.items(): if info.get("status") in ("pending", "running"): interrupted_ids.append(cmd_id) if not interrupted_ids: return log.info("found %d interrupted command(s) to recover", len(interrupted_ids)) for cmd_id in interrupted_ids: info = states.get(cmd_id) or {} cmd_type = info.get("command_type") if cmd_type == "server.reboot": pre_boot = info.get("pre_reboot_boot_time") current_boot = get_system_boot_time() is_real_reboot = False if pre_boot is not None and current_boot is not None: if current_boot > pre_boot + 0.1: is_real_reboot = True if is_real_reboot: log.info("recovering reboot command id=%s, system boot time verified. marking as succeeded", cmd_id) update_command_state( var_dir, cmd_id, status="success", updated_at=datetime.utcnow().isoformat() ) try: api_client.post_command_result( int(cmd_id), { "status": "succeeded", "result": {"success": True, "message": "Server rebooted successfully"} }, token ) log.info("posted recovery success result for command id=%s", cmd_id) except Exception as exc: log.warning("failed to post recovery result for command id=%s: %s", cmd_id, exc) else: log.warning("recovering reboot command id=%s failed validation: pre_boot=%s, current_boot=%s. marking as failed", cmd_id, pre_boot, current_boot) update_command_state( var_dir, cmd_id, status="failed", updated_at=datetime.utcnow().isoformat() ) try: api_client.post_command_result( int(cmd_id), { "status": "failed", "result": {"error": "Server did not reboot or boot time could not be verified"} }, token ) log.info("posted recovery failure result for command id=%s", cmd_id) except Exception as exc: log.warning("failed to post recovery result for command id=%s: %s", cmd_id, exc) else: log.info("recovering command id=%s, marking as failed", cmd_id) update_command_state( var_dir, cmd_id, status="failed", updated_at=datetime.utcnow().isoformat() ) try: api_client.post_command_result( int(cmd_id), { "status": "failed", "result": {"error": "Agent restarted during command execution"} }, token ) log.info("posted recovery failure result for command id=%s", cmd_id) except Exception as exc: log.warning("failed to post recovery result for command id=%s: %s", cmd_id, exc) class CommandWorkerManager: def __init__(self, config: Config, api_client: ApiClient) -> None: self.config = config self.api_client = api_client # Get limit from env var or default to 3. In tests (like pytest) run synchronously by default (max_workers=0). import sys is_test = "pytest" in sys.modules or "unittest" in sys.modules default_workers = "0" if is_test else "3" if os.environ.get("WOLFPANEL_TEST_ASYNC") == "1": default_workers = "3" self.max_workers = int(os.environ.get("WOLFPANEL_MAX_CONCURRENT_COMMANDS", default_workers)) self.queue = queue.Queue(maxsize=1000) self.workers = [] self.lock = threading.Lock() self.shutdown_event = threading.Event() self.updating = False self.active_commands = set() def start_workers(self) -> None: if self.max_workers == 0: return with self.lock: self.workers = [t for t in self.workers if t.is_alive()] needed = self.max_workers - len(self.workers) if needed <= 0: return log.info("starting %d command worker threads", needed) for i in range(needed): idx = len(self.workers) t = threading.Thread( target=self._worker_loop, name=f"WolfCommandWorker-{idx}", daemon=True ) t.start() self.workers.append(t) def enqueue(self, command: dict[str, Any]) -> None: if self.max_workers == 0: # Inline execution for synchronous tests try: worker_log = get_logger("worker") except Exception: class DummyLogger: def info(self, msg, *args): pass def error(self, msg, *args): pass def warning(self, msg, *args): pass worker_log = DummyLogger() self._execute_command(command, worker_log) return cmd_id = str(command.get("id")) if self.queue.full(): log.warning("command queue is full, skipping command id=%s", cmd_id) return try: update_command_state( self.config.var_dir, cmd_id, status="pending", command_type=command.get("command_type"), updated_at=datetime.utcnow().isoformat() ) except AttributeError: pass try: self.queue.put(command, block=False) except queue.Full: log.warning("command queue is full, skipping command id=%s", cmd_id) def shutdown(self) -> None: try: log.info("shutting down command workers") except Exception: pass self.shutdown_event.set() for t in self.workers: t.join(timeout=1.0) self.workers.clear() def _worker_loop(self) -> None: # We need a separate worker logger in jobs.log try: worker_log = get_logger("worker") except Exception: class DummyLogger: def info(self, msg, *args): pass def error(self, msg, *args): pass def warning(self, msg, *args): pass worker_log = DummyLogger() import time while not self.shutdown_event.is_set(): if self.updating: time.sleep(0.5) continue try: command = self.queue.get(timeout=1.0) except queue.Empty: continue try: self._execute_command(command, worker_log) except Exception as e: try: worker_log.error("Unhandled worker error: %s", e) except Exception: pass finally: self.queue.task_done() def _execute_command(self, command: dict[str, Any], worker_log) -> None: cmd_id = str(command.get("id")) with self.lock: self.active_commands.add(cmd_id) try: cmd_type = command.get("command_type") payload = command.get("payload") or {} token = load_agent_token(self.config) try: worker_log.info("processing command id=%s type=%s", cmd_id, cmd_type) except Exception: pass try: state_changes = { "status": "running", "updated_at": datetime.utcnow().isoformat() } if cmd_type == "server.reboot": state_changes["started_at"] = datetime.utcnow().isoformat() state_changes["pre_reboot_boot_time"] = get_system_boot_time() update_command_state( self.config.var_dir, cmd_id, **state_changes ) except AttributeError: pass # Notify backend that command is running try: self.api_client.post_command_result( int(cmd_id), {"status": "running"}, token ) except Exception as exc: try: worker_log.warning("failed to post running status for command id=%s: %s", cmd_id, exc) except Exception: pass if cmd_type == "agent.update": payload = dict(payload) payload["job_id"] = cmd_id # Determine handler handler = HANDLERS.get(cmd_type) if not handler: try: worker_log.error("unsupported command type: %s", cmd_type) except Exception: pass status_str = "failed" payload_result = {"error": f"Unsupported command type: {cmd_type}"} else: result_container = {} def run_handler(): try: if cmd_type in ("pipeline.run", "site.create", "site.delete", "agent.update"): res = handler(payload, self.config, self.api_client) else: res = handler(payload) result_container["result"] = res except Exception as exc: result_container["exception"] = exc handler_thread = threading.Thread( target=run_handler, name=f"CommandHandler-{cmd_id}", daemon=True ) handler_thread.start() timeout = int(payload.get("timeout", 600)) # Default 10 minutes handler_thread.join(timeout=timeout) if handler_thread.is_alive(): try: worker_log.error("command id=%s timed out after %s seconds", cmd_id, timeout) except Exception: pass # Attempt to terminate the thread using ctypes SetAsyncExc try: import ctypes if handler_thread.ident: res = ctypes.pythonapi.PyThreadState_SetAsyncExc( ctypes.c_long(handler_thread.ident), ctypes.py_object(SystemExit) ) if res > 1: ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(handler_thread.ident), None) elif handler_thread.ident: pass except Exception as ex: try: worker_log.warning("failed to raise exception in timed out thread: %s", ex) except Exception: pass status_str = "failed" payload_result = {"error": f"Command execution timed out after {timeout} seconds"} else: if "exception" in result_container: status_str = "failed" payload_result = {"error": f"Handler exception: {str(result_container['exception'])}"} else: res = result_container.get("result", {}) if res.get("success", True) is True: status_str = "succeeded" else: status_str = "failed" payload_result = res # Post the final result try: self.api_client.post_command_result( int(cmd_id), {"status": status_str, "result": payload_result}, token, ) try: worker_log.info("posted command result for id=%s status=%s", cmd_id, status_str) except Exception: pass except Exception as exc: try: worker_log.error("failed to post command result for id=%s: %s", cmd_id, exc) except Exception: pass local_status = "success" if status_str == "succeeded" else "failed" try: update_command_state( self.config.var_dir, cmd_id, status=local_status, updated_at=datetime.utcnow().isoformat() ) except AttributeError: pass finally: with self.lock: self.active_commands.discard(cmd_id) @atexit.register def shutdown_workers() -> None: global _worker_manager if _worker_manager is not None: _worker_manager.shutdown() def process_commands(config: Config, api_client: ApiClient) -> None: """Fetch pending commands from Central API and dispatch them to worker manager. """ if config.status == "revoked": log.warning("Command runner skipped: Agent is revoked.") return token = load_agent_token(config) if not (config.server_id and token): log.debug("Command runner skipped: Unregistered agent.") return global _worker_manager if _worker_manager is None: try: recover_interrupted_commands(config, api_client) except Exception as exc: log.error("failed to recover interrupted commands: %s", exc) _worker_manager = CommandWorkerManager(config, api_client) _worker_manager.start_workers() else: _worker_manager.config = config _worker_manager.api_client = api_client import sys is_test = "pytest" in sys.modules or "unittest" in sys.modules default_workers = "0" if is_test else "3" if os.environ.get("WOLFPANEL_TEST_ASYNC") == "1": default_workers = "3" _worker_manager.max_workers = int(os.environ.get("WOLFPANEL_MAX_CONCURRENT_COMMANDS", default_workers)) _worker_manager.start_workers() try: commands = api_client.get_pending_commands(token) except ApiError as exc: log.warning("failed to fetch pending commands: %s", exc) return if not commands: return log.info("fetched %d pending command(s)", len(commands)) for command in commands: try: command_id = command.get("id") command_type = command.get("command_type") if not command_id or not command_type: log.warning("skipped malformed command: %s", command) continue # Prevent duplicate processing of the same command cmd_id_str = str(command_id) with _state_lock: states = load_command_states(config.var_dir) if cmd_id_str in states: current_status = states[cmd_id_str].get("status") if current_status in ("pending", "running"): log.info("command id=%s is already %s, skipping", cmd_id_str, current_status) continue _worker_manager.enqueue(command) except Exception as exc: log.error("failed to queue command: %s", exc)