juc500/juc500_xfer/wormhole_kill.py

"""Detect and terminate official j5create / OTi Wormhole processes.

The JUC500 virtual CD (WORMHOLE) auto-runs Wormhole.exe / WormholeShell which
exclusively claims USB Interface 5. Our stack needs that interface free.
"""

from __future__ import annotations

import os
import signal
import subprocess
import sys
import time
from dataclasses import asdict, dataclass
from typing import Any

_IS_WIN = sys.platform == "win32"

# Exact / substring match against process image name (case-insensitive)
PROCESS_NAMES = (
    "wormhole.exe",
    "wormhold.exe",  # typo safety
    "wormholeld.exe",
    "wormholeshell",
    "wormhole",
    "u3wormhole.exe",
    "u3wormholeshell",
    "u3wormhole",
    "gobridge.exe",
    "gobridgedemon.exe",
    "gobridgedemon",
    "gobridgedemonu3.exe",
    "gobridgedemonu3",
    "downloadupdatetool.exe",
    "downloadupdatetool",
    "smartdatalink.exe",
    "smartlink.exe",
    "oti_kmloader.exe",
    "otigoBridge.exe",
    "gosuite.exe",
)

# Broader name hints (substring in process name)
NAME_HINTS = (
    "wormhole",
    "gobridge",
    "kaijet",
    "j5create",
    "smartdatalink",
    "smart data link",
    "oti_",
    "otisync",
)

# Do not kill ourselves
SELF_HINTS = ("juc500", "juc500_xfer", "juc500-portable")


@dataclass
class ProcInfo:
    pid: int
    name: str
    path: str = ""

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)


def _is_self(name: str, path: str = "") -> bool:
    blob = f"{name} {path}".lower()
    return any(s in blob for s in SELF_HINTS)


def _matches_wormhole(name: str, path: str = "") -> bool:
    n = (name or "").lower().strip()
    p = (path or "").lower()
    if not n:
        return False
    if _is_self(n, p):
        return False
    base = n if n.endswith(".exe") else n
    # exact-ish
    for cand in PROCESS_NAMES:
        c = cand.lower()
        if base == c or base == c.replace(".exe", "") or n == c:
            return True
    # path / hint
    blob = f"{n} {p}"
    return any(h in blob for h in NAME_HINTS)


def list_wormhole_processes() -> list[ProcInfo]:
    if _IS_WIN:
        return _list_windows()
    return _list_posix()


def _list_windows() -> list[ProcInfo]:
    # CSV: "name","pid","session","mem"
    try:
        proc = subprocess.run(
            ["tasklist", "/FO", "CSV", "/NH"],
            capture_output=True,
            text=True,
            timeout=20,
            check=False,
            creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
        )
    except Exception:
        return []
    out: list[ProcInfo] = []
    for line in (proc.stdout or "").splitlines():
        line = line.strip()
        if not line:
            continue
        # "Wormhole.exe","1234","Console","1","12,345 K"
        parts = [x.strip().strip('"') for x in line.split('","')]
        if len(parts) < 2:
            # fallback split
            parts = [x.strip().strip('"') for x in line.split(",")]
        if len(parts) < 2:
            continue
        name, pid_s = parts[0], parts[1]
        try:
            pid = int(pid_s)
        except ValueError:
            continue
        if _matches_wormhole(name):
            out.append(ProcInfo(pid=pid, name=name))
    # Enrich path via WMIC / PowerShell (optional)
    return _dedupe(out)


def _list_posix() -> list[ProcInfo]:
    try:
        proc = subprocess.run(
            ["ps", "-ax", "-o", "pid=,comm=,command="],
            capture_output=True,
            text=True,
            timeout=15,
            check=False,
        )
    except Exception:
        return []
    out: list[ProcInfo] = []
    my_pid = os.getpid()
    for line in (proc.stdout or "").splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            pid_s, rest = line.split(None, 1)
            pid = int(pid_s)
        except ValueError:
            continue
        if pid == my_pid:
            continue
        # comm is first token of rest sometimes; keep full command
        tokens = rest.split(None, 1)
        comm = tokens[0]
        cmd = tokens[1] if len(tokens) > 1 else ""
        if _matches_wormhole(comm, cmd) or _matches_wormhole(os.path.basename(comm), cmd):
            out.append(ProcInfo(pid=pid, name=os.path.basename(comm) or comm, path=cmd[:240]))
    return _dedupe(out)


def _dedupe(items: list[ProcInfo]) -> list[ProcInfo]:
    seen: set[int] = set()
    out: list[ProcInfo] = []
    for p in items:
        if p.pid in seen:
            continue
        seen.add(p.pid)
        out.append(p)
    return out


def kill_wormhole_processes(*, force: bool = True) -> dict[str, Any]:
    """Terminate official Wormhole processes. Returns status dict."""
    before = list_wormhole_processes()
    killed: list[dict[str, Any]] = []
    errors: list[str] = []

    for info in before:
        try:
            _kill_one(info.pid, force=force)
            killed.append(info.to_dict())
        except Exception as exc:  # noqa: BLE001
            errors.append(f"{info.name}({info.pid}): {exc}")

    # brief settle so USB handles are released
    if killed:
        time.sleep(0.8)

    after = list_wormhole_processes()
    # second pass for stubborn ones
    if after and force:
        for info in after:
            try:
                _kill_one(info.pid, force=True)
                if info.to_dict() not in killed:
                    killed.append(info.to_dict())
            except Exception as exc:  # noqa: BLE001
                errors.append(f"retry {info.name}({info.pid}): {exc}")
        time.sleep(0.5)
        after = list_wormhole_processes()

    return {
        "ok": len(after) == 0,
        "found": [p.to_dict() for p in before],
        "killed": killed,
        "remaining": [p.to_dict() for p in after],
        "errors": errors,
        "message": _summary(before, killed, after),
    }


def _summary(before: list[ProcInfo], killed: list[dict], after: list[ProcInfo]) -> str:
    if not before:
        return "실행 중인 Wormhole 관련 프로세스가 없습니다."
    if not after:
        names = ", ".join(sorted({k.get("name", "?") for k in killed})) or "프로세스"
        return f"Wormhole 종료 완료: {names}"
    rem = ", ".join(f"{p.name}({p.pid})" for p in after)
    return f"일부 종료되지 않음: {rem} (관리자 권한으로 다시 시도하거나 작업 관리자에서 종료)"


def _kill_one(pid: int, *, force: bool) -> None:
    if _IS_WIN:
        args = ["taskkill", "/PID", str(pid)]
        if force:
            args.append("/F")
        proc = subprocess.run(
            args,
            capture_output=True,
            text=True,
            timeout=15,
            check=False,
            creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
        )
        if proc.returncode != 0:
            err = (proc.stderr or proc.stdout or "taskkill failed").strip()
            # 128 = not found — treat as OK
            if proc.returncode == 128 or "not found" in err.lower():
                return
            raise RuntimeError(err)
        return

    sig = signal.SIGKILL if force else signal.SIGTERM
    try:
        os.kill(pid, sig)
    except ProcessLookupError:
        return
    if not force:
        time.sleep(0.4)
        try:
            os.kill(pid, 0)
        except ProcessLookupError:
            return
        os.kill(pid, signal.SIGKILL)


def wormhole_status() -> dict[str, Any]:
    procs = list_wormhole_processes()
    return {
        "running": len(procs) > 0,
        "count": len(procs),
        "processes": [p.to_dict() for p in procs],
        "message": (
            f"Wormhole 관련 프로세스 {len(procs)}개 실행 중 (Interface 5 점유 → 종료 권장)"
            if procs
            else "Wormhole 프로세스 없음"
        ),
    }