feat(v1/proxmox): storage + per-VM snapshot endpoints (direct httpx)

app/routes/v1/proxmox/__init__.py

@@ -1,8 +1,12 @@
 from fastapi import APIRouter
 
 from app.routes.v1.proxmox.hosts import router as hosts_router
+from app.routes.v1.proxmox.snapshots import router as snapshots_router
+from app.routes.v1.proxmox.storage import router as storage_router
 from app.routes.v1.proxmox.vms import router as vms_router
 
 router = APIRouter(prefix="/proxmox", tags=["v1 proxmox"])
 router.include_router(hosts_router)
 router.include_router(vms_router)
+router.include_router(storage_router)
+router.include_router(snapshots_router)

app/routes/v1/proxmox/_helpers.py

@@ -0,0 +1,80 @@
+"""Shared helpers for the v1 proxmox route modules (vms/hosts/storage/snapshots).
+
+Talk to PVE directly over httpx with a registered host's token. Extracted so
+sibling modules reuse one copy instead of cross-importing privates.
+"""
+from __future__ import annotations
+
+import json
+
+import httpx
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.db import get_session_factory
+from app.core.errors import AuthFailedError, Range42Error, VmidProtectedError
+from app.core.logging import get_logger
+from app.core.models import ProxmoxHost
+from app.core.vmid_guard import VmidProtectedError as GuardVmidProtected
+from app.core.vmid_guard import assert_vmid_safe
+
+log = get_logger(__name__)
+
+
+async def _session() -> AsyncSession:
+    async with get_session_factory()() as session:
+        yield session
+
+
+async def _get_host(host_id: str, session: AsyncSession) -> ProxmoxHost:
+    row = (
+        await session.execute(select(ProxmoxHost).where(ProxmoxHost.id == host_id))
+    ).scalar_one_or_none()
+    if row is None:
+        raise Range42Error(
+            error="not_found", code="NOT_FOUND", status=404,
+            message=f"Host {host_id} not found",
+        )
+    return row
+
+
+def _auth_headers(row: ProxmoxHost) -> dict[str, str]:
+    return {"Authorization": f"PVEAPIToken={row.token_ref}"}
+
+
+def _assert_vmid_safe(row: ProxmoxHost, vmid: int, action: str) -> None:
+    overrides = (
+        json.loads(row.protected_vmids_override_json)
+        if row.protected_vmids_override_json else None
+    )
+    try:
+        assert_vmid_safe(vmid, host_overrides=overrides)
+    except GuardVmidProtected as e:
+        raise VmidProtectedError(
+            message=f"VMID {vmid} is protected; '{action}' is refused",
+            details=e.details,
+        ) from e
+
+
+def _unreachable(row: ProxmoxHost, err: Exception) -> Range42Error:
+    log.warning("proxmox unreachable", host=row.id, err=str(err))
+    return Range42Error(
+        error="upstream_error", code="PROXMOX_UNREACHABLE", status=502,
+        message=f"Proxmox host {row.name} is unreachable",
+        details=[{"field": "api_url", "reason": str(err)[:200]}],
+    )
+
+
+def _raise_for_pve(row: ProxmoxHost, r: httpx.Response, action: str = "request") -> None:
+    """Map a non-200 PVE response to the canonical error classes."""
+    if r.status_code in (401, 403):
+        raise AuthFailedError(details=[{
+            "field": "token_ref",
+            "reason": f"Proxmox API rejected credentials ({r.status_code})",
+        }])
+    if r.status_code != 200:
+        raise Range42Error(
+            error="upstream_error", code="PROXMOX_ERROR", status=502,
+            message=f"Proxmox returned {r.status_code} for {action}",
+            details=[{"field": "host", "reason": (r.text or "")[:300]}],
+        )

app/routes/v1/proxmox/hosts.py

@@ -11,22 +11,17 @@
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.core.db import get_session_factory
 from app.core.errors import AuthFailedError, Range42Error
 from app.core.logging import get_logger
 from app.core.models import ProxmoxHost
+from app.routes.v1.proxmox._helpers import _session
 from app.schemas.v1.common import Page
 from app.schemas.v1.proxmox import HostHealth, HostIn, HostOut
 
 router = APIRouter()
 log = get_logger(__name__)
 
 
-async def _session() -> AsyncSession:
-    async with get_session_factory()() as session:
-        yield session
-
-
 def _row_to_out(row: ProxmoxHost) -> HostOut:
     overrides = (
         json.loads(row.protected_vmids_override_json)

app/routes/v1/proxmox/snapshots.py

@@ -0,0 +1,128 @@
+"""/v1/proxmox/hosts/{id}/vms/{vmid}/snapshots — per-VM PVE snapshots.
+
+Direct PVE httpx. NET-NEW vs v0: supports vmtype=lxc and the vmstate flag
+(v0 was qemu-only, no vmstate). create/delete/rollback return UPIDs.
+"""
+from __future__ import annotations
+
+from typing import Literal
+
+import httpx
+from fastapi import APIRouter, Depends, Path
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.routes.v1.proxmox._helpers import (
+    _assert_vmid_safe, _auth_headers, _get_host, _raise_for_pve, _session,
+    _unreachable,
+)
+from app.schemas.v1.common import Page
+from app.schemas.v1.proxmox import SnapshotItem, SnapshotCreateIn, VmActionResult
+
+router = APIRouter()
+
+
+def _snap_base(row, vmtype: str, vmid: int) -> str:
+    return (
+        f"{row.api_url.rstrip('/')}/api2/json/nodes/{row.node_name}"
+        f"/{vmtype}/{vmid}/snapshot"
+    )
+
+
+@router.get(
+    "/hosts/{host_id}/vms/{vmid}/snapshots", response_model=Page[SnapshotItem]
+)
+async def list_snapshots(
+    host_id: str,
+    vmid: int,
+    vmtype: Literal["qemu", "lxc"] = "qemu",
+    session: AsyncSession = Depends(_session),
+):
+    row = await _get_host(host_id, session)
+    try:
+        async with httpx.AsyncClient(verify=False, timeout=15) as cli:
+            r = await cli.get(_snap_base(row, vmtype, vmid), headers=_auth_headers(row))
+    except httpx.RequestError as e:
+        raise _unreachable(row, e) from e
+    _raise_for_pve(row, r, "snapshot list")
+    data = r.json().get("data", []) or []
+    items = [
+        SnapshotItem(
+            name=d["name"], description=d.get("description"),
+            snaptime=d.get("snaptime"),
+            vmstate=bool(d["vmstate"]) if d.get("vmstate") is not None else None,
+            parent=d.get("parent"),
+        )
+        for d in data
+        if d.get("name") and d["name"] != "current"
+    ]
+    return Page(items=items, total=len(items))
+
+
+@router.post("/hosts/{host_id}/vms/{vmid}/snapshots", response_model=VmActionResult)
+async def create_snapshot(
+    host_id: str,
+    vmid: int,
+    body: SnapshotCreateIn,
+    vmtype: Literal["qemu", "lxc"] = "qemu",
+    session: AsyncSession = Depends(_session),
+):
+    row = await _get_host(host_id, session)
+    form: dict[str, str | int] = {"snapname": body.snapname}
+    if body.description is not None:
+        form["description"] = body.description
+    # vmstate (save running RAM state) is a qemu-only PVE option; PVE rejects it
+    # on /lxc/.../snapshot, so omit it for LXC (matches community.proxmox).
+    if vmtype == "qemu" and body.vmstate is not None:
+        form["vmstate"] = 1 if body.vmstate else 0
+    try:
+        async with httpx.AsyncClient(verify=False, timeout=15) as cli:
+            r = await cli.post(_snap_base(row, vmtype, vmid),
+                               headers=_auth_headers(row), data=form)
+    except httpx.RequestError as e:
+        raise _unreachable(row, e) from e
+    _raise_for_pve(row, r, "snapshot create")
+    return VmActionResult(status="accepted", upid=r.json().get("data"))
+
+
+@router.delete(
+    "/hosts/{host_id}/vms/{vmid}/snapshots/{name}", response_model=VmActionResult
+)
+async def delete_snapshot(
+    host_id: str,
+    vmid: int,
+    vmtype: Literal["qemu", "lxc"] = "qemu",
+    name: str = Path(pattern=r"^[A-Za-z0-9_][A-Za-z0-9._-]*$"),
+    session: AsyncSession = Depends(_session),
+):
+    row = await _get_host(host_id, session)
+    url = f"{_snap_base(row, vmtype, vmid)}/{name}"
+    try:
+        async with httpx.AsyncClient(verify=False, timeout=15) as cli:
+            r = await cli.delete(url, headers=_auth_headers(row))
+    except httpx.RequestError as e:
+        raise _unreachable(row, e) from e
+    _raise_for_pve(row, r, "snapshot delete")
+    return VmActionResult(status="accepted", upid=r.json().get("data"))
+
+
+@router.post(
+    "/hosts/{host_id}/vms/{vmid}/snapshots/{name}/rollback",
+    response_model=VmActionResult,
+)
+async def rollback_snapshot(
+    host_id: str,
+    vmid: int,
+    vmtype: Literal["qemu", "lxc"] = "qemu",
+    name: str = Path(pattern=r"^[A-Za-z0-9_][A-Za-z0-9._-]*$"),
+    session: AsyncSession = Depends(_session),
+):
+    row = await _get_host(host_id, session)
+    _assert_vmid_safe(row, vmid, "rollback")
+    url = f"{_snap_base(row, vmtype, vmid)}/{name}/rollback"
+    try:
+        async with httpx.AsyncClient(verify=False, timeout=15) as cli:
+            r = await cli.post(url, headers=_auth_headers(row))
+    except httpx.RequestError as e:
+        raise _unreachable(row, e) from e
+    _raise_for_pve(row, r, "snapshot rollback")
+    return VmActionResult(status="accepted", upid=r.json().get("data"))

app/routes/v1/proxmox/storage.py

@@ -0,0 +1,113 @@
+"""/v1/proxmox/hosts/{id}/storage — pools + content listing + download-url.
+
+Direct PVE httpx. Replaces the v0 Ansible-driven storage surface.
+"""
+from __future__ import annotations
+
+from typing import Literal
+
+import httpx
+from fastapi import APIRouter, Depends, Path
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.routes.v1.proxmox._helpers import (
+    _auth_headers, _get_host, _raise_for_pve, _session, _unreachable,
+)
+from app.schemas.v1.common import Page
+from app.schemas.v1.proxmox import DownloadUrlIn, StorageContent, StoragePool, VmActionResult
+
+router = APIRouter()
+
+
+def _volid_name(volid: str) -> str:
+    """'local:iso/ubuntu.iso' -> 'ubuntu.iso'; 'local:vztmpl/x.tar.zst' -> 'x.tar.zst'."""
+    if "/" in volid:
+        return volid.rsplit("/", 1)[-1]
+    return volid.rsplit(":", 1)[-1]
+
+
+@router.get(
+    "/hosts/{host_id}/storage/{store}/content",
+    response_model=Page[StorageContent],
+)
+async def list_storage_content(
+    host_id: str,
+    content: Literal["iso", "vztmpl"] = "iso",
+    store: str = Path(pattern=r"^[A-Za-z0-9][A-Za-z0-9._-]*$"),
+    session: AsyncSession = Depends(_session),
+):
+    row = await _get_host(host_id, session)
+    url = (
+        f"{row.api_url.rstrip('/')}/api2/json/nodes/{row.node_name}"
+        f"/storage/{store}/content"
+    )
+    try:
+        async with httpx.AsyncClient(verify=False, timeout=15) as cli:
+            r = await cli.get(url, headers=_auth_headers(row),
+                              params={"content": content})
+    except httpx.RequestError as e:
+        raise _unreachable(row, e) from e
+    _raise_for_pve(row, r, "storage content")
+    data = r.json().get("data", []) or []
+    items = [
+        StorageContent(
+            volid=d["volid"], name=_volid_name(d["volid"]),
+            content=d.get("content", content), size=d.get("size"),
+            format=d.get("format"), vmid=d.get("vmid"),
+        )
+        for d in data
+    ]
+    return Page(items=items, total=len(items))
+
+
+@router.get("/hosts/{host_id}/storage", response_model=Page[StoragePool])
+async def list_storage(host_id: str, session: AsyncSession = Depends(_session)):
+    row = await _get_host(host_id, session)
+    url = f"{row.api_url.rstrip('/')}/api2/json/nodes/{row.node_name}/storage"
+    try:
+        async with httpx.AsyncClient(verify=False, timeout=15) as cli:
+            r = await cli.get(url, headers=_auth_headers(row))
+    except httpx.RequestError as e:
+        raise _unreachable(row, e) from e
+    _raise_for_pve(row, r, "storage list")
+    data = r.json().get("data", []) or []
+    items = [
+        StoragePool(
+            storage=d.get("storage"), type=d.get("type"), content=d.get("content"),
+            total=d.get("total"), used=d.get("used"), avail=d.get("avail"),
+            active=bool(d["active"]) if d.get("active") is not None else None,
+        )
+        for d in data
+    ]
+    return Page(items=items, total=len(items))
+
+
+@router.post(
+    "/hosts/{host_id}/storage/{store}/download-url",
+    response_model=VmActionResult,
+)
+async def storage_download_url(
+    host_id: str,
+    body: DownloadUrlIn,
+    store: str = Path(pattern=r"^[A-Za-z0-9][A-Za-z0-9._-]*$"),
+    session: AsyncSession = Depends(_session),
+):
+    row = await _get_host(host_id, session)
+    url = (
+        f"{row.api_url.rstrip('/')}/api2/json/nodes/{row.node_name}"
+        f"/storage/{store}/download-url"
+    )
+    form: dict[str, str] = {
+        "content": body.content, "filename": body.filename, "url": body.url,
+    }
+    if body.checksum is not None:
+        form["checksum"] = body.checksum
+    if body.checksum_algorithm is not None:
+        form["checksum-algorithm"] = body.checksum_algorithm
+    try:
+        async with httpx.AsyncClient(verify=False, timeout=15) as cli:
+            r = await cli.post(url, headers=_auth_headers(row), data=form)
+    except httpx.RequestError as e:
+        raise _unreachable(row, e) from e
+    _raise_for_pve(row, r, "download-url")
+    return VmActionResult(status="accepted", upid=r.json().get("data"))