""" Archive extraction, disk probing, and OS detection. Runs on the Proxmox host (srv2) where qemu-img, 7z, unzip, tar are available. """ from __future__ import annotations import logging import os import re import shutil import subprocess import threading import time from pathlib import Path from dataclasses import dataclass from typing import Optional STAGING_ROOT = Path(__file__).resolve().parent.parent / "tmp" STAGING_IN = STAGING_ROOT / "in" logger = logging.getLogger("backend.converter") DISK_EXTENSIONS = {".vmdk", ".qcow2", ".qcow", ".img", ".raw", ".vhd", ".vhdx"} ARCHIVE_EXTENSIONS = {".7z", ".zip", ".rar", ".tar.gz", ".tgz", ".tar", ".gz", ".bz2", ".xz"} MIN_DISK_BYTES = 1024 @dataclass class DiskInfo: path: Path format: str # from qemu-img size_bytes: int size_gb: float os_type: Optional[str] = None bootable: bool = False efi_detectable: bool = False # --------------------------------------------------------------------------- # Extraction # --------------------------------------------------------------------------- def extract_if_needed(filename: str) -> Path: """Extract an archive into STAGING_IN if it's compressed. Returns the path to the extracted directory (or the original file if not compressed). Accepts both absolute paths and paths relative to STAGING_IN.""" filepath = Path(filename) if not filepath.is_absolute(): filepath = STAGING_ROOT / filename if not filepath.exists(): raise FileNotFoundError(f"Source not found: {filepath}") # Detect archive type name = filepath.name.lower() ext = _detect_archive_ext(name) if ext is None: return filepath # not an archive, return as-is # Determine extraction tool tool_map = { ".7z": ("7z", ["7z", "x", "-y", str(filepath)]), ".zip": ("7z", ["7z", "x", "-y", str(filepath)]), ".rar": ("unrar", ["unrar", "x", "-y", str(filepath)]), ".tar.gz": ("tar", ["tar", "-xzf", str(filepath)]), ".tgz": ("tar", ["tar", "-xzf", str(filepath)]), ".tar": ("tar", ["tar", "-xf", str(filepath)]), ".gz": ("gunzip", ["gunzip", "-f", str(filepath)]), ".bz2": ("bunzip2", ["bunzip2", "-f", str(filepath)]), ".xz": ("xz", ["xz", "-d", str(filepath)]), } tool, cmd = tool_map[ext] if not shutil.which(tool): raise RuntimeError(f"Extraction tool '{tool}' not found on host") # Determine extraction base: use the file's parent directory # (handles both tmp/in/ and tmp/{guid}/in/ paths) extract_base = filepath.parent extract_base.mkdir(parents=True, exist_ok=True) # Directory stem: strip extension(s) stem = name for e in sorted(tool_map, key=len, reverse=True): if stem.endswith(e): stem = stem[:-len(e)] break # If already extracted, skip re-extraction expected_dir = extract_base / stem if expected_dir.is_dir() and any( f.suffix.lower() in DISK_EXTENSIONS for f in expected_dir.rglob("*") if f.is_file() ): logger.info("Already extracted: %s", expected_dir) return expected_dir # Snapshot before extraction before = set(extract_base.iterdir()) if extract_base.exists() else set() # Extract into the file's parent directory result = subprocess.run(cmd, cwd=str(extract_base), capture_output=True, text=True, timeout=600) if result.returncode != 0: raise RuntimeError(f"{tool} failed: {result.stderr.strip()[:500]}") # Determine what was created after = set(extract_base.iterdir()) new_items = after - before # If the archive had a single top-level directory matching the stem, use it expected_dir = extract_base / stem if expected_dir in new_items and expected_dir.is_dir(): result = expected_dir else: out_dir = extract_base / stem out_dir.mkdir(parents=True, exist_ok=True) for item in new_items: shutil.move(str(item), str(out_dir / item.name)) result = out_dir # Flatten nested single-child directories (archive quirk) result = _flatten_nested(result) return result def _flatten_nested(base: Path) -> Path: """If a directory has exactly one child that is also a directory, move the child's contents up and remove the child. Repeat until stable. Returns the (possibly changed) base path.""" changed = True while changed: changed = False if not base.is_dir(): break entries = [e for e in base.iterdir() if e.name not in ('.', '..')] if len(entries) == 1 and entries[0].is_dir(): inner = entries[0] # Move everything from inner into base for item in list(inner.iterdir()): shutil.move(str(item), str(base / item.name)) inner.rmdir() changed = True return base def _detect_archive_ext(name: str) -> Optional[str]: for ext in sorted(ARCHIVE_EXTENSIONS, key=len, reverse=True): if name.endswith(ext): return ext return None # --------------------------------------------------------------------------- # Disk discovery # --------------------------------------------------------------------------- def discover_disk(source: Path) -> DiskInfo: """Walk the source directory and find the primary disk image.""" if source.is_file(): return _probe(source) candidates = [] for dirpath, _, filenames in os.walk(source): for fname in filenames: fpath = Path(dirpath) / fname if fpath.suffix.lower() in DISK_EXTENSIONS and fpath.stat().st_size >= MIN_DISK_BYTES: candidates.append(fpath) if not candidates: raise FileNotFoundError(f"No disk image found in {source}") # Pick largest as primary candidates.sort(key=lambda p: p.stat().st_size, reverse=True) return _probe(candidates[0]) def _probe(disk_path: Path) -> DiskInfo: """Run qemu-img info and extract format + virtual size.""" result = subprocess.run( ["qemu-img", "info", "--output=json", str(disk_path)], capture_output=True, text=True, timeout=30, ) fmt = disk_path.suffix.lstrip(".").lower() virtual_size = disk_path.stat().st_size if result.returncode == 0: import json data = json.loads(result.stdout) fmt = data.get("format", fmt) virtual_size = data.get("virtual-size", virtual_size) size_gb = round(virtual_size / (1024 ** 3), 1) return DiskInfo( path=disk_path, format=fmt, size_bytes=virtual_size, size_gb=size_gb, ) # --------------------------------------------------------------------------- # OS detection # --------------------------------------------------------------------------- def detect_os(disk_path: Path, extract_dir: Path) -> Optional[str]: """Try multiple strategies to identify the guest OS. 1. Parse .vmx sidecar file for guestOS field 2. Read /etc/os-release from disk via libguestfs if available 3. Heuristic from filename / directory name """ # Strategy 1: .vmx sidecar vmx = _find_sidecar(extract_dir, ".vmx") if vmx: os_type = _parse_vmx_guest_os(vmx) if os_type: return os_type # Strategy 2: libguestfs (if available) if shutil.which("virt-inspector"): try: result = subprocess.run( ["virt-inspector", str(disk_path)], capture_output=True, text=True, timeout=60, ) if result.returncode == 0: import xml.etree.ElementTree as ET root = ET.fromstring(result.stdout) os_elem = root.find(".//operatingsystem") if os_elem is not None and os_elem.text: return f"{os_elem.text}" except Exception: pass # Strategy 3: filename heuristics name = disk_path.name.lower() os_hints = { "debian": "Debian", "ubuntu": "Ubuntu", "centos": "CentOS", "rhel": "RHEL", "fedora": "Fedora", "windows": "Windows", "win10": "Windows 10", "win11": "Windows 11", "alpine": "Alpine", "arch": "Arch Linux", "opensuse": "openSUSE", } for keyword, label in os_hints.items(): if keyword in name: return label # Also check parent directory name parent = str(extract_dir).lower() for keyword, label in os_hints.items(): if keyword in parent: return label return None def _find_sidecar(base: Path, ext: str) -> Optional[Path]: """Find first file with given extension in the directory tree.""" if not base.is_dir(): return None for dirpath, _, filenames in os.walk(base): for fname in filenames: if fname.lower().endswith(ext): return Path(dirpath) / fname return None def _parse_vmx_guest_os(vmx_path: Path) -> Optional[str]: """Parse VMware .vmx file for guestOS field.""" try: text = vmx_path.read_text(errors="ignore") except OSError: return None m = re.search(r'guestOS\s*=\s*"([^"]+)"', text, re.IGNORECASE) if not m: return None raw = m.group(1) # Map common VMware guestOS values mapping = { "debian": "Debian", "ubuntu": "Ubuntu", "centos": "CentOS", "rhel": "RHEL", "fedora": "Fedora", "windows": "Windows", "other": "Other Linux", "other-64": "Other Linux (64-bit)", "other26xlinux": "Linux 2.6.x", "other3xlinux": "Linux 3.x+", "other4xlinux": "Linux 4.x+", "other5xlinux": "Linux 5.x+", } raw_lower = raw.lower().replace("_", "").replace("-", "").replace(" ", "") for k, v in mapping.items(): if k in raw_lower: return v return raw.replace("-", " ").replace("_", " ").title() # --------------------------------------------------------------------------- # EFI detection # --------------------------------------------------------------------------- def detect_efi(disk_path: Path) -> Optional[bool]: """Check whether the disk has an EFI System Partition using guestfish. Returns True if a VFAT EFI partition is found, False if only non-EFI filesystems detected, None if guestfish isn't available or fails. """ if not shutil.which("guestfish"): return None try: result = subprocess.run( ["guestfish", "--ro", "-a", str(disk_path), "-i", "list-filesystems"], capture_output=True, text=True, timeout=30, ) if result.returncode != 0: return None if "vfat" in result.stdout.lower(): return True return False except Exception: return None # --------------------------------------------------------------------------- # Async analysis — background extraction + probing # --------------------------------------------------------------------------- _analyses: dict[str, dict] = {} _analyses_lock = threading.Lock() def start_analysis(vmid: int, source_filename: str) -> dict: analysis_id = f"analysis_{vmid}_{int(time.time())}" with _analyses_lock: _analyses[analysis_id] = { "vmid": vmid, "status": "queued", "message": "Queued", "result": None, "error": None, } thread = threading.Thread(target=_process_analysis, args=(analysis_id, vmid, source_filename), daemon=True) thread.start() return {"analysis_id": analysis_id, "status": "queued", "message": "Analysis queued"} def get_analysis_status(analysis_id: str) -> dict: with _analyses_lock: a = _analyses.get(analysis_id) if not a: raise KeyError(analysis_id) return { "analysis_id": analysis_id, "vmid": a["vmid"], "status": a["status"], "message": a.get("message", ""), "result": a.get("result"), "error_details": a.get("error"), } def _update_analysis(analysis_id: str, status: str, message: str = "", result=None, error: str = None): with _analyses_lock: a = _analyses.get(analysis_id) if not a: return a["status"] = status if message: a["message"] = message if result is not None: a["result"] = result if error is not None: a["error"] = error def _process_analysis(analysis_id: str, vmid: int, source_filename: str): try: _update_analysis(analysis_id, "processing", "Extracting archive...") source = extract_if_needed(source_filename) logger.info("Extraction done: %s", source) _update_analysis(analysis_id, "processing", "Extracting nested archives...") source = _extract_nested(source) _update_analysis(analysis_id, "processing", "Discovering disk image...") disk = discover_disk(source) logger.info("Disk found: %s (format=%s, size=%.1f GiB)", disk.path, disk.format, disk.size_gb) _update_analysis(analysis_id, "processing", "Detecting OS and boot type...") extract_dir = source if source.is_dir() else source.parent os_type = detect_os(disk.path, extract_dir) efi = detect_efi(disk.path) logger.info("Analysis done: os=%s efi=%s", os_type, efi) from models import AnalyzeResponse result = AnalyzeResponse( vmid=vmid, filename=disk.path.name, disk_format=disk.format, disk_size_gb=disk.size_gb, os_type=os_type, efi_detectable=efi, ) _update_analysis(analysis_id, "completed", "Analysis complete", result=result.model_dump()) except Exception as exc: logger.exception("Analysis %s failed", analysis_id) _update_analysis(analysis_id, "failed", "Analysis failed", error=str(exc)) # --------------------------------------------------------------------------- # Nested archive extraction — handles split zips, chained archives # --------------------------------------------------------------------------- def _extract_nested(base: Path) -> Path: """Recursively extract nested archives until disk images are found or no more archives remain. Handles split zips (.z01-.zNN + .zip).""" for _ in range(5): if not base.is_dir(): return base cand = _find_nested_archive(base) if cand is None: break logger.info("Extracting nested archive: %s", cand) _extract_one_archive(cand, base) try: cand.unlink() except Exception: pass base = _flatten_nested(base) return base def _find_nested_archive(base: Path) -> Optional[Path]: """Find the next archive to extract inside a directory. Prioritises split zip masters if .z01 segments are present.""" files = sorted([f for f in base.iterdir() if f.is_file()], key=lambda f: f.name) split_master = None has_segments = False for f in files: name = f.name.lower() if re.search(r'\.z\d{2,3}$', name): has_segments = True if name.endswith(".zip") and not re.search(r'\.z\d{2,3}$', name): split_master = f if has_segments and split_master: return split_master for f in files: if _detect_archive_ext(f.name.lower()) is not None: return f return None def _extract_one_archive(archive_path: Path, target_dir: Path): name = archive_path.name.lower() ext = _detect_archive_ext(name) tool_map = { ".7z": ("7z", ["7z", "x", "-y", str(archive_path)]), ".zip": ("7z", ["7z", "x", "-y", str(archive_path)]), ".rar": ("unrar", ["unrar", "x", "-y", str(archive_path)]), ".tar.gz": ("tar", ["tar", "-xzf", str(archive_path)]), ".tgz": ("tar", ["tar", "-xzf", str(archive_path)]), ".tar": ("tar", ["tar", "-xf", str(archive_path)]), ".gz": ("gunzip", ["gunzip", "-f", str(archive_path)]), ".bz2": ("bunzip2", ["bunzip2", "-f", str(archive_path)]), ".xz": ("xz", ["xz", "-d", str(archive_path)]), } tool, cmd = tool_map[ext] if not shutil.which(tool): raise RuntimeError(f"Extraction tool '{tool}' not found on host") result = subprocess.run(cmd, cwd=str(target_dir), capture_output=True, text=True, timeout=900) if result.returncode != 0: logger.warning("Nested extraction %s failed: %s", archive_path.name, result.stderr.strip()[:200])