4959004a3f
- Add core/icons.py — centralized icon text helper with emoji/symbol support - Add Windows SSH command sanitization in ssh.py (Linux→Windows auto-translation) - Improve embedded RDP: launch tab connect/disconnect, fullscreen toggle - Refactor sidebar: cleaner server type badges - Update server_dialog: adaptive fields per server type - Add setup_openssh.bat tool - Update skill-ssh.md and CLAUDE.md docs for Windows SSH support - Cleanup old releases, add v1.8.48-v1.8.52 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
244 lines
8.8 KiB
Python
244 lines
8.8 KiB
Python
"""
|
|
PowerShell/CMD tab — request-response terminal for WinRM servers.
|
|
|
|
No pyte needed: WinRM is not an interactive PTY, just command → output.
|
|
"""
|
|
|
|
import threading
|
|
import customtkinter as ctk
|
|
from core.winrm_client import WinRMClient
|
|
from core.i18n import t
|
|
from core.icons import icon_text
|
|
|
|
|
|
class PowershellTab(ctk.CTkFrame):
|
|
"""Simplified terminal for WinRM command execution (PS or CMD)."""
|
|
|
|
_MAX_HISTORY = 200
|
|
|
|
def __init__(self, master, store):
|
|
super().__init__(master, fg_color="transparent")
|
|
self.store = store
|
|
self._current_alias: str | None = None
|
|
self._client: WinRMClient | None = None
|
|
self._mode: str = "ps" # "ps" or "cmd"
|
|
self._history: list[str] = []
|
|
self._history_index: int = -1
|
|
self._running = False
|
|
|
|
self._build_ui()
|
|
|
|
# ── UI construction ──────────────────────────────────────────────
|
|
|
|
def _build_ui(self):
|
|
# Top bar: mode toggle
|
|
top = ctk.CTkFrame(self, fg_color="transparent")
|
|
top.pack(fill="x", padx=8, pady=(8, 0))
|
|
|
|
self._mode_var = ctk.StringVar(value="ps")
|
|
self._ps_radio = ctk.CTkRadioButton(
|
|
top, text=t("ps_mode_ps"), variable=self._mode_var,
|
|
value="ps", command=self._on_mode_changed,
|
|
)
|
|
self._ps_radio.pack(side="left", padx=(0, 12))
|
|
|
|
self._cmd_radio = ctk.CTkRadioButton(
|
|
top, text=t("ps_mode_cmd"), variable=self._mode_var,
|
|
value="cmd", command=self._on_mode_changed,
|
|
)
|
|
self._cmd_radio.pack(side="left")
|
|
|
|
# Output console
|
|
self._output = ctk.CTkTextbox(
|
|
self, font=ctk.CTkFont(family="Consolas", size=13),
|
|
state="disabled", wrap="word",
|
|
)
|
|
self._output.pack(fill="both", expand=True, padx=8, pady=8)
|
|
|
|
# Input row: entry + execute button
|
|
input_row = ctk.CTkFrame(self, fg_color="transparent")
|
|
input_row.pack(fill="x", padx=8, pady=(0, 4))
|
|
|
|
self._entry = ctk.CTkEntry(
|
|
input_row, placeholder_text="PS> ...",
|
|
font=ctk.CTkFont(family="Consolas", size=13),
|
|
)
|
|
self._entry.pack(side="left", fill="x", expand=True, padx=(0, 6))
|
|
self._entry.bind("<Return>", lambda e: self._execute())
|
|
self._entry.bind("<Up>", lambda e: self._history_navigate(-1))
|
|
self._entry.bind("<Down>", lambda e: self._history_navigate(1))
|
|
|
|
self._exec_btn = ctk.CTkButton(
|
|
input_row, text=icon_text("execute", t("ps_execute")), width=100,
|
|
command=self._execute,
|
|
)
|
|
self._exec_btn.pack(side="right")
|
|
|
|
# Status bar
|
|
self._status = ctk.CTkLabel(
|
|
self, text="", anchor="w",
|
|
font=ctk.CTkFont(size=11), text_color="#888888",
|
|
)
|
|
self._status.pack(fill="x", padx=10, pady=(0, 6))
|
|
|
|
# ── Public API ───────────────────────────────────────────────────
|
|
|
|
def set_server(self, alias: str | None):
|
|
"""Switch to a different server (or None to disconnect)."""
|
|
if alias == self._current_alias:
|
|
return
|
|
|
|
self._disconnect()
|
|
self._current_alias = alias
|
|
self._history.clear()
|
|
self._history_index = -1
|
|
|
|
if alias is None:
|
|
self._set_status(t("ps_disconnected"), "#888888")
|
|
return
|
|
|
|
self._connect(alias)
|
|
|
|
# ── Connection ───────────────────────────────────────────────────
|
|
|
|
def _connect(self, alias: str):
|
|
server = self.store.get_server(alias)
|
|
if not server:
|
|
self._set_status(t("server_not_found").format(alias=alias), "#ff4444")
|
|
return
|
|
|
|
self._set_status(t("ps_connecting").format(alias=alias), "#ccaa00")
|
|
|
|
def _do():
|
|
try:
|
|
client = WinRMClient(server)
|
|
client.connect()
|
|
self._client = client
|
|
self.after(0, lambda: self._set_status(
|
|
t("ps_connected").format(alias=alias), "#44cc44",
|
|
))
|
|
self.after(0, lambda: self._entry.focus())
|
|
except Exception as exc:
|
|
self.after(0, lambda: self._set_status(
|
|
t("ps_connect_failed").format(error=str(exc)), "#ff4444",
|
|
))
|
|
|
|
threading.Thread(target=_do, daemon=True).start()
|
|
|
|
def _disconnect(self):
|
|
if self._client:
|
|
try:
|
|
self._client.close()
|
|
except Exception:
|
|
pass
|
|
self._client = None
|
|
|
|
# ── Command execution ────────────────────────────────────────────
|
|
|
|
def _execute(self):
|
|
cmd = self._entry.get().strip()
|
|
if not cmd:
|
|
return
|
|
if not self._client:
|
|
self._set_status(t("ps_not_connected"), "#ff4444")
|
|
return
|
|
if self._running:
|
|
return
|
|
|
|
# Save to history
|
|
if not self._history or self._history[-1] != cmd:
|
|
self._history.append(cmd)
|
|
if len(self._history) > self._MAX_HISTORY:
|
|
self._history.pop(0)
|
|
self._history_index = -1
|
|
|
|
# Show command in output
|
|
prompt = "PS>" if self._mode == "ps" else "CMD>"
|
|
self._append_output(f"\n{prompt} {cmd}\n")
|
|
|
|
self._entry.delete(0, "end")
|
|
self._running = True
|
|
self._exec_btn.configure(state="disabled")
|
|
self._set_status(t("ps_running"), "#ccaa00")
|
|
|
|
mode = self._mode
|
|
client = self._client
|
|
|
|
def _run():
|
|
try:
|
|
if mode == "ps":
|
|
result = client.exec_ps(cmd)
|
|
else:
|
|
result = client.exec_cmd(cmd)
|
|
|
|
stdout = result.get("stdout", "")
|
|
stderr = result.get("stderr", "")
|
|
rc = result.get("return_code", None)
|
|
|
|
def _show():
|
|
if stdout:
|
|
self._append_output(stdout)
|
|
if stderr:
|
|
self._append_output(f"[STDERR] {stderr}")
|
|
if rc is not None and rc != 0:
|
|
self._append_output(f"[Exit code: {rc}]")
|
|
self._set_status(t("ps_done"), "#44cc44")
|
|
|
|
self.after(0, _show)
|
|
except Exception as exc:
|
|
self.after(0, lambda: self._append_output(
|
|
f"\n[ERROR] {exc}\n"
|
|
))
|
|
self.after(0, lambda: self._set_status(
|
|
t("ps_exec_error"), "#ff4444",
|
|
))
|
|
finally:
|
|
self._running = False
|
|
self.after(0, lambda: self._exec_btn.configure(state="normal"))
|
|
|
|
threading.Thread(target=_run, daemon=True).start()
|
|
|
|
# ── History navigation ───────────────────────────────────────────
|
|
|
|
def _history_navigate(self, direction: int):
|
|
"""Navigate command history. direction: -1 = older, +1 = newer."""
|
|
if not self._history:
|
|
self._set_status(t("ps_history_empty"), "#888888")
|
|
return
|
|
|
|
if self._history_index == -1:
|
|
if direction == -1:
|
|
self._history_index = len(self._history) - 1
|
|
else:
|
|
return
|
|
else:
|
|
self._history_index += direction
|
|
|
|
if self._history_index < 0:
|
|
self._history_index = 0
|
|
elif self._history_index >= len(self._history):
|
|
self._history_index = -1
|
|
self._entry.delete(0, "end")
|
|
return
|
|
|
|
self._entry.delete(0, "end")
|
|
self._entry.insert(0, self._history[self._history_index])
|
|
|
|
# ── Mode toggle ──────────────────────────────────────────────────
|
|
|
|
def _on_mode_changed(self):
|
|
self._mode = self._mode_var.get()
|
|
placeholder = "PS> ..." if self._mode == "ps" else "CMD> ..."
|
|
self._entry.configure(placeholder_text=placeholder)
|
|
|
|
# ── Helpers ──────────────────────────────────────────────────────
|
|
|
|
def _append_output(self, text: str):
|
|
self._output.configure(state="normal")
|
|
self._output.insert("end", text)
|
|
self._output.see("end")
|
|
self._output.configure(state="disabled")
|
|
|
|
def _set_status(self, text: str, color: str = "#888888"):
|
|
self._status.configure(text=text, text_color=color)
|