571 lines
19 KiB
Python
571 lines
19 KiB
Python
"""
|
|
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", ".vdi"}
|
|
|
|
ARCHIVE_EXTENSIONS = {".7z", ".zip", ".rar", ".tar.gz", ".tgz", ".tar", ".gz", ".bz2", ".xz", ".ova"}
|
|
|
|
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():
|
|
# Try finding by name — walk the parent directory
|
|
parent = filepath.parent
|
|
fname = filepath.name
|
|
if parent.exists() and parent.is_dir():
|
|
found = list(parent.rglob(fname))
|
|
if found:
|
|
filepath = found[0]
|
|
logger.info("Found by name: %s", filepath)
|
|
else:
|
|
raise FileNotFoundError(f"Source not found: {filepath}")
|
|
else:
|
|
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)]),
|
|
".ova": ("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."""
|
|
return _pick_largest(discover_all_disks(source))
|
|
|
|
|
|
def discover_all_disks(source: Path) -> list:
|
|
"""Walk the source directory and return all disk images found,
|
|
sorted by size descending."""
|
|
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}")
|
|
|
|
candidates.sort(key=lambda p: p.stat().st_size, reverse=True)
|
|
return [_probe(p) for p in candidates]
|
|
|
|
|
|
def _pick_largest(disks: list) -> DiskInfo:
|
|
if not disks:
|
|
raise FileNotFoundError("No disks")
|
|
return disks[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_config(vmx_path: Path) -> dict:
|
|
"""Parse VMware .vmx file for all useful VM configuration."""
|
|
result = {}
|
|
try:
|
|
text = vmx_path.read_text(errors="ignore")
|
|
except OSError:
|
|
return result
|
|
|
|
def _get(key: str) -> Optional[str]:
|
|
m = re.search(rf'{key}\s*=\s*"([^"]*)"', text, re.IGNORECASE)
|
|
return m.group(1) if m else None
|
|
|
|
raw_os = _get("guestOS")
|
|
if raw_os:
|
|
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_os.lower().replace("_", "").replace("-", "").replace(" ", "")
|
|
for k, v in mapping.items():
|
|
if k in raw_lower:
|
|
result["os_type"] = v
|
|
break
|
|
if "os_type" not in result:
|
|
result["os_type"] = raw_os.replace("-", " ").replace("_", " ").title()
|
|
|
|
cores = _get("numvcpus")
|
|
if cores:
|
|
try:
|
|
result["cpu_cores"] = int(cores)
|
|
except ValueError:
|
|
pass
|
|
|
|
mem = _get("memSize")
|
|
if mem:
|
|
try:
|
|
result["ram_mb"] = int(mem)
|
|
except ValueError:
|
|
pass
|
|
|
|
fw = _get("firmware")
|
|
if fw and fw.lower() == "efi":
|
|
result["boot_type"] = "uefi"
|
|
elif fw and fw.lower() == "bios":
|
|
result["boot_type"] = "legacy"
|
|
|
|
return result
|
|
|
|
|
|
def _parse_vmx_guest_os(vmx_path: Path) -> Optional[str]:
|
|
"""Parse VMware .vmx file for guestOS field (backward-compat)."""
|
|
cfg = _parse_vmx_config(vmx_path)
|
|
return cfg.get("os_type")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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...")
|
|
disks = discover_all_disks(source)
|
|
disk = disks[0]
|
|
logger.info("Disk found: %s (format=%s, size=%.1f GiB)", disk.path, disk.format, disk.size_gb)
|
|
logger.info("Total disks discovered: %d", len(disks))
|
|
|
|
_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
|
|
all_disks_data = [{
|
|
"filename": d.path.name,
|
|
"format": d.format,
|
|
"size_gb": d.size_gb,
|
|
"size_bytes": d.size_bytes,
|
|
} for d in disks]
|
|
|
|
# Detect OVF for qm importovf
|
|
ovf_path = None
|
|
vmx_config = {}
|
|
for dirpath, _, filenames in os.walk(source if source.is_dir() else source.parent):
|
|
for fname in filenames:
|
|
if fname.lower().endswith(".ovf") and not ovf_path:
|
|
ovf_path = str(Path(dirpath) / fname)
|
|
logger.info("OVF descriptor found: %s", ovf_path)
|
|
if fname.lower().endswith(".vmx"):
|
|
vmx_path = Path(dirpath) / fname
|
|
vmx_config = _parse_vmx_config(vmx_path)
|
|
logger.info("VMX config found: %s → %s", fname, vmx_config)
|
|
if ovf_path and vmx_config:
|
|
break
|
|
|
|
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,
|
|
all_disks=all_disks_data,
|
|
ovf_path=ovf_path,
|
|
vmx_config=vmx_config,
|
|
)
|
|
_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)]),
|
|
".ova": ("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])
|