"""OS Update Manager for Dangerous Pi. Provides visibility into OS-level package updates and allows triggering apt upgrades from the web UI. Works alongside unattended-upgrades which handles automatic security patches. """ import asyncio import os import platform import re import time from dataclasses import dataclass, field from pathlib import Path from typing import Optional, List, Dict, Any from .. import config @dataclass class OsPackageUpdate: """A single upgradable OS package.""" name: str current_version: str available_version: str architecture: str = "" origin: str = "" @dataclass class OsInfo: """OS-level system information.""" debian_version: str = "" debian_codename: str = "" kernel: str = "" architecture: str = "" uptime_seconds: float = 0 hostname: str = "" last_apt_update: Optional[str] = None auto_security_updates: bool = True reboot_required: bool = False class OsUpdateManager: """Manages OS-level package updates.""" def __init__(self): self._cache: List[OsPackageUpdate] = [] self._cache_time: float = 0 self._cache_ttl: float = 3600 # 1 hour self._upgrading: bool = False self._upgrade_output: List[str] = [] self._env_path = Path(os.getenv("ENV_FILE", "/opt/dangerous-pi/.env")) async def get_os_info(self) -> Dict[str, Any]: """Get OS-level system information.""" info = OsInfo() # Debian version try: os_release = Path("/etc/os-release") if os_release.exists(): content = os_release.read_text() for line in content.splitlines(): if line.startswith("VERSION_ID="): info.debian_version = line.split("=", 1)[1].strip('"') elif line.startswith("VERSION_CODENAME="): info.debian_codename = line.split("=", 1)[1].strip('"') except Exception: pass # Kernel info.kernel = platform.release() # Architecture info.architecture = platform.machine() # Uptime try: uptime_path = Path("/proc/uptime") if uptime_path.exists(): info.uptime_seconds = float(uptime_path.read_text().split()[0]) except Exception: pass # Hostname info.hostname = platform.node() # Last apt update (mtime of apt lists directory) try: apt_lists = Path("/var/lib/apt/lists") if apt_lists.exists(): mtime = apt_lists.stat().st_mtime from datetime import datetime, timezone info.last_apt_update = datetime.fromtimestamp( mtime, tz=timezone.utc ).isoformat() except Exception: pass # Auto security updates setting info.auto_security_updates = self._read_auto_updates_setting() # Reboot required info.reboot_required = Path("/var/run/reboot-required").exists() return { "debian_version": info.debian_version, "debian_codename": info.debian_codename, "kernel": info.kernel, "architecture": info.architecture, "uptime_seconds": info.uptime_seconds, "hostname": info.hostname, "last_apt_update": info.last_apt_update, "auto_security_updates": info.auto_security_updates, "reboot_required": info.reboot_required, } async def check_available_updates(self, force_refresh: bool = False) -> List[Dict[str, str]]: """Check for available OS package updates. Returns cached results unless force_refresh=True or cache has expired. """ now = time.monotonic() if not force_refresh and self._cache and (now - self._cache_time) < self._cache_ttl: return [self._package_to_dict(p) for p in self._cache] try: proc = await asyncio.create_subprocess_exec( "apt", "list", "--upgradable", "-qq", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=60) output = stdout.decode(errors="replace").strip() packages = [] for line in output.splitlines(): line = line.strip() if not line or line.startswith("Listing"): continue pkg = self._parse_apt_line(line) if pkg: packages.append(pkg) self._cache = packages self._cache_time = now except (asyncio.TimeoutError, Exception) as e: # Return stale cache on error rather than failing if not self._cache: return [{"name": "error", "current_version": "", "available_version": str(e), "architecture": "", "origin": ""}] return [self._package_to_dict(p) for p in self._cache] async def run_upgrade(self, security_only: bool = False) -> Dict[str, Any]: """Trigger an apt upgrade. Returns the result after completion. Only one upgrade can run at a time. Uses create_subprocess_exec (not shell) to avoid injection risks. """ if self._upgrading: return {"success": False, "error": "An upgrade is already in progress"} self._upgrading = True self._upgrade_output = [] try: if security_only: cmd = ["sudo", "unattended-upgrade", "--verbose"] else: cmd = ["sudo", "apt-get", "upgrade", "-y"] proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, env={**os.environ, "DEBIAN_FRONTEND": "noninteractive"}, ) while True: line = await proc.stdout.readline() if not line: break decoded = line.decode(errors="replace").rstrip() self._upgrade_output.append(decoded) await proc.wait() # Invalidate cache after upgrade self._cache = [] self._cache_time = 0 success = proc.returncode == 0 return { "success": success, "return_code": proc.returncode, "output": self._upgrade_output, "reboot_required": Path("/var/run/reboot-required").exists(), } except Exception as e: return {"success": False, "error": str(e), "output": self._upgrade_output} finally: self._upgrading = False async def toggle_auto_updates(self, enabled: bool) -> Dict[str, Any]: """Toggle automatic security updates by writing to .env and restarting the timer.""" try: self._write_auto_updates_setting(enabled) # Restart the toggle service to apply proc = await asyncio.create_subprocess_exec( "sudo", "systemctl", "restart", "dangerous-pi-auto-updates.service", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) await asyncio.wait_for(proc.communicate(), timeout=30) return { "success": True, "auto_security_updates": enabled, } except Exception as e: return {"success": False, "error": str(e)} @property def is_upgrading(self) -> bool: return self._upgrading @property def upgrade_output(self) -> List[str]: return list(self._upgrade_output) # ----------------------------------------------------------------------- # Internal helpers # ----------------------------------------------------------------------- def _parse_apt_line(self, line: str) -> Optional[OsPackageUpdate]: """Parse a line from `apt list --upgradable -qq`. Format: package/origin version arch [upgradable from: old_version] """ match = re.match( r"^(\S+?)(?:/(\S+))?\s+(\S+)\s+(\S+)\s+\[upgradable from:\s+(\S+)\]", line, ) if match: return OsPackageUpdate( name=match.group(1), origin=match.group(2) or "", available_version=match.group(3), architecture=match.group(4), current_version=match.group(5), ) return None def _package_to_dict(self, pkg: OsPackageUpdate) -> Dict[str, str]: return { "name": pkg.name, "current_version": pkg.current_version, "available_version": pkg.available_version, "architecture": pkg.architecture, "origin": pkg.origin, } def _read_auto_updates_setting(self) -> bool: """Read AUTO_SECURITY_UPDATES from .env file.""" try: if self._env_path.exists(): for line in self._env_path.read_text().splitlines(): line = line.strip() if line.startswith("AUTO_SECURITY_UPDATES="): val = line.split("=", 1)[1].strip().lower() return val != "false" except Exception: pass return True # Default: enabled def _write_auto_updates_setting(self, enabled: bool): """Write AUTO_SECURITY_UPDATES to .env file.""" value = "true" if enabled else "false" env_line = f"AUTO_SECURITY_UPDATES={value}" if not self._env_path.exists(): self._env_path.write_text(env_line + "\n") return lines = self._env_path.read_text().splitlines() found = False for i, line in enumerate(lines): if line.strip().startswith("AUTO_SECURITY_UPDATES="): lines[i] = env_line found = True break if not found: lines.append(env_line) self._env_path.write_text("\n".join(lines) + "\n") # Singleton _os_update_manager: Optional[OsUpdateManager] = None def get_os_update_manager() -> OsUpdateManager: """Get or create the OS update manager singleton.""" global _os_update_manager if _os_update_manager is None: _os_update_manager = OsUpdateManager() return _os_update_manager