Files
server-manager/gui/tabs/terminal_tab.py
chrome-storm-c442 bf39fd7b67 v1.2.0 + v1.3.0: Localization, About dialog, TOTP/2FA, stability improvements
v1.2.0:
- GUI localization (EN/RU/ZH) with language switcher and persistent selection
- About dialog (ⓘ) with app info, features, quick start guide
- core/i18n.py — internationalization module with t() function
- All GUI components translated via t() keys

v1.3.0:
- TOTP/2FA tab — Google Authenticator compatible codes with live 30s countdown,
  one-click copy, per-server secret management
- core/totp.py — TOTP module (pyotp, RFC 6238)
- core/logger.py — rotating file logger (5MB, 3 backups)
- Stronger Fernet encryption key with automatic migration from old key
- Thread-safe server store with locks, atomic writes, auto-restore on corruption
- Parallel status checks via ThreadPoolExecutor (up to 10 concurrent)
- SSH client: explicit channel cleanup, Unix key permissions
- Server dialog: port validation (1-65535), TOTP secret field
- Language change preserves active tab and server selection
- pyotp dependency added

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-23 11:07:51 -05:00

106 lines
3.9 KiB
Python

"""
Terminal tab — command input + output display.
"""
import threading
import customtkinter as ctk
from core.ssh_client import SSHClientWrapper
from core.i18n import t
class TerminalTab(ctk.CTkFrame):
def __init__(self, master, store):
super().__init__(master, fg_color="transparent")
self.store = store
self._current_alias: str | None = None
# Output
self.output = ctk.CTkTextbox(self, font=ctk.CTkFont(family="Consolas", size=12), state="disabled")
self.output.pack(fill="both", expand=True, padx=10, pady=(10, 5))
# Input row
input_frame = ctk.CTkFrame(self, fg_color="transparent")
input_frame.pack(fill="x", padx=10, pady=(0, 10))
self.sudo_var = ctk.BooleanVar(value=True)
self.sudo_check = ctk.CTkCheckBox(input_frame, text=t("sudo"), variable=self.sudo_var, width=60)
self.sudo_check.pack(side="left", padx=(0, 5))
self.cmd_entry = ctk.CTkEntry(input_frame, placeholder_text=t("enter_command"))
self.cmd_entry.pack(side="left", fill="x", expand=True, padx=(0, 5))
self.cmd_entry.bind("<Return>", lambda e: self._run_command())
self.run_btn = ctk.CTkButton(input_frame, text=t("run"), width=70, command=self._run_command)
self.run_btn.pack(side="left", padx=(0, 5))
self.clear_btn = ctk.CTkButton(input_frame, text=t("clear"), width=60, fg_color="#6b7280", command=self._clear)
self.clear_btn.pack(side="right")
def set_server(self, alias: str | None):
self._current_alias = alias
if alias:
server = self.store.get_server(alias)
user = server.get("user", "root") if server else "root"
self.sudo_var.set(user != "root")
def _append_output(self, text: str, color: str = "white"):
self.output.configure(state="normal")
self.output.insert("end", text)
self.output.configure(state="disabled")
self.output.see("end")
def _run_command(self):
if not self._current_alias:
self._append_output(t("no_server_selected") + "\n")
return
command = self.cmd_entry.get().strip()
if not command:
return
server = self.store.get_server(self._current_alias)
if not server:
self._append_output(t("server_not_found").format(alias=self._current_alias) + "\n")
return
self.cmd_entry.delete(0, "end")
use_sudo = self.sudo_var.get()
prefix = f"[{self._current_alias}]$ "
if use_sudo and server.get("user", "root") != "root":
prefix = f"[{self._current_alias}]# "
self._append_output(f"{prefix}{command}\n")
self.run_btn.configure(state="disabled", text="...")
def _exec():
try:
key_path = self.store.get_ssh_key_path()
wrapper = SSHClientWrapper(server, key_path)
out, err, code = wrapper.exec_command(command, use_sudo=use_sudo)
def _show():
if out:
self._append_output(out)
if not out.endswith("\n"):
self._append_output("\n")
if err:
self._append_output(f"STDERR: {err}\n")
if code != 0:
self._append_output(f"[exit code: {code}]\n")
self._append_output("\n")
self.run_btn.configure(state="normal", text=t("run"))
self.after(0, _show)
except Exception as e:
def _err():
self._append_output(f"[ERROR] {e}\n\n")
self.run_btn.configure(state="normal", text=t("run"))
self.after(0, _err)
threading.Thread(target=_exec, daemon=True).start()
def _clear(self):
self.output.configure(state="normal")
self.output.delete("1.0", "end")
self.output.configure(state="disabled")