diff --git a/src/zashterminal/filemanager/tftp_server.py b/src/zashterminal/filemanager/tftp_server.py new file mode 100644 index 0000000..e6b5742 --- /dev/null +++ b/src/zashterminal/filemanager/tftp_server.py @@ -0,0 +1,327 @@ +import os +import socket +import struct +import threading +from pathlib import Path +from typing import Callable, Optional, Tuple + +from ..utils.logger import get_logger + + +class TftpServerError(Exception): + """Base exception for TFTP server failures.""" + + +class TftpBindError(TftpServerError): + """Raised when the TFTP server cannot bind its UDP socket.""" + + +class TftpFileError(TftpServerError): + """Raised when the TFTP server cannot access its configured directories.""" + + +class TftpNetworkError(TftpServerError): + """Raised when the TFTP server hits an unexpected network failure.""" + + +class TftpServer: + """Small RFC 1350-compatible TFTP server for RRQ/WRQ octet transfers.""" + + OP_RRQ = 1 + OP_WRQ = 2 + OP_DATA = 3 + OP_ACK = 4 + OP_ERROR = 5 + + ERR_UNDEFINED = 0 + ERR_NOT_FOUND = 1 + ERR_ACCESS = 2 + ERR_DISK_FULL = 3 + ERR_ILLEGAL_OP = 4 + ERR_UNKNOWN_ID = 5 + ERR_EXISTS = 6 + + BLOCK_SIZE = 512 + DEFAULT_TIMEOUT = 2.0 + DEFAULT_RETRIES = 3 + + _ERROR_MESSAGES = { + ERR_UNDEFINED: "Undefined error", + ERR_NOT_FOUND: "File not found", + ERR_ACCESS: "Access violation", + ERR_DISK_FULL: "Disk full or allocation exceeded", + ERR_ILLEGAL_OP: "Illegal TFTP operation", + ERR_UNKNOWN_ID: "Unknown transfer ID", + ERR_EXISTS: "File already exists", + } + + def __init__( + self, + on_running_changed: Optional[Callable[[bool], None]] = None, + on_error: Optional[Callable[[Exception], None]] = None, + ): + self.logger = get_logger("zashterminal.filemanager.tftp_server") + self._on_running_changed = on_running_changed + self._on_error = on_error + self._stop_event = threading.Event() + self._thread: Optional[threading.Thread] = None + self._socket: Optional[socket.socket] = None + self._lock = threading.RLock() + self._running = False + self.port = 69 + self.upload_dir = Path.home() + self.download_dir = Path.home() + + @property + def is_running(self) -> bool: + with self._lock: + return self._running + + def start(self, port: int, upload_dir: str, download_dir: str) -> None: + if self.is_running: + self.stop() + + upload_path = Path(upload_dir).expanduser().resolve() + download_path = Path(download_dir).expanduser().resolve() + if not upload_path.is_dir() or not os.access(upload_path, os.R_OK): + raise TftpFileError(f"Upload directory is not readable: {upload_path}") + if not download_path.is_dir() or not os.access(download_path, os.W_OK): + raise TftpFileError(f"Download directory is not writable: {download_path}") + if port < 0 or port > 65535: + raise TftpBindError(f"Invalid UDP port: {port}") + + with self._lock: + self.port = port + self.upload_dir = upload_path + self.download_dir = download_path + self._stop_event.clear() + self._thread = threading.Thread( + target=self._serve, name="ZashTerminalTftpServer", daemon=True + ) + self._thread.start() + + def stop(self) -> None: + thread = None + with self._lock: + self._stop_event.set() + thread = self._thread + if self._socket is not None: + try: + self._socket.close() + except OSError: + pass + if thread and thread.is_alive() and thread is not threading.current_thread(): + thread.join(timeout=2.5) + with self._lock: + self._thread = None + + def _set_running(self, running: bool) -> None: + with self._lock: + if self._running == running: + return + self._running = running + if self._on_running_changed: + self._on_running_changed(running) + + def _emit_error(self, error: Exception) -> None: + self.logger.error(f"TFTP server error: {error}") + if self._on_error: + self._on_error(error) + + def _serve(self) -> None: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.settimeout(0.5) + try: + sock.bind(("", self.port)) + except OSError as exc: + sock.close() + self._set_running(False) + self._emit_error(TftpBindError(str(exc))) + return + + with self._lock: + self._socket = sock + self._set_running(True) + + try: + while not self._stop_event.is_set(): + try: + data, client = sock.recvfrom(2048) + except socket.timeout: + continue + except OSError: + if self._stop_event.is_set(): + break + raise + + threading.Thread( + target=self._handle_request, + args=(data, client), + name=f"ZashTerminalTftpTransfer-{client[0]}:{client[1]}", + daemon=True, + ).start() + except Exception as exc: + if not self._stop_event.is_set(): + self._emit_error(TftpNetworkError(str(exc))) + finally: + try: + sock.close() + except OSError: + pass + with self._lock: + if self._socket is sock: + self._socket = None + self._set_running(False) + + def _handle_request(self, data: bytes, client: Tuple[str, int]) -> None: + transfer_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + transfer_sock.settimeout(self.DEFAULT_TIMEOUT) + try: + try: + opcode, filename, mode = self._parse_request(data) + except ValueError: + self._send_error(transfer_sock, client, self.ERR_ILLEGAL_OP) + return + + if mode not in ("octet", "netascii"): + self._send_error(transfer_sock, client, self.ERR_ILLEGAL_OP) + return + + if opcode == self.OP_RRQ: + self._serve_read_request(transfer_sock, client, filename) + elif opcode == self.OP_WRQ: + self._serve_write_request(transfer_sock, client, filename) + else: + self._send_error(transfer_sock, client, self.ERR_ILLEGAL_OP) + except OSError as exc: + self.logger.warning(f"TFTP transfer failed for {client}: {exc}") + finally: + transfer_sock.close() + + def _parse_request(self, data: bytes) -> Tuple[int, str, str]: + if len(data) < 4: + raise ValueError("Packet too short") + opcode = struct.unpack("!H", data[:2])[0] + parts = data[2:].split(b"\0") + if len(parts) < 3 or not parts[0] or not parts[1]: + raise ValueError("Malformed request") + filename = parts[0].decode("utf-8", errors="strict") + mode = parts[1].decode("ascii", errors="strict").lower() + return opcode, filename, mode + + def _resolve_child(self, root: Path, requested: str) -> Optional[Path]: + requested = requested.replace("\\", "/").lstrip("/") + if not requested or "\0" in requested: + return None + candidate = (root / requested).resolve() + try: + candidate.relative_to(root) + except ValueError: + return None + return candidate + + def _serve_read_request( + self, sock: socket.socket, client: Tuple[str, int], filename: str + ) -> None: + path = self._resolve_child(self.upload_dir, filename) + if path is None: + self._send_error(sock, client, self.ERR_ACCESS) + return + if not path.is_file(): + self._send_error(sock, client, self.ERR_NOT_FOUND) + return + if not os.access(path, os.R_OK): + self._send_error(sock, client, self.ERR_ACCESS) + return + + block = 1 + try: + with path.open("rb") as file: + while not self._stop_event.is_set(): + payload = file.read(self.BLOCK_SIZE) + packet = struct.pack("!HH", self.OP_DATA, block) + payload + if not self._send_with_ack(sock, client, packet, block): + return + block = (block + 1) & 0xFFFF + if len(payload) < self.BLOCK_SIZE: + return + except OSError: + self._send_error(sock, client, self.ERR_ACCESS) + + def _serve_write_request( + self, sock: socket.socket, client: Tuple[str, int], filename: str + ) -> None: + path = self._resolve_child(self.download_dir, filename) + if path is None: + self._send_error(sock, client, self.ERR_ACCESS) + return + if path.exists(): + self._send_error(sock, client, self.ERR_EXISTS) + return + + try: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("xb") as file: + self._send_ack(sock, client, 0) + expected_block = 1 + while not self._stop_event.is_set(): + data, sender = sock.recvfrom(self.BLOCK_SIZE + 4) + if sender != client: + self._send_error(sock, sender, self.ERR_UNKNOWN_ID) + continue + if len(data) < 4: + self._send_error(sock, client, self.ERR_ILLEGAL_OP) + return + opcode, block = struct.unpack("!HH", data[:4]) + if opcode != self.OP_DATA: + self._send_error(sock, client, self.ERR_ILLEGAL_OP) + return + if block != expected_block: + self._send_ack(sock, client, (expected_block - 1) & 0xFFFF) + continue + payload = data[4:] + file.write(payload) + self._send_ack(sock, client, block) + expected_block = (expected_block + 1) & 0xFFFF + if len(payload) < self.BLOCK_SIZE: + return + except FileExistsError: + self._send_error(sock, client, self.ERR_EXISTS) + except OSError: + self._send_error(sock, client, self.ERR_ACCESS) + + def _send_with_ack( + self, sock: socket.socket, client: Tuple[str, int], packet: bytes, block: int + ) -> bool: + for _attempt in range(self.DEFAULT_RETRIES): + sock.sendto(packet, client) + try: + while not self._stop_event.is_set(): + data, sender = sock.recvfrom(4) + if sender != client: + self._send_error(sock, sender, self.ERR_UNKNOWN_ID) + continue + if len(data) < 4: + continue + opcode, ack_block = struct.unpack("!HH", data[:4]) + if opcode == self.OP_ACK and ack_block == block: + return True + if opcode == self.OP_ERROR: + return False + except socket.timeout: + continue + return False + + def _send_ack(self, sock: socket.socket, client: Tuple[str, int], block: int) -> None: + sock.sendto(struct.pack("!HH", self.OP_ACK, block), client) + + def _send_error( + self, sock: socket.socket, client: Tuple[str, int], code: int + ) -> None: + message = self._ERROR_MESSAGES.get( + code, self._ERROR_MESSAGES[self.ERR_UNDEFINED] + ) + packet = ( + struct.pack("!HH", self.OP_ERROR, code) + message.encode("ascii") + b"\0" + ) + sock.sendto(packet, client) diff --git a/src/zashterminal/settings/config.py b/src/zashterminal/settings/config.py index ae9a716..fd26465 100644 --- a/src/zashterminal/settings/config.py +++ b/src/zashterminal/settings/config.py @@ -237,6 +237,10 @@ def get_defaults() -> Dict[str, Any]: # Remote Editing "use_system_tmp_for_edit": False, "clear_remote_edit_files_on_exit": True, + # TFTP Server + "tftp_server_port": 6969, + "tftp_server_upload_dir": str(Path.home()), + "tftp_server_download_dir": str(Path.home()), # AI Assistant "ai_assistant_enabled": False, "ai_assistant_provider": "groq", diff --git a/src/zashterminal/ui/actions.py b/src/zashterminal/ui/actions.py index 4e24ac7..aeac444 100644 --- a/src/zashterminal/ui/actions.py +++ b/src/zashterminal/ui/actions.py @@ -73,6 +73,7 @@ def setup_actions(self): "toggle-broadcast": self.toggle_broadcast, "show-command-manager": self.show_command_manager, "import-securecrt-sessions": self.import_securecrt_sessions, + "toggle-tftp-server": self.toggle_tftp_server, "preferences": self.preferences, "shortcuts": self.shortcuts, "new-window": self.new_window, @@ -418,6 +419,28 @@ def show_command_manager(self, *_args): self._hide_tooltip() self.window._show_command_manager_dialog() + def toggle_tftp_server(self, *_args): + self._hide_tooltip() + if self.window.tftp_server.is_running: + self.window.tftp_server.stop() + return + + from .dialogs import TftpServerDialog + + def start_server(port: int, upload_dir: str, download_dir: str): + try: + self.window.tftp_server.start(port, upload_dir, download_dir) + except Exception as exc: + self.logger.error(f"Failed to start TFTP server: {exc}") + self.window._show_tftp_error(exc) + + dialog = TftpServerDialog( + self.window, + self.window.settings_manager, + start_server, + ) + dialog.present() + def preferences(self, *_args): self._hide_tooltip() from .dialogs import PreferencesDialog diff --git a/src/zashterminal/ui/dialogs/__init__.py b/src/zashterminal/ui/dialogs/__init__.py index ff18e95..784b307 100644 --- a/src/zashterminal/ui/dialogs/__init__.py +++ b/src/zashterminal/ui/dialogs/__init__.py @@ -6,6 +6,7 @@ from .preferences_dialog import PreferencesDialog from .session_edit_dialog import SessionEditDialog from .shortcuts_dialog import ShortcutsDialog +from .tftp_server_dialog import TftpServerDialog __all__ = [ "BaseDialog", @@ -18,4 +19,5 @@ "PreferencesDialog", "SessionEditDialog", "ShortcutsDialog", + "TftpServerDialog", ] diff --git a/src/zashterminal/ui/dialogs/tftp_server_dialog.py b/src/zashterminal/ui/dialogs/tftp_server_dialog.py new file mode 100644 index 0000000..6660187 --- /dev/null +++ b/src/zashterminal/ui/dialogs/tftp_server_dialog.py @@ -0,0 +1,130 @@ +from pathlib import Path + +import gi + +gi.require_version("Gtk", "4.0") +gi.require_version("Adw", "1") +from gi.repository import Adw, Gio, Gtk + +from ...utils.icons import icon_button +from ...utils.translation_utils import _ +from .base_dialog import BaseDialog + + +class TftpServerDialog(BaseDialog): + """Dialog for configuring and starting the embedded TFTP server.""" + + def __init__(self, parent_window, settings_manager, on_start): + super().__init__( + parent_window, + _("Start TFTP Server"), + auto_setup_toolbar=True, + default_width=560, + default_height=360, + ) + self.settings_manager = settings_manager + self._on_start = on_start + + start_button = Gtk.Button(label=_("Start")) + start_button.add_css_class("suggested-action") + start_button.connect("clicked", self._on_start_clicked) + self.add_header_button(start_button) + + self.port_row = Adw.SpinRow.new_with_range(0, 65535, 1) + self.port_row.set_title(_("TFTP Port")) + self.port_row.set_value(self.settings_manager.get("tftp_server_port", 6969)) + + upload_dir = self.settings_manager.get( + "tftp_server_upload_dir", str(Path.home()) + ) + download_dir = self.settings_manager.get( + "tftp_server_download_dir", str(Path.home()) + ) + self.upload_row = self._create_directory_row( + _("Upload Directory"), + _("Files requested by TFTP clients are read from this directory"), + upload_dir, + ) + self.download_row = self._create_directory_row( + _("Download Directory"), + _("Files sent by TFTP clients are written to this directory"), + download_dir, + ) + + group = Adw.PreferencesGroup( + title=_("TFTP Server"), + description=_( + "The TFTP server uses local directories for uploading and downloading files." + ), + ) + group.add(self.port_row) + group.add(self.upload_row) + group.add(self.download_row) + + page = Adw.PreferencesPage() + page.add(group) + self.set_body_content(page) + + def _create_directory_row( + self, title: str, subtitle: str, path: str + ) -> Adw.ActionRow: + row = Adw.ActionRow(title=title, subtitle=subtitle) + entry = Gtk.Entry(text=path, hexpand=True) + entry.set_valign(Gtk.Align.CENTER) + browse_button = icon_button("folder-open-symbolic") + browse_button.set_tooltip_text(_("Select Directory")) + browse_button.connect("clicked", self._on_browse_clicked, entry) + row.add_suffix(entry) + row.add_suffix(browse_button) + row.set_activatable_widget(entry) + row.entry = entry + return row + + def _on_browse_clicked(self, _button, entry: Gtk.Entry) -> None: + chooser = Gtk.FileChooserDialog( + title=_("Open Directory"), + transient_for=self, + modal=True, + action=Gtk.FileChooserAction.SELECT_FOLDER, + ) + chooser.add_css_class("zashterminal-dialog") + chooser.add_button(_("Cancel"), Gtk.ResponseType.CANCEL) + chooser.add_button(_("Select"), Gtk.ResponseType.ACCEPT) + try: + current = entry.get_text().strip() + if current: + chooser.set_current_folder(Gio.File.new_for_path(current)) + except Exception: + pass + + def on_response(dialog, response_id): + try: + if response_id == Gtk.ResponseType.ACCEPT: + folder = dialog.get_file() + if folder and folder.get_path(): + entry.set_text(folder.get_path()) + finally: + dialog.destroy() + + chooser.connect("response", on_response) + chooser.present() + + def _on_start_clicked(self, _button) -> None: + upload_dir = self.upload_row.entry.get_text().strip() + download_dir = self.download_row.entry.get_text().strip() + upload_path = Path(upload_dir).expanduser() + download_path = Path(download_dir).expanduser() + + if not upload_path.is_dir() or not download_path.is_dir(): + self._show_error_dialog( + _("Warning"), + _("Please select a valid directory!"), + ) + return + + port = int(self.port_row.get_value()) + self.settings_manager.set("tftp_server_port", port) + self.settings_manager.set("tftp_server_upload_dir", str(upload_path)) + self.settings_manager.set("tftp_server_download_dir", str(download_path)) + self._on_start(port, str(upload_path), str(download_path)) + self.close() diff --git a/src/zashterminal/ui/menus.py b/src/zashterminal/ui/menus.py index 11ff245..1f7814e 100644 --- a/src/zashterminal/ui/menus.py +++ b/src/zashterminal/ui/menus.py @@ -178,6 +178,11 @@ def create_main_popover(parent_window) -> tuple[Gtk.Popover, FontSizerWidget]: "label": _("Import SecureCRT Sessions"), "action": "win.import-securecrt-sessions", }, + { + "label": _("Start TFTP Server"), + "action": "win.toggle-tftp-server", + "dynamic_label": "tftp_server", + }, {"label": _("About"), "action": "app.about"}, "---", {"label": _("Quit"), "action": "app.quit"}, @@ -185,6 +190,7 @@ def create_main_popover(parent_window) -> tuple[Gtk.Popover, FontSizerWidget]: actions_that_close_menu = { "win.new-window", "win.import-securecrt-sessions", + "win.toggle-tftp-server", "win.preferences", "win.highlight-settings", "win.configure-ai", @@ -205,6 +211,9 @@ def create_main_popover(parent_window) -> tuple[Gtk.Popover, FontSizerWidget]: box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=24) button.set_child(box) action_label = Gtk.Label(label=item["label"], xalign=0.0, hexpand=True) + if item.get("dynamic_label") == "tftp_server": + popover.tftp_server_action_label = action_label + MainApplicationMenu.update_tftp_server_label(parent_window, popover) box.append(action_label) accels = app.get_accels_for_action(item["action"]) if accels: @@ -217,6 +226,19 @@ def create_main_popover(parent_window) -> tuple[Gtk.Popover, FontSizerWidget]: return popover, font_sizer_widget + @staticmethod + def update_tftp_server_label(parent_window, popover: Gtk.Popover) -> None: + label = getattr(popover, "tftp_server_action_label", None) + if not label: + return + + is_running = bool( + getattr(getattr(parent_window, "tftp_server", None), "is_running", False) + ) + label.set_text( + _("Stop TFTP Server") if is_running else _("Start TFTP Server") + ) + def create_session_menu( session_item, diff --git a/src/zashterminal/ui/window_ui.py b/src/zashterminal/ui/window_ui.py index 12f4a18..07d5388 100644 --- a/src/zashterminal/ui/window_ui.py +++ b/src/zashterminal/ui/window_ui.py @@ -768,7 +768,7 @@ def _on_menu_popover_show(self, popover: Gtk.Popover) -> None: real_popover, self.font_sizer_widget = MainApplicationMenu.create_main_popover( self.window ) - real_popover.connect("show", lambda p: self.tooltip_helper.hide()) + real_popover.connect("show", self._on_main_menu_popover_show) self._main_menu_popover = real_popover self.menu_button.set_popover(real_popover) @@ -783,11 +783,27 @@ def _ensure_main_menu_popover(self, button: Gtk.MenuButton) -> None: popover, self.font_sizer_widget = MainApplicationMenu.create_main_popover( self.window ) - popover.connect("show", lambda p: self.tooltip_helper.hide()) + popover.connect("show", self._on_main_menu_popover_show) self._main_menu_popover = popover self.menu_button.set_popover(popover) # Popover will be shown automatically by MenuButton + def _on_main_menu_popover_show(self, popover: Gtk.Popover) -> None: + self.tooltip_helper.hide() + from .menus import MainApplicationMenu + + MainApplicationMenu.update_tftp_server_label(self.window, popover) + + def update_tftp_server_menu_state(self) -> None: + if self._main_menu_popover is None: + return + + from .menus import MainApplicationMenu + + MainApplicationMenu.update_tftp_server_label( + self.window, self._main_menu_popover + ) + def _create_ai_chat_panel(self) -> None: """Create the AI chat panel widget (lazy initialization).""" if self.ai_chat_panel is not None: diff --git a/src/zashterminal/window.py b/src/zashterminal/window.py index b57d315..2caea60 100644 --- a/src/zashterminal/window.py +++ b/src/zashterminal/window.py @@ -17,6 +17,12 @@ from .sessions.tree import SessionTreeView from .settings.manager import SettingsManager from .state.window_state import WindowStateManager +from .filemanager.tftp_server import ( + TftpBindError, + TftpFileError, + TftpNetworkError, + TftpServer, +) from .terminal.ai_assistant import TerminalAiAssistant from .terminal.manager import TerminalManager from .terminal.tabs import TabManager @@ -51,6 +57,7 @@ def __init__(self, application, settings_manager: SettingsManager, **kwargs): self.layouts: List[LayoutItem] = [] self.active_temp_files = weakref.WeakKeyDictionary() self.command_manager_dialog = None # For Command Manager dialog + self.tftp_server = None # Search state tracking self.current_search_terminal = None @@ -206,6 +213,10 @@ def _create_managers_and_ui(self) -> None: # UI/View-Model Layer self.terminal_manager = TerminalManager(self, self.settings_manager) + self.tftp_server = TftpServer( + on_running_changed=self._on_tftp_server_running_changed, + on_error=self._on_tftp_server_error, + ) # Start terminal pre-creation in background for faster first tab if not self._is_for_detached_tab: self.terminal_manager.prepare_initial_terminal() @@ -1613,9 +1624,45 @@ def _perform_cleanup(self) -> None: for fm in self.tab_manager.file_managers.values(): fm.shutdown(None) + if self.tftp_server: + self.tftp_server.stop() + # Clean up CSS providers to prevent memory leaks self.settings_manager.cleanup_css_providers(self) + def _on_tftp_server_running_changed(self, running: bool) -> None: + def notify(): + title = _("TFTP server started.") if running else _("TFTP server stopped.") + if hasattr(self, "ui_builder") and self.ui_builder: + self.ui_builder.update_tftp_server_menu_state() + if hasattr(self, "toast_overlay") and self.toast_overlay: + self.toast_overlay.add_toast(Adw.Toast(title=title)) + return GLib.SOURCE_REMOVE + + GLib.idle_add(notify) + + def _on_tftp_server_error(self, error: Exception) -> None: + GLib.idle_add(self._show_tftp_error, error) + + def _show_tftp_error(self, error: Exception) -> bool: + if isinstance(error, TftpBindError): + message = _("TFTP server bind error!") + elif isinstance(error, TftpFileError): + message = _("TFTP server file error!") + elif isinstance(error, TftpNetworkError): + message = _("TFTP server network error!") + else: + message = _("TFTP server error!") + + dialog = Adw.MessageDialog( + transient_for=self, + title=_("Warning"), + body=message, + ) + dialog.add_response("ok", _("OK")) + dialog.present() + return GLib.SOURCE_REMOVE + def destroy(self) -> None: self._perform_cleanup() super().destroy()