""" Backend API — Proxmox Image Conversion Engine Runs on the Proxmox host (srv2) at http://10.2.0.2:9000 All API routes live under /api/v1/. Matches open-api.yaml v1.1.0 spec exactly. """ from __future__ import annotations import logging from logging.handlers import RotatingFileHandler from pathlib import Path from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from models import ( JobSubmissionRequest, JobStatusResponse, CleanupRequest, CleanupResponse, AnalyzeRequest, AnalyzeResponse, AnalyzeStatusResponse, CloneRequest, HealthResponse, ErrorResponse, ) from converter import extract_if_needed, discover_disk, detect_os, detect_efi, start_analysis, get_analysis_status from provisioner import submit_job, get_job_status, cleanup_staging, clone_vm # --------------------------------------------------------------------------- # Logging # --------------------------------------------------------------------------- logger = logging.getLogger("backend") logger.setLevel(logging.INFO) # Console → systemd journal _ch = logging.StreamHandler() _ch.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s")) logger.addHandler(_ch) # File handler → shared log directory LOG_DIR = Path(__file__).resolve().parent.parent / "logs" LOG_DIR.mkdir(parents=True, exist_ok=True) _fh = RotatingFileHandler( LOG_DIR / "vm-bench-backend.log", maxBytes=10 * 1024 * 1024, backupCount=5, ) _fh.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s %(message)s")) logger.addHandler(_fh) logger.info("Backend starting — log file: %s", LOG_DIR / "vm-bench-backend.log") # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- API_PREFIX = "/api/v1" # --------------------------------------------------------------------------- # App # --------------------------------------------------------------------------- app = FastAPI( title="Proxmox Image Conversion Engine", version="1.1.0", docs_url="/docs", ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _validate_filename(filename: str) -> None: """Reject filenames that attempt path traversal.""" if ".." in filename or filename.startswith("/"): raise HTTPException(status_code=400, detail="Invalid filename: path traversal not allowed") # --------------------------------------------------------------------------- # Routes — all under /api/v1/ # --------------------------------------------------------------------------- @app.get(f"{API_PREFIX}/health", response_model=HealthResponse) def health() -> HealthResponse: """Health check.""" return HealthResponse(status="ok") @app.get(f"{API_PREFIX}/storage/pools") def storage_pools(): """Return available Proxmox storage pools that can host VM disks.""" result = [] try: import subprocess out = subprocess.run(["/usr/sbin/pvesm", "status"], capture_output=True, text=True, timeout=10) for line in out.stdout.split("\n")[1:]: parts = line.split() if len(parts) < 3: continue name, stype, active = parts[0], parts[1], parts[2] if active != "active": continue if stype not in ("lvmthin", "zfspool", "rbd", "dir", "nfs"): continue result.append({"name": name, "type": stype}) except Exception as exc: logger.warning("Storage pool discovery failed: %s", exc) logger.info("Storage pools: %s", [p["name"] for p in result]) return {"pools": result} @app.get(f"{API_PREFIX}/sessions") def list_sessions(): """Return existing session directories with staged files that can be resumed.""" tmp = Path("/mnt/converter/tmp") if Path("/mnt/converter/tmp").exists() else Path(__file__).resolve().parent.parent / "tmp" sessions = [] try: for entry in sorted(tmp.iterdir(), reverse=True): if not entry.is_dir(): continue in_dir = entry / "in" if not in_dir.is_dir(): continue session_files = [] total_bytes = 0 has_disks = False all_files = sorted([f for f in in_dir.rglob("*") if f.is_file()], key=lambda f: f.stat().st_size, reverse=True) for f in all_files: sz = f.stat().st_size total_bytes += sz rel = str(f.relative_to(in_dir)) session_files.append({"name": rel, "size_bytes": sz, "size_gb": round(sz / (1024**3), 1)}) if f.suffix.lower() in {".vmdk", ".qcow2", ".img", ".raw", ".vhd", ".vhdx"}: has_disks = True if not session_files: continue archives = {".7z", ".zip", ".rar", ".tar.gz", ".tgz", ".tar", ".gz", ".bz2", ".xz", ".ova"} has_source = has_disks or any(f["name"].lower().endswith(tuple(archives)) for f in session_files) sessions.append({ "session_id": entry.name, "files": session_files[:10], "total_size_gb": round(total_bytes / (1024**3), 1), "has_disks": has_disks, "resumable": has_source, }) except Exception: pass return {"sessions": sessions} @app.post(f"{API_PREFIX}/analyze", response_model=AnalyzeStatusResponse, status_code=202) def analyze(req: AnalyzeRequest) -> AnalyzeStatusResponse: """Analyze a source file asynchronously: extraction + disk probing + OS detection.""" _validate_filename(req.source_filename) logger.info("Analyze request: vmid=%d file=%s", req.vmid, req.source_filename) try: result = start_analysis(req.vmid, req.source_filename) return AnalyzeStatusResponse( analysis_id=result["analysis_id"], vmid=req.vmid, status=result["status"], message=result["message"], ) except Exception as exc: logger.error("Analysis start failed: %s", exc) raise HTTPException(status_code=400, detail=str(exc)) @app.get(f"{API_PREFIX}/analyze/{{analysis_id}}", response_model=AnalyzeStatusResponse) def analyze_status(analysis_id: str) -> AnalyzeStatusResponse: """Poll async analysis status.""" try: result = get_analysis_status(analysis_id) return AnalyzeStatusResponse(**result) except KeyError: logger.warning("Analysis not found: %s", analysis_id) raise HTTPException(status_code=404, detail=f"Analysis not found: {analysis_id}") @app.post(f"{API_PREFIX}/jobs", response_model=JobStatusResponse, status_code=202) def create_job(req: JobSubmissionRequest) -> JobStatusResponse: """Submit a conversion + provisioning job.""" logger.info("Job submit: vmid=%d name=%s disk=%s storage=%s", req.vmid, req.vm_name, req.boot_disk.source_filename, req.target_storage) try: result = submit_job(req) logger.info("Job queued: %s", result.job_id) return result except Exception as exc: logger.error("Job submission failed: %s", exc) raise HTTPException(status_code=400, detail=str(exc)) @app.get(f"{API_PREFIX}/jobs/{{job_id}}", response_model=JobStatusResponse) def job_status(job_id: str) -> JobStatusResponse: """Get current status of a conversion job.""" try: result = get_job_status(job_id) return result except Exception: logger.warning("Job not found: %s", job_id) raise HTTPException(status_code=404, detail=f"Job not found: {job_id}") @app.post(f"{API_PREFIX}/jobs/{{job_id}}/cleanup", response_model=CleanupResponse) def cleanup_job(job_id: str, req: CleanupRequest) -> CleanupResponse: """Delete or preserve staging files for a job.""" try: vmid = int(job_id.split("_")[1]) except (IndexError, ValueError): vmid = 0 logger.info("Cleanup: job=%s vmid=%d delete=%s", job_id, vmid, req.delete_staging_files) result = cleanup_staging(vmid, req.delete_staging_files, req.session_id) return CleanupResponse( job_id=result["job_id"], action_taken=result["action_taken"], message=result["message"], ) @app.post(f"{API_PREFIX}/clone", response_model=JobStatusResponse, status_code=202) def clone_vm_endpoint(req: CloneRequest) -> JobStatusResponse: """Clone an existing VM with a new ID and name.""" logger.info("Clone request: %d → %d (%s)", req.source_vmid, req.target_vmid, req.target_name) try: return clone_vm(req) except Exception as exc: logger.error("Clone failed: %s", exc) raise HTTPException(status_code=400, detail=str(exc)) # --------------------------------------------------------------------------- # Run # --------------------------------------------------------------------------- if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=9000)