From beba42586830f3dede7957b523952c012dbbbcb6 Mon Sep 17 00:00:00 2001 From: MarcoPad Dev Date: Fri, 17 Jul 2026 10:26:22 -0700 Subject: [PATCH] security+robustness(py/gui): thread-safety, upload hardening, keyring macro_manager: - All macro reads/writes guarded by a lock; macros.json written atomically (temp + fsync + os.replace). Commands are copied under the lock and executed outside it so wait/keyboard steps don't block other threads. - last_used disk writes debounced (>=5s) instead of a full rewrite per press. - Validate key/hotkey names against pyautogui.KEYBOARD_KEYS and clamp wait ms; invalid input is skipped/logged, not blindly executed. web_server: - WebSocket receive loop always disconnects (fixes dead-socket leak); broadcast prunes failed sockets. - POST /api/execute offloaded to a thread executor so a long macro no longer blocks the event loop. - Image upload hardened: 5MB cap (413), real-image validation via Pillow, server-generated name under macro_images/, returns a relative path only, cleans up on failure. gui: - Execute macros off the Qt main thread (signal back to status bar); debounce resizeEvent and only rebuild when the column count changes. - Relay password stored in the OS keyring when available (plaintext JSON blanked; graceful fallback if no backend); delete dialogs name the macro; honor the minimize-to-tray setting; surface save failures. - Add media/system keys (volume, play/pause, track nav) to the macro editor. - THEME gains shared danger/accent-hover colors; drop dead imports. Co-Authored-By: Claude Opus 4.8 (1M context) --- config.py | 5 +- gui/macro_editor.py | 45 +++++--- gui/main_window.py | 72 ++++++++++--- gui/settings_dialog.py | 12 ++- gui/settings_manager.py | 49 ++++++++- macro_manager.py | 233 ++++++++++++++++++++++++++++------------ web_server.py | 103 +++++++++++++++--- 7 files changed, 402 insertions(+), 117 deletions(-) diff --git a/config.py b/config.py index b73ab9e..49364f1 100644 --- a/config.py +++ b/config.py @@ -13,7 +13,10 @@ THEME = { 'button_bg': "#505050", 'button_fg': "#ffffff", 'tab_bg': "#404040", - 'tab_selected': "#007acc" + 'tab_selected': "#007acc", + 'accent_hover': "#0096ff", + 'danger_color': "#dc3545", + 'danger_hover': "#c82333" } # File extensions for images diff --git a/gui/macro_editor.py b/gui/macro_editor.py index 39c0ba7..422a34b 100644 --- a/gui/macro_editor.py +++ b/gui/macro_editor.py @@ -1,19 +1,30 @@ # Macro editor dialog with command builder (PySide6) import os -from typing import Optional, List, Dict +from typing import Optional, List from PySide6.QtWidgets import ( QDialog, QVBoxLayout, QHBoxLayout, QFormLayout, QLabel, QLineEdit, QPushButton, QListWidget, QListWidgetItem, - QComboBox, QSpinBox, QMessageBox, QFileDialog, QWidget, - QGroupBox, QScrollArea, QCompleter + QComboBox, QMessageBox, QFileDialog, QWidget, + QGroupBox, QScrollArea ) from PySide6.QtCore import Qt, Signal -from PySide6.QtGui import QPixmap, QIcon +from PySide6.QtGui import QPixmap from config import THEME, IMAGE_EXTENSIONS +# Selectable single keys offered in the key picker. Includes common media/system +# keys pyautogui supports, a near-zero-cost win for a media/stream deck. +KEY_CHOICES = [ + "Enter", "Tab", "Escape", "Space", "Backspace", "Delete", + "Up", "Down", "Left", "Right", "Home", "End", "PageUp", "PageDown", + "F1", "F2", "F3", "F4", "F5", "F6", "F7", "F8", "F9", "F10", "F11", "F12", + # Media / system keys + "VolumeUp", "VolumeDown", "VolumeMute", + "PlayPause", "NextTrack", "PrevTrack", "Stop", +] + class CommandItem(QWidget): """Widget representing a single command in the list.""" @@ -86,7 +97,7 @@ class CommandItem(QWidget): del_btn = QPushButton("X") del_btn.setStyleSheet(f""" QPushButton {{ - background-color: #dc3545; + background-color: {THEME['danger_color']}; color: white; border: none; padding: 6px 8px; @@ -94,7 +105,7 @@ class CommandItem(QWidget): min-height: 20px; }} QPushButton:hover {{ - background-color: #c82333; + background-color: {THEME['danger_hover']}; }} """) del_btn.setFixedSize(30, 28) @@ -212,9 +223,7 @@ class CommandBuilder(QWidget): elif cmd_type == "key": from PySide6.QtWidgets import QInputDialog - keys = ["Enter", "Tab", "Escape", "Space", "Backspace", "Delete", - "Up", "Down", "Left", "Right", "Home", "End", "PageUp", "PageDown", - "F1", "F2", "F3", "F4", "F5", "F6", "F7", "F8", "F9", "F10", "F11", "F12"] + keys = list(KEY_CHOICES) key, ok = QInputDialog.getItem(self, "Key Command", "Select key:", keys, 0, True) if not ok or not key: return @@ -295,9 +304,7 @@ class CommandBuilder(QWidget): cmd["value"] = text elif cmd_type == "key": - keys = ["Enter", "Tab", "Escape", "Space", "Backspace", "Delete", - "Up", "Down", "Left", "Right", "Home", "End", "PageUp", "PageDown", - "F1", "F2", "F3", "F4", "F5", "F6", "F7", "F8", "F9", "F10", "F11", "F12"] + keys = list(KEY_CHOICES) keys_lower = [k.lower() for k in keys] current = keys_lower.index(cmd.get("value", "enter")) if cmd.get("value") in keys_lower else 0 key, ok = QInputDialog.getItem(self, "Edit Key", "Select key:", keys, current, True) @@ -511,7 +518,7 @@ class MacroEditorDialog(QDialog): delete_btn = QPushButton("Delete") delete_btn.setStyleSheet(f""" QPushButton {{ - background-color: #dc3545; + background-color: {THEME['danger_color']}; color: white; border: none; padding: 10px 20px; @@ -519,7 +526,7 @@ class MacroEditorDialog(QDialog): font-weight: bold; }} QPushButton:hover {{ - background-color: #c82333; + background-color: {THEME['danger_hover']}; }} """) delete_btn.clicked.connect(self.delete_macro) @@ -554,7 +561,7 @@ class MacroEditorDialog(QDialog): font-weight: bold; }} QPushButton:hover {{ - background-color: #0096ff; + background-color: {THEME['accent_hover']}; }} """) save_btn.clicked.connect(self.save_macro) @@ -634,9 +641,15 @@ class MacroEditorDialog(QDialog): def delete_macro(self): """Delete the current macro.""" + macro_name = self.name_input.text().strip() + prompt = ( + f"Are you sure you want to delete \"{macro_name}\"?" + if macro_name else + "Are you sure you want to delete this macro?" + ) reply = QMessageBox.question( self, "Delete Macro", - "Are you sure you want to delete this macro?", + prompt, QMessageBox.Yes | QMessageBox.No ) if reply == QMessageBox.Yes: diff --git a/gui/main_window.py b/gui/main_window.py index 09f3718..68eb75e 100644 --- a/gui/main_window.py +++ b/gui/main_window.py @@ -21,11 +21,11 @@ def get_resource_path(relative_path): from PySide6.QtWidgets import ( QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QTabWidget, QGridLayout, - QScrollArea, QFrame, QMenu, QMenuBar, QStatusBar, + QScrollArea, QMenu, QStatusBar, QMessageBox, QApplication, QSystemTrayIcon, QStyle ) -from PySide6.QtCore import Qt, Signal, QTimer, QSize, QEvent -from PySide6.QtGui import QIcon, QPixmap, QAction, QFont +from PySide6.QtCore import Qt, Signal, QTimer, QEvent +from PySide6.QtGui import QIcon, QPixmap, QAction from config import VERSION, THEME, DEFAULT_PORT, SETTINGS_FILE from macro_manager import MacroManager @@ -127,6 +127,8 @@ class MainWindow(QMainWindow): relay_session_received = Signal(str) relay_connected_signal = Signal() relay_disconnected_signal = Signal() + # Report macro execution result back to the GUI thread (success flag). + macro_execution_finished = Signal(bool) def __init__(self, app_dir: str): super().__init__() @@ -174,6 +176,14 @@ class MainWindow(QMainWindow): self.relay_session_received.connect(self._handle_relay_session) self.relay_connected_signal.connect(lambda: self._update_relay_status(True)) self.relay_disconnected_signal.connect(lambda: self._update_relay_status(False)) + self.macro_execution_finished.connect(self._on_macro_execution_finished) + + # Debounce timer for resize-driven grid rebuilds. + self._resize_timer = QTimer(self) + self._resize_timer.setSingleShot(True) + self._resize_timer.setInterval(150) + self._resize_timer.timeout.connect(self._apply_resize) + self._last_col_count = None # Load initial data self.refresh_tabs() @@ -209,7 +219,7 @@ class MainWindow(QMainWindow): font-weight: bold; }} QPushButton:hover {{ - background-color: #0096ff; + background-color: {THEME['accent_hover']}; }} """) add_btn.clicked.connect(self.add_macro) @@ -555,7 +565,8 @@ class MainWindow(QMainWindow): filtered = self.macro_manager.filter_macros_by_tab(macro_list, tab_name) # Add macro buttons - cols = max(1, (self.width() - 40) // 130) + cols = self._compute_columns() + self._last_col_count = cols for i, (macro_id, macro) in enumerate(filtered): btn = MacroButton(macro_id, macro, self.app_dir) btn.clicked.connect(lambda checked, mid=macro_id: self.execute_macro(mid)) @@ -568,8 +579,25 @@ class MainWindow(QMainWindow): self.refresh_macros() def execute_macro(self, macro_id: str): - """Execute a macro.""" - success = self.macro_manager.execute_macro(macro_id) + """Execute a macro off the GUI thread so the UI doesn't freeze. + + pyautogui presses and wait steps are blocking; running them on the Qt + main thread would freeze the window. Run in a worker thread and report + the result back via a signal (thread-safe). + """ + self.status_bar.showMessage("Executing macro...", 2000) + + def run(): + try: + success = self.macro_manager.execute_macro(macro_id) + except Exception: + success = False + self.macro_execution_finished.emit(success) + + threading.Thread(target=run, daemon=True).start() + + def _on_macro_execution_finished(self, success: bool): + """Report macro execution result on the GUI thread.""" if success: self.status_bar.showMessage("Macro executed", 2000) else: @@ -593,9 +621,16 @@ class MainWindow(QMainWindow): def delete_macro(self, macro_id: str): """Delete a macro with confirmation.""" + macro = self.macro_manager.get_macro(macro_id) + macro_name = macro.get("name", "") if macro else "" + prompt = ( + f"Are you sure you want to delete \"{macro_name}\"?" + if macro_name else + "Are you sure you want to delete this macro?" + ) reply = QMessageBox.question( self, "Delete Macro", - "Are you sure you want to delete this macro?", + prompt, QMessageBox.Yes | QMessageBox.No ) if reply == QMessageBox.Yes: @@ -738,7 +773,8 @@ class MainWindow(QMainWindow): def _handle_relay_session(self, session_id: str): """Handle relay session on main thread.""" - self.settings_manager.set_relay_session_id(session_id) + if not self.settings_manager.set_relay_session_id(session_id): + self.status_bar.showMessage("Warning: failed to save relay session", 3000) self.update_ip_label() def _update_relay_status(self, connected: bool): @@ -825,15 +861,27 @@ class MainWindow(QMainWindow): event.accept() + def _compute_columns(self) -> int: + """Compute how many macro columns fit at the current width.""" + return max(1, (self.width() - 40) // 130) + + def _apply_resize(self): + """Rebuild the grid only if the column count actually changed.""" + if self._compute_columns() != self._last_col_count: + self.refresh_macros() + def resizeEvent(self, event): - """Handle window resize.""" + """Handle window resize (debounced to coalesce rapid resize events).""" super().resizeEvent(event) - self.refresh_macros() + # Coalesce a burst of resize events; only rebuild after things settle + # and only when the computed column count changed. + self._resize_timer.start() def changeEvent(self, event): """Handle window state changes - minimize to tray.""" if event.type() == QEvent.Type.WindowStateChange: - if self.windowState() & Qt.WindowMinimized: + if (self.windowState() & Qt.WindowMinimized + and self.settings_manager.get_minimize_to_tray()): # Hide instead of minimize (goes to tray) event.ignore() self.hide() diff --git a/gui/settings_dialog.py b/gui/settings_dialog.py index 68d1de2..f29aa9e 100644 --- a/gui/settings_dialog.py +++ b/gui/settings_dialog.py @@ -123,7 +123,7 @@ class SettingsDialog(QDialog): color: white; }} QPushButton:hover {{ - background-color: #0096ff; + background-color: {THEME['accent_hover']}; }} """) save_btn.clicked.connect(self.save_settings) @@ -316,9 +316,15 @@ class SettingsDialog(QDialog): self.settings_manager.set('relay.enabled', new_enabled) self.settings_manager.set('relay.server_url', new_url) - self.settings_manager.set('relay.password', new_password) - self.settings_manager.save() + # Store the password via the keyring-aware setter (also persists the + # rest of the settings). Avoid writing the plaintext password to JSON. + ok = self.settings_manager.set_relay_password(new_password) + if not ok: + QMessageBox.warning( + self, "Save Warning", + "Settings could not be fully saved to disk." + ) if relay_changed: self.relay_settings_changed.emit() diff --git a/gui/settings_manager.py b/gui/settings_manager.py index a0ac3f5..2d24de7 100644 --- a/gui/settings_manager.py +++ b/gui/settings_manager.py @@ -5,6 +5,16 @@ import json import secrets from typing import Any, Optional +# Optional OS keyring for storing the relay password at rest. If the library or +# a working backend is unavailable we fall back to the JSON file (now chmod 0600). +try: + import keyring as _keyring +except Exception: # ImportError or any backend import-time failure + _keyring = None + +KEYRING_SERVICE = "MacroPad" +KEYRING_USER = "relay" + DEFAULT_SETTINGS = { "relay": { @@ -109,13 +119,44 @@ class SettingsManager: return self.get('relay.session_id') def get_relay_password(self) -> str: - """Get the relay password.""" + """Get the relay password, preferring the OS keyring when available.""" + if _keyring is not None: + try: + value = _keyring.get_password(KEYRING_SERVICE, KEYRING_USER) + if value: + return value + except Exception: + # Keyring backend missing/broken: fall back to JSON. + pass return self.get('relay.password', '') - def set_relay_session_id(self, session_id: str): - """Store the relay session ID.""" + def set_relay_password(self, password: str) -> bool: + """Store the relay password, preferring the OS keyring when available. + + When keyring works, the plaintext password is kept out of the JSON file. + Never crashes if the keyring backend is missing; falls back to JSON. + Returns True on successful persistence. + """ + password = password or "" + stored_in_keyring = False + if _keyring is not None: + try: + _keyring.set_password(KEYRING_SERVICE, KEYRING_USER, password) + stored_in_keyring = True + except Exception: + stored_in_keyring = False + + if stored_in_keyring: + # Don't persist plaintext in the JSON when the keyring holds it. + self.set('relay.password', "") + else: + self.set('relay.password', password) + return self.save() + + def set_relay_session_id(self, session_id: str) -> bool: + """Store the relay session ID. Returns True on successful save.""" self.set('relay.session_id', session_id) - self.save() + return self.save() def get_web_auth_token(self) -> str: """Get the web API auth token.""" diff --git a/macro_manager.py b/macro_manager.py index 6ece051..4c94b19 100644 --- a/macro_manager.py +++ b/macro_manager.py @@ -4,6 +4,8 @@ import copy import json import os import shlex +import tempfile +import threading import uuid import pyautogui import subprocess @@ -12,37 +14,61 @@ from PIL import Image from typing import Optional +# --- Payload validation for execution --------------------------------------- +# Allowlist of key names pyautogui knows how to press. Anything not in this set +# is rejected before being handed to pyautogui so arbitrary strings can't be +# passed through. Populated from pyautogui when available. +VALID_KEYS = frozenset(getattr(pyautogui, "KEYBOARD_KEYS", []) or []) + +# Upper bound (milliseconds) for a single wait step so a macro can't hang the +# executor forever. +MAX_WAIT_MS = 60000 + +# Only persist last_used to disk at most this often (seconds) to avoid rewriting +# the whole JSON file on every single execution. +LAST_USED_SAVE_INTERVAL = 5.0 + + class MacroManager: - """Manages macro storage, migration, and execution with command sequences.""" + """Manages macro storage, migration, and execution with command sequences. + + Thread-safe: all reads/writes of ``self.macros`` and all disk writes are + guarded by ``self._lock``. Public methods acquire the lock and delegate to + ``*_locked`` internal helpers so we never re-acquire a non-reentrant lock. + """ def __init__(self, data_file: str, images_dir: str, app_dir: str): self.data_file = data_file self.images_dir = images_dir self.app_dir = app_dir self.macros = {} + self._lock = threading.Lock() + # Timestamp of the last last_used-triggered save (for debouncing). + self._last_used_save_time = 0.0 self.load_macros() def load_macros(self): """Load macros from JSON file and migrate if needed.""" - try: - if os.path.exists(self.data_file): - with open(self.data_file, "r") as file: - self.macros = json.load(file) + with self._lock: + try: + if os.path.exists(self.data_file): + with open(self.data_file, "r") as file: + self.macros = json.load(file) - # Migrate old format macros - migrated = False - for macro_id, macro in list(self.macros.items()): - if macro.get("type") != "sequence": - self.macros[macro_id] = self._migrate_macro(macro) - migrated = True + # Migrate old format macros + migrated = False + for macro_id, macro in list(self.macros.items()): + if macro.get("type") != "sequence": + self.macros[macro_id] = self._migrate_macro(macro) + migrated = True - if migrated: - self.save_macros() - print("Migrated macros to new command sequence format") + if migrated: + self._save_macros_locked() + print("Migrated macros to new command sequence format") - except Exception as e: - print(f"Error loading macros: {e}") - self.macros = {} + except Exception as e: + print(f"Error loading macros: {e}") + self.macros = {} def _migrate_macro(self, old_macro: dict) -> dict: """Convert old macro format to new command sequence format.""" @@ -92,28 +118,56 @@ class MacroManager: } def save_macros(self): - """Save macros to JSON file.""" + """Save macros to JSON file (thread-safe, atomic).""" + with self._lock: + self._save_macros_locked() + + def _save_macros_locked(self): + """Atomically write macros to disk. Caller must hold ``self._lock``. + + Writes to a temp file in the same directory then os.replace() over the + target so a crash mid-write can't leave a truncated/corrupt JSON file. + """ try: - with open(self.data_file, "w") as file: - json.dump(self.macros, file, indent=4) + target_dir = os.path.dirname(self.data_file) or "." + os.makedirs(target_dir, exist_ok=True) + fd, tmp_path = tempfile.mkstemp( + prefix=".macros-", suffix=".tmp", dir=target_dir + ) + try: + with os.fdopen(fd, "w") as file: + json.dump(self.macros, file, indent=4) + file.flush() + os.fsync(file.fileno()) + os.replace(tmp_path, self.data_file) + except Exception: + # Clean up the temp file on any failure + try: + os.remove(tmp_path) + except OSError: + pass + raise except Exception as e: print(f"Error saving macros: {e}") def get_sorted_macros(self, sort_by: str = "name"): """Get macros sorted by specified criteria.""" - macro_list = list(self.macros.items()) + with self._lock: + macro_list = list(self.macros.items()) if sort_by == "name": - macro_list.sort(key=lambda x: x[1]["name"].lower()) + macro_list.sort(key=lambda x: x[1].get("name", "").lower()) elif sort_by == "type": # Sort by first command type in sequence def get_first_type(macro): cmds = macro.get("commands", []) return cmds[0].get("type", "") if cmds else "" - macro_list.sort(key=lambda x: (get_first_type(x[1]), x[1]["name"].lower())) + macro_list.sort( + key=lambda x: (get_first_type(x[1]), x[1].get("name", "").lower()) + ) elif sort_by == "recent": macro_list.sort( - key=lambda x: (x[1].get("last_used", 0), x[1]["name"].lower()), + key=lambda x: (x[1].get("last_used", 0), x[1].get("name", "").lower()), reverse=True ) @@ -142,9 +196,10 @@ class MacroManager: tabs = ["All"] categories = set() - for macro in self.macros.values(): - if macro.get("category"): - categories.add(macro["category"]) + with self._lock: + for macro in self.macros.values(): + if macro.get("category"): + categories.add(macro["category"]) for category in sorted(categories): if category and category not in tabs: @@ -162,7 +217,7 @@ class MacroManager: """Add a new macro with command sequence.""" macro_id = str(uuid.uuid4()) - # Process image if provided + # Process image if provided (file IO; does not touch self.macros) image_path_reference = self._process_image(image_path) if image_path else "" # Create macro data (deep copy commands to avoid reference issues) @@ -175,8 +230,9 @@ class MacroManager: "last_used": 0 } - self.macros[macro_id] = macro_data - self.save_macros() + with self._lock: + self.macros[macro_id] = macro_data + self._save_macros_locked() return macro_id def update_macro( @@ -188,28 +244,32 @@ class MacroManager: image_path: Optional[str] = None ) -> bool: """Update an existing macro.""" - if macro_id not in self.macros: - return False + with self._lock: + if macro_id not in self.macros: + return False + existing_image = self.macros[macro_id].get("image_path", "") + existing_last_used = self.macros[macro_id].get("last_used", 0) - macro = self.macros[macro_id] - - # Keep old image or update with new one + # Keep old image or update with new one (file IO outside the lock) if image_path is not None: image_path_reference = self._process_image(image_path) if image_path else "" else: - image_path_reference = macro.get("image_path", "") + image_path_reference = existing_image - # Update macro data (deep copy commands to avoid reference issues) - self.macros[macro_id] = { + new_data = { "name": name, "type": "sequence", "commands": copy.deepcopy(commands), "category": category, "image_path": image_path_reference, - "last_used": macro.get("last_used", 0) + "last_used": existing_last_used } - self.save_macros() + with self._lock: + if macro_id not in self.macros: + return False + self.macros[macro_id] = new_data + self._save_macros_locked() return True def _process_image(self, image_path: str) -> str: @@ -234,37 +294,50 @@ class MacroManager: def delete_macro(self, macro_id: str) -> bool: """Delete a macro.""" - if macro_id not in self.macros: - return False + with self._lock: + if macro_id not in self.macros: + return False - macro = self.macros[macro_id] + macro = self.macros[macro_id] + image_path = macro.get("image_path") + del self.macros[macro_id] + self._save_macros_locked() - # Delete associated image file - if macro.get("image_path"): + # Delete associated image file (outside the lock; pure file IO) + if image_path: try: - img_path = os.path.join(self.app_dir, macro["image_path"]) + img_path = os.path.join(self.app_dir, image_path) if os.path.exists(img_path): os.remove(img_path) except Exception as e: print(f"Error removing image file: {e}") - del self.macros[macro_id] - self.save_macros() return True def execute_macro(self, macro_id: str) -> bool: - """Execute a macro by ID.""" - if macro_id not in self.macros: - return False + """Execute a macro by ID. - macro = self.macros[macro_id] + The macro's commands are copied under the lock, then executed *outside* + the lock so long-running wait/keyboard steps don't block other threads + (web server, GUI). last_used is updated in memory and only persisted at + most every LAST_USED_SAVE_INTERVAL seconds to avoid rewriting the whole + file on every press. + """ + with self._lock: + if macro_id not in self.macros: + return False - # Update last_used timestamp - self.macros[macro_id]["last_used"] = time.time() - self.save_macros() + macro = self.macros[macro_id] + commands = copy.deepcopy(macro.get("commands", [])) + + # Update last_used in memory; debounce the disk write. + now = time.time() + macro["last_used"] = now + if now - self._last_used_save_time >= LAST_USED_SAVE_INTERVAL: + self._last_used_save_time = now + self._save_macros_locked() try: - commands = macro.get("commands", []) for cmd in commands: self._execute_command(cmd) return True @@ -272,6 +345,22 @@ class MacroManager: print(f"Error executing macro: {e}") return False + @staticmethod + def _valid_keys(keys) -> list: + """Filter a list of key names to those pyautogui recognises.""" + result = [] + for k in keys: + if not isinstance(k, str): + continue + name = k.strip().lower() + if not name: + continue + if VALID_KEYS and name not in VALID_KEYS: + print(f"Skipping invalid key: {name!r}") + continue + result.append(name) + return result + def _execute_command(self, cmd: dict): """Execute a single command from a sequence.""" cmd_type = cmd.get("type", "") @@ -294,10 +383,11 @@ class MacroManager: pyautogui.typewrite(value, interval=0.02) elif cmd_type == "key": - # Press a single key + # Press a single key (validated against pyautogui's known keys) key = cmd.get("value", "") - if key: - pyautogui.press(key) + valid = self._valid_keys([key]) if key else [] + if valid: + pyautogui.press(valid[0]) elif cmd_type == "hotkey": # Press key combination @@ -306,14 +396,22 @@ class MacroManager: # Ensure keys is a list, not a string if isinstance(keys, str): keys = [k.strip().lower() for k in keys.split(",")] - # Small delay before hotkey for reliability on Windows - time.sleep(0.05) - pyautogui.hotkey(*keys, interval=0.05) + valid = self._valid_keys(keys) + # Only fire if every requested key was valid; a partial hotkey + # is worse than nothing. + if valid and len(valid) == len([k for k in keys if str(k).strip()]): + # Small delay before hotkey for reliability on Windows + time.sleep(0.05) + pyautogui.hotkey(*valid, interval=0.05) elif cmd_type == "wait": - # Delay in milliseconds - ms = cmd.get("ms", 0) + # Delay in milliseconds (bounded, non-negative) + try: + ms = int(cmd.get("ms", 0)) + except (TypeError, ValueError): + ms = 0 if ms > 0: + ms = min(ms, MAX_WAIT_MS) time.sleep(ms / 1000.0) elif cmd_type == "app": @@ -366,8 +464,11 @@ class MacroManager: def get_macro(self, macro_id: str) -> Optional[dict]: """Get a macro by ID.""" - return self.macros.get(macro_id) + with self._lock: + macro = self.macros.get(macro_id) + return copy.deepcopy(macro) if macro is not None else None def get_all_macros(self) -> dict: """Get all macros.""" - return self.macros.copy() + with self._lock: + return copy.deepcopy(self.macros) diff --git a/web_server.py b/web_server.py index 017ca68..5ba71d6 100644 --- a/web_server.py +++ b/web_server.py @@ -1,7 +1,9 @@ # FastAPI web server for MacroPad +import io import os import sys +import uuid import hmac import asyncio from typing import List, Optional @@ -11,10 +13,23 @@ from fastapi import FastAPI, WebSocket, WebSocketDisconnect, UploadFile, File, H from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse, HTMLResponse, JSONResponse from pydantic import BaseModel +from PIL import Image import uvicorn from config import DEFAULT_PORT, VERSION +# Max accepted upload size for images (bytes). +MAX_IMAGE_UPLOAD_BYTES = 5 * 1024 * 1024 + +# Map Pillow image formats to safe file extensions we're willing to persist. +_IMAGE_FORMAT_EXT = { + "PNG": ".png", + "JPEG": ".jpg", + "GIF": ".gif", + "BMP": ".bmp", + "WEBP": ".webp", +} + def get_resource_path(relative_path): """Get the path to a bundled resource file.""" @@ -73,12 +88,16 @@ class ConnectionManager: self.active_connections.remove(websocket) async def broadcast(self, message: dict): - """Send message to all connected clients.""" - for connection in self.active_connections: + """Send message to all connected clients, pruning dead sockets.""" + dead = [] + # Iterate over a snapshot so the list isn't mutated during iteration. + for connection in list(self.active_connections): try: await connection.send_json(message) except Exception: - pass + dead.append(connection) + for connection in dead: + self.disconnect(connection) class WebServer: @@ -190,7 +209,12 @@ class WebServer: @app.post("/api/execute") async def execute_macro(request: ExecuteRequest): """Execute a macro by ID.""" - success = self.macro_manager.execute_macro(request.macro_id) + # execute_macro is synchronous and may sleep (wait steps); run it in + # a thread executor so it doesn't block the event loop / other traffic. + loop = asyncio.get_event_loop() + success = await loop.run_in_executor( + None, self.macro_manager.execute_macro, request.macro_id + ) if success: # Broadcast execution to all connected clients await self.manager.broadcast({ @@ -239,18 +263,60 @@ class WebServer: @app.post("/api/upload-image") async def upload_image(file: UploadFile = File(...)): - """Upload an image for a macro.""" - if not file.content_type or not file.content_type.startswith("image/"): - raise HTTPException(status_code=400, detail="File must be an image") + """Upload an image for a macro. - # Save to temp location - import tempfile - import shutil + Hardened: enforces a size cap, validates the bytes are a real image + with Pillow (content_type is untrusted), and saves into the app's + macro_images/ directory under a server-generated uuid name. Returns + only a relative reference, never an absolute server path. + """ + # Read with a hard size cap (reject oversize before buffering it all). + data = b"" + chunk_size = 64 * 1024 + while True: + chunk = await file.read(chunk_size) + if not chunk: + break + data += chunk + if len(data) > MAX_IMAGE_UPLOAD_BYTES: + raise HTTPException(status_code=413, detail="Image too large") - ext = os.path.splitext(file.filename)[1] if file.filename else ".png" - with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as tmp: - shutil.copyfileobj(file.file, tmp) - return {"path": tmp.name} + if not data: + raise HTTPException(status_code=400, detail="Empty upload") + + # Validate it's really an image (don't trust content_type). + try: + Image.open(io.BytesIO(data)).verify() + # verify() leaves the image unusable; reopen to read the format. + img_format = (Image.open(io.BytesIO(data)).format or "").upper() + except Exception: + raise HTTPException(status_code=400, detail="File is not a valid image") + + ext = _IMAGE_FORMAT_EXT.get(img_format) + if not ext: + raise HTTPException(status_code=400, detail="Unsupported image format") + + images_dir = os.path.join(self.app_dir, "macro_images") + os.makedirs(images_dir, exist_ok=True) + + filename = f"{uuid.uuid4().hex}{ext}" + dest_path = os.path.join(images_dir, filename) + rel_path = os.path.join("macro_images", filename) + + try: + with open(dest_path, "wb") as out: + out.write(data) + except Exception: + # Clean up a partial file on failure. + try: + if os.path.exists(dest_path): + os.remove(dest_path) + except OSError: + pass + raise HTTPException(status_code=500, detail="Failed to save image") + + # Return only a relative, opaque reference. + return {"path": rel_path.replace(os.sep, "/")} @app.get("/api/image/{image_path:path}") async def get_image(image_path: str): @@ -276,11 +342,18 @@ class WebServer: await self.manager.connect(websocket) try: while True: + # receive_json raises on malformed input / non-JSON frames; + # any such error must still tear the connection down cleanly. data = await websocket.receive_json() # Handle incoming messages if needed - if data.get("type") == "ping": + if isinstance(data, dict) and data.get("type") == "ping": await websocket.send_json({"type": "pong"}) except WebSocketDisconnect: + pass + except Exception: + # Malformed input or any other error: don't leak the connection. + pass + finally: self.manager.disconnect(websocket) self.app = app