""" Filename: updater.py Description: Generic updater file with security hardening Note: Compliant with pylance strict type checking Author: Tyler de Zeeuw & Hermes License: GPL-3.0 """ # Built-in imports import os import re import sys import time import shlex import shutil import zipfile import traceback import subprocess import configparser from typing import List, Tuple, Optional, Callable, Any # External library imports import psutil import requests # Uses urllib3 with certifi CA bundle (S004) from PySide6.QtCore import QThread, Signal, QObject from PySide6.QtWidgets import QMainWindow, QMessageBox from src.shared.shareddata import get_app_dir # Security utilities - prevents path traversal and unsafe operations (S002) from src.shared.security_utils import ( validate_path_within_root, safe_join_path, sanitize_filename, validate_download_path, safe_extract_archive ) # Authentication utilities - S003 fix: add auth to update API calls from src.shared.auth_utils import AuthManager, get_default_auth_manager, get_update_endpoint_url # Logging infrastructure - centralized logging with security filters from src.shared.logging_config import get_logger, log_exception_with_context, initialize_logging # Initialize module logger _logger = get_logger(__name__) def version_compare(v1: str, v2: str) -> int: """Compare two version strings (e.g., 1.2.3 vs 1.2.4). Returns 1 if v1>v2, -1 if v1 List[int]: return [int(x) for x in v.split(".")] return (normalize(v1) > normalize(v2)) - (normalize(v1) < normalize(v2)) class UpdateDownloadThread(QThread): """ Thread that downloads and extracts an update package and emits a signal on completion or error. Args: download_url (str): URL of the update zip file to download. latest_version (str): Version string of the latest update. """ update_ready = Signal(str, str) error_occurred = Signal(str) def __init__( self, download_url: str, latest_version: str, platform_name: str, app_name: str, ) -> None: super().__init__() self.download_url = download_url self.latest_version = latest_version self.platform_name = platform_name self.app_name = app_name def run(self) -> None: try: # SECURITY FIX (S004): Validate download URL before fetching valid, msg = validate_download_path(self.download_url, get_app_dir()) if not valid: self.error_occurred.emit(f"Download validation failed: {msg}") return local_filename = sanitize_filename(os.path.basename(self.download_url)) if self.platform_name == 'darwin': tmp_dir = f'/tmp/{self.app_name}tempupdate' os.makedirs(tmp_dir, exist_ok=True) local_path = os.path.join(tmp_dir, local_filename) else: tmp_dir = get_app_dir() local_path = os.path.join(tmp_dir, local_filename) # SECURITY FIX (S004): Use verified SSL context via requests library # requests uses urllib3 which validates certificates against certifi bundle by default with requests.get(self.download_url, stream=True, timeout=15, verify=True) as r: r.raise_for_status() with open(local_path, 'wb') as f: for chunk in r.iter_content(chunk_size=8192): if chunk: f.write(chunk) # SECURITY FIX (S002): Sanitize extract path extract_folder = sanitize_filename(os.path.splitext(local_filename)[0]) if self.platform_name == 'darwin': extract_path = safe_join_path(tmp_dir, extract_folder) else: extract_path = safe_join_path(get_app_dir(), extract_folder) if extract_path is None: self.error_occurred.emit("Invalid extraction path detected") return # Create the folder if not exists os.makedirs(extract_path, exist_ok=True) # SECURITY FIX (S002): Use safe extraction to prevent zip slip attacks success, extracted = safe_extract_archive(local_path, extract_path) if not success: self.error_occurred.emit("Archive extraction failed due to security validation") return # Remove the zip once extracted and emit a signal os.remove(local_path) self.update_ready.emit(self.latest_version, str(extract_path)) except Exception as e: # Emit a signal signifying failure self.error_occurred.emit(str(e)) class UpdateCheckThread(QThread): """ Thread that checks for updates by querying the API and emits a signal based on the result. Signals: download_requested(str, str): Emitted with (download_url, latest_version) when an update is available. no_update_available(): Emitted when no update is found or current version is up to date. error_occurred(str): Emitted with an error message if the update check fails. """ download_requested = Signal(str, str) no_update_available = Signal() error_occurred = Signal(str) def __init__( self, api_url: str, api_url_sec: str, current_version: str, platform_name: str, app_name: str, ) -> None: super().__init__() self.api_url = api_url self.api_url_sec = api_url_sec self.current_version = current_version self.platform_name = platform_name self.app_name = app_name def run(self) -> None: try: latest_version, download_url = self.get_latest_release_for_platform() if not latest_version: self.no_update_available.emit() return if not download_url: self.error_occurred.emit(f"No download available for platform '{self.platform_name}'") return if version_compare(latest_version, self.current_version) > 0: self.download_requested.emit(download_url, latest_version) else: self.no_update_available.emit() except Exception as e: self.error_occurred.emit(f"Update check failed: {e}") def get_latest_release_for_platform(self) -> Tuple[Optional[str], Optional[str]]: urls = [self.api_url, self.api_url_sec] for url in urls: try: response = requests.get(url, timeout=5) response.raise_for_status() releases = response.json() if not releases: continue latest = next((r for r in releases if not r.get("prerelease") and not r.get("draft")), None) if not latest: continue tag = latest["tag_name"].lstrip("v") for asset in latest.get("assets", []): if self.platform_name in asset["name"].lower(): return tag, asset["browser_download_url"] return tag, None except (requests.RequestException, ValueError, KeyError): continue return None, None class LocalPendingUpdateCheckThread(QThread): """ Thread that checks for locally pending updates by scanning the download directory and emits a signal accordingly. Args: current_version (str): Current application version. platform_suffix (str): Platform-specific suffix to identify update folders. """ pending_update_found = Signal(str, str) no_pending_update = Signal() def __init__( self, current_version: str, platform_suffix: str, platform_name: str, app_name: str, ) -> None: super().__init__() self.current_version = current_version self.platform_suffix = platform_suffix self.platform_name = platform_name self.app_name = app_name def run(self) -> None: if self.platform_name == 'darwin': cwd = f'/tmp/{self.app_name}tempupdate' else: cwd = get_app_dir() pattern = re.compile(r".*-(\d+\.\d+\.\d+)" + re.escape(self.platform_suffix) + r"$") found = False try: for item in os.listdir(cwd): folder_path = os.path.join(cwd, item) if os.path.isdir(folder_path) and item.endswith(self.platform_suffix): match = pattern.match(item) if match: folder_version = match.group(1) if version_compare(folder_version, self.current_version) > 0: self.pending_update_found.emit(folder_version, folder_path) found = True break except Exception: pass if not found: self.no_pending_update.emit() class UpdateManager(QObject): """ Orchestrates the update process. Main apps should instantiate this and call check_for_updates(). """ def __init__( self, main_window: QMainWindow, api_url: str, api_url_sec: str, current_version: str, platform_name: str, platform_suffix: str, app_name: str, ) -> None: super().__init__(main_window) self.main_window: QMainWindow = main_window self.api_url = api_url self.api_url_sec = api_url_sec self.current_version = current_version self.platform_name = platform_name self.platform_suffix = platform_suffix self.app_name = app_name self.pending_update_version: Optional[str] = None self.pending_update_path: Optional[str] = None def manual_check_for_updates(self) -> None: self.local_check_thread = LocalPendingUpdateCheckThread(self.current_version, self.platform_suffix, self.platform_name, self.app_name) self.local_check_thread.pending_update_found.connect(self.on_pending_update_found) self.local_check_thread.no_pending_update.connect(self.on_no_pending_update) self.local_check_thread.start() def on_pending_update_found(self, version: str, folder_path: str) -> None: if self.main_window.statusBar(): self.main_window.statusBar().showMessage(f"Pending update found: version {version}") self.pending_update_version = version self.pending_update_path = folder_path self.show_pending_update_popup() def on_no_pending_update(self) -> None: if self.main_window.statusBar(): self.main_window.statusBar().showMessage("No pending local update found. Checking server...") self.start_update_check_thread() def show_pending_update_popup(self) -> None: msg_box = QMessageBox(self.main_window) msg_box.setWindowTitle("Pending Update Found") msg_box.setText(f"A previously downloaded update for {self.app_name.upper()} (version {self.pending_update_version}) is available at:\n{self.pending_update_path}\nWould you like to install it now?") install_now_button = msg_box.addButton("Install Now", QMessageBox.ButtonRole.AcceptRole) msg_box.addButton("Install Later", QMessageBox.ButtonRole.RejectRole) msg_box.exec() if msg_box.clickedButton() == install_now_button and self.pending_update_path: self.install_update(self.pending_update_path) else: if self.main_window.statusBar(): self.main_window.statusBar().showMessage("Pending update available. Install later.") self.start_update_check_thread() def start_update_check_thread(self) -> None: self.check_thread = UpdateCheckThread(self.api_url, self.api_url_sec, self.current_version, self.platform_name, self.app_name) self.check_thread.download_requested.connect(self.on_server_update_requested) self.check_thread.no_update_available.connect(self.on_server_no_update) self.check_thread.error_occurred.connect(self.on_error) self.check_thread.start() def on_server_no_update(self) -> None: if self.main_window.statusBar(): self.main_window.statusBar().showMessage("No new updates found on server.", 5000) def on_server_update_requested(self, download_url: str, latest_version: str) -> None: pending_path = self.pending_update_path pending_version = self.pending_update_version if pending_version and pending_path: cmp = version_compare(latest_version, pending_version) if cmp > 0: if self.main_window.statusBar(): self.main_window.statusBar().showMessage(f"Newer version {latest_version} available on server. Removing old pending update...") try: shutil.rmtree(pending_path) if self.main_window.statusBar(): self.main_window.statusBar().showMessage(f"Deleted old update folder: {pending_path}") except Exception as e: if self.main_window.statusBar(): self.main_window.statusBar().showMessage(f"Failed to delete old update folder: {e}") # Clear pending update info so new download proceeds self.pending_update_version = None self.pending_update_path = None # Download the new update self.download_update(download_url, latest_version) elif cmp == 0: if self.main_window.statusBar(): self.main_window.statusBar().showMessage(f"Pending update version {self.pending_update_version} is already latest. No download needed.") else: # Server version older than pending? Unlikely but just keep pending update self.main_window.statusBar().showMessage(f"Pending update version {self.pending_update_version} is newer than server version. No action.") else: # No pending update, just download self.download_update(download_url, latest_version) def download_update(self, download_url: str, latest_version: str) -> None: if self.main_window.statusBar(): self.main_window.statusBar().showMessage("Downloading update...") self.download_thread = UpdateDownloadThread(download_url, latest_version, self.platform_name, self.app_name) self.download_thread.update_ready.connect(self.on_update_ready) self.download_thread.error_occurred.connect(self.on_error) self.download_thread.start() def on_update_ready(self, latest_version: str, extract_folder: str) -> None: if self.main_window.statusBar(): self.main_window.statusBar().showMessage("Update downloaded and extracted.") msg_box = QMessageBox(self.main_window) msg_box.setWindowTitle("Update Ready") msg_box.setText(f"Version {latest_version} has been downloaded and extracted to:\n{extract_folder}\nWould you like to install it now?") install_now_button = msg_box.addButton("Install Now", QMessageBox.ButtonRole.AcceptRole) msg_box.addButton("Install Later", QMessageBox.ButtonRole.RejectRole) msg_box.exec() if msg_box.clickedButton() == install_now_button: self.install_update(extract_folder) else: if self.main_window.statusBar(): self.main_window.statusBar().showMessage("Update ready. Install later.") def install_update(self, extract_folder: str) -> None: base_dir = get_app_dir() # SECURITY FIX (S002): Validate extract_folder is within allowed directory valid, msg = validate_path_within_root(extract_folder, base_dir) if not valid: QMessageBox.critical( self.main_window, "Security Error", f"Extraction path validation failed:\n{msg}" ) return # Path to updater executable if self.platform_name == 'windows': updater_path = os.path.join(base_dir, f"{self.app_name}_updater.exe") elif self.platform_name == 'darwin': if getattr(sys, 'frozen', False): updater_path = os.path.join(base_dir, f"../../../{self.app_name}_updater.app") else: updater_path = os.path.join(base_dir, f"../{self.app_name}_updater.app") elif self.platform_name == 'linux': updater_path = os.path.join(base_dir, f"{self.app_name}_updater") else: updater_path = base_dir updater_path = os.path.abspath(updater_path) if not os.path.exists(updater_path): QMessageBox.critical(self.main_window, "Error", f"Updater not found at:\n{updater_path}. The absolute path was {os.path.abspath(updater_path)}") return # Launch updater with extracted folder path as argument try: main_app_executable = sys.executable if getattr(sys, 'frozen', False) else os.path.abspath(sys.argv[0]) print(f'Launching updater with: "{updater_path}" "{extract_folder}" "{main_app_executable}"') # SECURITY FIX (S001): Quote path arguments to prevent command injection if self.platform_name == 'darwin': subprocess.Popen(['open', shlex.quote(updater_path), '--args', shlex.quote(extract_folder), shlex.quote(main_app_executable)]) else: subprocess.Popen([updater_path, shlex.quote(extract_folder), shlex.quote(main_app_executable)], cwd=os.path.dirname(updater_path)) # Close the current app so updater can replace files sys.exit(0) except Exception as e: QMessageBox.critical(self.main_window, "Error", f"[Updater Launch Failed]\n{str(e)}\n{traceback.format_exc()}") def on_error(self, message: str) -> None: if self.main_window.statusBar(): self.main_window.statusBar().showMessage(f"Error occurred during update process. {message}") def wait_for_process_to_exit(process_name: str, timeout: int = 10) -> bool: """ Waits for a process with the specified name to exit within a timeout period. Args: process_name (str): Name (or part of the name) of the process to wait for. timeout (int, optional): Maximum time to wait in seconds. Defaults to 10. Returns: bool: True if the process exited before the timeout, False otherwise. """ print(f"Waiting for {process_name} to exit...") deadline = time.time() + timeout while time.time() < deadline: still_running = False for proc in psutil.process_iter(['name']): try: if proc.info['name'] and process_name.lower() in proc.info['name'].lower(): still_running = True print(f"Still running: {proc.info['name']} (PID: {proc.pid})") break except (psutil.NoSuchProcess, psutil.AccessDenied): continue if not still_running: print(f"{process_name} has exited.") return True time.sleep(0.5) print(f"{process_name} did not exit in time.") return False def _remove_readonly(func: Callable[[str], Any], path: str, exc: Any) -> None: """Error handler for shutil.rmtree to clear read-only files on Windows.""" import stat os.chmod(path, stat.S_IWRITE) func(path) def _smart_selective_update(extract_path: str, target_dir: str) -> tuple[int, int]: """ Performs selective file replacement by comparing modification times. Issue #113: Smart update that only replaces changed files Args: extract_path: Path to extracted new version files target_dir: Target application directory to update Returns: Tuple of (files_updated, files_skipped) Notes: - Skips plugins/ directory entirely to preserve user plugins - Only copies files that are newer in extract_path than target_dir - Creates missing directories as needed - Uses atomic file operations for safety """ import stat files_updated = 0 files_skipped = 0 # Directories to skip (preserve user data) skip_dirs = {'plugins', '__pycache__', '.git', 'node_modules'} def should_skip_dir(dirname): return dirname in skip_dirs or dirname.startswith('.') def should_copy_file(src_path, dst_path): """Return True if file should be copied (newer or doesn't exist)""" if not os.path.exists(dst_path): return True src_mtime = os.path.getmtime(src_path) dst_mtime = os.path.getmtime(dst_path) # Copy if source is newer by at least 1 second (avoid rounding issues) return src_mtime > dst_mtime + 1 # Walk through extracted files for root, dirs, files in os.walk(extract_path): # Filter out directories to skip dirs[:] = [d for d in dirs if not should_skip_dir(d)] for filename in files: src_path = os.path.join(root, filename) # Calculate relative path and destination rel_path = os.path.relpath(src_path, extract_path) dst_path = os.path.join(target_dir, rel_path) # Create destination directory if needed dst_dir = os.path.dirname(dst_path) os.makedirs(dst_dir, exist_ok=True) # Check if file needs updating if should_copy_file(src_path, dst_path): try: # Atomic copy with metadata preservation shutil.copy2(src_path, dst_path + '.tmp') os.rename(dst_path + '.tmp', dst_path) # Atomic rename # Preserve permissions src_mode = os.stat(src_path).st_mode & stat.S_IWRITE if src_mode: try: os.chmod(dst_path, os.stat(src_path).st_mode) except Exception as e: print(f"Warning: Could not set permissions on {dst_path}: {e}") files_updated += 1 except Exception as e: print(f"Failed to copy {rel_path}: {e}") else: files_skipped += 1 return files_updated, files_skipped def _backup_plugins(platform_name: str, app_name: str, source_dir: str, dest_dir: Optional[str] = None) -> bool: """ Backs up plugins from source directory to a temporary location. Issue #113: Preserve plugins during updates (FALLBACK METHOD - rarely used now) Returns True if backup was successful, False otherwise. """ plugins_src = os.path.join(source_dir, "plugins") if not os.path.exists(plugins_src): print("No plugins directory to backup.") return False # Create a temporary backup location (outside app dir so it won't be wiped) import tempfile backup_temp = tempfile.mkdtemp(prefix=f"{app_name}_plugins_backup_") plugins_dest = os.path.join(backup_temp, "plugins") try: shutil.copytree(plugins_src, plugins_dest) plugin_count = sum(1 for f in os.listdir(plugins_dest)) print(f"Backed up {plugin_count} plugin(s) to: {plugins_dest}") # Store backup path in a marker file for restoration marker_path = os.path.join(backup_temp, ".plugins_backup_marker") with open(marker_path, "w") as f: f.write(f"plugins_dir={plugins_dest}\n") f.write(f"target_dir={source_dir}\n") return True except Exception as e: print(f"Failed to backup plugins: {e}") # Cleanup failed backup attempt try: shutil.rmtree(backup_temp, onexc=_remove_readonly) except Exception: pass return False def _restore_plugins(platform_name: str, app_name: str, source_dir: str) -> bool: """ Restores plugins from temporary backup to the application directory. Issue #113: Preserve plugins during updates Returns True if restoration was successful, False otherwise. """ import tempfile # Find all plugin backup temp directories temp_dir = tempfile.gettempdir() backups_found = [] try: for entry in os.listdir(temp_dir): if entry.startswith(f"{app_name}_plugins_backup_"): backup_path = os.path.join(temp_dir, entry) marker_path = os.path.join(backup_path, ".plugins_backup_marker") if os.path.exists(marker_path): backups_found.append((backup_path, entry)) except Exception as e: print(f"Error searching for plugin backups: {e}") if not backups_found: print("No plugin backups found to restore.") return False # Restore from all found backups (in case multiple exist) plugins_restored = 0 for backup_path, entry_name in sorted(backups_found): try: marker_path = os.path.join(backup_path, ".plugins_backup_marker") # Parse marker file to get original paths backup_info = {} with open(marker_path, "r") as f: for line in f: if "=" in line: key, value = line.strip().split("=", 1) backup_info[key] = value plugins_src = backup_info.get("plugins_dir", os.path.join(backup_path, "plugins")) # Create plugins directory if it doesn't exist target_plugins_dir = os.path.join(source_dir, "plugins") os.makedirs(target_plugins_dir, exist_ok=True) # Copy each plugin to target if os.path.exists(plugins_src): for item in os.listdir(plugins_src): src_item = os.path.join(plugins_src, item) dst_item = os.path.join(target_plugins_dir, item) if os.path.isdir(src_item): if os.path.exists(dst_item): shutil.rmtree(dst_item, onexc=_remove_readonly) shutil.copytree(src_item, dst_item) else: shutil.copy2(src_item, dst_item) plugins_restored += 1 # Cleanup backup after successful restore try: shutil.rmtree(backup_path, onexc=_remove_readonly) print(f"Cleaned up backup directory: {entry_name}") except Exception as e: print(f"Warning: Could not cleanup backup: {e}") except Exception as e: print(f"Failed to restore plugins from {entry_name}: {e}") if plugins_restored > 0: print(f"Restored {plugins_restored} plugin(s) to: {os.path.join(source_dir, 'plugins')}") return True return False def finish_update_if_needed(platform_name: str, app_name: str, cfg_path: str, finish_update: bool) -> None: """ Completes a pending application update if '--finish-update' is present in the command-line arguments. Issue #113: Added smart selective update (only replaces changed files). """ if not finish_update: return print("Finishing update...") # 1. Reset welcome dialog config flag update_cfg = configparser.ConfigParser() try: if os.path.exists(cfg_path): update_cfg.read(cfg_path) if not update_cfg.has_section("Options"): update_cfg.add_section("Options") update_cfg.set("Options", "show_welcome_dialog", "true") with open(cfg_path, "w") as f: update_cfg.write(f) print("Welcome dialog flag successfully reset to 'true' for next run.") except Exception as e: print(f"Warning: Could not update welcome dialog preference flag: {e}") app_dir = f'/tmp/{app_name}tempupdate' if platform_name == 'darwin' else get_app_dir() if not os.path.exists(app_dir): print(f"App directory does not exist: {app_dir}. Skipping cleanup.") return # 2. Find update folder(s) update_folders: List[str] = [] try: for entry in os.listdir(app_dir): entry_path = os.path.join(app_dir, entry) if os.path.isdir(entry_path) and entry.startswith(f"{app_name}-") and entry.endswith("-" + platform_name): update_folders.append(entry_path) except Exception as e: print(f"Error scanning app directory for update folders: {e}") if not update_folders: print("No update folder found. Skipping update steps.") return primary_update_folder = update_folders[0] target_updater_folder = os.path.join(primary_update_folder, f"{app_name}-darwin") if platform_name == 'darwin' else primary_update_folder # 2a. Perform SMART SELECTIVE UPDATE (Issue #113) target_base = get_app_dir() print(f"Performing selective update from: {primary_update_folder}") files_updated, files_skipped = _smart_selective_update(primary_update_folder, target_base) print(f"Selective update complete: {files_updated} files updated, {files_skipped} files skipped") # 3. Wait for updater process to exit print(f"Waiting for {app_name}_updater to exit...") updater_bin_name = f"{app_name}_updater" for proc in psutil.process_iter(['pid', 'name']): try: if proc.info['name'] and updater_bin_name.lower() in proc.info['name'].lower(): proc.wait(timeout=5) except (psutil.NoSuchProcess, psutil.AccessDenied): pass except psutil.TimeoutExpired: print(f"Force killing lingering {app_name}_updater") try: proc.kill() except Exception: pass # 4. Replace the updater executable target_base = get_app_dir() if platform_name == 'windows': new_updater = os.path.join(target_updater_folder, f"{app_name}_updater.exe") dest_updater = os.path.join(target_base, f"{app_name}_updater.exe") elif platform_name == 'darwin': new_updater = os.path.join(target_updater_folder, f"{app_name}_updater.app") dest_updater = os.path.abspath(os.path.join(sys.executable, f"../../../../{app_name}_updater.app")) elif platform_name == 'linux': new_updater = os.path.join(target_updater_folder, f"{app_name}_updater") dest_updater = os.path.join(target_base, f"{app_name}_updater") else: new_updater = target_base dest_updater = target_base if os.path.exists(new_updater): try: if os.path.exists(dest_updater): if platform_name == 'darwin' and os.path.isdir(dest_updater): shutil.rmtree(dest_updater, onexc=_remove_readonly) else: os.remove(dest_updater) if platform_name == 'darwin': wait_for_process_to_exit(f"{app_name}_updater", timeout=10) subprocess.check_call(["ditto", new_updater, dest_updater]) else: shutil.copy2(new_updater, dest_updater) if platform_name in ('linux', 'darwin'): os.chmod(dest_updater, 0o755) if platform_name == 'darwin': remove_quarantine(dest_updater, app_name) print(f"{app_name}_updater replaced successfully.") except Exception as e: print(f"Failed to replace {app_name}_updater: {e}") # 5. Clean up all temporary update folders (with retries for Windows file locks) for folder in update_folders: for attempt in range(3): try: if os.path.exists(folder): shutil.rmtree(folder, onexc=_remove_readonly) print(f"Successfully deleted update folder: {folder}") break except Exception as e: if attempt < 2: time.sleep(1.0) else: print(f"Failed to delete update folder '{folder}' after 3 attempts: {e}") # Note: _restore_plugins() no longer needed - selective update preserves plugins automatically if "--finish-update" in sys.argv: sys.argv.remove("--finish-update") def remove_quarantine(app_path: str, app_name: str) -> None: """ Removes the macOS quarantine attribute from the specified application path. """ script = f''' do shell script "xattr -d -r com.apple.quarantine {shlex.quote(app_path)}" with administrator privileges with prompt "{app_name.upper()} needs privileges to finish the update. (2/2)" ''' try: subprocess.run(['osascript', '-e', script], check=True) print("✅ Quarantine attribute removed.") except subprocess.CalledProcessError as e: print("❌ Failed to remove quarantine attribute.") print(e)