"""Update Manager for Dangerous Pi. Handles checking for updates from GitHub releases, downloading, and applying updates to the system. Supports per-component updates for pm3, frontend, backend, and theme components. """ import asyncio import hashlib import json import os import platform import re import shutil import sys import tempfile from dataclasses import dataclass, field, asdict from datetime import datetime, timedelta from enum import Enum from pathlib import Path from typing import Optional, Dict, Any, List import aiohttp from .. import config # --------------------------------------------------------------------------- # Enums # --------------------------------------------------------------------------- class UpdateStatus(str, Enum): """Update status enum.""" IDLE = "idle" CHECKING = "checking" AVAILABLE = "available" DOWNLOADING = "downloading" INSTALLING = "installing" COMPLETE = "complete" FAILED = "failed" ROLLING_BACK = "rolling_back" class ComponentId(str, Enum): """Known updateable component identifiers.""" PM3 = "pm3" FRONTEND = "frontend" BACKEND = "backend" THEME = "theme" # --------------------------------------------------------------------------- # Component manifest dataclasses # --------------------------------------------------------------------------- @dataclass class ComponentInfo: """Information about an installed component.""" component_id: str version: str installed_at: str checksum: Optional[str] = None install_path: str = "" platform: Optional[str] = None python_version: Optional[str] = None metadata: Dict[str, Any] = field(default_factory=dict) @dataclass class SystemManifest: """Complete manifest of all installed components.""" schema_version: int = 1 system_version: str = "0.1.0" components: Dict[str, ComponentInfo] = field(default_factory=dict) plugins: Dict[str, Dict[str, Any]] = field(default_factory=dict) last_updated: str = "" # --------------------------------------------------------------------------- # Release / progress dataclasses # --------------------------------------------------------------------------- @dataclass class AvailableComponent: """A component update available for download.""" component_id: str current_version: Optional[str] available_version: str download_url: str asset_name: str asset_size: int checksum: str changelog: str platform: Optional[str] = None python_version: Optional[str] = None compatible: bool = True incompatible_reason: Optional[str] = None @dataclass class ReleaseInfo: """GitHub release information.""" version: str tag_name: str published_at: str changelog: str is_prerelease: bool components: Dict[str, AvailableComponent] = field(default_factory=dict) # Legacy fields for backward compatibility download_url: str = "" asset_name: str = "" asset_size: int = 0 checksum: Optional[str] = None @dataclass class ComponentProgress: """Progress for a single component update.""" component_id: str status: UpdateStatus download_progress: float = 0.0 error_message: Optional[str] = None @dataclass class UpdateProgress: """Update progress information.""" status: UpdateStatus current_version: str available_version: Optional[str] = None download_progress: float = 0.0 error_message: Optional[str] = None last_check: Optional[str] = None component_progress: Dict[str, ComponentProgress] = field(default_factory=dict) active_component: Optional[str] = None class UpdateManager: """Manages system updates from GitHub releases. Supports per-component updates for pm3, frontend, backend, and theme. Tracks installed component versions via a manifest file. """ MANIFEST_PATH = Path(config.COMPONENT_MANIFEST_PATH) BACKUP_BASE = Path("/opt/dangerous-pi/backups") INSTALL_DIR = Path("/opt/dangerous-pi") def __init__(self): """Initialize the update manager.""" self._current_version = config.VERSION self._github_repo = config.GITHUB_REPO self._github_api_base = "https://api.github.com" self._status = UpdateStatus.IDLE self._progress = UpdateProgress( status=UpdateStatus.IDLE, current_version=self._current_version ) self._latest_release: Optional[ReleaseInfo] = None self._available_components: Dict[str, AvailableComponent] = {} self._update_lock = asyncio.Lock() self._download_path: Optional[Path] = None self._check_interval = config.UPDATE_CHECK_INTERVAL # Load or create component manifest self._manifest = self._load_manifest() self._current_version = self._manifest.system_version # ------------------------------------------------------------------ # Manifest I/O # ------------------------------------------------------------------ def _load_manifest(self) -> SystemManifest: """Load component manifest from disk, migrating from legacy if needed.""" if self.MANIFEST_PATH.exists(): try: data = json.loads(self.MANIFEST_PATH.read_text()) return self._parse_manifest(data) except (json.JSONDecodeError, KeyError, TypeError) as e: print(f"Warning: failed to parse manifest, migrating: {e}") return self._migrate_from_legacy() def _parse_manifest(self, data: Dict[str, Any]) -> SystemManifest: """Parse a manifest dict into a SystemManifest.""" components = {} for comp_id, comp_data in data.get("components", {}).items(): components[comp_id] = ComponentInfo( component_id=comp_data.get("component_id", comp_id), version=comp_data.get("version", "unknown"), installed_at=comp_data.get("installed_at", ""), checksum=comp_data.get("checksum"), install_path=comp_data.get("install_path", ""), platform=comp_data.get("platform"), python_version=comp_data.get("python_version"), metadata=comp_data.get("metadata", {}), ) return SystemManifest( schema_version=data.get("schema_version", 1), system_version=data.get("system_version", config.VERSION), components=components, plugins=data.get("plugins", {}), last_updated=data.get("last_updated", ""), ) def _save_manifest(self) -> None: """Write the current manifest to disk.""" self._manifest.last_updated = datetime.utcnow().isoformat() data = { "schema_version": self._manifest.schema_version, "system_version": self._manifest.system_version, "components": { comp_id: asdict(comp) for comp_id, comp in self._manifest.components.items() }, "plugins": self._manifest.plugins, "last_updated": self._manifest.last_updated, } # Ensure parent directory exists self.MANIFEST_PATH.parent.mkdir(parents=True, exist_ok=True) self.MANIFEST_PATH.write_text(json.dumps(data, indent=2)) def _migrate_from_legacy(self) -> SystemManifest: """Create a component manifest from legacy VERSION files.""" # Read legacy system version legacy_version_file = self.INSTALL_DIR / "VERSION" if legacy_version_file.exists(): system_version = legacy_version_file.read_text().strip() else: system_version = config.VERSION # Detect installed PM3 version pm3_home = Path(config.PM3_CLIENT_PATH).parent.parent pm3_version_file = pm3_home / "VERSION.txt" pm3_version = "unknown" if pm3_version_file.exists(): pm3_version = pm3_version_file.read_text().strip().split("\n")[0] now = datetime.utcnow().isoformat() py_version = f"{sys.version_info.major}.{sys.version_info.minor}" manifest = SystemManifest( schema_version=1, system_version=system_version, components={ ComponentId.PM3: ComponentInfo( component_id=ComponentId.PM3, version=pm3_version, installed_at=now, install_path=str(pm3_home), platform=f"{platform.machine()}-linux", python_version=py_version, metadata={"migrated_from_legacy": True}, ), ComponentId.FRONTEND: ComponentInfo( component_id=ComponentId.FRONTEND, version=system_version, installed_at=now, install_path=str(self.INSTALL_DIR / "app" / "frontend" / "build"), metadata={"migrated_from_legacy": True}, ), ComponentId.BACKEND: ComponentInfo( component_id=ComponentId.BACKEND, version=system_version, installed_at=now, install_path=str(self.INSTALL_DIR / "app" / "backend"), metadata={"migrated_from_legacy": True}, ), ComponentId.THEME: ComponentInfo( component_id=ComponentId.THEME, version=system_version, installed_at=now, install_path=str(self.INSTALL_DIR / "app" / "frontend" / "themes"), metadata={"migrated_from_legacy": True}, ), }, last_updated=now, ) self._manifest = manifest self._save_manifest() print(f"✅ Migrated to component manifest (system v{system_version})") return manifest # ------------------------------------------------------------------ # Manifest accessors # ------------------------------------------------------------------ def get_manifest(self) -> SystemManifest: """Return the current system manifest.""" return self._manifest def get_component_version(self, component_id: str) -> Optional[str]: """Get the installed version of a component.""" comp = self._manifest.components.get(component_id) return comp.version if comp else None @staticmethod def get_system_info() -> Dict[str, str]: """Detect the current system's platform and Python version.""" return { "arch": platform.machine(), "python_version": f"{sys.version_info.major}.{sys.version_info.minor}", "python_abi": f"cp{sys.version_info.major}{sys.version_info.minor}", "os": "linux", } # ------------------------------------------------------------------ # Periodic checks # ------------------------------------------------------------------ async def start_periodic_checks(self): """Start periodic update checks in background.""" while True: try: await self.check_for_updates() await asyncio.sleep(self._check_interval) except asyncio.CancelledError: break except Exception as e: print(f"Error in periodic update check: {e}") await asyncio.sleep(self._check_interval) # ------------------------------------------------------------------ # Check for updates (component-aware) # ------------------------------------------------------------------ async def check_for_updates(self) -> Dict[str, Any]: """Check for available updates from GitHub. Compares each component's installed version against the latest release manifest. Returns per-component update availability. """ async with self._update_lock: try: self._status = UpdateStatus.CHECKING self._progress.status = UpdateStatus.CHECKING release = await self._fetch_latest_release() if not release: self._status = UpdateStatus.IDLE self._progress.status = UpdateStatus.IDLE self._progress.last_check = datetime.utcnow().isoformat() return { "update_available": False, "current_version": self._current_version, "message": "No releases found", "components": [], } # Trigger plugin hooks (non-critical) try: from .plugin_manager import get_plugin_manager await get_plugin_manager().trigger_hook("update_check") except Exception: pass # Check for plugin updates plugin_updates = [] try: from .plugin_manager import get_plugin_manager pm = get_plugin_manager() plugin_update_infos = await pm.check_plugin_updates() for pu in plugin_update_infos: plugin_updates.append({ "plugin_id": pu.plugin_id, "current_version": pu.current_version, "available_version": pu.available_version, "changelog": pu.changelog, "download_size": pu.download_size, "source_repo": pu.source_repo, }) except Exception as e: print(f"Warning: plugin update check failed: {e}") # Build per-component update list self._latest_release = release self._available_components = {} component_updates = [] any_update = False for comp_id, available in release.components.items(): installed_ver = self.get_component_version(comp_id) available.current_version = installed_ver if installed_ver and not self._is_newer_version( available.available_version, installed_ver ): continue # Already up to date self._available_components[comp_id] = available any_update = True component_updates.append({ "component_id": comp_id, "current_version": installed_ver, "available_version": available.available_version, "changelog": available.changelog, "download_size": available.asset_size, "compatible": available.compatible, "incompatible_reason": available.incompatible_reason, }) # Also check system-level version (legacy compat) system_update = self._is_newer_version( release.version, self._current_version ) now = datetime.utcnow().isoformat() self._progress.last_check = now if any_update or system_update: self._status = UpdateStatus.AVAILABLE self._progress.status = UpdateStatus.AVAILABLE self._progress.available_version = release.version else: self._status = UpdateStatus.IDLE self._progress.status = UpdateStatus.IDLE return { "update_available": any_update or system_update or len(plugin_updates) > 0, "current_version": self._current_version, "latest_version": release.version, "release_date": release.published_at, "changelog": release.changelog, "is_prerelease": release.is_prerelease, "download_size": release.asset_size, "components": component_updates, "plugins": plugin_updates, } except Exception as e: self._status = UpdateStatus.FAILED self._progress.status = UpdateStatus.FAILED self._progress.error_message = str(e) raise Exception(f"Failed to check for updates: {e}") # ------------------------------------------------------------------ # Download (component-level and legacy) # ------------------------------------------------------------------ async def download_component(self, component_id: str) -> bool: """Download a specific component update.""" async with self._update_lock: available = self._available_components.get(component_id) if not available: raise ValueError(f"No update available for {component_id}") if not available.compatible: raise ValueError( f"Component {component_id} is not compatible: " f"{available.incompatible_reason}" ) try: self._progress.active_component = component_id comp_progress = ComponentProgress( component_id=component_id, status=UpdateStatus.DOWNLOADING, ) self._progress.component_progress[component_id] = comp_progress self._status = UpdateStatus.DOWNLOADING self._progress.status = UpdateStatus.DOWNLOADING temp_dir = Path(tempfile.mkdtemp( prefix=f"dangerous-pi-{component_id}-" )) download_path = temp_dir / available.asset_name async with aiohttp.ClientSession() as session: async with session.get(available.download_url) as response: if response.status != 200: raise Exception( f"Download failed with status {response.status}" ) total_size = int( response.headers.get('content-length', 0) ) downloaded = 0 with open(download_path, 'wb') as f: async for chunk in response.content.iter_chunked(8192): f.write(chunk) downloaded += len(chunk) if total_size > 0: pct = (downloaded / total_size) * 100 comp_progress.download_progress = pct self._progress.download_progress = pct # Verify checksum if available.checksum: if not self._verify_checksum(download_path, available.checksum): raise Exception("Checksum verification failed") comp_progress.download_progress = 100.0 comp_progress.status = UpdateStatus.IDLE self._progress.download_progress = 100.0 # Store path for install step available.metadata = {"download_path": str(download_path)} # type: ignore[attr-defined] return True except Exception as e: comp_progress.status = UpdateStatus.FAILED comp_progress.error_message = str(e) self._status = UpdateStatus.FAILED self._progress.status = UpdateStatus.FAILED self._progress.error_message = str(e) if 'temp_dir' in locals(): shutil.rmtree(temp_dir, ignore_errors=True) raise Exception(f"Failed to download {component_id}: {e}") async def download_update(self) -> bool: """Download all available component updates (legacy compat).""" for comp_id in list(self._available_components.keys()): available = self._available_components[comp_id] if available.compatible: await self.download_component(comp_id) return True # ------------------------------------------------------------------ # Install (component-level and legacy) # ------------------------------------------------------------------ async def install_component(self, component_id: str) -> bool: """Install a downloaded component update.""" async with self._update_lock: available = self._available_components.get(component_id) if not available: raise ValueError(f"No update available for {component_id}") download_path_str = getattr(available, 'metadata', {}).get( 'download_path' ) if hasattr(available, 'metadata') else None if not download_path_str: raise ValueError(f"Component {component_id} not downloaded yet") download_path = Path(download_path_str) if not download_path.exists(): raise ValueError(f"Download file missing for {component_id}") try: self._progress.active_component = component_id comp_progress = self._progress.component_progress.get( component_id, ComponentProgress(component_id=component_id, status=UpdateStatus.INSTALLING) ) comp_progress.status = UpdateStatus.INSTALLING self._progress.component_progress[component_id] = comp_progress self._status = UpdateStatus.INSTALLING self._progress.status = UpdateStatus.INSTALLING # Backup current component await self._backup_component(component_id) # Extract to staging directory staging_dir = Path(tempfile.mkdtemp( prefix=f"dangerous-pi-stage-{component_id}-" )) await self._run_command( f"tar -xzf {download_path} -C {staging_dir}" ) # Run component-specific installer installer = self._get_installer(component_id) await installer(staging_dir, available) # Update manifest now = datetime.utcnow().isoformat() comp_info = self._manifest.components.get( component_id, ComponentInfo( component_id=component_id, version="unknown", installed_at=now, ) ) comp_info.version = available.available_version comp_info.installed_at = now comp_info.checksum = available.checksum self._manifest.components[component_id] = comp_info self._save_manifest() # Cleanup shutil.rmtree(download_path.parent, ignore_errors=True) shutil.rmtree(staging_dir, ignore_errors=True) comp_progress.status = UpdateStatus.COMPLETE self._status = UpdateStatus.COMPLETE self._progress.status = UpdateStatus.COMPLETE # Remove from available self._available_components.pop(component_id, None) return True except Exception as e: comp_progress.status = UpdateStatus.FAILED comp_progress.error_message = str(e) self._status = UpdateStatus.FAILED self._progress.status = UpdateStatus.FAILED self._progress.error_message = str(e) # Attempt rollback try: await self.rollback_component(component_id) except Exception as rb_err: print(f"Rollback failed for {component_id}: {rb_err}") raise Exception(f"Failed to install {component_id}: {e}") async def install_update(self) -> bool: """Install all downloaded component updates (legacy compat).""" for comp_id in list(self._available_components.keys()): await self.install_component(comp_id) # Update system version if self._latest_release: self._current_version = self._latest_release.version self._manifest.system_version = self._latest_release.version self._progress.current_version = self._latest_release.version self._save_manifest() return True # ------------------------------------------------------------------ # Component-specific installers # ------------------------------------------------------------------ def _get_installer(self, component_id: str): """Return the installer function for a component.""" installers = { ComponentId.PM3: self._install_pm3, ComponentId.FRONTEND: self._install_frontend, ComponentId.BACKEND: self._install_backend, ComponentId.THEME: self._install_theme, } installer = installers.get(component_id) if not installer: raise ValueError(f"Unknown component: {component_id}") return installer async def _install_pm3(self, staging_dir: Path, info: AvailableComponent): """Install PM3 client + firmware from pre-built binary.""" pm3_home = Path(config.PM3_CLIENT_PATH).parent.parent # The tarball extracts to pm3/ subdirectory source = staging_dir / "pm3" if not source.exists(): # Try finding the extracted directory dirs = [d for d in staging_dir.iterdir() if d.is_dir()] source = dirs[0] if dirs else staging_dir # Copy files preserving structure for item in source.rglob("*"): if item.is_file(): rel = item.relative_to(source) dest = pm3_home / rel dest.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(item, dest) # Ensure _pm3.so symlink exists pyscripts = pm3_home / "client" / "pyscripts" pm3_so = pyscripts / "_pm3.so" lib_so = pyscripts / "libpm3rrg_rdv4.so" if lib_so.exists() and not pm3_so.exists(): pm3_so.symlink_to("libpm3rrg_rdv4.so") # Make executables executable client_bin = pm3_home / "client" / "proxmark3" if client_bin.exists(): client_bin.chmod(0o755) for flash_script in pm3_home.glob("pm3-flash-*"): flash_script.chmod(0o755) async def _install_frontend(self, staging_dir: Path, info: AvailableComponent): """Install frontend build output.""" target = self.INSTALL_DIR / "app" / "frontend" / "build" source = staging_dir / "frontend" / "build" if not source.exists(): source = staging_dir / "frontend" if target.exists(): shutil.rmtree(target) shutil.copytree(source, target) async def _install_backend(self, staging_dir: Path, info: AvailableComponent): """Install backend Python source and update dependencies.""" target = self.INSTALL_DIR / "app" / "backend" source = staging_dir / "backend" / "app" / "backend" if not source.exists(): source = staging_dir / "backend" if target.exists(): shutil.rmtree(target) shutil.copytree(source, target) # Update pip dependencies if requirements.txt included req_file = staging_dir / "backend" / "requirements.txt" if req_file.exists(): shutil.copy2(req_file, self.INSTALL_DIR / "requirements.txt") try: await self._run_command( f"pip install -r {self.INSTALL_DIR / 'requirements.txt'} " f"--quiet --disable-pip-version-check" ) except Exception as e: print(f"Warning: pip install failed: {e}") async def _install_theme(self, staging_dir: Path, info: AvailableComponent): """Install theme token CSS files.""" target = self.INSTALL_DIR / "app" / "frontend" / "themes" source = staging_dir / "theme" if not source.exists(): source = staging_dir target.mkdir(parents=True, exist_ok=True) # Copy theme directories (dt/, classic/, supra/, themes.json) for item in source.iterdir(): dest = target / item.name if item.name == "COMPONENT_VERSION": continue if item.is_dir(): if dest.exists(): shutil.rmtree(dest) shutil.copytree(item, dest) elif item.is_file(): shutil.copy2(item, dest) # ------------------------------------------------------------------ # Backup & rollback # ------------------------------------------------------------------ async def _backup_component(self, component_id: str) -> Path: """Backup current component before installing update.""" backup_dir = self.BACKUP_BASE / component_id if backup_dir.exists(): shutil.rmtree(backup_dir) backup_dir.mkdir(parents=True, exist_ok=True) comp = self._manifest.components.get(component_id) if comp and comp.install_path: source = Path(comp.install_path) if source.exists(): shutil.copytree(source, backup_dir / "files", dirs_exist_ok=True) # Save component info for rollback info_file = backup_dir / "component-info.json" info_file.write_text(json.dumps(asdict(comp), indent=2)) return backup_dir async def rollback_component(self, component_id: str) -> bool: """Rollback a component to its backup.""" backup_dir = self.BACKUP_BASE / component_id if not backup_dir.exists(): raise ValueError(f"No backup found for {component_id}") self._status = UpdateStatus.ROLLING_BACK self._progress.status = UpdateStatus.ROLLING_BACK info_file = backup_dir / "component-info.json" if not info_file.exists(): raise ValueError(f"Backup metadata missing for {component_id}") old_info = json.loads(info_file.read_text()) target = Path(old_info["install_path"]) # Restore files backup_files = backup_dir / "files" if backup_files.exists(): if target.exists(): shutil.rmtree(target) shutil.copytree(backup_files, target) # Restore manifest entry self._manifest.components[component_id] = ComponentInfo(**old_info) self._save_manifest() self._status = UpdateStatus.IDLE self._progress.status = UpdateStatus.IDLE return True # ------------------------------------------------------------------ # Progress & release notes # ------------------------------------------------------------------ async def get_progress(self) -> UpdateProgress: """Get current update progress.""" return self._progress async def get_release_notes(self, version: Optional[str] = None) -> str: """Get release notes for a specific version.""" try: if version: release = await self._fetch_release_by_version(version) else: release = await self._fetch_latest_release() return release.changelog if release else "No release notes available" except Exception as e: raise Exception(f"Failed to get release notes: {e}") # ------------------------------------------------------------------ # GitHub API # ------------------------------------------------------------------ async def _fetch_latest_release(self) -> Optional[ReleaseInfo]: """Fetch latest release from GitHub API.""" url = f"{self._github_api_base}/repos/{self._github_repo}/releases/latest" async with aiohttp.ClientSession() as session: async with session.get(url) as response: if response.status == 404: return None if response.status != 200: raise Exception( f"GitHub API returned status {response.status}" ) data = await response.json() return await self._parse_release_data(data, session) async def _fetch_release_by_version(self, version: str) -> Optional[ReleaseInfo]: """Fetch specific release by version tag.""" tag = version if version.startswith('v') else f'v{version}' url = f"{self._github_api_base}/repos/{self._github_repo}/releases/tags/{tag}" async with aiohttp.ClientSession() as session: async with session.get(url) as response: if response.status == 404: return None if response.status != 200: raise Exception( f"GitHub API returned status {response.status}" ) data = await response.json() return await self._parse_release_data(data, session) async def _parse_release_data( self, data: Dict[str, Any], session: aiohttp.ClientSession ) -> ReleaseInfo: """Parse GitHub release API response. Supports two formats: 1. New-style: release has a release-manifest.json asset 2. Legacy: release has a single .tar.gz asset """ version = data['tag_name'].lstrip('v') assets = data.get('assets', []) # Look for release-manifest.json (new-style) manifest_asset = None legacy_asset = None for asset in assets: name = asset['name'] if name == 'release-manifest.json': manifest_asset = asset elif name.endswith('.tar.gz') and legacy_asset is None: legacy_asset = asset release = ReleaseInfo( version=version, tag_name=data['tag_name'], published_at=data['published_at'], changelog=data.get('body', ''), is_prerelease=data.get('prerelease', False), ) if manifest_asset: # New-style: parse component manifest await self._parse_component_manifest( manifest_asset, assets, release, session ) elif legacy_asset: # Legacy: single tarball release.download_url = legacy_asset['browser_download_url'] release.asset_name = legacy_asset['name'] release.asset_size = legacy_asset['size'] return release async def _parse_component_manifest( self, manifest_asset: Dict[str, Any], all_assets: List[Dict[str, Any]], release: ReleaseInfo, session: aiohttp.ClientSession, ) -> None: """Download and parse release-manifest.json, populating release.components.""" manifest_url = manifest_asset['browser_download_url'] async with session.get(manifest_url) as resp: if resp.status != 200: return manifest_data = await resp.json() # Build asset URL lookup asset_urls = {a['name']: a for a in all_assets} sys_info = self.get_system_info() for comp_id, comp_data in manifest_data.get("components", {}).items(): comp_ver = comp_data.get("version", release.version) comp_assets = comp_data.get("assets", {}) changelog = comp_data.get("changelog", "") # Select the right asset variant for this system selected_key, selected_asset = self._select_asset( comp_id, comp_assets, sys_info ) if not selected_asset: continue filename = selected_asset.get("filename", "") gh_asset = asset_urls.get(filename, {}) available = AvailableComponent( component_id=comp_id, current_version=self.get_component_version(comp_id), available_version=comp_ver, download_url=gh_asset.get("browser_download_url", ""), asset_name=filename, asset_size=selected_asset.get("size", gh_asset.get("size", 0)), checksum=selected_asset.get("checksum_sha256", ""), changelog=changelog, platform=selected_asset.get("platform"), python_version=selected_asset.get("python_version"), compatible=True, ) # Sum total asset size for legacy compat release.asset_size += available.asset_size release.components[comp_id] = available # Check for incompatible pm3 variants if ComponentId.PM3 not in release.components: # Check if pm3 exists but no compatible variant pm3_data = manifest_data.get("components", {}).get(ComponentId.PM3) if pm3_data and pm3_data.get("assets"): release.components[ComponentId.PM3] = AvailableComponent( component_id=ComponentId.PM3, current_version=self.get_component_version(ComponentId.PM3), available_version=pm3_data.get("version", release.version), download_url="", asset_name="", asset_size=0, checksum="", changelog=pm3_data.get("changelog", ""), compatible=False, incompatible_reason=( f"No binary for {sys_info['arch']}-linux " f"with Python {sys_info['python_version']}" ), ) def _select_asset( self, component_id: str, assets: Dict[str, Any], sys_info: Dict[str, str], ) -> tuple: """Select the correct asset variant for this system.""" if not assets: return None, None # For pm3, match platform + python ABI if component_id == ComponentId.PM3: key = f"{sys_info['arch']}-linux-{sys_info['python_abi']}" if key in assets: return key, assets[key] return None, None # For other components, use "default" key or first available if "default" in assets: return "default", assets["default"] # Take the first (and usually only) asset first_key = next(iter(assets)) return first_key, assets[first_key] # ------------------------------------------------------------------ # Utilities # ------------------------------------------------------------------ def _is_newer_version(self, version1: str, version2: str) -> bool: """Compare two version strings. Returns True if version1 > version2.""" def parse_version(v: str) -> tuple: v = v.lstrip('v') parts = re.split(r'[-+]', v)[0] return tuple(map(int, parts.split('.'))) try: return parse_version(version1) > parse_version(version2) except (ValueError, AttributeError): # Fall back to string comparison for non-semver (e.g. PM3 tags) return version1 != version2 def _verify_checksum(self, file_path: Path, expected_checksum: str) -> bool: """Verify SHA256 checksum of a file.""" sha256 = hashlib.sha256() with open(file_path, 'rb') as f: while True: data = f.read(65536) if not data: break sha256.update(data) return sha256.hexdigest().lower() == expected_checksum.lower() async def _run_command(self, command: str) -> str: """Run a shell command asynchronously.""" process = await asyncio.create_subprocess_shell( command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, stderr = await process.communicate() if process.returncode != 0: raise Exception(f"Command failed: {stderr.decode()}") return stdout.decode() # Global update manager instance _update_manager: Optional[UpdateManager] = None def get_update_manager() -> UpdateManager: """Get the global update manager instance.""" global _update_manager if _update_manager is None: _update_manager = UpdateManager() return _update_manager