From cbe1e73565d80656931cf7a89cd4fb6a3a262323 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 9 Dec 2025 05:11:54 +0000 Subject: [PATCH 1/6] Initial plan From a4cac82272f37cd9e2ac788017c10d1fe3f3b3ee Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 9 Dec 2025 05:16:17 +0000 Subject: [PATCH 2/6] Add new feature modules from PR #66 Co-authored-by: AdmGenSameer <154604600+AdmGenSameer@users.noreply.github.com> --- archpkg/config_manager.py | 135 +++++++++ archpkg/download_manager.py | 353 +++++++++++++++++++++++ archpkg/github_install.py | 511 ++++++++++++++++++++++++++++++++++ archpkg/installed_apps.py | 276 ++++++++++++++++++ archpkg/security.py | 264 ++++++++++++++++++ archpkg/templates/home.html | 332 ++++++++++++++++++++++ archpkg/templates/search.html | 218 +++++++++++++++ archpkg/update_manager.py | 225 +++++++++++++++ archpkg/web.py | 123 ++++++++ pyproject.toml | 8 +- 10 files changed, 2443 insertions(+), 2 deletions(-) create mode 100644 archpkg/config_manager.py create mode 100644 archpkg/download_manager.py create mode 100644 archpkg/github_install.py create mode 100644 archpkg/installed_apps.py create mode 100644 archpkg/security.py create mode 100644 archpkg/templates/home.html create mode 100644 archpkg/templates/search.html create mode 100644 archpkg/update_manager.py create mode 100644 archpkg/web.py diff --git a/archpkg/config_manager.py b/archpkg/config_manager.py new file mode 100644 index 0000000..2d12a0f --- /dev/null +++ b/archpkg/config_manager.py @@ -0,0 +1,135 @@ +#!/usr/bin/env python3 +""" +Configuration management for archpkg-helper +""" + +import json +import os +from pathlib import Path +from typing import Dict, Any, Optional +from dataclasses import dataclass, asdict +from archpkg.logging_config import get_logger + +logger = get_logger(__name__) + +@dataclass +class UserConfig: + """User configuration settings""" + auto_update_enabled: bool = False + auto_update_mode: str = "manual" # "automatic" or "manual" + update_check_interval_hours: int = 24 + background_download_enabled: bool = True + notification_enabled: bool = True + +class ConfigManager: + """Manages user configuration with atomic file operations""" + + def __init__(self): + self.config_dir = Path.home() / ".archpkg" + self.config_file = self.config_dir / "config.json" + self._ensure_config_dir() + + def _ensure_config_dir(self) -> None: + """Ensure configuration directory exists""" + self.config_dir.mkdir(parents=True, exist_ok=True) + + def _atomic_write(self, data: Dict[str, Any]) -> None: + """Atomically write configuration to file""" + temp_file = self.config_file.with_suffix('.tmp') + + try: + # Write to temporary file first + with open(temp_file, 'w', encoding='utf-8') as f: + json.dump(data, f, indent=2, ensure_ascii=False) + + # Atomic move to final location + temp_file.replace(self.config_file) + logger.info(f"Configuration saved to {self.config_file}") + + except Exception as e: + # Clean up temp file on error + if temp_file.exists(): + temp_file.unlink() + logger.error(f"Failed to save configuration: {e}") + raise + + def load_config(self) -> UserConfig: + """Load configuration from file""" + if not self.config_file.exists(): + logger.info("No configuration file found, using defaults") + return UserConfig() + + try: + with open(self.config_file, 'r', encoding='utf-8') as f: + data = json.load(f) + + # Create config object from loaded data + config = UserConfig() + for key, value in data.items(): + if hasattr(config, key): + setattr(config, key, value) + + logger.info("Configuration loaded successfully") + return config + + except Exception as e: + logger.warning(f"Failed to load configuration, using defaults: {e}") + return UserConfig() + + def save_config(self, config: UserConfig) -> None: + """Save configuration to file atomically""" + data = asdict(config) + self._atomic_write(data) + + def get_config_value(self, key: str) -> Any: + """Get a specific configuration value""" + config = self.load_config() + return getattr(config, key, None) + + def set_config_value(self, key: str, value: Any) -> None: + """Set a specific configuration value""" + config = self.load_config() + if hasattr(config, key): + setattr(config, key, value) + self.save_config(config) + logger.info(f"Configuration updated: {key} = {value}") + else: + raise ValueError(f"Unknown configuration key: {key}") + + def show_config(self) -> str: + """Get formatted configuration display""" + config = self.load_config() + lines = [ + f"Configuration file: {self.config_file}", + "", + "Current settings:", + f" Auto-update enabled: {config.auto_update_enabled}", + f" Auto-update mode: {config.auto_update_mode}", + f" Update check interval: {config.update_check_interval_hours} hours", + f" Background download: {config.background_download_enabled}", + f" Notifications: {config.notification_enabled}", + ] + return "\n".join(lines) + +# Global configuration manager instance +config_manager = ConfigManager() + +def get_user_config() -> UserConfig: + """Get current user configuration""" + return config_manager.load_config() + +def save_user_config(config: UserConfig) -> None: + """Save user configuration""" + config_manager.save_config(config) + +def set_config_option(key: str, value: Any) -> None: + """Set a configuration option""" + config_manager.set_config_value(key, value) + +def get_config_option(key: str) -> Any: + """Get a configuration option""" + return config_manager.get_config_value(key) + +def show_current_config() -> str: + """Show current configuration""" + return config_manager.show_config() \ No newline at end of file diff --git a/archpkg/download_manager.py b/archpkg/download_manager.py new file mode 100644 index 0000000..f618997 --- /dev/null +++ b/archpkg/download_manager.py @@ -0,0 +1,353 @@ +#!/usr/bin/env python3 +""" +Background downloading and installation for archpkg-helper +""" + +import os +import hashlib +import tempfile +import threading +import time +from typing import Dict, List, Optional, Any, Callable +from pathlib import Path +from datetime import datetime, timezone +from urllib.request import urlopen, Request +from urllib.error import URLError, HTTPError +from archpkg.logging_config import get_logger +from archpkg.installed_apps import ( + get_all_installed_packages, + update_package_info, + get_packages_with_updates +) +from archpkg.config_manager import get_user_config + +logger = get_logger(__name__) + +class DownloadManager: + """Manages background downloads with resumability""" + + def __init__(self): + self.active_downloads: Dict[str, Dict[str, Any]] = {} + self.download_dir = Path.home() / ".archpkg" / "downloads" + self.download_dir.mkdir(parents=True, exist_ok=True) + + def start_download(self, package_name: str, download_url: str, + callback: Optional[Callable] = None) -> str: + """Start a background download""" + if package_name in self.active_downloads: + logger.warning(f"Download already in progress for {package_name}") + return self.active_downloads[package_name]["download_id"] + + download_id = f"{package_name}_{int(datetime.now(timezone.utc).timestamp())}" + temp_file = self.download_dir / f"{download_id}.tmp" + + download_info = { + "download_id": download_id, + "package_name": package_name, + "url": download_url, + "temp_file": temp_file, + "status": "starting", + "progress": 0, + "total_size": 0, + "downloaded_size": 0, + "callback": callback, + "thread": None + } + + self.active_downloads[package_name] = download_info + + # Start download in background thread + thread = threading.Thread( + target=self._download_worker, + args=(download_info,), + daemon=True, + name=f"download-{package_name}" + ) + download_info["thread"] = thread + thread.start() + + logger.info(f"Started download for {package_name}: {download_id}") + return download_id + + def _download_worker(self, download_info: Dict[str, Any]) -> None: + """Background download worker""" + package_name = download_info["package_name"] + url = download_info["url"] + temp_file = download_info["temp_file"] + + try: + download_info["status"] = "downloading" + + # Create request with resume support + headers = {} + if temp_file.exists(): + # Resume download + downloaded_size = temp_file.stat().st_size + headers["Range"] = f"bytes={downloaded_size}-" + download_info["downloaded_size"] = downloaded_size + logger.info(f"Resuming download for {package_name} from {downloaded_size} bytes") + + req = Request(url, headers=headers) + + with urlopen(req) as response: + total_size = int(response.headers.get("content-length", 0)) + download_info["total_size"] = total_size + + if total_size > 0 and download_info["downloaded_size"] > 0: + # Verify we're resuming correctly + if response.code != 206: # 206 Partial Content + logger.warning(f"Server doesn't support resume for {package_name}, restarting") + download_info["downloaded_size"] = 0 + temp_file.unlink(missing_ok=True) + + mode = "ab" if download_info["downloaded_size"] > 0 else "wb" + + with open(temp_file, mode) as f: + downloaded = download_info["downloaded_size"] + + while True: + chunk = response.read(8192) + if not chunk: + break + + f.write(chunk) + downloaded += len(chunk) + download_info["downloaded_size"] = downloaded + + if total_size > 0: + progress = (downloaded / total_size) * 100 + download_info["progress"] = progress + + download_info["status"] = "completed" + logger.info(f"Download completed for {package_name}") + + # Notify callback if provided + if download_info["callback"]: + try: + download_info["callback"](download_info) + except Exception as e: + logger.error(f"Download callback error for {package_name}: {e}") + + except HTTPError as e: + download_info["status"] = "failed" + download_info["error"] = f"HTTP {e.code}: {e.reason}" + logger.error(f"Download failed for {package_name}: {e}") + except URLError as e: + download_info["status"] = "failed" + download_info["error"] = str(e.reason) + logger.error(f"Download failed for {package_name}: {e}") + except Exception as e: + download_info["status"] = "failed" + download_info["error"] = str(e) + logger.error(f"Download failed for {package_name}: {e}") + + def get_download_status(self, package_name: str) -> Optional[Dict[str, Any]]: + """Get status of a download""" + return self.active_downloads.get(package_name) + + def cancel_download(self, package_name: str) -> bool: + """Cancel a download""" + if package_name not in self.active_downloads: + return False + + download_info = self.active_downloads[package_name] + download_info["status"] = "cancelled" + + # Clean up temp file + temp_file = download_info["temp_file"] + if temp_file.exists(): + temp_file.unlink() + + del self.active_downloads[package_name] + logger.info(f"Download cancelled for {package_name}") + return True + + def get_completed_downloads(self) -> List[Dict[str, Any]]: + """Get list of completed downloads""" + return [ + info for info in self.active_downloads.values() + if info["status"] == "completed" + ] + + def cleanup_old_downloads(self, days_old: int = 7) -> None: + """Clean up old temporary download files""" + cutoff_time = datetime.now(timezone.utc).timestamp() - (days_old * 24 * 3600) + + for temp_file in self.download_dir.glob("*.tmp"): + if temp_file.stat().st_mtime < cutoff_time: + temp_file.unlink() + logger.debug(f"Cleaned up old download file: {temp_file}") + +class UpdateInstaller: + """Handles installation of downloaded updates""" + + def __init__(self): + self.download_manager = DownloadManager() + + def install_updates(self, package_names: Optional[List[str]] = None) -> Dict[str, Any]: + """Install updates for specified packages or all available updates""" + config = get_user_config() + + if package_names is None: + # Get all packages with available updates + packages_with_updates = get_packages_with_updates() + package_names = [p.name for p in packages_with_updates] + + if not package_names: + return {"status": "success", "installed": 0, "message": "No updates available"} + + installed_count = 0 + failed_count = 0 + results = [] + + for package_name in package_names: + try: + result = self._install_single_update(package_name) + results.append(result) + + if result["status"] == "success": + installed_count += 1 + else: + failed_count += 1 + + except Exception as e: + logger.error(f"Failed to install update for {package_name}: {e}") + results.append({ + "package": package_name, + "status": "error", + "error": str(e) + }) + failed_count += 1 + + return { + "status": "success", + "installed": installed_count, + "failed": failed_count, + "results": results + } + + def _install_single_update(self, package_name: str) -> Dict[str, Any]: + """Install update for a single package""" + logger.info(f"Installing update for {package_name}") + + # This is a placeholder - actual installation would depend on the package source + # For now, we'll just mark it as installed + + try: + # In a real implementation, this would: + # 1. Get the package info + # 2. Run the appropriate installation command based on source + # 3. Verify installation + # 4. Update the installed package info + + update_package_info( + package_name, + installed_version="latest", # Would be actual version + update_available=False, + last_updated=datetime.now(timezone.utc).isoformat() + ) + + return { + "package": package_name, + "status": "success", + "message": "Update installed successfully" + } + + except Exception as e: + return { + "package": package_name, + "status": "failed", + "error": str(e) + } + +class BackgroundUpdateService: + """Complete background update service""" + + def __init__(self): + self.download_manager = DownloadManager() + self.update_installer = UpdateInstaller() + self.is_running = False + self.background_thread = None + + def start_service(self) -> None: + """Start the background update service""" + if self.is_running: + return + + config = get_user_config() + if not config.auto_update_enabled: + logger.info("Auto-update not enabled") + return + + self.is_running = True + + self.background_thread = threading.Thread( + target=self._background_worker, + daemon=True, + name="archpkg-background-updates" + ) + self.background_thread.start() + + logger.info("Background update service started") + + def stop_service(self) -> None: + """Stop the background update service""" + self.is_running = False + if self.background_thread: + self.background_thread.join(timeout=5) + logger.info("Background update service stopped") + + def _background_worker(self) -> None: + """Background worker for automatic updates""" + logger.info("Background update worker started") + + while self.is_running: + try: + config = get_user_config() + + if config.auto_install_updates: + # Automatically install available updates + packages_with_updates = get_packages_with_updates() + if packages_with_updates: + logger.info(f"Auto-installing {len(packages_with_updates)} updates") + result = self.update_installer.install_updates() + logger.info(f"Auto-install result: {result}") + + # Clean up old downloads periodically + self.download_manager.cleanup_old_downloads() + + # Sleep for check interval + sleep_time = config.update_check_interval_hours * 3600 + time.sleep(min(sleep_time, 3600)) # Sleep in 1-hour chunks + + except Exception as e: + logger.error(f"Background update worker error: {e}") + time.sleep(3600) # Wait 1 hour before retrying + + logger.info("Background update worker stopped") + +# Global instances +download_manager = DownloadManager() +update_installer = UpdateInstaller() +background_update_service = BackgroundUpdateService() + +def start_download(package_name: str, download_url: str, + callback: Optional[Callable] = None) -> str: + """Start a background download""" + return download_manager.start_download(package_name, download_url, callback) + +def get_download_status(package_name: str) -> Optional[Dict[str, Any]]: + """Get download status""" + return download_manager.get_download_status(package_name) + +def install_updates(package_names: Optional[List[str]] = None) -> Dict[str, Any]: + """Install updates""" + return update_installer.install_updates(package_names) + +def start_background_update_service() -> None: + """Start background update service""" + background_update_service.start_service() + +def stop_background_update_service() -> None: + """Stop background update service""" + background_update_service.stop_service() \ No newline at end of file diff --git a/archpkg/github_install.py b/archpkg/github_install.py new file mode 100644 index 0000000..39ff12f --- /dev/null +++ b/archpkg/github_install.py @@ -0,0 +1,511 @@ +#!/usr/bin/env python3 +""" +GitHub repository installation module for archpkg-helper +""" + +import os +import sys +import shutil +import tempfile +import subprocess +import re +from typing import Optional, Dict, List, Tuple, Any +from pathlib import Path +from abc import ABC, abstractmethod + +from archpkg.logging_config import get_logger + +logger = get_logger(__name__) + +class ProjectTypeHandler(ABC): + """Abstract base class for project type handlers""" + + @property + @abstractmethod + def name(self) -> str: + """Name of the project type""" + pass + + @property + @abstractmethod + def indicators(self) -> List[str]: + """Files that indicate this project type""" + pass + + @abstractmethod + def can_handle(self, repo_path: Path) -> bool: + """Check if this handler can handle the project""" + pass + + @abstractmethod + def build_and_install(self, repo_path: Path, temp_dir: Path) -> bool: + """Build and install the project""" + pass + + def log_command(self, command: List[str], description: str) -> None: + """Log a command execution""" + logger.info(f"{description}: {' '.join(command)}") + print(f" → {description}: {' '.join(command)}") + +class PythonHandler(ProjectTypeHandler): + """Handler for Python projects""" + + @property + def name(self) -> str: + return "Python" + + @property + def indicators(self) -> List[str]: + return ["setup.py", "pyproject.toml", "requirements.txt", "Pipfile"] + + def can_handle(self, repo_path: Path) -> bool: + return any((repo_path / indicator).exists() for indicator in self.indicators) + + def build_and_install(self, repo_path: Path, temp_dir: Path) -> bool: + try: + # Change to repo directory + os.chdir(repo_path) + + # Check for setup.py or pyproject.toml + if (repo_path / "setup.py").exists(): + self.log_command(["pip", "install", "."], "Installing Python package") + result = subprocess.run(["pip", "install", "."], capture_output=True, text=True) + elif (repo_path / "pyproject.toml").exists(): + self.log_command(["pip", "install", "."], "Installing Python package (PEP 517)") + result = subprocess.run(["pip", "install", "."], capture_output=True, text=True) + else: + print(" ⚠ No setup.py or pyproject.toml found, installing requirements if present") + if (repo_path / "requirements.txt").exists(): + self.log_command(["pip", "install", "-r", "requirements.txt"], "Installing requirements") + result = subprocess.run(["pip", "install", "-r", "requirements.txt"], capture_output=True, text=True) + else: + print(" ✗ No installation method found for Python project") + return False + + if result.returncode == 0: + print(" ✓ Python package installed successfully") + return True + else: + print(f" ✗ Python installation failed: {result.stderr}") + return False + + except Exception as e: + print(f" ✗ Python installation error: {e}") + return False + +class NodeJSHandler(ProjectTypeHandler): + """Handler for Node.js projects""" + + @property + def name(self) -> str: + return "Node.js" + + @property + def indicators(self) -> List[str]: + return ["package.json", "yarn.lock", "package-lock.json"] + + def can_handle(self, repo_path: Path) -> bool: + return (repo_path / "package.json").exists() + + def build_and_install(self, repo_path: Path, temp_dir: Path) -> bool: + try: + os.chdir(repo_path) + + # Check if yarn or npm should be used + use_yarn = (repo_path / "yarn.lock").exists() + + if use_yarn: + # Install dependencies with yarn + self.log_command(["yarn", "install"], "Installing dependencies with yarn") + result = subprocess.run(["yarn", "install"], capture_output=True, text=True) + if result.returncode != 0: + print(f" ✗ Yarn install failed: {result.stderr}") + return False + + # Build if build script exists + if self._has_build_script(): + self.log_command(["yarn", "build"], "Building project with yarn") + result = subprocess.run(["yarn", "build"], capture_output=True, text=True) + if result.returncode != 0: + print(f" ✗ Yarn build failed: {result.stderr}") + return False + + # Install globally if it's a CLI tool + if self._is_cli_tool(): + self.log_command(["yarn", "global", "add", "."], "Installing CLI tool globally") + result = subprocess.run(["yarn", "global", "add", "."], capture_output=True, text=True) + else: + print(" ⚠ Not a CLI tool, skipping global installation") + return True + + else: + # Use npm + self.log_command(["npm", "install"], "Installing dependencies with npm") + result = subprocess.run(["npm", "install"], capture_output=True, text=True) + if result.returncode != 0: + print(f" ✗ NPM install failed: {result.stderr}") + return False + + # Build if build script exists + if self._has_build_script(): + self.log_command(["npm", "run", "build"], "Building project with npm") + result = subprocess.run(["npm", "run", "build"], capture_output=True, text=True) + if result.returncode != 0: + print(f" ✗ NPM build failed: {result.stderr}") + return False + + # Install globally if it's a CLI tool + if self._is_cli_tool(): + self.log_command(["npm", "install", "-g", "."], "Installing CLI tool globally") + result = subprocess.run(["npm", "install", "-g", "."], capture_output=True, text=True) + else: + print(" ⚠ Not a CLI tool, skipping global installation") + return True + + if result.returncode == 0: + print(" ✓ Node.js package installed successfully") + return True + else: + print(f" ✗ Node.js installation failed: {result.stderr}") + return False + + except Exception as e: + print(f" ✗ Node.js installation error: {e}") + return False + + def _has_build_script(self) -> bool: + """Check if package.json has a build script""" + try: + import json + with open("package.json", "r") as f: + data = json.load(f) + return "scripts" in data and "build" in data["scripts"] + except: + return False + + def _is_cli_tool(self) -> bool: + """Check if this is a CLI tool by looking for bin field""" + try: + import json + with open("package.json", "r") as f: + data = json.load(f) + return "bin" in data or ("name" in data and data["name"].startswith("@")) + except: + return False + +class CMakeHandler(ProjectTypeHandler): + """Handler for CMake projects""" + + @property + def name(self) -> str: + return "CMake" + + @property + def indicators(self) -> List[str]: + return ["CMakeLists.txt", "cmake"] + + def can_handle(self, repo_path: Path) -> bool: + return (repo_path / "CMakeLists.txt").exists() + + def build_and_install(self, repo_path: Path, temp_dir: Path) -> bool: + try: + os.chdir(repo_path) + + # Create build directory + build_dir = repo_path / "build" + build_dir.mkdir(exist_ok=True) + os.chdir(build_dir) + + # Configure with CMake + self.log_command(["cmake", ".."], "Configuring with CMake") + result = subprocess.run(["cmake", ".."], capture_output=True, text=True) + if result.returncode != 0: + print(f" ✗ CMake configure failed: {result.stderr}") + return False + + # Build + self.log_command(["make", "-j$(nproc)"], "Building with make") + result = subprocess.run(["make", f"-j{os.cpu_count() or 1}"], capture_output=True, text=True) + if result.returncode != 0: + print(f" ✗ Make build failed: {result.stderr}") + return False + + # Install + self.log_command(["sudo", "make", "install"], "Installing with make") + result = subprocess.run(["sudo", "make", "install"], capture_output=True, text=True) + if result.returncode == 0: + print(" ✓ CMake project installed successfully") + return True + else: + print(f" ✗ Make install failed: {result.stderr}") + return False + + except Exception as e: + print(f" ✗ CMake installation error: {e}") + return False + +class MakefileHandler(ProjectTypeHandler): + """Handler for Makefile projects""" + + @property + def name(self) -> str: + return "Makefile" + + @property + def indicators(self) -> List[str]: + return ["Makefile", "makefile"] + + def can_handle(self, repo_path: Path) -> bool: + return (repo_path / "Makefile").exists() or (repo_path / "makefile").exists() + + def build_and_install(self, repo_path: Path, temp_dir: Path) -> bool: + try: + os.chdir(repo_path) + + # Try make install first + self.log_command(["make", "install"], "Installing with make") + result = subprocess.run(["make", "install"], capture_output=True, text=True) + + if result.returncode == 0: + print(" ✓ Makefile project installed successfully") + return True + else: + # Try with sudo + self.log_command(["sudo", "make", "install"], "Installing with sudo make") + result = subprocess.run(["sudo", "make", "install"], capture_output=True, text=True) + if result.returncode == 0: + print(" ✓ Makefile project installed successfully") + return True + else: + print(f" ✗ Make install failed: {result.stderr}") + return False + + except Exception as e: + print(f" ✗ Makefile installation error: {e}") + return False + +class GoHandler(ProjectTypeHandler): + """Handler for Go projects""" + + @property + def name(self) -> str: + return "Go" + + @property + def indicators(self) -> List[str]: + return ["go.mod", "main.go", ".go"] + + def can_handle(self, repo_path: Path) -> bool: + return (repo_path / "go.mod").exists() or any(f.suffix == ".go" for f in repo_path.glob("*.go")) + + def build_and_install(self, repo_path: Path, temp_dir: Path) -> bool: + try: + os.chdir(repo_path) + + # Install with go install + self.log_command(["go", "install", "."], "Installing Go package") + result = subprocess.run(["go", "install", "."], capture_output=True, text=True) + + if result.returncode == 0: + print(" ✓ Go package installed successfully") + return True + else: + print(f" ✗ Go install failed: {result.stderr}") + return False + + except Exception as e: + print(f" ✗ Go installation error: {e}") + return False + +class RustHandler(ProjectTypeHandler): + """Handler for Rust projects""" + + @property + def name(self) -> str: + return "Rust" + + @property + def indicators(self) -> List[str]: + return ["Cargo.toml", "Cargo.lock"] + + def can_handle(self, repo_path: Path) -> bool: + return (repo_path / "Cargo.toml").exists() + + def build_and_install(self, repo_path: Path, temp_dir: Path) -> bool: + try: + os.chdir(repo_path) + + # Install with cargo + self.log_command(["cargo", "install", "--path", "."], "Installing Rust package") + result = subprocess.run(["cargo", "install", "--path", "."], capture_output=True, text=True) + + if result.returncode == 0: + print(" ✓ Rust package installed successfully") + return True + else: + print(f" ✗ Cargo install failed: {result.stderr}") + return False + + except Exception as e: + print(f" ✗ Rust installation error: {e}") + return False + +class ProjectTypeRegistry: + """Registry for project type handlers""" + + def __init__(self): + self.handlers: List[ProjectTypeHandler] = [ + PythonHandler(), + NodeJSHandler(), + CMakeHandler(), + MakefileHandler(), + GoHandler(), + RustHandler(), + ] + + def detect_project_type(self, repo_path: Path) -> Optional[ProjectTypeHandler]: + """Detect the project type and return appropriate handler""" + for handler in self.handlers: + if handler.can_handle(repo_path): + return handler + return None + + def get_supported_types(self) -> List[str]: + """Get list of supported project types""" + return [handler.name for handler in self.handlers] + +def clone_repository(repo_url: str, temp_dir: Path) -> Optional[Path]: + """Clone a GitHub repository to a temporary directory""" + try: + print(f"📥 Cloning repository: {repo_url}") + + # Use GitPython if available, otherwise use subprocess + try: + from git import Repo + logger.info(f"Cloning {repo_url} to {temp_dir}") + Repo.clone_from(repo_url, temp_dir) + print(" ✓ Repository cloned successfully") + return temp_dir + except ImportError: + # Fallback to subprocess + logger.info(f"Cloning {repo_url} to {temp_dir} using subprocess") + result = subprocess.run(["git", "clone", repo_url, str(temp_dir)], capture_output=True, text=True) + if result.returncode == 0: + print(" ✓ Repository cloned successfully") + return temp_dir + else: + print(f" ✗ Git clone failed: {result.stderr}") + return None + + except Exception as e: + print(f" ✗ Repository cloning error: {e}") + return None + +def validate_github_url(url_or_repo: str) -> Optional[str]: + """Validate and convert GitHub URL or user/repo format to full URL""" + # Handle github:user/repo format + if url_or_repo.startswith("github:"): + repo = url_or_repo[7:] # Remove "github:" prefix + if "/" not in repo: + print(" ✗ Invalid GitHub repo format. Use: github:user/repo") + return None + return f"https://github.com/{repo}.git" + + # Handle full GitHub URLs + if url_or_repo.startswith("https://github.com/"): + if not url_or_repo.endswith(".git"): + url_or_repo += ".git" + return url_or_repo + + # Handle git@github.com:user/repo.git format + if url_or_repo.startswith("git@github.com:"): + return url_or_repo + + print(" ✗ Invalid GitHub URL format. Use: github:user/repo or https://github.com/user/repo") + return None + +def install_from_github(repo_spec: str) -> bool: + """Main function to install from GitHub repository""" + print(f"\n🔧 Installing from GitHub: {repo_spec}") + + # Validate the repo specification + repo_url = validate_github_url(repo_spec) + if not repo_url: + return False + + # Create temporary directory + temp_dir = None + try: + temp_dir = Path(tempfile.mkdtemp(prefix="archpkg-github-")) + print(f"📁 Working in temporary directory: {temp_dir}") + + # Clone the repository + repo_path = clone_repository(repo_url, temp_dir) + if not repo_path: + return False + + # Detect project type + registry = ProjectTypeRegistry() + handler = registry.detect_project_type(repo_path) + + if not handler: + # List found files to help user + files = list(repo_path.glob("*")) + file_names = [f.name for f in files[:10]] # Show first 10 files + print(f" ✗ Unsupported project type") + print(f" 📄 Found files: {', '.join(file_names)}") + if len(files) > 10: + print(f" ... and {len(files) - 10} more files") + print(f" 💡 Supported types: {', '.join(registry.get_supported_types())}") + print(f" 💡 Try manual installation or request support for this project type") + return False + + print(f" 🔍 Detected project type: {handler.name}") + + # Build and install + success = handler.build_and_install(repo_path, temp_dir) + + if success: + print(f"🎉 Successfully installed {repo_spec}!") + return True + else: + print(f"❌ Failed to install {repo_spec}") + return False + + except Exception as e: + print(f" ✗ Unexpected error during installation: {e}") + return False + + finally: + # Clean up temporary directory + if temp_dir and temp_dir.exists(): + print(f"🧹 Cleaning up temporary directory: {temp_dir}") + try: + shutil.rmtree(temp_dir) + print(" ✓ Cleanup completed") + except Exception as e: + print(f" ⚠ Cleanup warning: {e}") + +def check_dependencies() -> Dict[str, bool]: + """Check if required dependencies are available""" + deps = { + "git": False, + "python": False, + "node": False, + "npm": False, + "yarn": False, + "cmake": False, + "make": False, + "go": False, + "cargo": False, + } + + # Check each dependency + for dep in deps.keys(): + try: + result = subprocess.run([dep, "--version"], capture_output=True, timeout=5) + deps[dep] = result.returncode == 0 + except (subprocess.TimeoutExpired, FileNotFoundError, subprocess.SubprocessError): + deps[dep] = False + + return deps \ No newline at end of file diff --git a/archpkg/installed_apps.py b/archpkg/installed_apps.py new file mode 100644 index 0000000..37ec249 --- /dev/null +++ b/archpkg/installed_apps.py @@ -0,0 +1,276 @@ +#!/usr/bin/env python3 +""" +Installed applications tracking for archpkg-helper +""" + +import json +import os +from pathlib import Path +from typing import Dict, List, Optional, Any +from dataclasses import dataclass, asdict +from datetime import datetime, timezone +from archpkg.logging_config import get_logger + +logger = get_logger(__name__) + +@dataclass +class InstalledPackage: + """Represents an installed package""" + name: str + version: Optional[str] = None + source: str = "unknown" + install_date: str = "" + last_update_check: Optional[str] = None + available_version: Optional[str] = None + update_available: bool = False + install_method: str = "archpkg" # "archpkg", "github", "manual" + +class InstalledAppsManager: + """Manages tracking of installed applications""" + + def __init__(self): + self.config_dir = Path.home() / ".archpkg" + self.installed_file = self.config_dir / "installed.json" + self._ensure_config_dir() + + def _ensure_config_dir(self) -> None: + """Ensure configuration directory exists""" + self.config_dir.mkdir(parents=True, exist_ok=True) + + def _atomic_write(self, data: Dict[str, Any]) -> None: + """Atomically write installed apps data to file""" + temp_file = self.installed_file.with_suffix('.tmp') + + try: + # Write to temporary file first + with open(temp_file, 'w', encoding='utf-8') as f: + json.dump(data, f, indent=2, ensure_ascii=False) + + # Atomic move to final location + temp_file.replace(self.installed_file) + logger.debug(f"Installed apps data saved to {self.installed_file}") + + except Exception as e: + # Clean up temp file on error + if temp_file.exists(): + temp_file.unlink() + logger.error(f"Failed to save installed apps data: {e}") + raise + + def _load_installed_data(self) -> Dict[str, Dict[str, Any]]: + """Load installed packages data from file""" + if not self.installed_file.exists(): + logger.debug("No installed apps file found, starting fresh") + return {} + + try: + with open(self.installed_file, 'r', encoding='utf-8') as f: + data = json.load(f) + logger.debug(f"Loaded installed apps data with {len(data)} packages") + return data + + except Exception as e: + logger.warning(f"Failed to load installed apps data, starting fresh: {e}") + return {} + + def _save_installed_data(self, data: Dict[str, Dict[str, Any]]) -> None: + """Save installed packages data to file atomically""" + self._atomic_write(data) + + def add_package(self, package: InstalledPackage) -> None: + """Add a package to the installed list""" + data = self._load_installed_data() + + # Set install date if not provided + if not package.install_date: + package.install_date = datetime.now(timezone.utc).isoformat() + + data[package.name] = asdict(package) + self._save_installed_data(data) + logger.info(f"Added package to tracking: {package.name} ({package.source})") + + def remove_package(self, package_name: str) -> bool: + """Remove a package from the installed list""" + data = self._load_installed_data() + + if package_name in data: + del data[package_name] + self._save_installed_data(data) + logger.info(f"Removed package from tracking: {package_name}") + return True + + logger.warning(f"Package not found in tracking: {package_name}") + return False + + def get_package(self, package_name: str) -> Optional[InstalledPackage]: + """Get information about an installed package""" + data = self._load_installed_data() + + if package_name in data: + pkg_data = data[package_name] + return InstalledPackage(**pkg_data) + + return None + + def get_all_packages(self) -> List[InstalledPackage]: + """Get all installed packages""" + data = self._load_installed_data() + packages = [] + + for pkg_data in data.values(): + packages.append(InstalledPackage(**pkg_data)) + + return packages + + def update_package_info(self, package_name: str, **updates) -> bool: + """Update information for an installed package""" + data = self._load_installed_data() + + if package_name in data: + # Update the package data + data[package_name].update(updates) + + # Update last update check timestamp if we're checking for updates + if 'last_update_check' not in updates: + data[package_name]['last_update_check'] = datetime.now(timezone.utc).isoformat() + + self._save_installed_data(data) + logger.debug(f"Updated package info: {package_name}") + return True + + logger.warning(f"Package not found for update: {package_name}") + return False + + def get_packages_needing_update_check(self, max_age_hours: int = 24) -> List[InstalledPackage]: + """Get packages that need update checking""" + packages = self.get_all_packages() + needing_check = [] + + now = datetime.now(timezone.utc) + + for pkg in packages: + needs_check = True + + if pkg.last_update_check: + try: + last_check = datetime.fromisoformat(pkg.last_update_check.replace('Z', '+00:00')) + hours_since_check = (now - last_check).total_seconds() / 3600 + + if hours_since_check < max_age_hours: + needs_check = False + except ValueError: + # Invalid timestamp, needs check + pass + + if needs_check: + needing_check.append(pkg) + + logger.debug(f"Found {len(needing_check)} packages needing update check") + return needing_check + + def get_packages_with_updates(self) -> List[InstalledPackage]: + """Get packages that have available updates""" + packages = self.get_all_packages() + with_updates = [pkg for pkg in packages if pkg.update_available] + + logger.debug(f"Found {len(with_updates)} packages with available updates") + return with_updates + + def mark_update_available(self, package_name: str, available_version: str) -> bool: + """Mark that an update is available for a package""" + return self.update_package_info( + package_name, + available_version=available_version, + update_available=True, + last_update_check=datetime.now(timezone.utc).isoformat() + ) + + def mark_update_installed(self, package_name: str, new_version: str) -> bool: + """Mark that an update has been installed for a package""" + return self.update_package_info( + package_name, + version=new_version, + available_version=None, + update_available=False, + last_update_check=datetime.now(timezone.utc).isoformat() + ) + + def get_stats(self) -> Dict[str, int]: + """Get statistics about installed packages""" + packages = self.get_all_packages() + with_updates = len([p for p in packages if p.update_available]) + + return { + 'total_packages': len(packages), + 'packages_with_updates': with_updates, + 'packages_up_to_date': len(packages) - with_updates, + } + + def show_installed_packages(self) -> str: + """Get formatted display of installed packages""" + packages = self.get_all_packages() + + if not packages: + return "No packages installed via archpkg." + + lines = [f"Installed packages ({len(packages)}):", ""] + + for pkg in packages: + status = "✓ Up to date" + if pkg.update_available: + status = f"⚠ Update available: {pkg.available_version}" + + lines.append(f" {pkg.name} ({pkg.version or 'unknown'}) - {pkg.source}") + lines.append(f" Status: {status}") + lines.append(f" Installed: {pkg.install_date}") + if pkg.last_update_check: + lines.append(f" Last checked: {pkg.last_update_check}") + lines.append("") + + return "\n".join(lines) + +# Global installed apps manager instance +installed_apps_manager = InstalledAppsManager() + +def add_installed_package(package: InstalledPackage) -> None: + """Add a package to the installed list""" + installed_apps_manager.add_package(package) + +def remove_installed_package(package_name: str) -> bool: + """Remove a package from the installed list""" + return installed_apps_manager.remove_package(package_name) + +def get_installed_package(package_name: str) -> Optional[InstalledPackage]: + """Get information about an installed package""" + return installed_apps_manager.get_package(package_name) + +def get_all_installed_packages() -> List[InstalledPackage]: + """Get all installed packages""" + return installed_apps_manager.get_all_packages() + +def update_package_info(package_name: str, **updates) -> bool: + """Update information for an installed package""" + return installed_apps_manager.update_package_info(package_name, **updates) + +def get_packages_needing_update_check(max_age_hours: int = 24) -> List[InstalledPackage]: + """Get packages that need update checking""" + return installed_apps_manager.get_packages_needing_update_check(max_age_hours) + +def get_packages_with_updates() -> List[InstalledPackage]: + """Get packages that have available updates""" + return installed_apps_manager.get_packages_with_updates() + +def get_installed_stats() -> Dict[str, int]: + """Get statistics about installed packages""" + return installed_apps_manager.get_stats() + +def track_package(package_name: str, source: str, version: Optional[str] = None) -> None: + """Convenience function to track a newly installed package""" + package = InstalledPackage( + name=package_name, + source=source, + version=version or "latest", + install_date=datetime.now(timezone.utc).isoformat(), + install_method="archpkg" + ) + add_installed_package(package) \ No newline at end of file diff --git a/archpkg/security.py b/archpkg/security.py new file mode 100644 index 0000000..6484cc7 --- /dev/null +++ b/archpkg/security.py @@ -0,0 +1,264 @@ +#!/usr/bin/env python3 +""" +Security validations for archpkg-helper +""" + +import hashlib +import hmac +import os +from typing import Optional, Dict, Any +from pathlib import Path +from archpkg.logging_config import get_logger + +logger = get_logger(__name__) + +class SecurityValidator: + """Handles security validations for package updates""" + + def __init__(self): + self.supported_hash_algorithms = { + 'md5': hashlib.md5, + 'sha1': hashlib.sha1, + 'sha256': hashlib.sha256, + 'sha512': hashlib.sha512 + } + + def validate_checksum(self, file_path: Path, expected_hash: str, + algorithm: str = 'sha256') -> bool: + """Validate file checksum against expected hash""" + if algorithm not in self.supported_hash_algorithms: + logger.error(f"Unsupported hash algorithm: {algorithm}") + return False + + if not file_path.exists(): + logger.error(f"File does not exist: {file_path}") + return False + + try: + hash_func = self.supported_hash_algorithms[algorithm]() + + with open(file_path, 'rb') as f: + for chunk in iter(lambda: f.read(4096), b""): + hash_func.update(chunk) + + calculated_hash = hash_func.hexdigest() + is_valid = calculated_hash.lower() == expected_hash.lower() + + if is_valid: + logger.info(f"Checksum validation passed for {file_path.name}") + else: + logger.error(f"Checksum validation failed for {file_path.name}") + logger.debug(f"Expected: {expected_hash}") + logger.debug(f"Calculated: {calculated_hash}") + + return is_valid + + except Exception as e: + logger.error(f"Error validating checksum for {file_path}: {e}") + return False + + def generate_checksum(self, file_path: Path, algorithm: str = 'sha256') -> Optional[str]: + """Generate checksum for a file""" + if algorithm not in self.supported_hash_algorithms: + logger.error(f"Unsupported hash algorithm: {algorithm}") + return None + + if not file_path.exists(): + logger.error(f"File does not exist: {file_path}") + return None + + try: + hash_func = self.supported_hash_algorithms[algorithm]() + + with open(file_path, 'rb') as f: + for chunk in iter(lambda: f.read(4096), b""): + hash_func.update(chunk) + + return hash_func.hexdigest() + + except Exception as e: + logger.error(f"Error generating checksum for {file_path}: {e}") + return None + +class PackageSecurityValidator: + """Validates package security and integrity""" + + def __init__(self): + self.security_validator = SecurityValidator() + self.trusted_keys: Dict[str, str] = {} # Package name -> expected public key + + def validate_package_source(self, package_name: str, source: str) -> Dict[str, Any]: + """Validate that the package source is trusted""" + result = { + "valid": False, + "reason": "", + "warnings": [] + } + + # Basic source validation + trusted_sources = ['pacman', 'aur', 'flatpak', 'snap', 'apt', 'dnf'] + + if source not in trusted_sources: + result["reason"] = f"Unknown package source: {source}" + logger.warning(f"Package {package_name} from unknown source: {source}") + return result + + # Additional validation for AUR (less trusted) + if source == 'aur': + result["warnings"].append( + "AUR packages are user-contributed and may pose security risks" + ) + logger.warning(f"AUR package {package_name} - additional caution advised") + + result["valid"] = True + logger.info(f"Package source validation passed for {package_name} from {source}") + return result + + def validate_download_integrity(self, download_path: Path, + expected_checksums: Optional[Dict[str, str]] = None) -> Dict[str, Any]: + """Validate downloaded package integrity""" + result = { + "valid": False, + "checksums_validated": [], + "errors": [] + } + + if not download_path.exists(): + result["errors"].append(f"Download file does not exist: {download_path}") + return result + + # If no expected checksums provided, we can only do basic validation + if not expected_checksums: + logger.warning("No expected checksums provided - limited validation possible") + result["valid"] = True # Allow but warn + result["warnings"] = ["No checksum validation available"] + return result + + # Validate against provided checksums + all_valid = True + for algorithm, expected_hash in expected_checksums.items(): + is_valid = self.security_validator.validate_checksum( + download_path, expected_hash, algorithm + ) + + result["checksums_validated"].append({ + "algorithm": algorithm, + "valid": is_valid + }) + + if not is_valid: + all_valid = False + result["errors"].append(f"{algorithm} checksum validation failed") + + result["valid"] = all_valid + + if all_valid: + logger.info(f"Download integrity validation passed for {download_path.name}") + else: + logger.error(f"Download integrity validation failed for {download_path.name}") + + return result + + def validate_installation_safety(self, package_name: str, install_command: str) -> Dict[str, Any]: + """Validate that the installation command is safe to execute""" + result = { + "safe": False, + "warnings": [], + "blocked": False, + "reason": "" + } + + # Dangerous command patterns to block + dangerous_patterns = [ + 'rm -rf /', + 'rm -rf /*', + 'dd if=', + 'mkfs', + 'fdisk', + 'format', + 'wget.*|.*curl.*|.*bash', + 'chmod.*777', + 'chown.*root', + 'sudo.*su', + 'passwd', + 'shadow', + 'sudoers' + ] + + command_lower = install_command.lower() + + for pattern in dangerous_patterns: + if pattern in command_lower: + result["blocked"] = True + result["reason"] = f"Command contains dangerous pattern: {pattern}" + logger.error(f"Blocked dangerous install command for {package_name}: {pattern}") + return result + + # Check for sudo usage (warn but allow) + if 'sudo' in command_lower: + result["warnings"].append("Command uses sudo - ensure you have appropriate permissions") + + # Check for network downloads in install commands (warn) + if 'wget' in command_lower or 'curl' in command_lower: + result["warnings"].append("Command downloads from network - verify source trustworthiness") + + result["safe"] = True + logger.info(f"Installation safety validation passed for {package_name}") + return result + +class UpdateSecurityManager: + """Manages security for the update process""" + + def __init__(self): + self.package_validator = PackageSecurityValidator() + + def pre_update_validation(self, package_name: str, source: str, + install_command: str) -> Dict[str, Any]: + """Perform all security validations before allowing an update""" + validation_result = { + "approved": False, + "source_valid": False, + "command_safe": False, + "warnings": [], + "errors": [] + } + + # Validate package source + source_validation = self.package_validator.validate_package_source(package_name, source) + validation_result["source_valid"] = source_validation["valid"] + validation_result["warnings"].extend(source_validation.get("warnings", [])) + + if not source_validation["valid"]: + validation_result["errors"].append(f"Source validation failed: {source_validation['reason']}") + return validation_result + + # Validate installation command safety + command_validation = self.package_validator.validate_installation_safety(package_name, install_command) + validation_result["command_safe"] = command_validation["safe"] + validation_result["warnings"].extend(command_validation.get("warnings", [])) + + if command_validation["blocked"]: + validation_result["errors"].append(f"Command blocked: {command_validation['reason']}") + return validation_result + + # All validations passed + validation_result["approved"] = True + logger.info(f"Pre-update security validation passed for {package_name}") + + return validation_result + + def validate_downloaded_package(self, package_name: str, download_path: Path, + expected_checksums: Optional[Dict[str, str]] = None) -> Dict[str, Any]: + """Validate a downloaded package before installation""" + return self.package_validator.validate_download_integrity(download_path, expected_checksums) + +# Global security manager instance +security_manager = UpdateSecurityManager() + +def validate_update_security(package_name: str, source: str, install_command: str) -> Dict[str, Any]: + """Validate security for a package update""" + return security_manager.pre_update_validation(package_name, source, install_command) + +def validate_download_integrity(download_path: Path, expected_checksums: Optional[Dict[str, str]] = None) -> Dict[str, Any]: + """Validate download integrity""" + return security_manager.validate_downloaded_package("unknown", download_path, expected_checksums) \ No newline at end of file diff --git a/archpkg/templates/home.html b/archpkg/templates/home.html new file mode 100644 index 0000000..ff4d525 --- /dev/null +++ b/archpkg/templates/home.html @@ -0,0 +1,332 @@ + + + + + + ArchPkg Helper - Cross-Distro Package Manager + + + + + + + +
+
+

+ + Discover & Install Packages Across Linux Distributions +

+

+ A powerful cross-distro command-line utility that simplifies package discovery and installation + across Arch Linux, Ubuntu, Fedora, and more. No more guessing package names or commands! +

+ + Try Package Search + + + View on GitHub + +
+
+ +
+
+
+
+

What is ArchPkg Helper?

+

+ ArchPkg Helper is your universal companion for package management on Linux. + Whether you're using Arch Linux, Ubuntu, Fedora, or any other distribution, + our tool helps you find and install software with ease. +

+
+
+ +
+
+
+
+
+ +
+
Unified Search
+

+ Search across multiple package managers simultaneously. + Find packages from pacman, AUR, apt, dnf, flatpak, and snap in one place. +

+
+
+
+
+
+
+
+ +
+
Auto-Generated Commands
+

+ Get ready-to-use installation commands tailored to your distribution. + No more manual command crafting or syntax errors. +

+
+
+
+
+
+
+
+ +
+
Cross-Distro Support
+

+ Works seamlessly across all major Linux distributions. + Automatically detects your system and provides appropriate commands. +

+
+
+
+
+ +
+
+
+
+

How It Works

+
+
+
+
+ 1 +
+
+
Search
+

Enter a package name or keyword in our search interface

+
+
+
+
+ 2 +
+
+
Discover
+

Browse results from multiple package sources with descriptions

+
+
+
+
+ 3 +
+
+
Install
+

Copy and run the generated command for your distribution

+
+
+
+
+
+
+
+
+ +
+
+
+
+

Supported Package Managers

+

+ Comprehensive support for all major Linux package management systems +

+
+
+ +
+
+
+
+ +
pacman
+ Arch Linux +
+
+
+
+
+
+ +
AUR
+ Arch User Repo +
+
+
+
+
+
+ +
apt
+ Ubuntu/Debian +
+
+
+
+
+
+ +
dnf
+ Fedora +
+
+
+
+
+
+ +
Flatpak
+ Universal +
+
+
+
+
+
+ +
Snap
+ Universal +
+
+
+
+
+
+ +
+
+
+
+

Why Choose ArchPkg Helper?

+
+
+ +
+
+
+
+ +
+
+
Save Time
+

+ No more searching through multiple sources or remembering complex commands. + Get everything you need in seconds. +

+
+
+
+
+
+
+ +
+
+
User-Friendly
+

+ Perfect for both Linux newcomers and experienced users. + Clear interface and helpful guidance. +

+
+
+
+
+
+
+ +
+
+
Reliable
+

+ Uses official package repositories and APIs. + Always up-to-date with the latest package information. +

+
+
+
+
+
+
+ +
+
+
Open Source
+

+ Free and open source software. Contribute to the project + and help make Linux package management even better. +

+
+
+
+
+
+
+ + + + + + \ No newline at end of file diff --git a/archpkg/templates/search.html b/archpkg/templates/search.html new file mode 100644 index 0000000..774d701 --- /dev/null +++ b/archpkg/templates/search.html @@ -0,0 +1,218 @@ + + + + + + Search Packages - ArchPkg Helper + + + + + + + +
+
+
+

+ + Package Search +

+

+ Search for packages across multiple Linux distributions and package managers +

+
+ +
+
+
+
+ + +
+
+
+
+ +
+ +

Searching packages...

+
+ + + + +
+
+ + + + + + + + + \ No newline at end of file diff --git a/archpkg/update_manager.py b/archpkg/update_manager.py new file mode 100644 index 0000000..6f83094 --- /dev/null +++ b/archpkg/update_manager.py @@ -0,0 +1,225 @@ +#!/usr/bin/env python3 +""" +Update checking and management for archpkg-helper +""" + +import threading +import time +from typing import List, Dict, Optional, Any +from datetime import datetime, timezone +from archpkg.logging_config import get_logger +from archpkg.installed_apps import ( + get_all_installed_packages, + update_package_info, + InstalledPackage, + get_packages_needing_update_check +) +from archpkg.search_aur import search_aur +from archpkg.search_pacman import search_pacman +from archpkg.search_flatpak import search_flatpak +from archpkg.search_snap import search_snap +from archpkg.search_apt import search_apt +from archpkg.search_dnf import search_dnf +from archpkg.config_manager import get_user_config + +logger = get_logger(__name__) + +class UpdateChecker: + """Handles checking for package updates""" + + def __init__(self): + self.is_checking = False + self.last_check_time = None + + def check_for_updates(self, packages: Optional[List[InstalledPackage]] = None) -> Dict[str, Any]: + """Check for updates for installed packages""" + if self.is_checking: + logger.warning("Update check already in progress") + return {"status": "busy", "message": "Update check already in progress"} + + self.is_checking = True + self.last_check_time = datetime.now(timezone.utc) + + try: + if packages is None: + packages = get_all_installed_packages() + + if not packages: + logger.info("No packages to check for updates") + return {"status": "success", "checked": 0, "updates_found": 0} + + logger.info(f"Checking for updates for {len(packages)} packages") + + updates_found = 0 + checked_count = 0 + + for package in packages: + try: + has_update, latest_version = self._check_single_package(package) + checked_count += 1 + + if has_update: + update_package_info( + package.name, + available_version=latest_version, + update_available=True, + last_update_check=datetime.now(timezone.utc).isoformat() + ) + updates_found += 1 + logger.info(f"Update found for {package.name}: {latest_version}") + else: + update_package_info( + package.name, + update_available=False, + last_update_check=datetime.now(timezone.utc).isoformat() + ) + + except Exception as e: + logger.error(f"Failed to check updates for {package.name}: {e}") + continue + + result = { + "status": "success", + "checked": checked_count, + "updates_found": updates_found, + "timestamp": self.last_check_time.isoformat() + } + + logger.info(f"Update check completed: {checked_count} checked, {updates_found} updates found") + return result + + finally: + self.is_checking = False + + def _check_single_package(self, package: InstalledPackage) -> tuple[bool, Optional[str]]: + """Check for updates for a single package""" + logger.debug(f"Checking updates for {package.name} from {package.source}") + + # Search for the package in the same source it was installed from + results = [] + + try: + if package.source == "pacman": + results = search_pacman(package.name) + elif package.source == "aur": + results = search_aur(package.name) + elif package.source == "flatpak": + results = search_flatpak(package.name) + elif package.source == "snap": + results = search_snap(package.name) + elif package.source == "apt": + results = search_apt(package.name) + elif package.source == "dnf": + results = search_dnf(package.name) + else: + logger.warning(f"Unknown package source: {package.source}") + return False, None + + except Exception as e: + logger.error(f"Failed to search for {package.name}: {e}") + return False, None + + if not results: + logger.debug(f"No results found for {package.name}") + return False, None + + # Find the best match (usually the first result) + latest_name, latest_desc, latest_source = results[0] + + # For now, we'll assume any difference means an update is available + # In a real implementation, we'd need proper version comparison + # For this demo, we'll just check if the package exists + has_update = True # Simplified logic + latest_version = "latest" # Would need proper version parsing + + return has_update, latest_version + +class BackgroundUpdateManager: + """Manages background update checking and downloading""" + + def __init__(self): + self.update_checker = UpdateChecker() + self.background_thread = None + self.is_running = False + self.check_interval_hours = 24 + + def start_background_service(self) -> None: + """Start the background update service""" + if self.is_running: + logger.warning("Background update service already running") + return + + config = get_user_config() + if not config.auto_update_enabled: + logger.info("Auto-update not enabled, not starting background service") + return + + self.check_interval_hours = config.update_check_interval_hours + self.is_running = True + + self.background_thread = threading.Thread( + target=self._background_worker, + daemon=True, + name="archpkg-update-checker" + ) + self.background_thread.start() + + logger.info("Background update service started") + + def stop_background_service(self) -> None: + """Stop the background update service""" + self.is_running = False + if self.background_thread: + self.background_thread.join(timeout=5) + logger.info("Background update service stopped") + + def _background_worker(self) -> None: + """Background worker thread""" + logger.info("Background update worker started") + + while self.is_running: + try: + # Check if we need to run update checks + packages_to_check = get_packages_needing_update_check(self.check_interval_hours) + + if packages_to_check: + logger.info(f"Background update check: {len(packages_to_check)} packages need checking") + result = self.update_checker.check_for_updates(packages_to_check) + + if result.get("updates_found", 0) > 0: + logger.info(f"Background update check found {result['updates_found']} updates") + # Could send notifications here + + # Sleep for the check interval + sleep_time = self.check_interval_hours * 3600 + time.sleep(min(sleep_time, 300)) # Sleep in 5-minute chunks to allow quick shutdown + + except Exception as e: + logger.error(f"Background update worker error: {e}") + time.sleep(300) # Wait 5 minutes before retrying + + logger.info("Background update worker stopped") + + def trigger_manual_check(self) -> Dict[str, Any]: + """Manually trigger an update check""" + logger.info("Manual update check triggered") + return self.update_checker.check_for_updates() + +# Global update manager instance +update_manager = BackgroundUpdateManager() + +def check_for_updates(packages: Optional[List[InstalledPackage]] = None) -> Dict[str, Any]: + """Check for updates for installed packages""" + return update_manager.update_checker.check_for_updates(packages) + +def start_background_updates() -> None: + """Start background update service""" + update_manager.start_background_service() + +def stop_background_updates() -> None: + """Stop background update service""" + update_manager.stop_background_service() + +def trigger_update_check() -> Dict[str, Any]: + """Trigger a manual update check""" + return update_manager.trigger_manual_check() \ No newline at end of file diff --git a/archpkg/web.py b/archpkg/web.py new file mode 100644 index 0000000..48056ac --- /dev/null +++ b/archpkg/web.py @@ -0,0 +1,123 @@ +#!/usr/bin/env python3 +""" +Web interface for archpkg-helper +""" + +from flask import Flask, render_template, request, jsonify +import os +from archpkg.config import DISTRO_MAP +from archpkg.search_aur import search_aur +from archpkg.search_pacman import search_pacman +from archpkg.search_flatpak import search_flatpak +from archpkg.search_snap import search_snap +from archpkg.search_apt import search_apt +from archpkg.search_dnf import search_dnf +from archpkg.command_gen import generate_command + +app = Flask(__name__) + +@app.route('/') +def home(): + """Home page with description of the tool""" + return render_template('home.html') + +@app.route('/search') +def search_page(): + """Search interface""" + return render_template('search.html') + +@app.route('/api/search') +def api_search(): + """API endpoint for package search""" + query = request.args.get('q', '') + if not query: + return jsonify({'error': 'Query parameter required'}), 400 + + results = [] + + # Search across all package managers + try: + pacman_results = search_pacman(query) + for name, desc, source in pacman_results: + command = generate_command(name, source) + results.append({ + 'name': name, + 'description': desc or 'No description available', + 'manager': source, + 'command': command or 'Command not available' + }) + except Exception: + pass + + try: + aur_results = search_aur(query) + for name, desc, source in aur_results: + command = generate_command(name, source) + results.append({ + 'name': name, + 'description': desc or 'No description available', + 'manager': source, + 'command': command or 'Command not available' + }) + except Exception: + pass + + try: + flatpak_results = search_flatpak(query) + for name, desc, source in flatpak_results: + command = generate_command(name, source) + results.append({ + 'name': name, + 'description': desc or 'No description available', + 'manager': source, + 'command': command or 'Command not available' + }) + except Exception: + pass + + try: + snap_results = search_snap(query) + for name, desc, source in snap_results: + command = generate_command(name, source) + results.append({ + 'name': name, + 'description': desc or 'No description available', + 'manager': source, + 'command': command or 'Command not available' + }) + except Exception: + pass + + try: + apt_results = search_apt(query) + for name, desc, source in apt_results: + command = generate_command(name, source) + results.append({ + 'name': name, + 'description': desc or 'No description available', + 'manager': source, + 'command': command or 'Command not available' + }) + except Exception: + pass + + try: + dnf_results = search_dnf(query) + for name, desc, source in dnf_results: + command = generate_command(name, source) + results.append({ + 'name': name, + 'description': desc or 'No description available', + 'manager': source, + 'command': command or 'Command not available' + }) + except Exception: + pass + + # Sort by relevance (simple: prefer exact matches, then shorter names) + results.sort(key=lambda x: (x['name'].lower().find(query.lower()), len(x['name']))) + + return jsonify({'results': results[:50]}) # Limit results + +if __name__ == '__main__': + app.run(debug=True) \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index e17c807..b76532d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,9 +14,13 @@ dependencies = [ "requests", "rich", "typer[all]", - "distro" + "distro", + "flask" ] +[project.optional-dependencies] +github = ["GitPython"] + [project.scripts] archpkg = "archpkg.cli:main" @@ -24,4 +28,4 @@ archpkg = "archpkg.cli:main" packages = ["archpkg"] [tool.setuptools.package-data] -archpkg = ["*.yaml", "*.yml", "*.json"] +archpkg = ["*.yaml", "*.yml", "*.json", "templates/*.html"] From be28d0cb76796db82e93507edbcd9f06dbc66522 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 9 Dec 2025 05:21:03 +0000 Subject: [PATCH 3/6] Integrate new CLI commands while preserving existing functionality Co-authored-by: AdmGenSameer <154604600+AdmGenSameer@users.noreply.github.com> --- archpkg/cli.py | 480 ++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 479 insertions(+), 1 deletion(-) diff --git a/archpkg/cli.py b/archpkg/cli.py index 6ca59b1..40824e1 100755 --- a/archpkg/cli.py +++ b/archpkg/cli.py @@ -26,6 +26,11 @@ from archpkg.logging_config import get_logger, PackageHelperLogger from archpkg.suggest import suggest_apps, list_purposes from archpkg.cache import get_cache_manager, CacheConfig +from archpkg.github_install import install_from_github, validate_github_url +from archpkg.config_manager import get_user_config, set_config_option +from archpkg.update_manager import check_for_updates, trigger_update_check +from archpkg.download_manager import install_updates, start_background_update_service, stop_background_update_service +from archpkg.installed_apps import track_package, get_all_installed_packages, get_packages_with_updates, InstalledPackage console = Console() logger = get_logger(__name__) @@ -352,6 +357,399 @@ def handle_upgrade_command() -> None: border_style="red" )) +def handle_install_command(args) -> None: + """Handle the install command for one or more packages.""" + packages = args.packages + logger.info(f"Install command for packages: {packages}") + + # Check if it's a GitHub installation + if len(packages) == 1 and (packages[0].startswith('github:') or packages[0].startswith('https://github.com/')): + github_url = validate_github_url(packages[0]) + if github_url: + success = install_from_github(packages[0]) + if success: + console.print("[bold green]✓ GitHub installation completed![/bold green]") + else: + console.print("[bold red]✗ GitHub installation failed.[/bold red]") + sys.exit(1) + else: + console.print("[red]Invalid GitHub URL format[/red]") + sys.exit(1) + return + + # Determine tracking preference + should_track = args.track and not args.no_track + + # Normal package installation + detected = detect_distro() + + if len(packages) == 1: + # Single package installation + package_name = packages[0] + logger.info(f"Installing single package: {package_name}") + + # Search for the package + results = [] + if detected == "arch": + try: + results.extend(search_aur(package_name)) + except Exception: + pass + try: + results.extend(search_pacman(package_name)) + except Exception: + pass + elif detected == "debian": + try: + results.extend(search_apt(package_name)) + except Exception: + pass + elif detected == "fedora": + try: + results.extend(search_dnf(package_name)) + except Exception: + pass + + # Universal package managers + try: + results.extend(search_flatpak(package_name)) + except Exception: + pass + try: + results.extend(search_snap(package_name)) + except Exception: + pass + + if not results: + console.print(f"[red]No packages found for '{package_name}'[/red]") + sys.exit(1) + + top_matches = get_top_matches(package_name, results, limit=5) + if not top_matches: + console.print(f"[red]No suitable matches found for '{package_name}'[/red]") + sys.exit(1) + + # Display options + table = Table(title="Found Packages") + table.add_column("Index", style="cyan", no_wrap=True) + table.add_column("Package Name", style="green") + table.add_column("Source", style="blue") + table.add_column("Description", style="yellow") + + for idx, (pkg, desc, source) in enumerate(top_matches, 1): + table.add_row(str(idx), pkg, source, desc or "No description") + + console.print(table) + + try: + choice = input("\nSelect a package to install [1-5 or press Enter to cancel]: ") + if not choice.strip(): + console.print("[yellow]Installation cancelled.[/yellow]") + return + + choice = int(choice) + if not (1 <= choice <= len(top_matches)): + console.print("[red]Invalid choice.[/red]") + return + + pkg, desc, source = top_matches[choice - 1] + command = generate_command(pkg, source) + + if not command: + console.print(f"[red]Cannot generate install command for {source}[/red]") + return + + console.print(f"\n[bold green]Install Command:[/bold green] {command}") + console.print("[bold yellow]Press Enter to install, or Ctrl+C to cancel...[/bold yellow]") + input() + + console.print("[blue]Running install command...[/blue]") + exit_code = os.system(command) + + if exit_code == 0: + console.print(f"[bold green]✓ Successfully installed {pkg}![/bold green]") + if should_track: + track_package(pkg, source, "latest") + console.print("[dim]Package tracked for updates[/dim]") + else: + console.print(f"[red]✗ Installation failed with exit code {exit_code}[/red]") + sys.exit(1) + + except KeyboardInterrupt: + console.print("\n[yellow]Installation cancelled.[/yellow]") + return + except ValueError: + console.print("[red]Invalid input. Please enter a number.[/red]") + return + else: + # Batch installation + logger.info(f"Batch installation for packages: {packages}") + batch_install_packages(packages, should_track) + +def batch_install_packages(package_names: List[str], should_track: bool = True) -> None: + """Install multiple packages in batch mode with progress tracking.""" + logger.info(f"Starting batch installation for packages: {package_names}") + + if not package_names: + console.print("[red]No packages specified for batch installation.[/red]") + return + + detected = detect_distro() + console.print(f"\n[bold cyan]Batch Installation Mode[/bold cyan]") + console.print(f"Target platform: [cyan]{detected}[/cyan]") + console.print(f"Packages to install: [yellow]{', '.join(package_names)}[/yellow]\n") + + # Validate all packages first + console.print("[blue]Validating packages...[/blue]") + validated_packages = [] + validation_errors = [] + + for i, pkg_name in enumerate(package_names, 1): + console.print(f" [{i}/{len(package_names)}] Checking '{pkg_name}'...") + + results = [] + + if detected == "arch": + try: + results.extend(search_aur(pkg_name)) + except Exception: + pass + try: + results.extend(search_pacman(pkg_name)) + except Exception: + pass + elif detected == "debian": + try: + results.extend(search_apt(pkg_name)) + except Exception: + pass + elif detected == "fedora": + try: + results.extend(search_dnf(pkg_name)) + except Exception: + pass + + try: + results.extend(search_flatpak(pkg_name)) + except Exception: + pass + try: + results.extend(search_snap(pkg_name)) + except Exception: + pass + + if not results: + validation_errors.append(f"'{pkg_name}': No packages found") + console.print(f" [red]✗[/red] No packages found") + continue + + top_matches = get_top_matches(pkg_name, results, limit=1) + if not top_matches: + validation_errors.append(f"'{pkg_name}': No suitable matches found") + console.print(f" [red]✗[/red] No suitable matches found") + continue + + pkg, desc, source = top_matches[0] + command = generate_command(pkg, source) + + if not command: + validation_errors.append(f"'{pkg_name}': Cannot generate install command") + console.print(f" [red]✗[/red] Cannot generate install command") + continue + + validated_packages.append((pkg_name, pkg, desc, source, command)) + console.print(f" [green]✓[/green] Found: {pkg} ({source})") + + if validation_errors: + console.print(f"\n[red]Validation failed for {len(validation_errors)} package(s):[/red]") + for error in validation_errors: + console.print(f" - {error}") + console.print("\n[yellow]Batch installation cancelled.[/yellow]") + return + + console.print(f"\n[green]✓ All {len(validated_packages)} packages validated![/green]\n") + + # Proceed with installation + console.print("[blue]Starting batch installation...[/blue]\n") + + successful_installs = [] + failed_installs = [] + + for i, (original_name, pkg, desc, source, command) in enumerate(validated_packages, 1): + console.print(f"[{i}/{len(validated_packages)}] Installing [bold cyan]{pkg}[/bold cyan] ({source})...") + console.print(f" Command: [dim]{command}[/dim]") + + exit_code = os.system(command) + + if exit_code == 0: + console.print(f" [green]✓[/green] Successfully installed {pkg}") + successful_installs.append(pkg) + if should_track: + try: + track_package(pkg, source, "latest") + except Exception as e: + logger.warning(f"Failed to track package {pkg}: {e}") + else: + console.print(f" [red]✗[/red] Failed to install {pkg}") + failed_installs.append((pkg, exit_code)) + + # Summary + console.print(f"\n[bold]Batch Installation Summary:[/bold]") + console.print(f"Total packages: {len(validated_packages)}") + console.print(f"[green]Successful: {len(successful_installs)}[/green]") + if successful_installs: + console.print(f" - {', '.join(successful_installs)}") + if failed_installs: + console.print(f"[red]Failed: {len(failed_installs)}[/red]") + for pkg, code in failed_installs: + console.print(f" - {pkg} (exit code: {code})") + +def handle_update_command(args) -> None: + """Handle the update command.""" + packages = args.packages if args.packages else None + check_only = args.check_only + + if check_only: + console.print("[blue]Checking for updates...[/blue]") + result = trigger_update_check() + + if result["status"] == "success": + updates_found = result.get("updates_found", 0) + if updates_found > 0: + console.print(f"[green]Found {updates_found} update(s) available![/green]") + console.print("Run 'archpkg update' to install them.") + else: + console.print("[green]All packages are up to date![/green]") + else: + console.print(f"[red]Update check failed: {result.get('message', 'Unknown error')}[/red]") + else: + console.print("[blue]Installing updates...[/blue]") + result = install_updates(packages) + + if result["status"] == "success": + installed = result.get("installed", 0) + failed = result.get("failed", 0) + + if installed > 0: + console.print(f"[green]Successfully installed {installed} update(s)[/green]") + + if failed > 0: + console.print(f"[red]Failed to install {failed} update(s)[/red]") + for item in result.get("results", []): + if item.get("status") == "failed": + console.print(f" - {item['package']}: {item.get('error', 'Unknown error')}") + + if installed == 0 and failed == 0: + console.print("[green]No updates available[/green]") + else: + console.print("[red]Update installation failed[/red]") + +def handle_config_command(args) -> None: + """Handle the config command.""" + if args.list: + config = get_user_config() + console.print("[bold]Current Configuration:[/bold]") + for key, value in config.__dict__.items(): + if not key.startswith('_'): + console.print(f" {key}: {value}") + return + + if not args.key: + console.print("[red]Error: Specify a configuration key or use --list[/red]") + return + + if args.value is None: + # Get value + config = get_user_config() + if hasattr(config, args.key): + console.print(f"{args.key}: {getattr(config, args.key)}") + else: + console.print(f"[red]Unknown configuration key: {args.key}[/red]") + else: + # Set value + try: + value = args.value + # Convert to appropriate type + if value.lower() in ('true', 'false'): + value = value.lower() == 'true' + elif value.isdigit(): + value = int(value) + elif value.replace('.', '').isdigit(): + value = float(value) + + set_config_option(args.key, value) + console.print(f"[green]Updated {args.key} = {value}[/green]") + except Exception as e: + console.print(f"[red]Failed to update configuration: {e}[/red]") + +def handle_list_installed_command(args) -> None: + """Handle the list-installed command.""" + packages = get_all_installed_packages() + + if not packages: + console.print("[yellow]No packages are currently being tracked for updates.[/yellow]") + console.print("Use 'archpkg install --track ' to track packages.") + return + + table = Table(title="Tracked Installed Packages") + table.add_column("Package Name", style="green") + table.add_column("Source", style="blue") + table.add_column("Installed Version", style="cyan") + table.add_column("Update Available", style="yellow") + table.add_column("Last Updated", style="magenta") + + for pkg in packages: + update_status = "[green]No[/green]" if not pkg.update_available else "[red]Yes[/red]" + last_updated = pkg.install_date or "Never" + table.add_row( + pkg.name, + pkg.source, + pkg.version or "Unknown", + update_status, + last_updated + ) + + console.print(table) + +def handle_service_command(args) -> None: + """Handle the service command.""" + action = args.action + + if action == "start": + start_background_update_service() + console.print("[green]Background update service started[/green]") + elif action == "stop": + stop_background_update_service() + console.print("[green]Background update service stopped[/green]") + elif action == "status": + config = get_user_config() + if config.auto_update_enabled: + console.print("[green]Background update service is enabled[/green]") + console.print(f"Check interval: {config.update_check_interval_hours} hours") + console.print(f"Auto-install: {'Enabled' if config.auto_install_updates else 'Disabled'}") + else: + console.print("[yellow]Background update service is disabled[/yellow]") + +def handle_web_command(args) -> None: + """Handle the web command.""" + try: + from archpkg.web import app as web_app + console.print(Panel( + f"[bold cyan]🌐 Starting Web Interface[/bold cyan]\n\n" + f"Host: [yellow]{args.host}[/yellow]\n" + f"Port: [yellow]{args.port}[/yellow]\n\n" + f"Access the interface at:\n" + f"[cyan]http://{args.host}:{args.port}/[/cyan]\n\n" + f"Press [bold]Ctrl+C[/bold] to stop the server.", + title="Web Interface", + border_style="cyan" + )) + web_app.run(host=args.host, port=args.port, debug=False) + except KeyboardInterrupt: + console.print("\n[yellow]Web server stopped.[/yellow]") + except Exception as e: + console.print(f"[red]Failed to start web server: {e}[/red]") + def handle_search_errors(source_name: str, error: Exception) -> None: """Centralized error handling for search operations.""" PackageHelperLogger.log_exception(logger, f"{source_name} search failed", error) @@ -532,6 +930,63 @@ def main() -> None: formatter_class=argparse.RawDescriptionHelpFormatter ) + # Install command + install_parser = subparsers.add_parser( + 'install', + help='Install one or more packages', + formatter_class=argparse.RawDescriptionHelpFormatter + ) + install_parser.add_argument('packages', type=str, nargs='+', help='Package name(s) to install') + install_parser.add_argument('--source', '-s', type=str, help='Package source (auto-detected if not specified)') + install_parser.add_argument('--track', action='store_true', default=True, help='Track installation for updates (default: enabled)') + install_parser.add_argument('--no-track', action='store_true', help='Do not track installation for updates') + + # Update command + update_parser = subparsers.add_parser( + 'update', + help='Check for and install package updates', + formatter_class=argparse.RawDescriptionHelpFormatter + ) + update_parser.add_argument('packages', type=str, nargs='*', help='Specific packages to update (all if not specified)') + update_parser.add_argument('--check-only', '-c', action='store_true', help='Only check for updates, do not install') + + # Config command + config_parser = subparsers.add_parser( + 'config', + help='Manage archpkg configuration settings', + formatter_class=argparse.RawDescriptionHelpFormatter + ) + config_parser.add_argument('key', type=str, nargs='?', help='Configuration key to get/set') + config_parser.add_argument('value', type=str, nargs='?', help='Value to set (if setting)') + config_parser.add_argument('--list', '-l', action='store_true', help='List all configuration settings') + + # List-installed command + list_installed_parser = subparsers.add_parser( + 'list-installed', + help='List all installed packages being tracked for updates', + formatter_class=argparse.RawDescriptionHelpFormatter + ) + + # Service command + service_parser = subparsers.add_parser( + 'service', + help='Manage the background update service', + formatter_class=argparse.RawDescriptionHelpFormatter + ) + service_parser.add_argument('action', type=str, choices=['start', 'stop', 'status'], help='Action to perform') + + # Web command + web_parser = subparsers.add_parser( + 'web', + help='Start the web interface', + formatter_class=argparse.RawDescriptionHelpFormatter + ) + web_parser.add_argument('--port', '-p', type=int, default=5000, help='Port to run the web server on (default: 5000)') + web_parser.add_argument('--host', type=str, default='127.0.0.1', help='Host to bind to (default: 127.0.0.1)') + + # GitHub install command (as a special format of install) + # This will be handled in the install command by checking for github: prefix + # Handle help manually to show custom format if '--help' in sys.argv or '-h' in sys.argv: parser.print_help() @@ -601,6 +1056,24 @@ def main() -> None: elif args.command == 'upgrade': handle_upgrade_command() return + elif args.command == 'install': + handle_install_command(args) + return + elif args.command == 'update': + handle_update_command(args) + return + elif args.command == 'config': + handle_config_command(args) + return + elif args.command == 'list-installed': + handle_list_installed_command(args) + return + elif args.command == 'service': + handle_service_command(args) + return + elif args.command == 'web': + handle_web_command(args) + return elif args.command == 'search' or args.command is None: # Default to search behavior for backward compatibility handle_search_command(args, cache_manager) @@ -610,8 +1083,13 @@ def main() -> None: "[red]Unknown command.[/red]\n\n" "[bold cyan]Available commands:[/bold cyan]\n" "- [cyan]archpkg search firefox[/cyan] - Search for packages by name\n" + "- [cyan]archpkg install firefox[/cyan] - Install one or more packages\n" + "- [cyan]archpkg update[/cyan] - Check for and install package updates\n" "- [cyan]archpkg suggest video editing[/cyan] - Get app suggestions by purpose\n" - "- [cyan]archpkg suggest --list[/cyan] - List all available purposes\n" + "- [cyan]archpkg config --list[/cyan] - View configuration settings\n" + "- [cyan]archpkg list-installed[/cyan] - List tracked installed packages\n" + "- [cyan]archpkg service start[/cyan] - Start background update service\n" + "- [cyan]archpkg web[/cyan] - Start web interface\n" "- [cyan]archpkg upgrade[/cyan] - Upgrade archpkg tool from GitHub\n" "- [cyan]archpkg --help[/cyan] - Show help information", title="Invalid Command", From 43ed81d497ce7bd10686ea048982d8c507e23e4c Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 9 Dec 2025 05:26:13 +0000 Subject: [PATCH 4/6] Address code review feedback - improve security, logging, and documentation Co-authored-by: AdmGenSameer <154604600+AdmGenSameer@users.noreply.github.com> --- archpkg/cli.py | 4 ++-- archpkg/download_manager.py | 25 +++++++++++++++---------- archpkg/security.py | 35 +++++++++++++++++++++-------------- archpkg/update_manager.py | 14 +++++++++----- archpkg/web.py | 26 ++++++++++++++------------ 5 files changed, 61 insertions(+), 43 deletions(-) diff --git a/archpkg/cli.py b/archpkg/cli.py index 40824e1..89edb00 100755 --- a/archpkg/cli.py +++ b/archpkg/cli.py @@ -377,8 +377,8 @@ def handle_install_command(args) -> None: sys.exit(1) return - # Determine tracking preference - should_track = args.track and not args.no_track + # Determine tracking preference (track is default, no_track overrides it) + should_track = not args.no_track # Normal package installation detected = detect_distro() diff --git a/archpkg/download_manager.py b/archpkg/download_manager.py index f618997..7e8a8eb 100644 --- a/archpkg/download_manager.py +++ b/archpkg/download_manager.py @@ -227,22 +227,27 @@ def install_updates(self, package_names: Optional[List[str]] = None) -> Dict[str } def _install_single_update(self, package_name: str) -> Dict[str, Any]: - """Install update for a single package""" + """Install update for a single package + + NOTE: This is a simplified implementation for demonstration purposes. + A production implementation should: + 1. Retrieve actual package metadata including version information + 2. Generate and execute the appropriate install command for the package manager + 3. Verify installation success through package manager queries + 4. Handle rollback on failure + 5. Support different package sources (pacman, apt, dnf, etc.) + """ logger.info(f"Installing update for {package_name}") - - # This is a placeholder - actual installation would depend on the package source - # For now, we'll just mark it as installed + logger.warning("Update installation uses simplified logic - manual installation recommended for production use") try: - # In a real implementation, this would: - # 1. Get the package info - # 2. Run the appropriate installation command based on source - # 3. Verify installation - # 4. Update the installed package info + # TODO: Implement actual package installation logic + # For now, we just update the tracking information + # In production, this should execute package manager commands update_package_info( package_name, - installed_version="latest", # Would be actual version + installed_version="unknown", # Indicates version detection not implemented update_available=False, last_updated=datetime.now(timezone.utc).isoformat() ) diff --git a/archpkg/security.py b/archpkg/security.py index 6484cc7..21f01c9 100644 --- a/archpkg/security.py +++ b/archpkg/security.py @@ -169,26 +169,33 @@ def validate_installation_safety(self, package_name: str, install_command: str) } # Dangerous command patterns to block + # NOTE: This is a basic pattern matching implementation + # A production system should use: + # 1. Proper shell command parsing + # 2. Regex patterns for more flexible matching + # 3. Whitelist of allowed commands rather than blacklist + # 4. Integration with security scanning tools + import re + dangerous_patterns = [ - 'rm -rf /', - 'rm -rf /*', - 'dd if=', - 'mkfs', - 'fdisk', - 'format', - 'wget.*|.*curl.*|.*bash', - 'chmod.*777', - 'chown.*root', - 'sudo.*su', - 'passwd', - 'shadow', - 'sudoers' + r'rm\s+-rf\s+/', # Recursive delete from root + r'dd\s+if=', # Disk operations + r'mkfs', # Filesystem formatting + r'fdisk', # Disk partitioning + r'format', # Formatting operations + r'wget.*\|.*bash', # Pipe to bash + r'curl.*\|.*bash', # Pipe to bash + r'chmod\s+777', # Overly permissive permissions + r'chown.*root', # Change ownership to root + r'passwd', # Password changes + r'shadow', # Shadow file access + r'sudoers', # Sudoers file modification ] command_lower = install_command.lower() for pattern in dangerous_patterns: - if pattern in command_lower: + if re.search(pattern, command_lower): result["blocked"] = True result["reason"] = f"Command contains dangerous pattern: {pattern}" logger.error(f"Blocked dangerous install command for {package_name}: {pattern}") diff --git a/archpkg/update_manager.py b/archpkg/update_manager.py index 6f83094..8771aad 100644 --- a/archpkg/update_manager.py +++ b/archpkg/update_manager.py @@ -126,11 +126,15 @@ def _check_single_package(self, package: InstalledPackage) -> tuple[bool, Option # Find the best match (usually the first result) latest_name, latest_desc, latest_source = results[0] - # For now, we'll assume any difference means an update is available - # In a real implementation, we'd need proper version comparison - # For this demo, we'll just check if the package exists - has_update = True # Simplified logic - latest_version = "latest" # Would need proper version parsing + # NOTE: This is a simplified implementation for demonstration purposes + # A production implementation should: + # 1. Parse and compare semantic versions (e.g., using packaging.version) + # 2. Query package manager APIs for actual version information + # 3. Handle different versioning schemes across package managers + # For now, we conservatively report updates may be available + logger.warning(f"Update check for {package.name} uses simplified logic - manual verification recommended") + has_update = False # Changed to False to avoid false positives + latest_version = "unknown" # Indicates version comparison not implemented return has_update, latest_version diff --git a/archpkg/web.py b/archpkg/web.py index 48056ac..a731721 100644 --- a/archpkg/web.py +++ b/archpkg/web.py @@ -13,8 +13,10 @@ from archpkg.search_apt import search_apt from archpkg.search_dnf import search_dnf from archpkg.command_gen import generate_command +from archpkg.logging_config import get_logger app = Flask(__name__) +logger = get_logger(__name__) @app.route('/') def home(): @@ -46,8 +48,8 @@ def api_search(): 'manager': source, 'command': command or 'Command not available' }) - except Exception: - pass + except Exception as e: + logger.warning(f"Pacman search failed for query '{query}': {e}") try: aur_results = search_aur(query) @@ -59,8 +61,8 @@ def api_search(): 'manager': source, 'command': command or 'Command not available' }) - except Exception: - pass + except Exception as e: + logger.warning(f"AUR search failed for query '{query}': {e}") try: flatpak_results = search_flatpak(query) @@ -72,8 +74,8 @@ def api_search(): 'manager': source, 'command': command or 'Command not available' }) - except Exception: - pass + except Exception as e: + logger.warning(f"Flatpak search failed for query '{query}': {e}") try: snap_results = search_snap(query) @@ -85,8 +87,8 @@ def api_search(): 'manager': source, 'command': command or 'Command not available' }) - except Exception: - pass + except Exception as e: + logger.warning(f"Snap search failed for query '{query}': {e}") try: apt_results = search_apt(query) @@ -98,8 +100,8 @@ def api_search(): 'manager': source, 'command': command or 'Command not available' }) - except Exception: - pass + except Exception as e: + logger.warning(f"APT search failed for query '{query}': {e}") try: dnf_results = search_dnf(query) @@ -111,8 +113,8 @@ def api_search(): 'manager': source, 'command': command or 'Command not available' }) - except Exception: - pass + except Exception as e: + logger.warning(f"DNF search failed for query '{query}': {e}") # Sort by relevance (simple: prefer exact matches, then shorter names) results.sort(key=lambda x: (x['name'].lower().find(query.lower()), len(x['name']))) From 642faf63298b8a27b4fd47e6370a2b6515c9da4d Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 9 Dec 2025 05:28:06 +0000 Subject: [PATCH 5/6] Fix Flask debug mode security issue - disable debug in production Co-authored-by: AdmGenSameer <154604600+AdmGenSameer@users.noreply.github.com> --- archpkg/web.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/archpkg/web.py b/archpkg/web.py index a731721..a775817 100644 --- a/archpkg/web.py +++ b/archpkg/web.py @@ -122,4 +122,5 @@ def api_search(): return jsonify({'results': results[:50]}) # Limit results if __name__ == '__main__': - app.run(debug=True) \ No newline at end of file + # Debug mode disabled for security - never enable in production + app.run(debug=False) \ No newline at end of file From 52e7179aaf6fd6b07e06246b0cf05839e731bbc7 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 9 Dec 2025 05:35:07 +0000 Subject: [PATCH 6/6] Fix command recognition - add new commands to known commands list Co-authored-by: AdmGenSameer <154604600+AdmGenSameer@users.noreply.github.com> --- archpkg/cli.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/archpkg/cli.py b/archpkg/cli.py index 89edb00..752bc61 100755 --- a/archpkg/cli.py +++ b/archpkg/cli.py @@ -880,7 +880,8 @@ def main() -> None: break # If we found a positional arg and it's not a known command, inject 'search' - if first_arg_idx < len(sys.argv) and sys.argv[first_arg_idx] not in ['search', 'suggest', 'upgrade', '-h', '--help']: + known_commands = ['search', 'suggest', 'upgrade', 'install', 'update', 'config', 'list-installed', 'service', 'web', '-h', '--help'] + if first_arg_idx < len(sys.argv) and sys.argv[first_arg_idx] not in known_commands: sys.argv.insert(first_arg_idx, 'search') parser = argparse.ArgumentParser(