aidev/juc500
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add Wormhole auto-kill so Interface 5 can be claimed

Browse Source 
Cable autorun holds IF5; kill on launch/connect and via GUI button/CLI.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
joyharam
2026-07-15 10:58:35 +09:00
parent 2035c0b4fe
commit d9c0b8078c
9 changed files with 398 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
juc500_xfer/cli.py
+21
View File
@@ -11,6 +11,7 @@ from pathlib import Path
from . import __version__
from .protocol import Session
from .usb_link import list_devices, Juc500Link, doctor
from .wormhole_kill import kill_wormhole_processes, wormhole_status
def _progress(prefix: str):
@@ -61,12 +62,29 @@ def cmd_doctor(_args: argparse.Namespace) -> int:
print("Windows PnP VID_0711:")
for row in rep.pnp_vid0711:
print(f" - {row}")
wh = wormhole_status()
print(f"Wormhole 프로세스: {wh['count']} — {wh['message']}")
for p in wh.get("processes") or []:
print(f" - {p.get('name')} pid={p.get('pid')}")
print("안내:")
for h in rep.hints:
print(f" * {h}")
return 0 if rep.ok else 1
def cmd_kill_wormhole(_args: argparse.Namespace) -> int:
result = kill_wormhole_processes(force=True)
print(result.get("message") or "")
if result.get("killed"):
for p in result["killed"]:
print(f" 종료: {p.get('name')} pid={p.get('pid')}")
if result.get("remaining"):
for p in result["remaining"]:
print(f" 잔여: {p.get('name')} pid={p.get('pid')}")
return 1
return 0
def cmd_probe(args: argparse.Namespace) -> int:
with Juc500Link(serial=args.serial) as link:
print(f"열림: serial={link.serial!r}")
@@ -143,6 +161,9 @@ def build_parser() -> argparse.ArgumentParser:
s = sub.add_parser("doctor", help="USB/libusb/WinUSB 진단")
s.set_defaults(func=cmd_doctor)
s = sub.add_parser("kill-wormhole", help="공식 Wormhole/GoBridge 프로세스 종료")
s.set_defaults(func=cmd_kill_wormhole)
s = sub.add_parser("probe", help="USB 인터페이스 점유 및 송수신 프로브")
s.add_argument("--serial", default=None)
s.set_defaults(func=cmd_probe)
juc500_xfer/gui_server.py
+27
View File
@@ -17,6 +17,7 @@ from typing import Any, Optional
from .protocol import Session
from .usb_link import Juc500Link, list_devices, doctor
from .wormhole_kill import kill_wormhole_processes, wormhole_status
DEFAULT_GUI_PORT = 8765
@@ -68,6 +69,7 @@ class GuiState:
self.last_progress = {"done": 0, "total": 0, "label": "", "error": None, "busy": False}
self.connecting = False
self.log_lines: list[str] = []
self.auto_kill_wormhole = True
def log(self, msg: str) -> None:
self.log_lines.append(msg)
@@ -141,6 +143,13 @@ def connect_usb(timeout: float = 45.0) -> dict[str, Any]:
STATE.auto_status = "USB 링크 여는 중…"
try:
if STATE.auto_kill_wormhole:
STATE.auto_status = "공식 Wormhole 종료 중…"
STATE.log("공식 Wormhole 프로세스 종료 시도")
result = kill_wormhole_processes(force=True)
STATE.log(result.get("message") or "")
if result.get("killed"):
time.sleep(0.4)
devices = list_devices()
if not devices:
raise RuntimeError("JUC500(Smart Data Link) 장치가 없습니다. 케이블을 연결하세요.")
@@ -442,6 +451,7 @@ class Handler(BaseHTTPRequestHandler):
"pid": f"{cable.pid:04x}" if cable else None,
},
"usb_error": list_err,
"wormhole": wormhole_status(),
"inputshare": {
"supported": False,
"enabled": False,
@@ -456,6 +466,10 @@ class Handler(BaseHTTPRequestHandler):
self._json(200, doctor().to_dict())
return
if path == "/api/wormhole":
self._json(200, wormhole_status())
return
if path == "/api/settings/ports":
# Reuse settings modal: expose local_root / serial as "ports"-like rows
self._json(
@@ -548,6 +562,12 @@ class Handler(BaseHTTPRequestHandler):
self._json(200, {"ok": True})
return
if path in ("/api/wormhole/kill", "/api/wormhole/stop"):
result = kill_wormhole_processes(force=True)
STATE.log(result.get("message") or "wormhole kill")
self._json(200, result)
return
if path == "/api/auto/start":
self._json(200, {"ok": True})
return
@@ -644,6 +664,13 @@ def run_gui(open_browser: bool = True, port: int | None = None, host: str = "127
preferred = port or STATE.gui_port or DEFAULT_GUI_PORT
listen_port = _pick_port(host, preferred)
STATE.gui_port = listen_port
# Cable autorun Wormhole often starts with the volume — clear it on launch.
try:
result = kill_wormhole_processes(force=True)
if result.get("killed"):
print(result.get("message"))
except Exception as exc: # noqa: BLE001
print(f"Wormhole 자동 종료 건너뜀: {exc}")
server = ThreadingHTTPServer((host, listen_port), Handler)
url = f"http://{host}:{listen_port}/"
print(f"JUC500 GUI: {url}")
juc500_xfer/static/app.js
+47 -2
View File
@@ -450,6 +450,47 @@ function updateBadge(connected, text) {
badge.className = connected ? "auto-badge ok" : "auto-badge wait";
}
function updateWormholeUI(info) {
const badge = $("wormholeBadge");
const btn = $("killWormholeBtn");
if (!badge) return;
if (!info) {
badge.textContent = "Wormhole —";
badge.className = "auto-badge wait";
return;
}
if (info.running) {
badge.textContent = `Wormhole ${info.count}개 실행중`;
badge.className = "auto-badge wait";
badge.title = (info.processes || [])
.map((p) => `${p.name} (${p.pid})`)
.join("\n") || info.message || "";
if (btn) btn.disabled = false;
} else {
badge.textContent = "Wormhole 없음";
badge.className = "auto-badge ok";
badge.title = info.message || "공식 Wormhole 프로세스 없음";
if (btn) btn.disabled = false;
}
}
async function killWormhole() {
setStatus("공식 Wormhole 종료 중…");
try {
const data = await api("/api/wormhole/kill", { method: "POST" });
updateWormholeUI({
running: (data.remaining || []).length > 0,
count: (data.remaining || []).length,
processes: data.remaining || [],
message: data.message,
});
setStatus(data.message || "Wormhole 종료 완료");
await refreshStatus();
} catch (e) {
setStatus(e.message || String(e), true);
}
}
function sortEntries(entries, sort) {
const dir = sort.dir;
const key = sort.key;
@@ -503,6 +544,7 @@ async function refreshStatus() {
if (s.local_root) $("localRootHint").textContent = s.local_root;
updateInputShareUI(s.inputshare);
updateWormholeUI(s.wormhole);
if (state.connected) {
updateBadge(true, "연결됨");
@@ -536,8 +578,8 @@ async function connectOneClick() {
state.connecting = true;
$("connectBtn").disabled = true;
updateBadge(false, "연결 중…");
setStatus("상대 JUC500 프로그램 대기 중…");
$("peerInfo").textContent = "USB 핸드셰이크 중… 상대에서도 GUI를 실행하세요";
setStatus("공식 Wormhole 종료 후 상대 프로그램 대기 중…");
$("peerInfo").textContent = "Wormhole 정리 · USB 핸드셰이크 중…";
try {
const data = await api("/api/connect/auto", { method: "POST" });
@@ -1099,6 +1141,9 @@ function bindSort(prefix, getSort, setSort, rerender) {
function bindUi() {
$("connectBtn").onclick = () => connectOneClick();
$("disconnectBtn").onclick = () => disconnect().catch((e) => setStatus(e.message, true));
if ($("killWormholeBtn")) {
$("killWormholeBtn").onclick = () => killWormhole();
}
$("upBtn").onclick = () => goUp().catch((e) => setStatus(e.message, true));
$("refreshBtn").onclick = () => {
if (!state.connected) {
juc500_xfer/static/index.html
+2
View File
@@ -15,6 +15,8 @@
</div>
<div class="conn">
<span id="autoBadge" class="auto-badge wait">미연결</span>
<span id="wormholeBadge" class="auto-badge wait" title="공식 Wormhole 프로세스">Wormhole —</span>
<button id="killWormholeBtn" type="button" class="ghost" title="케이블 자동실행 Wormhole 종료">Wormhole 종료</button>
<span id="inputBadge" class="auto-badge wait" title="마우스·키보드 공유">입력공유 미지원</span>
<button id="inputShareBtn" type="button" class="ghost" disabled>입력공유</button>
<div class="control-switch" title="입력공유 미지원">
juc500_xfer/usb_link.py
+19 -1
View File
@@ -203,7 +203,7 @@ def _claim_failure_message(ifnum: int, exc: BaseException) -> str:
f"{base}\n\n"
"Windows가 Interface 5를 다른 드라이버(공식 Wormhole 등)로 붙잡고 있습니다.\n"
"아래를 순서대로 하세요:\n"
"1) 작업 관리자에서 Wormhole / SmartDataLink / j5create 관련 프로그램 모두 종료\n"
"1) GUI의 'Wormhole 종료' 버튼 또는 `python -m juc500_xfer kill-wormhole`\n"
"2) Zadig(https://zadig.akeo.ie/) 실행 (관리자)\n"
"3) Options → List All Devices\n"
"4) 목록에서 'Smart Data Link (Interface 5)' 또는 InstanceId에 MI_05 가 있는 항목 선택\n"
@@ -464,6 +464,24 @@ class Juc500Link:
try:
usb.util.claim_interface(self._dev, ifnum)
except usb.core.USBError as exc:
# Official Wormhole often holds IF5 — kill and retry once.
if _is_access_denied(exc) and ifnum == IF_VENDOR:
try:
from .wormhole_kill import kill_wormhole_processes
kill_wormhole_processes(force=True)
time.sleep(0.6)
if not _IS_WIN:
try:
if self._dev.is_kernel_driver_active(ifnum):
self._dev.detach_kernel_driver(ifnum)
except (NotImplementedError, usb.core.USBError):
pass
usb.util.claim_interface(self._dev, ifnum)
self._claimed.append(ifnum)
return
except Exception:
pass
if required:
raise RuntimeError(_claim_failure_message(ifnum, exc)) from exc
return
juc500_xfer/wormhole_kill.py
+274
View File
@@ -0,0 +1,274 @@
"""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 프로세스 없음"
),
}