1544 lines
63 KiB
Python
1544 lines
63 KiB
Python
import base64
|
||
import codecs
|
||
import time
|
||
import traceback
|
||
from typing import Any, Callable, Dict, List, Optional, Union
|
||
|
||
import paramiko
|
||
|
||
from config import (
|
||
EMULATOR_ADB_IP_OFFSET,
|
||
LOCAL_IMPORT_DIR,
|
||
MUMU_BRIDGE_IP_OFFSET,
|
||
MUMU_NETWORK_GATEWAYS,
|
||
MUMU_NETWORK_SCRIPT_PATH,
|
||
MUMU_RECOVER_AHK_EXE,
|
||
MUMU_RECOVER_SCRIPT_PATH,
|
||
MUMU_MANAGER_PATH,
|
||
MUMU_RESTART_SETTLE_SECONDS,
|
||
MUMU_VM_INDEX,
|
||
NET_BRIDGE_CARD,
|
||
RUNBATCH_FORCE_KILL_IMAGES,
|
||
RUNBATCH_LEGACY_END_WORKER_IPS,
|
||
RUN_PIPELINE_DEFAULT_OPTIONS,
|
||
SSH_ACTION_TIMEOUT_SECONDS,
|
||
SSH_CONNECT_TIMEOUT_SECONDS,
|
||
SSH_DEFAULT_PASSWORD,
|
||
SHARE_SMB_USER,
|
||
SHARE_SMB_PASSWORD,
|
||
extract_worker_ip,
|
||
get_repo_url,
|
||
get_clean_backup_path,
|
||
get_share_smb_target,
|
||
)
|
||
from log_manager import logger
|
||
|
||
RemoteCommand = Union[str, Dict[str, str]]
|
||
MUMU_NETWORK_GATEWAY_RULES = ";".join(f"{prefix}={gateway}" for prefix, gateway in MUMU_NETWORK_GATEWAYS.items())
|
||
|
||
class RemoteWorkerController:
|
||
def __init__(
|
||
self,
|
||
worker_inventory: List[Dict[str, Any]],
|
||
*,
|
||
ssh_connect_timeout: int = SSH_CONNECT_TIMEOUT_SECONDS,
|
||
ssh_action_timeout: int = SSH_ACTION_TIMEOUT_SECONDS,
|
||
default_password: str = SSH_DEFAULT_PASSWORD,
|
||
):
|
||
self.worker_inventory = {item["worker_id"]: dict(item) for item in worker_inventory}
|
||
self.ssh_connect_timeout = ssh_connect_timeout
|
||
self.ssh_action_timeout = ssh_action_timeout
|
||
self.default_password = default_password
|
||
|
||
def get_worker(self, worker_id: str) -> Dict[str, Any]:
|
||
worker = self.worker_inventory.get(worker_id)
|
||
if not worker:
|
||
raise KeyError(f"unknown worker_id: {worker_id}")
|
||
return worker
|
||
|
||
def run_action(
|
||
self,
|
||
worker_id: str,
|
||
action: str,
|
||
timeout: Optional[int] = None,
|
||
options: Optional[Dict[str, Any]] = None,
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
|
||
) -> Dict[str, Any]:
|
||
worker = self.get_worker(worker_id)
|
||
if action == "clone":
|
||
return self._run_clone(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
if action == "pull":
|
||
return self._run_pull(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
if action == "setup":
|
||
return self._run_setup(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
if action == "pull_pcap_files":
|
||
return self._run_pull_pcap_files(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
if action == "stop_worker":
|
||
return self._run_stop_worker(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
if action == "status":
|
||
return self._run_status(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
if action == "reboot":
|
||
return self._run_reboot(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
if action == "recover_mumu":
|
||
return self._run_recover_mumu(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
if action == "restart_mumu":
|
||
return self._run_restart_mumu(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
if action == "start_worker":
|
||
return self._run_start_worker(
|
||
worker_id,
|
||
worker,
|
||
timeout=timeout,
|
||
progress_callback=progress_callback,
|
||
)
|
||
if action == "run_pipeline":
|
||
return self._run_pipeline(worker_id, worker, timeout=timeout, options=options, progress_callback=progress_callback)
|
||
if action == "execute_command":
|
||
return self._run_execute_command(worker_id, worker, timeout=timeout, options=options, progress_callback=progress_callback)
|
||
if action == "fix_adb_connection":
|
||
return self._run_fix_adb_connection(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
if action == "configure_mumu_network":
|
||
return self._run_configure_mumu_network(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
if action == "recover_mumu_full":
|
||
return self._run_recover_mumu_full(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
raise ValueError(f"unsupported action: {action}")
|
||
|
||
def _run_clone(
|
||
self,
|
||
worker_id: str,
|
||
worker: Dict[str, Any],
|
||
*,
|
||
timeout: Optional[int] = None,
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
|
||
) -> Dict[str, Any]:
|
||
return self._run_ssh_command(
|
||
worker_id,
|
||
worker,
|
||
"clone",
|
||
self._clone_command(worker),
|
||
timeout=timeout,
|
||
progress_callback=progress_callback,
|
||
)
|
||
|
||
def _run_execute_command(
|
||
self,
|
||
worker_id: str,
|
||
worker: Dict[str, Any],
|
||
*,
|
||
timeout: Optional[int] = None,
|
||
options: Optional[Dict[str, Any]] = None,
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
|
||
) -> Dict[str, Any]:
|
||
command = (options or {}).get("command", "")
|
||
if not command:
|
||
return {
|
||
"ok": False,
|
||
"worker_id": worker_id,
|
||
"action": "execute_command",
|
||
"message": "no command provided",
|
||
}
|
||
return self._run_ssh_command(
|
||
worker_id,
|
||
worker,
|
||
"execute_command",
|
||
command,
|
||
timeout=timeout,
|
||
progress_callback=progress_callback,
|
||
)
|
||
|
||
def _run_fix_adb_connection(
|
||
self,
|
||
worker_id: str,
|
||
worker: Dict[str, Any],
|
||
*,
|
||
timeout: Optional[int] = None,
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
|
||
) -> Dict[str, Any]:
|
||
host = str(worker.get("ssh_host") or "").strip()
|
||
parts = host.split(".")
|
||
if len(parts) != 4:
|
||
return {
|
||
"ok": False,
|
||
"worker_id": worker_id,
|
||
"action": "fix_adb_connection",
|
||
"message": f"无法解析 Worker IP: {host}",
|
||
}
|
||
try:
|
||
last_octet = int(parts[3]) + EMULATOR_ADB_IP_OFFSET
|
||
except (ValueError, IndexError):
|
||
return {
|
||
"ok": False,
|
||
"worker_id": worker_id,
|
||
"action": "fix_adb_connection",
|
||
"message": f"无法计算模拟器 IP: {host}",
|
||
}
|
||
emulator_ip = f"{parts[0]}.{parts[1]}.{parts[2]}.{last_octet}"
|
||
command = f"adb disconnect && adb connect {emulator_ip}:5555"
|
||
return self._run_ssh_command(
|
||
worker_id,
|
||
worker,
|
||
"fix_adb_connection",
|
||
command,
|
||
timeout=timeout,
|
||
progress_callback=progress_callback,
|
||
)
|
||
|
||
def _run_configure_mumu_network(
|
||
self,
|
||
worker_id: str,
|
||
worker: Dict[str, Any],
|
||
*,
|
||
timeout: Optional[int] = None,
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
|
||
) -> Dict[str, Any]:
|
||
manager = MUMU_MANAGER_PATH
|
||
vm_index = MUMU_VM_INDEX
|
||
default_adapter = NET_BRIDGE_CARD
|
||
script_path = MUMU_NETWORK_SCRIPT_PATH
|
||
command = " ".join(
|
||
[
|
||
self._cmd_quote(script_path),
|
||
self._cmd_quote(manager),
|
||
str(vm_index),
|
||
self._cmd_quote(default_adapter),
|
||
str(MUMU_BRIDGE_IP_OFFSET),
|
||
self._cmd_quote(MUMU_NETWORK_GATEWAY_RULES),
|
||
]
|
||
)
|
||
inner = f"if not exist {self._cmd_quote(script_path)} (echo [ERROR] script not found: {script_path} 1>&2 & exit /b 1) & call {command}"
|
||
return self._run_ssh_command(
|
||
worker_id, worker, "configure_mumu_network",
|
||
f"cmd.exe /d /q /s /c {self._cmd_quote(inner)}",
|
||
timeout=timeout, progress_callback=progress_callback,
|
||
)
|
||
|
||
def _run_pull(
|
||
self,
|
||
worker_id: str,
|
||
worker: Dict[str, Any],
|
||
*,
|
||
timeout: Optional[int] = None,
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
|
||
) -> Dict[str, Any]:
|
||
return self._run_ssh_command(
|
||
worker_id,
|
||
worker,
|
||
"pull",
|
||
self._pull_command(worker),
|
||
timeout=timeout,
|
||
progress_callback=progress_callback,
|
||
)
|
||
|
||
def _run_setup(
|
||
self,
|
||
worker_id: str,
|
||
worker: Dict[str, Any],
|
||
*,
|
||
timeout: Optional[int] = None,
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
|
||
) -> Dict[str, Any]:
|
||
return self._run_ssh_command(
|
||
worker_id,
|
||
worker,
|
||
"setup",
|
||
self._setup_command(worker),
|
||
timeout=timeout,
|
||
progress_callback=progress_callback,
|
||
)
|
||
|
||
def _run_stop_worker(
|
||
self,
|
||
worker_id: str,
|
||
worker: Dict[str, Any],
|
||
*,
|
||
timeout: Optional[int] = None,
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
|
||
) -> Dict[str, Any]:
|
||
return self._run_ssh_command(
|
||
worker_id,
|
||
worker,
|
||
"stop_worker",
|
||
self._stop_worker_command(worker),
|
||
timeout=timeout,
|
||
progress_callback=progress_callback,
|
||
)
|
||
|
||
def _run_pull_pcap_files(
|
||
self,
|
||
worker_id: str,
|
||
worker: Dict[str, Any],
|
||
*,
|
||
timeout: Optional[int] = None,
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
|
||
) -> Dict[str, Any]:
|
||
pull_result = self._run_pull_pcap_files_once(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
steps: List[Dict[str, Any]] = [self._step_result("pull_pcap_files", pull_result)]
|
||
if not pull_result.get("ok"):
|
||
return {
|
||
"ok": False,
|
||
"worker_id": worker_id,
|
||
"action": "pull_pcap_files",
|
||
"message": pull_result.get("message", "pull_pcap_files failed"),
|
||
"failed_step": "pull_pcap_files",
|
||
"steps": steps,
|
||
"stdout": pull_result.get("stdout", ""),
|
||
"stderr": pull_result.get("stderr", ""),
|
||
}
|
||
|
||
return {
|
||
"ok": True,
|
||
"worker_id": worker_id,
|
||
"action": "pull_pcap_files",
|
||
"message": pull_result.get("message", "pull_pcap_files completed"),
|
||
"steps": steps,
|
||
"stdout": pull_result.get("stdout", ""),
|
||
"stderr": pull_result.get("stderr", ""),
|
||
}
|
||
|
||
def _run_pull_pcap_files_once(
|
||
self,
|
||
worker_id: str,
|
||
worker: Dict[str, Any],
|
||
*,
|
||
timeout: Optional[int] = None,
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
|
||
) -> Dict[str, Any]:
|
||
return self._run_ssh_command(
|
||
worker_id,
|
||
worker,
|
||
"pull_pcap_files",
|
||
self._pull_pcap_files_command(worker),
|
||
timeout=timeout,
|
||
progress_callback=progress_callback,
|
||
)
|
||
|
||
def _run_status(
|
||
self,
|
||
worker_id: str,
|
||
worker: Dict[str, Any],
|
||
*,
|
||
timeout: Optional[int] = None,
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
|
||
) -> Dict[str, Any]:
|
||
return self._run_ssh_command(
|
||
worker_id,
|
||
worker,
|
||
"status",
|
||
self._status_command(worker),
|
||
timeout=timeout,
|
||
progress_callback=progress_callback,
|
||
)
|
||
|
||
def _run_reboot(
|
||
self,
|
||
worker_id: str,
|
||
worker: Dict[str, Any],
|
||
*,
|
||
timeout: Optional[int] = None,
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
|
||
) -> Dict[str, Any]:
|
||
return self._run_ssh_command(
|
||
worker_id,
|
||
worker,
|
||
"reboot",
|
||
self._reboot_command(),
|
||
timeout=timeout,
|
||
progress_callback=progress_callback,
|
||
)
|
||
|
||
def _run_recover_mumu(
|
||
self,
|
||
worker_id: str,
|
||
worker: Dict[str, Any],
|
||
*,
|
||
timeout: Optional[int] = None,
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
|
||
) -> Dict[str, Any]:
|
||
command = self._recover_mumu_command(worker)
|
||
primary_timeout = (timeout or self.ssh_action_timeout) // 2
|
||
result = self._run_ssh_command(
|
||
worker_id, worker, "recover_mumu",
|
||
command, timeout=primary_timeout,
|
||
progress_callback=progress_callback,
|
||
)
|
||
if result.get("ok"):
|
||
return result
|
||
logger.warning(
|
||
"[WorkerAction] recover_mumu 首次失败 worker=%s, 自动重试1次",
|
||
worker_id,
|
||
)
|
||
self._emit_progress(
|
||
progress_callback, worker_id, "recover_mumu", "retry",
|
||
"MuMu 镜像恢复首次失败，自动重试 1 次", status="running",
|
||
)
|
||
result = self._run_ssh_command(
|
||
worker_id, worker, "recover_mumu",
|
||
command, timeout=primary_timeout,
|
||
progress_callback=progress_callback,
|
||
)
|
||
if not result.get("ok"):
|
||
logger.warning(
|
||
"[WorkerAction] recover_mumu 重试依然失败 worker=%s message=%s",
|
||
worker_id, result.get("message", ""),
|
||
)
|
||
return result
|
||
|
||
def _run_recover_mumu_full(
|
||
self,
|
||
worker_id: str,
|
||
worker: Dict[str, Any],
|
||
*,
|
||
timeout: Optional[int] = None,
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
|
||
) -> Dict[str, Any]:
|
||
steps: List[Dict[str, Any]] = []
|
||
|
||
recover_result = self._run_recover_mumu(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
steps.append(self._step_result("recover_mumu", recover_result))
|
||
if not recover_result.get("ok"):
|
||
return self._pipeline_failure(worker_id, {}, steps, "recover_mumu", recover_result)
|
||
|
||
restart_result = self._run_restart_mumu(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
steps.append(self._step_result("restart_mumu", restart_result))
|
||
if not restart_result.get("ok"):
|
||
return self._pipeline_failure(worker_id, {}, steps, "restart_mumu", restart_result)
|
||
|
||
# wait for MuMu to boot before configuring network
|
||
self._wait_for_mumu_restart(worker_id, progress_callback=progress_callback)
|
||
|
||
network_result = self._run_configure_mumu_network(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
steps.append(self._step_result("configure_mumu_network", network_result))
|
||
|
||
adb_result = self._run_fix_adb_connection(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
steps.append(self._step_result("fix_adb_connection", adb_result))
|
||
|
||
return {
|
||
"ok": True,
|
||
"worker_id": worker_id,
|
||
"action": "recover_mumu_full",
|
||
"message": "完整恢复完成（镜像恢复 → 重启 → 网络配置 → ADB连接）",
|
||
"steps": steps,
|
||
"stdout": recover_result.get("stdout", ""),
|
||
"stderr": recover_result.get("stderr", ""),
|
||
}
|
||
|
||
def _run_restart_mumu(
|
||
self,
|
||
worker_id: str,
|
||
worker: Dict[str, Any],
|
||
*,
|
||
timeout: Optional[int] = None,
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
|
||
) -> Dict[str, Any]:
|
||
return self._run_ssh_command(
|
||
worker_id,
|
||
worker,
|
||
"restart_mumu",
|
||
self._restart_mumu_task_command(worker),
|
||
timeout=timeout,
|
||
progress_callback=progress_callback,
|
||
)
|
||
|
||
def _run_start_worker(
|
||
self,
|
||
worker_id: str,
|
||
worker: Dict[str, Any],
|
||
*,
|
||
timeout: Optional[int] = None,
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
|
||
) -> Dict[str, Any]:
|
||
return self._run_ssh_command(
|
||
worker_id,
|
||
worker,
|
||
"start_worker",
|
||
self._start_worker_command(worker),
|
||
timeout=timeout,
|
||
progress_callback=progress_callback,
|
||
)
|
||
|
||
def _run_pipeline(
|
||
self,
|
||
worker_id: str,
|
||
worker: Dict[str, Any],
|
||
*,
|
||
timeout: Optional[int] = None,
|
||
options: Optional[Dict[str, Any]] = None,
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
|
||
) -> Dict[str, Any]:
|
||
spec = dict(RUN_PIPELINE_DEFAULT_OPTIONS)
|
||
if options:
|
||
spec.update(options)
|
||
steps: List[Dict[str, Any]] = []
|
||
|
||
if not any(
|
||
spec.get(key)
|
||
for key in (
|
||
"clone_if_missing",
|
||
"stop_worker",
|
||
"git_pull",
|
||
"setup",
|
||
"pull_pcap_files",
|
||
"recover_mumu",
|
||
"restart_mumu",
|
||
"start_worker",
|
||
)
|
||
):
|
||
return {
|
||
"ok": True,
|
||
"worker_id": worker_id,
|
||
"action": "run_pipeline",
|
||
"message": "no pipeline steps selected",
|
||
"steps": steps,
|
||
"selected_options": spec,
|
||
}
|
||
|
||
if spec.get("clone_if_missing"):
|
||
result = self._run_ssh_command(
|
||
worker_id,
|
||
worker,
|
||
"clone_if_missing",
|
||
self._clone_command(worker),
|
||
timeout=timeout,
|
||
progress_callback=progress_callback,
|
||
)
|
||
steps.append(self._step_result("clone_if_missing", result))
|
||
if not result.get("ok"):
|
||
return self._pipeline_failure(worker_id, spec, steps, "clone_if_missing", result)
|
||
|
||
if spec.get("stop_worker"):
|
||
result = self._run_stop_worker(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
steps.append(self._step_result("stop_worker", result))
|
||
if not result.get("ok"):
|
||
return self._pipeline_failure(worker_id, spec, steps, "stop_worker", result)
|
||
|
||
if spec.get("git_pull"):
|
||
result = self._run_pull(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
steps.append(self._step_result("git_pull", result))
|
||
if not result.get("ok"):
|
||
return self._pipeline_failure(worker_id, spec, steps, "git_pull", result)
|
||
|
||
if spec.get("setup"):
|
||
result = self._run_setup(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
steps.append(self._step_result("setup", result))
|
||
if not result.get("ok"):
|
||
return self._pipeline_failure(worker_id, spec, steps, "setup", result)
|
||
|
||
pcap_pulled = False
|
||
restart_ready = False
|
||
if spec.get("pull_pcap_files") and spec.get("recover_mumu"):
|
||
if not restart_ready:
|
||
result = self._run_restart_mumu(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
steps.append(self._step_result("restart_mumu_before_pull_pcap_files", result))
|
||
if not result.get("ok"):
|
||
return self._pipeline_failure(worker_id, spec, steps, "restart_mumu_before_pull_pcap_files", result)
|
||
self._wait_for_mumu_restart(worker_id, progress_callback=progress_callback)
|
||
restart_ready = True
|
||
result = self._run_pull_pcap_files_once(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
steps.append(self._step_result("pull_pcap_files", result))
|
||
if not result.get("ok"):
|
||
return self._pipeline_failure(worker_id, spec, steps, "pull_pcap_files", result)
|
||
pcap_pulled = True
|
||
|
||
if spec.get("recover_mumu"):
|
||
result = self._run_recover_mumu(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
steps.append(self._step_result("recover_mumu", result))
|
||
if not result.get("ok"):
|
||
return self._pipeline_failure(worker_id, spec, steps, "recover_mumu", result)
|
||
restart_ready = False
|
||
|
||
if spec.get("restart_mumu"):
|
||
result = self._run_restart_mumu(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
steps.append(self._step_result("restart_mumu", result))
|
||
if not result.get("ok"):
|
||
return self._pipeline_failure(worker_id, spec, steps, "restart_mumu", result)
|
||
self._run_configure_mumu_network(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
self._run_fix_adb_connection(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
if spec.get("start_worker") or (spec.get("pull_pcap_files") and not pcap_pulled):
|
||
self._wait_for_mumu_restart(worker_id, progress_callback=progress_callback)
|
||
restart_ready = True
|
||
|
||
if spec.get("start_worker"):
|
||
if spec.get("pull_pcap_files") and not pcap_pulled:
|
||
if not restart_ready:
|
||
result = self._run_restart_mumu(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
steps.append(self._step_result("restart_mumu_before_pull_pcap_files", result))
|
||
if not result.get("ok"):
|
||
return self._pipeline_failure(worker_id, spec, steps, "restart_mumu_before_pull_pcap_files", result)
|
||
self._wait_for_mumu_restart(worker_id, progress_callback=progress_callback)
|
||
restart_ready = True
|
||
result = self._run_pull_pcap_files_once(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
steps.append(self._step_result("pull_pcap_files", result))
|
||
if not result.get("ok"):
|
||
return self._pipeline_failure(worker_id, spec, steps, "pull_pcap_files", result)
|
||
pcap_pulled = True
|
||
result = self._run_start_worker(
|
||
worker_id,
|
||
worker,
|
||
timeout=timeout,
|
||
progress_callback=progress_callback,
|
||
)
|
||
steps.append(self._step_result("start_worker", result))
|
||
if not result.get("ok"):
|
||
return self._pipeline_failure(worker_id, spec, steps, "start_worker", result)
|
||
|
||
if spec.get("pull_pcap_files") and not pcap_pulled:
|
||
if not restart_ready:
|
||
result = self._run_restart_mumu(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
steps.append(self._step_result("restart_mumu_before_pull_pcap_files", result))
|
||
if not result.get("ok"):
|
||
return self._pipeline_failure(worker_id, spec, steps, "restart_mumu_before_pull_pcap_files", result)
|
||
self._wait_for_mumu_restart(worker_id, progress_callback=progress_callback)
|
||
result = self._run_pull_pcap_files_once(worker_id, worker, timeout=timeout, progress_callback=progress_callback)
|
||
steps.append(self._step_result("pull_pcap_files", result))
|
||
if not result.get("ok"):
|
||
return self._pipeline_failure(worker_id, spec, steps, "pull_pcap_files", result)
|
||
|
||
return {
|
||
"ok": True,
|
||
"worker_id": worker_id,
|
||
"action": "run_pipeline",
|
||
"message": "pipeline completed",
|
||
"steps": steps,
|
||
"selected_options": spec,
|
||
}
|
||
|
||
@staticmethod
|
||
def _step_result(step: str, result: Dict[str, Any]) -> Dict[str, Any]:
|
||
item = {
|
||
"step": step,
|
||
"ok": result.get("ok"),
|
||
"message": result.get("message", ""),
|
||
}
|
||
command = str(result.get("command", "") or "").strip()
|
||
if command:
|
||
item["command"] = command
|
||
# 保留 stdout/stderr，确保失败步骤的详细报错信息不会丢失
|
||
for key in ("stdout", "stderr"):
|
||
value = str(result.get(key, "") or "").strip()
|
||
if value:
|
||
item[key] = value
|
||
return item
|
||
|
||
@staticmethod
|
||
def _pipeline_failure(
|
||
worker_id: str,
|
||
spec: Dict[str, Any],
|
||
steps: List[Dict[str, Any]],
|
||
failed_step: str,
|
||
result: Dict[str, Any],
|
||
) -> Dict[str, Any]:
|
||
return {
|
||
"ok": False,
|
||
"worker_id": worker_id,
|
||
"action": "run_pipeline",
|
||
"message": result.get("message", f"{failed_step} failed"),
|
||
"failed_step": failed_step,
|
||
"steps": steps,
|
||
"selected_options": spec,
|
||
"stdout": result.get("stdout", ""),
|
||
"stderr": result.get("stderr", ""),
|
||
}
|
||
|
||
def _clone_command(self, worker: Dict[str, Any]) -> str:
|
||
repo_dir = self._repo_dir(worker)
|
||
repo_url = get_repo_url(worker["ssh_target"])
|
||
return (
|
||
f'if exist "{repo_dir}\\.git" '
|
||
f'(cd /d "{repo_dir}" && git config lfs.fetchexclude "" && git lfs pull) '
|
||
f'else (git clone "{repo_url}" "{repo_dir}" && cd /d "{repo_dir}" && git config lfs.fetchexclude "" && git lfs pull)'
|
||
)
|
||
|
||
def _pull_command(self, worker: Dict[str, Any]) -> str:
|
||
repo_dir = self._repo_dir(worker)
|
||
is_physical = str(worker.get("device_type") or "").strip().lower() == "physical"
|
||
if is_physical:
|
||
pull_cmd = "git stash && git pull && (git stash pop || ver>nul)"
|
||
else:
|
||
pull_cmd = "git reset --hard HEAD && git pull"
|
||
return (
|
||
f'if exist "{repo_dir}\\.git" '
|
||
f'(cd /d "{repo_dir}" && {pull_cmd} && git config lfs.fetchexclude "" && git lfs pull) '
|
||
f'else (echo repository is not initialized 1>&2 & exit /b 1)'
|
||
)
|
||
|
||
def _setup_command(self, worker: Dict[str, Any]) -> str:
|
||
repo_dir = self._repo_dir(worker)
|
||
bootstrap_python = self._bootstrap_python(worker)
|
||
# 纯 CLI 操作，无需 GUI/用户会话，直接通过 SSH 执行
|
||
cmd_line = (
|
||
f'cd /d "{repo_dir}" && '
|
||
f'"{bootstrap_python}" -m pip config set global.index-url https://mirrors.aliyun.com/pypi/simple && '
|
||
f'"{bootstrap_python}" -m pip config set install.trusted-host mirrors.aliyun.com && '
|
||
f'"{bootstrap_python}" setup.py'
|
||
)
|
||
return self._repo_required_command(worker, cmd_line)
|
||
|
||
def _stop_worker_command(self, worker: Dict[str, Any]) -> RemoteCommand:
|
||
if self._use_legacy_runbatch_end(worker):
|
||
return self._legacy_runbatch_end_command()
|
||
return self._runbatch_taskkill_command()
|
||
|
||
def _status_command(self, worker: Dict[str, Any]) -> RemoteCommand:
|
||
repo_dir = self._repo_dir(worker)
|
||
venv_python = self._venv_python(worker)
|
||
script = "\n".join(
|
||
[
|
||
f"$repoDir = {self._powershell_quote(repo_dir)}",
|
||
f"$venvPython = {self._powershell_quote(venv_python)}",
|
||
(
|
||
"$workerRunning = @("
|
||
"Get-CimInstance Win32_Process -ErrorAction SilentlyContinue | "
|
||
"Where-Object { "
|
||
"$_.Name -in @('python.exe', 'pythonw.exe') -and "
|
||
"$_.CommandLine -like '*batch_run.py*' "
|
||
"}"
|
||
").Count -gt 0"
|
||
),
|
||
"Write-Output ('repo_dir=' + $repoDir)",
|
||
"Write-Output ('repo_exists=' + [int](Test-Path -LiteralPath (Join-Path $repoDir '.git')))",
|
||
"Write-Output ('venv_exists=' + [int](Test-Path -LiteralPath $venvPython))",
|
||
"Write-Output ('worker_running=' + [int]$workerRunning)",
|
||
]
|
||
)
|
||
return self._powershell_stdin_command(script)
|
||
|
||
def _reboot_command(self) -> str:
|
||
return '"C:\\Windows\\System32\\shutdown.exe" /r /f /t 10 /c "Restart initiated by remote controller"'
|
||
|
||
def _recover_mumu_command(self, worker: Dict[str, Any]) -> str:
|
||
manager = MUMU_MANAGER_PATH
|
||
vm_index = MUMU_VM_INDEX
|
||
source = get_clean_backup_path(worker["ssh_target"])
|
||
local_import_dir = LOCAL_IMPORT_DIR
|
||
script_path = MUMU_RECOVER_SCRIPT_PATH
|
||
script_call = " ".join(
|
||
[
|
||
self._cmd_quote(script_path),
|
||
self._cmd_quote(source),
|
||
self._cmd_quote(local_import_dir),
|
||
self._cmd_quote(manager),
|
||
str(vm_index),
|
||
self._cmd_quote(NET_BRIDGE_CARD),
|
||
self._cmd_quote(MUMU_RECOVER_AHK_EXE),
|
||
str(MUMU_BRIDGE_IP_OFFSET),
|
||
self._cmd_quote(MUMU_NETWORK_GATEWAY_RULES),
|
||
]
|
||
)
|
||
inner_command = (
|
||
f"if not exist {self._cmd_quote(script_path)} "
|
||
f"(echo [ERROR] Recover script not found: {script_path} 1>&2 & exit /b 1) & call {script_call}"
|
||
)
|
||
return f"cmd.exe /d /q /s /c {self._cmd_quote(inner_command)}"
|
||
|
||
def _restart_mumu_command(self, worker: Dict[str, Any]) -> str:
|
||
manager = MUMU_MANAGER_PATH
|
||
vm_index = MUMU_VM_INDEX
|
||
del worker
|
||
return f'"{manager}" control -v {vm_index} restart'
|
||
|
||
def _restart_mumu_task_command(self, worker: Dict[str, Any]) -> RemoteCommand:
|
||
return self._scheduled_task_run_command(
|
||
"RestartMuMu",
|
||
self._restart_mumu_command(worker),
|
||
direct_command=True,
|
||
)
|
||
|
||
def _wait_for_mumu_restart(
|
||
self,
|
||
worker_id: str,
|
||
*,
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
|
||
) -> None:
|
||
message = f"等待 MuMu 重启稳定 {MUMU_RESTART_SETTLE_SECONDS} 秒"
|
||
self._emit_progress(progress_callback, worker_id, "run_pipeline", "wait_mumu_restart", message, status="running")
|
||
time.sleep(MUMU_RESTART_SETTLE_SECONDS)
|
||
self._emit_progress(progress_callback, worker_id, "run_pipeline", "wait_mumu_restart", "MuMu 重启等待完成", status="success")
|
||
|
||
def _start_worker_command(self, worker: Dict[str, Any]) -> RemoteCommand:
|
||
repo_dir = self._repo_dir(worker)
|
||
batch_script = f"{self._repo_dir(worker)}\\batch_run.py"
|
||
venv_python = self._venv_python(worker)
|
||
inner_command = f"{venv_python} {batch_script}"
|
||
return self._scheduled_task_run_command(
|
||
"RunBatch",
|
||
inner_command,
|
||
keep_open=True,
|
||
stop_existing=True,
|
||
unlimited_execution_time=True,
|
||
working_directory=repo_dir,
|
||
)
|
||
|
||
def _pull_pcap_files_command(self, worker: Dict[str, Any]) -> str:
|
||
utils_dir = f"{self._repo_dir(worker)}\\utils_android"
|
||
# ADB 拉取 pcap 文件，纯 CLI 操作，无需计划任务
|
||
inner_command = f'cd /d "{utils_dir}" && python pull_pcapdroid_files.py'
|
||
return self._repo_required_command(worker, inner_command)
|
||
|
||
def _repo_required_command(self, worker: Dict[str, Any], command: str) -> str:
|
||
repo_dir = self._repo_dir(worker)
|
||
return f'if not exist "{repo_dir}\\.git" (echo repository is not initialized 1>&2 & exit /b 1) && {command}'
|
||
|
||
def _repo_dir(self, worker: Dict[str, Any]) -> str:
|
||
return str(worker["repo_dir"]).replace("/", "\\")
|
||
|
||
@staticmethod
|
||
def _legacy_runbatch_end_command() -> str:
|
||
return (
|
||
'cmd.exe /d /v:on /s /c "set STOPPED=0'
|
||
' & schtasks /end /tn ""RunBatch"" >nul 2>&1 && set STOPPED=1'
|
||
' & if !STOPPED!==1 (echo worker stopped) else (echo worker is not running)"'
|
||
)
|
||
|
||
def _runbatch_taskkill_command(self) -> str:
|
||
command_script = self._runbatch_taskkill_command_script()
|
||
return f"cmd.exe /d /v:on /q /s /c {self._cmd_quote(command_script)}"
|
||
|
||
def _runbatch_taskkill_command_script(self) -> str:
|
||
command_parts = [
|
||
"set STOPPED=0",
|
||
'schtasks /end /tn "RunBatch" >nul 2>&1 && set STOPPED=1',
|
||
]
|
||
for image_name in RUNBATCH_FORCE_KILL_IMAGES:
|
||
command_parts.append(f"taskkill /F /T /IM {image_name} >nul 2>&1 && set STOPPED=1")
|
||
command_parts.extend(
|
||
[
|
||
'if !STOPPED!==1 (echo worker stopped) else (echo worker is not running)',
|
||
"start /min taskkill /F /T /IM cmd.exe",
|
||
]
|
||
)
|
||
return " & ".join(command_parts)
|
||
|
||
@staticmethod
|
||
def _use_legacy_runbatch_end(worker: Dict[str, Any]) -> bool:
|
||
for key in ("worker_id", "ssh_host", "ssh_target"):
|
||
worker_ip = extract_worker_ip(worker.get(key, ""))
|
||
if worker_ip in RUNBATCH_LEGACY_END_WORKER_IPS:
|
||
return True
|
||
return False
|
||
|
||
@staticmethod
|
||
def _cmd_escape(value: str) -> str:
|
||
return str(value).replace('"', '""')
|
||
|
||
@classmethod
|
||
def _cmd_quote(cls, value: str) -> str:
|
||
return f'"{cls._cmd_escape(value)}"'
|
||
|
||
@classmethod
|
||
def _cmd_task_command(cls, inner_command: str) -> str:
|
||
del cls
|
||
return f'C:\\Windows\\System32\\cmd.exe /d /q /s /c "{inner_command}"'
|
||
|
||
@staticmethod
|
||
def _cmd_sleep(seconds: int) -> str:
|
||
return f"ping -n {max(1, int(seconds)) + 1} 127.0.0.1 >nul"
|
||
|
||
@staticmethod
|
||
def _powershell_quote(value: str) -> str:
|
||
return "'" + str(value).replace("'", "''") + "'"
|
||
|
||
@staticmethod
|
||
def _cmd_switch_with_command(cmd_switch: str, inner_command: str) -> str:
|
||
return f"{cmd_switch} {inner_command}"
|
||
|
||
@staticmethod
|
||
def _powershell_encoded_command(script: str) -> str:
|
||
payload = base64.b64encode(str(script).encode("utf-16le")).decode("ascii")
|
||
return f"powershell -NoProfile -ExecutionPolicy Bypass -EncodedCommand {payload}"
|
||
|
||
@staticmethod
|
||
def _powershell_stdin_command(script: str) -> Dict[str, str]:
|
||
return {
|
||
"command": "powershell -NoProfile -ExecutionPolicy Bypass -Command -",
|
||
"stdin": str(script),
|
||
}
|
||
|
||
@staticmethod
|
||
def _task_signal_name(task_name: str) -> str:
|
||
sanitized = "".join(ch if ch.isalnum() else "_" for ch in str(task_name))
|
||
return f"AUTOOL_{sanitized}_DONE"
|
||
|
||
def _scheduled_task_runner_script(
|
||
self,
|
||
task_name: str,
|
||
inner_command: str,
|
||
*,
|
||
wait: bool,
|
||
working_directory: Optional[str],
|
||
) -> str:
|
||
artifact_dir = "%TEMP%\\autool-dispatcher-remote-worker"
|
||
stdout_path = f"{artifact_dir}\\{task_name}.stdout.log"
|
||
stderr_path = f"{artifact_dir}\\{task_name}.stderr.log"
|
||
exit_path = f"{artifact_dir}\\{task_name}.exitcode"
|
||
lines = [
|
||
"@echo off",
|
||
"setlocal",
|
||
f'if not exist "{artifact_dir}" mkdir "{artifact_dir}"',
|
||
]
|
||
if working_directory:
|
||
lines.append(f'cd /d "{working_directory}"')
|
||
if wait:
|
||
lines.extend(
|
||
[
|
||
f'del /f /q "{stdout_path}" "{stderr_path}" "{exit_path}" >nul 2>&1',
|
||
f'({inner_command}) 1>"{stdout_path}" 2>"{stderr_path}"',
|
||
'set "RC=%ERRORLEVEL%"',
|
||
f'> "{exit_path}" echo %RC%',
|
||
f'waitfor /SI {self._task_signal_name(task_name)} >nul 2>&1',
|
||
"exit /b %RC%",
|
||
]
|
||
)
|
||
else:
|
||
lines.extend(
|
||
[
|
||
inner_command,
|
||
"exit /b %ERRORLEVEL%",
|
||
]
|
||
)
|
||
return "\r\n".join(lines) + "\r\n"
|
||
|
||
def _scheduled_task_command_via_cmd(
|
||
self,
|
||
task_name: str,
|
||
inner_command: str,
|
||
*,
|
||
wait: bool,
|
||
keep_open: bool,
|
||
stop_existing: bool,
|
||
working_directory: Optional[str],
|
||
) -> RemoteCommand:
|
||
artifact_dir = "%TEMP%\\autool-dispatcher-remote-worker"
|
||
runner_path = f"{artifact_dir}\\{task_name}.cmd"
|
||
stdout_path = f"{artifact_dir}\\{task_name}.stdout.log"
|
||
stderr_path = f"{artifact_dir}\\{task_name}.stderr.log"
|
||
exit_path = f"{artifact_dir}\\{task_name}.exitcode"
|
||
signal_name = self._task_signal_name(task_name)
|
||
cmd_switch = "/k" if keep_open else "/c"
|
||
runner_script = self._scheduled_task_runner_script(
|
||
task_name,
|
||
inner_command,
|
||
wait=wait,
|
||
working_directory=working_directory,
|
||
)
|
||
task_command = f'C:\\Windows\\System32\\cmd.exe {cmd_switch} {runner_path}'
|
||
command_parts = [
|
||
'if not exist "%TEMP%\\autool-dispatcher-remote-worker" mkdir "%TEMP%\\autool-dispatcher-remote-worker"',
|
||
f'type con > {self._cmd_quote(runner_path)}',
|
||
f'schtasks /create /tn {self._cmd_quote(task_name)} /tr {self._cmd_quote(task_command)} /sc ONCE /st 00:00 /rl HIGHEST /it /f >nul',
|
||
"if errorlevel 1 exit /b %ERRORLEVEL%",
|
||
]
|
||
if stop_existing:
|
||
command_parts.append(f'schtasks /end /tn {self._cmd_quote(task_name)} >nul 2>&1')
|
||
if wait:
|
||
command_parts.extend(
|
||
[
|
||
f'del /f /q "{stdout_path}" "{stderr_path}" "{exit_path}" >nul 2>&1',
|
||
f'schtasks /run /tn {self._cmd_quote(task_name)} >nul',
|
||
"if errorlevel 1 exit /b %ERRORLEVEL%",
|
||
f'waitfor {signal_name} /t {int(self.ssh_action_timeout)} >nul',
|
||
"if errorlevel 1 exit /b %ERRORLEVEL%",
|
||
f'set /p TASK_RC=<{self._cmd_quote(exit_path)}',
|
||
f'if exist "{stdout_path}" type "{stdout_path}"',
|
||
f'if exist "{stderr_path}" type "{stderr_path}" 1>&2',
|
||
"exit /b %TASK_RC%",
|
||
]
|
||
)
|
||
else:
|
||
command_parts.extend(
|
||
[
|
||
f'schtasks /run /tn {self._cmd_quote(task_name)} >nul',
|
||
"if errorlevel 1 exit /b %ERRORLEVEL%",
|
||
f'echo {task_name} started',
|
||
]
|
||
)
|
||
command_script = " & ".join(command_parts)
|
||
return {
|
||
"command": f"cmd.exe /d /q /s /c {self._cmd_quote(command_script)}",
|
||
"stdin": runner_script,
|
||
}
|
||
|
||
def _scheduled_task_direct_run_command_via_cmd(
|
||
self,
|
||
task_name: str,
|
||
task_command: str,
|
||
*,
|
||
stop_existing: bool = False,
|
||
) -> str:
|
||
command_parts = [
|
||
(
|
||
f'schtasks /create /tn {self._cmd_quote(task_name)} '
|
||
f'/tr {self._cmd_quote(task_command)} /sc ONCE /st 00:00 /rl HIGHEST /it /f >nul'
|
||
),
|
||
"if errorlevel 1 exit /b %ERRORLEVEL%",
|
||
]
|
||
if stop_existing:
|
||
command_parts.append(f'schtasks /end /tn {self._cmd_quote(task_name)} >nul 2>&1')
|
||
command_parts.extend(
|
||
[
|
||
f'schtasks /run /tn {self._cmd_quote(task_name)} >nul',
|
||
"if errorlevel 1 exit /b %ERRORLEVEL%",
|
||
f"echo {task_name} started",
|
||
]
|
||
)
|
||
command_script = " & ".join(command_parts)
|
||
return f"cmd.exe /d /q /s /c {self._cmd_quote(command_script)}"
|
||
|
||
def _scheduled_task_direct_wait_command_via_cmd(
|
||
self,
|
||
task_name: str,
|
||
inner_command: str,
|
||
) -> str:
|
||
artifact_dir = "%TEMP%\\autool-dispatcher-remote-worker"
|
||
stdout_path = f"{artifact_dir}\\{task_name}.stdout.log"
|
||
stderr_path = f"{artifact_dir}\\{task_name}.stderr.log"
|
||
exit_path = f"{artifact_dir}\\{task_name}.exitcode"
|
||
loop_limit = max(1, int(self.ssh_action_timeout // 2))
|
||
task_command = self._cmd_task_command(
|
||
(
|
||
f'({inner_command}) 1>"{stdout_path}" 2>"{stderr_path}"'
|
||
f' & echo %ERRORLEVEL%>"{exit_path}"'
|
||
)
|
||
)
|
||
command_parts = [
|
||
f'if not exist "{artifact_dir}" mkdir "{artifact_dir}"',
|
||
f'del /f /q "{stdout_path}" "{stderr_path}" "{exit_path}" >nul 2>&1',
|
||
(
|
||
f'schtasks /create /tn {self._cmd_quote(task_name)} '
|
||
f'/tr {self._cmd_quote(task_command)} /sc ONCE /st 00:00 /rl HIGHEST /it /f >nul'
|
||
),
|
||
"if errorlevel 1 exit /b %ERRORLEVEL%",
|
||
f'schtasks /run /tn {self._cmd_quote(task_name)} >nul',
|
||
"if errorlevel 1 exit /b %ERRORLEVEL%",
|
||
(
|
||
f'for /l %i in (1,1,{loop_limit}) do @if exist "{exit_path}" '
|
||
f'(goto {task_name}_done) else @ping -n 3 127.0.0.1 >nul'
|
||
),
|
||
"exit /b 124",
|
||
f":{task_name}_done",
|
||
f'set /p TASK_RC=<{self._cmd_quote(exit_path)}',
|
||
f'if exist "{stdout_path}" type "{stdout_path}"',
|
||
f'if exist "{stderr_path}" type "{stderr_path}" 1>&2',
|
||
"exit /b %TASK_RC%",
|
||
]
|
||
command_script = " & ".join(command_parts)
|
||
return f"cmd.exe /d /q /s /c {self._cmd_quote(command_script)}"
|
||
|
||
def _scheduled_task_command_via_powershell(
|
||
self,
|
||
task_name: str,
|
||
inner_command: str,
|
||
*,
|
||
wait: bool,
|
||
keep_open: bool,
|
||
hidden: bool,
|
||
stop_existing: bool,
|
||
unlimited_execution_time: bool,
|
||
working_directory: Optional[str],
|
||
) -> RemoteCommand:
|
||
stdout_path = f"%TEMP%\\autool-dispatcher-remote-worker\\{task_name}.stdout.log"
|
||
stderr_path = f"%TEMP%\\autool-dispatcher-remote-worker\\{task_name}.stderr.log"
|
||
task_inner_command = inner_command
|
||
if wait:
|
||
task_inner_command = f'({inner_command}) 1>"{stdout_path}" 2>"{stderr_path}"'
|
||
cmd_switch = "/k" if keep_open else "/c"
|
||
if hidden:
|
||
hidden_script = " ".join(
|
||
[
|
||
f"$cmdCommand = {self._powershell_quote(task_inner_command)};",
|
||
f"& cmd.exe {cmd_switch} $cmdCommand;",
|
||
"exit $LASTEXITCODE;",
|
||
]
|
||
)
|
||
task_execute = "powershell.exe"
|
||
task_arguments = (
|
||
"-NoProfile -WindowStyle Hidden -ExecutionPolicy Bypass "
|
||
f"-EncodedCommand {base64.b64encode(hidden_script.encode('utf-16le')).decode('ascii')}"
|
||
)
|
||
else:
|
||
task_execute = "C:\\Windows\\System32\\cmd.exe"
|
||
task_arguments = self._cmd_switch_with_command(cmd_switch, task_inner_command)
|
||
script_parts = [
|
||
f"$taskName = {self._powershell_quote(task_name)};",
|
||
f"$taskExecute = {self._powershell_quote(task_execute)};",
|
||
f"$taskArguments = {self._powershell_quote(task_arguments)};",
|
||
f"$stopExisting = {'$true' if stop_existing else '$false'};",
|
||
f"$unlimitedExecutionTime = {'$true' if unlimited_execution_time else '$false'};",
|
||
f"$workingDirectory = {self._powershell_quote(working_directory or '')};",
|
||
"$needsCreate = $true;",
|
||
"try {",
|
||
" $task = Get-ScheduledTask -TaskName $taskName -ErrorAction Stop;",
|
||
" $action = @($task.Actions)[0];",
|
||
" $settings = $task.Settings;",
|
||
" $actionMatches = $action.Execute -eq $taskExecute -and $action.Arguments -eq $taskArguments;",
|
||
" $currentWorkingDirectory = '';",
|
||
" if ($null -ne $action.WorkingDirectory) { $currentWorkingDirectory = $action.WorkingDirectory };",
|
||
" $workingDirectoryMatches = $currentWorkingDirectory -eq $workingDirectory;",
|
||
" $executionTimeMatches = (-not $unlimitedExecutionTime) -or $settings.ExecutionTimeLimit -eq 'PT0S';",
|
||
" $multipleInstancesMatches = (-not $stopExisting) -or $settings.MultipleInstances -eq 'StopExisting';",
|
||
" if ($actionMatches -and $workingDirectoryMatches -and $executionTimeMatches -and $multipleInstancesMatches) { $needsCreate = $false };",
|
||
"} catch {};",
|
||
"if ($needsCreate) {",
|
||
" if ($workingDirectory) {",
|
||
" $action = New-ScheduledTaskAction -Execute $taskExecute -Argument $taskArguments -WorkingDirectory $workingDirectory;",
|
||
" } else {",
|
||
" $action = New-ScheduledTaskAction -Execute $taskExecute -Argument $taskArguments;",
|
||
" };",
|
||
" $trigger = New-ScheduledTaskTrigger -Once -At ([datetime]'2099-01-01T00:00:00');",
|
||
" $currentUser = [System.Security.Principal.WindowsIdentity]::GetCurrent().Name;",
|
||
" $principal = New-ScheduledTaskPrincipal -UserId $currentUser -LogonType Interactive -RunLevel Highest;",
|
||
" if ($unlimitedExecutionTime -and $stopExisting) {",
|
||
" $taskSettings = New-ScheduledTaskSettingsSet -ExecutionTimeLimit ([TimeSpan]::Zero) -MultipleInstances StopExisting;",
|
||
" } elseif ($unlimitedExecutionTime) {",
|
||
" $taskSettings = New-ScheduledTaskSettingsSet -ExecutionTimeLimit ([TimeSpan]::Zero);",
|
||
" } elseif ($stopExisting) {",
|
||
" $taskSettings = New-ScheduledTaskSettingsSet -MultipleInstances StopExisting;",
|
||
" } else {",
|
||
" $taskSettings = $null;",
|
||
" };",
|
||
" if ($null -ne $taskSettings) {",
|
||
" Register-ScheduledTask -TaskName $taskName -Action $action -Trigger $trigger -Principal $principal -Settings $taskSettings -Force -ErrorAction Stop | Out-Null;",
|
||
" } else {",
|
||
" Register-ScheduledTask -TaskName $taskName -Action $action -Trigger $trigger -Principal $principal -Force -ErrorAction Stop | Out-Null;",
|
||
" };",
|
||
"};",
|
||
]
|
||
if stop_existing and task_name == "RunBatch":
|
||
script_parts.append("& schtasks /end /tn $taskName 2>$null | Out-Null;")
|
||
if wait:
|
||
script_parts.extend(
|
||
[
|
||
(
|
||
"$artifactDir = "
|
||
"[System.IO.Path]::GetDirectoryName("
|
||
"[System.Environment]::ExpandEnvironmentVariables("
|
||
f"{self._powershell_quote(stdout_path)}));"
|
||
),
|
||
"New-Item -ItemType Directory -Path $artifactDir -Force | Out-Null;",
|
||
(
|
||
"$stdoutPath = "
|
||
f"[System.Environment]::ExpandEnvironmentVariables({self._powershell_quote(stdout_path)});"
|
||
),
|
||
(
|
||
"$stderrPath = "
|
||
f"[System.Environment]::ExpandEnvironmentVariables({self._powershell_quote(stderr_path)});"
|
||
),
|
||
"Remove-Item -LiteralPath $stdoutPath, $stderrPath -Force -ErrorAction SilentlyContinue;",
|
||
"$baseline = [datetime]::MinValue;",
|
||
"try { $baseline = (Get-ScheduledTaskInfo -TaskName $taskName -ErrorAction Stop).LastRunTime } catch {};",
|
||
]
|
||
)
|
||
script_parts.extend(
|
||
[
|
||
"& schtasks /run /tn $taskName | Out-Null;",
|
||
"if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE };",
|
||
]
|
||
)
|
||
if wait:
|
||
script_parts.extend(
|
||
[
|
||
"while ($true) {",
|
||
" $task = Get-ScheduledTask -TaskName $taskName -ErrorAction Stop;",
|
||
" $info = Get-ScheduledTaskInfo -TaskName $taskName -ErrorAction Stop;",
|
||
" $hasRun = $info.LastRunTime -gt $baseline;",
|
||
" if ($hasRun -and $task.State -ne 'Running') { break };",
|
||
" Start-Sleep -Milliseconds 500;",
|
||
"};",
|
||
"if (Test-Path -LiteralPath $stdoutPath) {",
|
||
" [Console]::Out.Write([System.IO.File]::ReadAllText($stdoutPath));",
|
||
"};",
|
||
"if (Test-Path -LiteralPath $stderrPath) {",
|
||
" [Console]::Error.Write([System.IO.File]::ReadAllText($stderrPath));",
|
||
"};",
|
||
"exit [int]$info.LastTaskResult;",
|
||
]
|
||
)
|
||
else:
|
||
script_parts.append(f"Write-Output {self._powershell_quote(task_name + ' started')};")
|
||
script = " ".join(script_parts)
|
||
return self._powershell_stdin_command(script)
|
||
|
||
def _scheduled_task_run_command(
|
||
self,
|
||
task_name: str,
|
||
inner_command: str,
|
||
*,
|
||
keep_open: bool = False,
|
||
hidden: bool = False,
|
||
stop_existing: bool = False,
|
||
unlimited_execution_time: bool = False,
|
||
working_directory: Optional[str] = None,
|
||
direct_command: bool = False,
|
||
) -> RemoteCommand:
|
||
return self._scheduled_task_command(
|
||
task_name,
|
||
inner_command,
|
||
wait=False,
|
||
keep_open=keep_open,
|
||
hidden=hidden,
|
||
stop_existing=stop_existing,
|
||
unlimited_execution_time=unlimited_execution_time,
|
||
working_directory=working_directory,
|
||
direct_command=direct_command,
|
||
)
|
||
|
||
def _scheduled_task_wait_command(
|
||
self,
|
||
task_name: str,
|
||
inner_command: str,
|
||
*,
|
||
hidden: bool = True,
|
||
direct_command: bool = False,
|
||
) -> RemoteCommand:
|
||
return self._scheduled_task_command(
|
||
task_name,
|
||
inner_command,
|
||
wait=True,
|
||
keep_open=False,
|
||
hidden=hidden,
|
||
stop_existing=False,
|
||
unlimited_execution_time=False,
|
||
working_directory=None,
|
||
direct_command=direct_command,
|
||
)
|
||
|
||
def _scheduled_task_stop_command(
|
||
self,
|
||
task_name: str,
|
||
success_message: str,
|
||
idle_message: str,
|
||
) -> RemoteCommand:
|
||
command_script = (
|
||
f'schtasks /query /tn {self._cmd_quote(task_name)} >nul 2>&1'
|
||
f' || (echo {idle_message} & exit /b 0)'
|
||
f' & schtasks /end /tn {self._cmd_quote(task_name)} >nul 2>&1'
|
||
f' || (echo {idle_message} & exit /b 0)'
|
||
f' & echo {success_message}'
|
||
)
|
||
return f"cmd.exe /d /q /s /c {self._cmd_quote(command_script)}"
|
||
|
||
def _scheduled_task_command(
|
||
self,
|
||
task_name: str,
|
||
inner_command: str,
|
||
*,
|
||
wait: bool,
|
||
keep_open: bool,
|
||
hidden: bool,
|
||
stop_existing: bool,
|
||
unlimited_execution_time: bool,
|
||
working_directory: Optional[str],
|
||
direct_command: bool = False,
|
||
) -> RemoteCommand:
|
||
if not unlimited_execution_time and not direct_command:
|
||
del hidden
|
||
return self._scheduled_task_command_via_cmd(
|
||
task_name,
|
||
inner_command,
|
||
wait=wait,
|
||
keep_open=keep_open,
|
||
stop_existing=stop_existing,
|
||
working_directory=working_directory,
|
||
)
|
||
return self._scheduled_task_command_via_powershell(
|
||
task_name,
|
||
inner_command,
|
||
wait=wait,
|
||
keep_open=keep_open,
|
||
hidden=hidden,
|
||
stop_existing=stop_existing,
|
||
unlimited_execution_time=unlimited_execution_time,
|
||
working_directory=working_directory,
|
||
)
|
||
|
||
def _pull_pcap_script_path(self, worker: Dict[str, Any]) -> str:
|
||
return f"{self._repo_dir(worker)}\\utils_android\\pull_pcapdroid_files.py"
|
||
|
||
def _venv_python(self, worker: Dict[str, Any]) -> str:
|
||
return f"{self._repo_dir(worker)}\\venv\\Scripts\\python.exe"
|
||
|
||
def _bootstrap_python(self, worker: Dict[str, Any]) -> str:
|
||
return str(worker["python_exe"]).replace("/", "\\")
|
||
|
||
def _run_ssh_command(
|
||
self,
|
||
worker_id: str,
|
||
worker: Dict[str, Any],
|
||
action: str,
|
||
command: RemoteCommand,
|
||
*,
|
||
timeout: Optional[int] = None,
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
|
||
) -> Dict[str, Any]:
|
||
ssh = paramiko.SSHClient()
|
||
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
||
target = f"{worker['ssh_host']}:{int(worker['ssh_port'])}"
|
||
if isinstance(command, dict):
|
||
command_text = str(command.get("command") or "")
|
||
stdin_text = str(command.get("stdin") or "")
|
||
else:
|
||
command_text = str(command)
|
||
stdin_text = ""
|
||
|
||
if SHARE_SMB_USER and SHARE_SMB_PASSWORD:
|
||
share_target = get_share_smb_target(worker["ssh_target"])
|
||
repo_url = get_repo_url(worker["ssh_target"])
|
||
share_markers = []
|
||
if share_target:
|
||
share_markers.extend([share_target.lower(), share_target.replace("\\\\", "//", 1).lower()])
|
||
if repo_url.startswith("//"):
|
||
repo_host = repo_url[2:].split("/", 1)[0].strip()
|
||
if repo_host:
|
||
share_markers.extend([f"//{repo_host}".lower(), f"\\\\{repo_host}".lower()])
|
||
if not share_target:
|
||
share_target = f"\\\\{repo_host}"
|
||
combined_command = "\n".join(part for part in (command_text, stdin_text) if part).lower()
|
||
requires_share_auth = action in {"clone", "pull", "recover_mumu"} or any(
|
||
marker in combined_command for marker in share_markers
|
||
)
|
||
if share_target and requires_share_auth:
|
||
net_use_cmd = f'net use {share_target} "{SHARE_SMB_PASSWORD}" /user:"{SHARE_SMB_USER}" >nul 2>&1'
|
||
command_text = f"{net_use_cmd} & {command_text}"
|
||
|
||
logger.info(
|
||
"[WorkerAction] start worker=%s action=%s target=%s timeout=%s",
|
||
worker_id,
|
||
action,
|
||
target,
|
||
timeout or self.ssh_action_timeout,
|
||
)
|
||
try:
|
||
self._emit_progress(progress_callback, worker_id, action, "ssh_connect", "建立 SSH 连接", status="running")
|
||
try:
|
||
ssh.connect(
|
||
hostname=worker["ssh_host"],
|
||
port=int(worker["ssh_port"]),
|
||
username=worker["ssh_user"],
|
||
password=worker.get("ssh_password") or self.default_password,
|
||
timeout=self.ssh_connect_timeout,
|
||
auth_timeout=self.ssh_connect_timeout,
|
||
banner_timeout=self.ssh_connect_timeout,
|
||
look_for_keys=True,
|
||
allow_agent=True,
|
||
)
|
||
except Exception as exc:
|
||
detail = traceback.format_exc().strip()
|
||
message = f"SSH 连接失败: {exc}"
|
||
self._emit_progress(progress_callback, worker_id, action, "ssh_connect", message, status="failed")
|
||
logger.error(
|
||
"[WorkerAction] ssh_connect failed worker=%s action=%s target=%s err=%s",
|
||
worker_id,
|
||
action,
|
||
target,
|
||
self._preview_text(detail),
|
||
)
|
||
return {
|
||
"ok": False,
|
||
"worker_id": worker_id,
|
||
"action": action,
|
||
"stage": "ssh_connect",
|
||
"message": message,
|
||
"stderr": detail,
|
||
"command": command_text,
|
||
}
|
||
self._emit_progress(progress_callback, worker_id, action, "ssh_connect", "SSH 已连接", status="success")
|
||
self._emit_progress(progress_callback, worker_id, action, "ssh_exec", "发送远端命令", status="running")
|
||
try:
|
||
stdin, stdout, stderr = ssh.exec_command(command_text, timeout=timeout or self.ssh_action_timeout)
|
||
if stdin_text:
|
||
stdin.write(stdin_text)
|
||
if not stdin_text.endswith("\n"):
|
||
stdin.write("\n")
|
||
stdin.flush()
|
||
stdin.channel.shutdown_write()
|
||
except Exception as exc:
|
||
detail = traceback.format_exc().strip()
|
||
message = f"远端命令启动失败: {exc}"
|
||
self._emit_progress(progress_callback, worker_id, action, "ssh_exec", message, status="failed")
|
||
logger.error(
|
||
"[WorkerAction] ssh_exec failed worker=%s action=%s target=%s err=%s",
|
||
worker_id,
|
||
action,
|
||
target,
|
||
self._preview_text(detail),
|
||
)
|
||
return {
|
||
"ok": False,
|
||
"worker_id": worker_id,
|
||
"action": action,
|
||
"stage": "ssh_exec",
|
||
"message": message,
|
||
"stderr": detail,
|
||
"command": command_text,
|
||
}
|
||
stdin.close()
|
||
self._emit_progress(progress_callback, worker_id, action, "ssh_exec", "命令已启动，等待远端反馈", status="success")
|
||
channel = stdout.channel
|
||
deadline = time.time() + float(timeout or self.ssh_action_timeout)
|
||
stdout_chunks = bytearray()
|
||
stderr_chunks = bytearray()
|
||
while True:
|
||
while channel.recv_ready():
|
||
stdout_chunks.extend(channel.recv(4096))
|
||
while channel.recv_stderr_ready():
|
||
stderr_chunks.extend(channel.recv_stderr(4096))
|
||
if channel.exit_status_ready():
|
||
break
|
||
if time.time() > deadline:
|
||
stdout_text = self._decode_remote_output(stdout_chunks).strip()
|
||
stderr_text = self._decode_remote_output(stderr_chunks).strip()
|
||
message = f"{action} 执行超时"
|
||
self._emit_progress(progress_callback, worker_id, action, "wait_remote_exit", message, status="failed")
|
||
logger.error(
|
||
"[WorkerAction] timeout worker=%s action=%s target=%s stdout=%s stderr=%s",
|
||
worker_id,
|
||
action,
|
||
target,
|
||
self._preview_text(stdout_text),
|
||
self._preview_text(stderr_text),
|
||
)
|
||
return {
|
||
"ok": False,
|
||
"worker_id": worker_id,
|
||
"action": action,
|
||
"stage": "wait_remote_exit",
|
||
"message": f"{action} timed out",
|
||
"stdout": stdout_text,
|
||
"stderr": stderr_text,
|
||
"command": command_text,
|
||
}
|
||
time.sleep(0.1)
|
||
|
||
while channel.recv_ready():
|
||
stdout_chunks.extend(channel.recv(4096))
|
||
while channel.recv_stderr_ready():
|
||
stderr_chunks.extend(channel.recv_stderr(4096))
|
||
exit_status = channel.recv_exit_status()
|
||
stdout_text = self._decode_remote_output(stdout_chunks).strip()
|
||
stderr_text = self._decode_remote_output(stderr_chunks).strip()
|
||
result = self._format_result(worker_id, worker, action, exit_status, stdout_text, stderr_text, command_text)
|
||
clean_stdout = result.get("stdout", "")
|
||
clean_stderr = result.get("stderr", "")
|
||
result_message = self._build_result_progress_message(result, exit_status, clean_stdout, clean_stderr)
|
||
self._emit_progress(
|
||
progress_callback,
|
||
worker_id,
|
||
action,
|
||
"remote_result",
|
||
result_message,
|
||
status="success" if result.get("ok") else "failed",
|
||
)
|
||
log_method = logger.info if result.get("ok") else logger.error
|
||
log_method(
|
||
"[WorkerAction] finish worker=%s action=%s target=%s ok=%s failed_step=%s stage=%s msg=%s stdout=%s stderr=%s",
|
||
worker_id,
|
||
action,
|
||
target,
|
||
result.get("ok"),
|
||
result.get("failed_step", ""),
|
||
result.get("stage", ""),
|
||
self._preview_text(result.get("message", "")),
|
||
self._preview_text(clean_stdout),
|
||
self._preview_text(clean_stderr),
|
||
)
|
||
return result
|
||
finally:
|
||
ssh.close()
|
||
|
||
@staticmethod
|
||
def _preview_text(value: Any, limit: int = 240) -> str:
|
||
text = str(value or "").replace("\r", " ").replace("\n", " | ").strip()
|
||
if len(text) <= limit:
|
||
return text
|
||
return text[:limit] + "..."
|
||
|
||
@staticmethod
|
||
def _decode_remote_output(payload: Any) -> str:
|
||
data = bytes(payload or b"")
|
||
if not data:
|
||
return ""
|
||
if data.startswith((codecs.BOM_UTF16_LE, codecs.BOM_UTF16_BE)):
|
||
for encoding in ("utf-16", "utf-16le", "utf-16be"):
|
||
try:
|
||
return data.decode(encoding)
|
||
except UnicodeDecodeError:
|
||
continue
|
||
if data.count(b"\x00") * 5 >= len(data):
|
||
for encoding in ("utf-16le", "utf-16be"):
|
||
try:
|
||
return data.decode(encoding)
|
||
except UnicodeDecodeError:
|
||
continue
|
||
for encoding in ("utf-8-sig", "utf-8", "gb18030", "cp936", "cp1252"):
|
||
try:
|
||
return data.decode(encoding)
|
||
except UnicodeDecodeError:
|
||
continue
|
||
return data.decode("utf-8", errors="replace")
|
||
|
||
@classmethod
|
||
def _build_result_progress_message(
|
||
cls,
|
||
result: Dict[str, Any],
|
||
exit_status: int,
|
||
stdout_text: str,
|
||
stderr_text: str,
|
||
) -> str:
|
||
parts = [f"exit={exit_status}"]
|
||
message = cls._preview_text(result.get("message", ""), limit=120)
|
||
stdout_preview = cls._preview_text(stdout_text, limit=120)
|
||
stderr_preview = cls._preview_text(stderr_text, limit=120)
|
||
if message:
|
||
parts.append(f"msg={message}")
|
||
if stdout_preview:
|
||
parts.append(f"stdout={stdout_preview}")
|
||
if stderr_preview:
|
||
parts.append(f"stderr={stderr_preview}")
|
||
if len(parts) == 1:
|
||
parts.append("msg=empty")
|
||
return " | ".join(parts)
|
||
|
||
@staticmethod
|
||
def _emit_progress(
|
||
progress_callback: Optional[Callable[[str, Dict[str, Any]], None]],
|
||
worker_id: str,
|
||
action: str,
|
||
step: str,
|
||
message: str,
|
||
**extra: Any,
|
||
) -> None:
|
||
if not progress_callback:
|
||
return
|
||
payload = {"progress": True, "action": action, "step": step, "message": message, **extra}
|
||
progress_callback(worker_id, payload)
|
||
|
||
def _format_result(
|
||
self,
|
||
worker_id: str,
|
||
worker: Dict[str, Any],
|
||
action: str,
|
||
exit_status: int,
|
||
stdout: str,
|
||
stderr: str,
|
||
command: str,
|
||
) -> Dict[str, Any]:
|
||
del worker
|
||
clean_stdout = str(stdout or "").strip()
|
||
clean_stderr = str(stderr or "").strip()
|
||
default_message = f"{action} completed" if exit_status == 0 else f"{action} failed (exit={exit_status})"
|
||
if exit_status == 0:
|
||
# 成功时优先显示 stdout
|
||
message = clean_stdout or clean_stderr or default_message
|
||
else:
|
||
# 失败时优先显示 stderr（真正的报错），其次是 stdout（可能包含合并后的报错），最后是默认消息
|
||
message = clean_stderr or clean_stdout or default_message
|
||
return {
|
||
"ok": exit_status == 0,
|
||
"worker_id": worker_id,
|
||
"action": action,
|
||
"message": message,
|
||
"command": command,
|
||
"stdout": clean_stdout,
|
||
"stderr": clean_stderr,
|
||
"raw_stdout": stdout,
|
||
"raw_stderr": stderr,
|
||
}
|