fix(reinstall): use OSTemplate.vmid (not template_vmid) and convert memory bytes->MB

- AttributeError: OSTemplate has no 'template_vmid' field; the actual column is 'vmid'.
- Use OSTemplate.get_vmid_for_node() to support cross-node replicated templates,
  falling back to source_node/vmid otherwise.
- VMInstance.memory is stored in bytes, but Proxmox config expects MB.
This commit is contained in:
markmorado
2026-04-29 23:48:48 +02:00
parent 5169bf80a3
commit 89d70654fc
45 changed files with 2946 additions and 741 deletions

View File

@@ -13,6 +13,7 @@ from ..db import get_db
from ..auth import get_current_user, PermissionChecker
from ..models import User, AuditLog
from ..logging_service import LoggingService
from ..i18n import I18nService
router = APIRouter()
@@ -178,15 +179,15 @@ async def cleanup_logs(
@router.get("/api/levels")
async def get_log_levels():
async def get_log_levels(lang: str = Query("ru")):
"""Get available log levels"""
return {
"levels": [
{"value": "debug", "label": "Debug", "color": "#6c757d"},
{"value": "info", "label": "Info", "color": "#0dcaf0"},
{"value": "warning", "label": "Warning", "color": "#ffc107"},
{"value": "error", "label": "Error", "color": "#dc3545"},
{"value": "critical", "label": "Critical", "color": "#dc3545"}
{"value": "debug", "label": I18nService.get("log_level_debug", lang), "color": "#6c757d"},
{"value": "info", "label": I18nService.get("log_level_info", lang), "color": "#0dcaf0"},
{"value": "warning", "label": I18nService.get("log_level_warning", lang), "color": "#ffc107"},
{"value": "error", "label": I18nService.get("log_level_error", lang), "color": "#dc3545"},
{"value": "critical", "label": I18nService.get("log_level_error", lang), "color": "#dc3545"}
]
}

View File

@@ -10,7 +10,7 @@ import httpx
import websockets
from ...db import get_db
from ...models import ProxmoxServer, VMInstance, User, IPAMAllocation, IPAMNetwork, VMSnapshotArchive, ProxmoxTask
from ...models import ProxmoxServer, VMInstance, User, IPAMAllocation, IPAMNetwork, VMSnapshotArchive, ProxmoxTask, DeployTask
from ...schemas import ProxmoxServerCreate, ProxmoxServerUpdate, ProxmoxServerResponse
from ...proxmox_client import ProxmoxClient, get_proxmox_resources
from ...auth import get_current_user, PermissionChecker, require_permission, check_permission
@@ -211,17 +211,20 @@ def get_all_tasks(
"""
bulk_q = db.query(TaskQueue).filter(TaskQueue.user_id == current_user.id)
proxmox_q = db.query(ProxmoxTask).filter(ProxmoxTask.user_id == current_user.id)
deploy_q = db.query(DeployTask).filter(DeployTask.user_id == current_user.id)
if status_filter:
# Normalise statuses: proxmox uses running/completed/failed; bulk uses the same + pending/cancelled
bulk_q = bulk_q.filter(TaskQueue.status == status_filter)
proxmox_q = proxmox_q.filter(ProxmoxTask.status == status_filter)
deploy_q = deploy_q.filter(DeployTask.status == status_filter)
bulk_tasks = bulk_q.order_by(TaskQueue.created_at.desc()).limit(limit).all()
proxmox_tasks = proxmox_q.order_by(ProxmoxTask.created_at.desc()).limit(limit).all()
deploy_tasks = deploy_q.order_by(DeployTask.created_at.desc()).limit(limit).all()
all_tasks = sorted(
[t.to_dict() for t in bulk_tasks] + [t.to_dict() for t in proxmox_tasks],
[t.to_dict() for t in bulk_tasks] + [t.to_dict() for t in proxmox_tasks] + [t.to_dict() for t in deploy_tasks],
key=lambda t: t.get("created_at") or "",
reverse=True,
)[:limit]
@@ -245,7 +248,12 @@ def get_active_count(
ProxmoxTask.status == "running",
).scalar() or 0
return {"count": bulk_count + proxmox_count}
deploy_count = db.query(func.count(DeployTask.id)).filter(
DeployTask.user_id == current_user.id,
DeployTask.status.in_(["pending", "running"]),
).scalar() or 0
return {"count": bulk_count + proxmox_count + deploy_count}
# ─── Proxmox UPID task log from DB ──────────────────────────────────────────
@@ -292,6 +300,10 @@ def clear_completed_tasks(
ProxmoxTask.user_id == current_user.id,
ProxmoxTask.status.in_(done_statuses),
).delete(synchronize_session=False)
deploy_deleted = db.query(DeployTask).filter(
DeployTask.user_id == current_user.id,
DeployTask.status.in_(done_statuses),
).delete(synchronize_session=False)
db.commit()
return {"deleted": bulk_deleted + proxmox_deleted}
return {"deleted": bulk_deleted + proxmox_deleted + deploy_deleted}

View File

@@ -425,14 +425,27 @@ def reinstall_vm_endpoint(
from ...models import OSTemplate
tpl = db.query(OSTemplate).filter(OSTemplate.id == cached.template_id).first()
if not tpl or not tpl.template_vmid:
if not tpl or not tpl.vmid:
raise HTTPException(status_code=400, detail="Template not found or invalid")
# Resolve template VMID and source node (with cross-node replication support)
local_template_vmid = tpl.get_vmid_for_node(node)
if local_template_vmid:
template_source_node = node
template_vmid = local_template_vmid
else:
template_source_node = tpl.get_source_node() or tpl.node or node
template_vmid = tpl.vmid
if not template_vmid:
raise HTTPException(status_code=400, detail="Template VMID is not available for this node")
name = cached.name
cores = cached.cores
memory = cached.memory
description = cached.description
is_lxc = (cached.vm_type == 'lxc')
# VMInstance.memory is stored in bytes; Proxmox expects MB
memory_mb = int(memory) // (1024 * 1024) if memory else None
# 1) stop
try:
@@ -470,13 +483,13 @@ def reinstall_vm_endpoint(
try:
if is_lxc:
upid = client.clone_container(
node=tpl.node or node, vmid=tpl.template_vmid, new_vmid=vmid,
node=template_source_node, vmid=template_vmid, new_vmid=vmid,
hostname=name, full=True, target_node=node,
description=description,
)
else:
upid = client.clone_vm(
node=tpl.node or node, vmid=tpl.template_vmid, new_vmid=vmid,
node=template_source_node, vmid=template_vmid, new_vmid=vmid,
name=name, full=True, target_node=node,
description=description,
)
@@ -495,13 +508,13 @@ def reinstall_vm_endpoint(
if is_lxc:
cfg = {}
if cores: cfg['cores'] = cores
if memory: cfg['memory'] = int(memory)
if memory_mb: cfg['memory'] = memory_mb
if cfg:
client.update_container_config(node, vmid, cfg)
else:
cfg = {}
if cores: cfg['cores'] = cores
if memory: cfg['memory'] = int(memory)
if memory_mb: cfg['memory'] = memory_mb
if cfg:
client.update_vm_config(node, vmid, cfg)
except Exception as _ce:
@@ -512,7 +525,7 @@ def reinstall_vm_endpoint(
resource_type='lxc' if is_lxc else 'vm', resource_id=vmid,
username=current_user.username, resource_name=name,
server_id=server_id, server_name=server.name, node_name=node,
details={'template_id': tpl.id, 'template_vmid': tpl.template_vmid},
details={'template_id': tpl.id, 'template_vmid': template_vmid},
success=True,
)
return {"status": "success", "vmid": vmid, "upid": upid}

View File

@@ -3,14 +3,19 @@ API endpoints for OS Templates management
Allows creating VM templates, groups, and deploying VMs from templates
"""
import asyncio
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from fastapi import APIRouter, Depends, Request, HTTPException, status
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from sqlalchemy.orm import Session
from loguru import logger
from typing import List
from typing import List, Optional
from ..db import get_db, SessionLocal
from ..models import ProxmoxServer, OSTemplateGroup, OSTemplate, VMInstance, User
from ..models import ProxmoxServer, OSTemplateGroup, OSTemplate, VMInstance, User, DeployTask
from ..schemas import (
OSTemplateGroupCreate, OSTemplateGroupUpdate, OSTemplateGroupResponse,
OSTemplateCreate, OSTemplateUpdate, OSTemplateResponse, OSTemplateWithGroup,
@@ -21,9 +26,39 @@ from ..api.proxmox import get_next_vmid, save_vm_instance, get_vm_instance
from ..auth import get_current_user, PermissionChecker
from ..ipam_service import IPAMService
from ..logging_service import LoggingService
from ..websocket_manager import broadcast_task_update
router = APIRouter()
# Thread pool for blocking deploy operations
_deploy_executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="deploy")
def _utcnow() -> datetime:
return datetime.now(timezone.utc)
# ── Deploy Task response schema ──────────────────────────────────────────────
class DeployTaskStartResponse(BaseModel):
task_id: int
status: str
name: str
class DeployTaskStatusResponse(BaseModel):
id: int
status: str
step: Optional[str]
progress: int
name: str
vmid: Optional[int]
node: Optional[str]
error_message: Optional[str]
created_at: Optional[str]
started_at: Optional[str]
completed_at: Optional[str]
# ==================== Template Groups CRUD ====================
@@ -639,134 +674,135 @@ def auto_import_templates(
# ==================== VM Deployment ====================
@router.post("/api/deploy", response_model=VMDeployResponse)
async def deploy_vm_from_template(
deploy_data: VMDeployRequest,
db: Session = Depends(get_db),
current_user: User = Depends(PermissionChecker("vms.create"))
):
"""
Развернуть новую VM из шаблона.
Клонирует шаблон и настраивает параметры VM.
Поддерживает развертывание на любую ноду кластера (cross-node deployment).
"""
# Get template
template = db.query(OSTemplate).filter(OSTemplate.id == deploy_data.template_id).first()
if not template:
raise HTTPException(status_code=404, detail="Шаблон не найден")
if not template.is_active:
raise HTTPException(status_code=400, detail="Шаблон неактивен")
# Get server
server = db.query(ProxmoxServer).filter(ProxmoxServer.id == template.server_id).first()
if not server:
raise HTTPException(status_code=404, detail="Proxmox сервер не найден")
# Validate resources
cores = deploy_data.cores or template.default_cores
memory = deploy_data.memory or template.default_memory
disk = deploy_data.disk or template.default_disk
if cores < template.min_cores:
raise HTTPException(status_code=400, detail=f"Минимум {template.min_cores} ядер CPU")
if memory < template.min_memory:
raise HTTPException(status_code=400, detail=f"Минимум {template.min_memory} MB памяти")
if disk < template.min_disk:
raise HTTPException(status_code=400, detail=f"Минимум {template.min_disk} GB диска")
def _update_deploy_task(task_id: int, status: str, step: str, progress: int,
vmid: int = None, node: str = None, error: str = None):
"""Update deploy task state in DB and broadcast via WebSocket."""
db = SessionLocal()
try:
# Connect to Proxmox
task = db.query(DeployTask).filter(DeployTask.id == task_id).first()
if not task:
return
task.status = status
task.step = step
task.progress = progress
if vmid is not None:
task.vmid = vmid
if node is not None:
task.node = node
if error is not None:
task.error_message = error
if status == 'running' and not task.started_at:
task.started_at = _utcnow()
if status in ('completed', 'failed'):
task.completed_at = _utcnow()
db.commit()
try:
broadcast_task_update(task.user_id, "deploy_task_update", task.to_dict())
except Exception:
pass
finally:
db.close()
def _do_deploy_sync(task_id: int, deploy_data: VMDeployRequest,
user_id: int, username: str, user_ssh_key: str):
"""Blocking deploy logic — runs in a thread pool."""
db = SessionLocal()
server = None
template = None
deploy_node = None
new_vmid = None
try:
_update_deploy_task(task_id, 'running', 'Подготовка...', 5)
template = db.query(OSTemplate).filter(OSTemplate.id == deploy_data.template_id).first()
if not template:
_update_deploy_task(task_id, 'failed', 'Шаблон не найден', 0, error='Шаблон не найден')
return
server = db.query(ProxmoxServer).filter(ProxmoxServer.id == template.server_id).first()
if not server:
_update_deploy_task(task_id, 'failed', 'Сервер не найден', 0, error='Сервер Proxmox не найден')
return
cores = deploy_data.cores or template.default_cores
memory = deploy_data.memory or template.default_memory
disk = deploy_data.disk or template.default_disk
# Connect
_update_deploy_task(task_id, 'running', 'Подключение к Proxmox...', 10)
if server.use_password:
client = ProxmoxClient(
host=server.ip_address,
user=server.api_user,
password=server.password,
verify_ssl=server.verify_ssl
host=server.ip_address, user=server.api_user,
password=server.password, verify_ssl=server.verify_ssl
)
else:
client = ProxmoxClient(
host=server.ip_address,
user=server.api_user,
token_name=server.api_token_name,
token_value=server.api_token_value,
host=server.ip_address, user=server.api_user,
token_name=server.api_token_name, token_value=server.api_token_value,
verify_ssl=server.verify_ssl
)
if not client.is_connected():
raise HTTPException(status_code=503, detail="Не удалось подключиться к Proxmox серверу")
# Determine source node (where template exists) and target node (where to deploy)
_update_deploy_task(task_id, 'failed', 'Ошибка подключения к Proxmox', 10,
error='Не удалось подключиться к Proxmox серверу')
return
# Locate template
_update_deploy_task(task_id, 'running', 'Поиск шаблона...', 20)
source_node = template.get_source_node()
target_node = deploy_data.target_node or source_node
# Check if template exists on source node
template_vmid = template.vmid
template_status = client.get_vm_status(source_node, template_vmid)
if not template_status:
# Try to find the template on other nodes
found = client.find_template_on_nodes(template_vmid)
if found:
source_node = found['node']
logger.info(f"Template {template_vmid} found on alternative node: {source_node}")
else:
raise HTTPException(
status_code=404,
detail=f"Шаблон VMID {template_vmid} не найден. Проверьте настройки шаблона."
)
# For shared storage: template is accessible from all nodes
# For local storage: template needs to be on the target node or replicated
#
# In a cluster with shared storage, we can clone from source_node
# but specify target_node for the new VM
clone_from_node = source_node
clone_template_vmid = template_vmid
# Log cross-node deployment
if target_node and target_node != source_node:
logger.info(f"Cross-node deployment: template on {source_node}, target node {target_node}")
# Get next VMID from database (range 10000-19999) or use provided VMID
_update_deploy_task(task_id, 'failed', 'Шаблон не найден на Proxmox', 20,
error=f'VMID {template_vmid} не найден на сервере')
return
# Allocate VMID
_update_deploy_task(task_id, 'running', 'Генерация VMID...', 25)
if deploy_data.vmid:
# Используем указанный VMID (для переустановки)
new_vmid = deploy_data.vmid
logger.info(f"Using provided VMID: {new_vmid} for reinstall")
else:
# Генерируем новый случайный VMID для новой VM
new_vmid = get_next_vmid(db, server.id)
logger.info(f"Generated new VMID: {new_vmid} for server {server.id}")
# Clone VM from template
# For shared storage: clone from source node, target specifies where VM will run
# Clone
_update_deploy_task(task_id, 'running', f'Клонирование шаблона → VMID {new_vmid}...', 30)
clone_upid = client.clone_vm_from_template(
node=clone_from_node,
template_vmid=clone_template_vmid,
node=source_node,
template_vmid=template_vmid,
new_vmid=new_vmid,
name=deploy_data.name,
full_clone=True,
description=f"Deployed from template {template.name} by {current_user.username}",
description=f"Deployed from template {template.name} by {username}",
target_storage=deploy_data.target_storage,
target_node=target_node if target_node != clone_from_node else None
target_node=target_node if target_node != source_node else None
)
if not clone_upid:
raise HTTPException(status_code=500, detail="Ошибка клонирования VM")
# Wait for clone to complete
clone_success = client.wait_for_task(clone_from_node, clone_upid, timeout=300)
_update_deploy_task(task_id, 'failed', 'Ошибка клонирования', 30,
error='Proxmox не вернул UPID задачи клонирования')
return
# Wait for clone — progress from 30 to 65
_update_deploy_task(task_id, 'running', 'Клонирование диска...', 45, vmid=new_vmid)
clone_success = client.wait_for_task(source_node, clone_upid, timeout=300)
if not clone_success:
raise HTTPException(status_code=500, detail="Таймаут клонирования VM")
# Use target_node for all subsequent operations
_update_deploy_task(task_id, 'failed', 'Таймаут клонирования', 45,
error='Клонирование заняло слишком долго (>5 мин)')
return
deploy_node = target_node
# IPAM Integration - Auto-allocate IP if network specified but no IP provided
_update_deploy_task(task_id, 'running', 'Клонирование завершено', 65, vmid=new_vmid, node=deploy_node)
# IPAM
allocated_ip = deploy_data.ip_address
ipam_allocation = None
if deploy_data.ipam_network_id and not deploy_data.ip_address:
_update_deploy_task(task_id, 'running', 'Выделение IP-адреса (IPAM)...', 68, vmid=new_vmid, node=deploy_node)
ipam = IPAMService(db)
ipam_allocation, error = ipam.auto_allocate_ip(
network_id=deploy_data.ipam_network_id,
@@ -776,25 +812,19 @@ async def deploy_vm_from_template(
proxmox_server_id=server.id,
proxmox_vmid=new_vmid,
proxmox_node=deploy_node,
allocated_by=current_user.username,
allocated_by=username,
notes=f"Auto-allocated for VM deployment from template {template.name}"
)
if error:
logger.error(f"IPAM auto-allocation failed: {error}")
# Continue without IPAM - let user know
else:
if not error:
allocated_ip = ipam_allocation.ip_address
logger.info(f"IPAM allocated IP {allocated_ip} for VM {deploy_data.name}")
# Get gateway from network if not provided
if not deploy_data.gateway:
from ..models import IPAMNetwork
network = db.query(IPAMNetwork).filter(IPAMNetwork.id == deploy_data.ipam_network_id).first()
if network and network.gateway:
deploy_data.gateway = network.gateway
net = db.query(IPAMNetwork).filter(IPAMNetwork.id == deploy_data.ipam_network_id).first()
if net and net.gateway:
deploy_data.gateway = net.gateway
elif deploy_data.ip_address and deploy_data.ipam_network_id:
# Register manually specified IP in IPAM
ipam = IPAMService(db)
ipam_allocation, error = ipam.allocate_ip(
ipam_allocation, _ = ipam.allocate_ip(
ip_address=deploy_data.ip_address,
network_id=deploy_data.ipam_network_id,
pool_id=deploy_data.ipam_pool_id,
@@ -803,64 +833,47 @@ async def deploy_vm_from_template(
proxmox_server_id=server.id,
proxmox_vmid=new_vmid,
proxmox_node=deploy_node,
allocated_by=current_user.username,
allocated_by=username,
notes=f"Manually assigned for VM deployment from template {template.name}"
)
if error:
logger.warning(f"IPAM registration failed for manually specified IP: {error}")
# Get network mask (CIDR) and gateway from IPAM network if IP is specified
network_mask = "24" # Default mask
# Determine network mask & gateway
network_mask = "24"
gateway = deploy_data.gateway
# Try to get proper network mask and gateway from IPAM
if allocated_ip and (deploy_data.ipam_network_id or ipam_allocation):
try:
from ..models import IPAMNetwork
network_id = deploy_data.ipam_network_id
if ipam_allocation and hasattr(ipam_allocation, 'network_id'):
network_id = ipam_allocation.network_id
if network_id:
network = db.query(IPAMNetwork).filter(IPAMNetwork.id == network_id).first()
if network:
# Extract mask from CIDR (e.g., "10.10.10.0/24" -> "24")
if network.network and '/' in network.network:
network_mask = network.network.split('/')[1]
logger.info(f"Using network mask /{network_mask} from IPAM network {network.name}")
# Get gateway if not already specified
if not gateway and network.gateway:
gateway = network.gateway
logger.info(f"Using gateway {gateway} from IPAM network {network.name}")
net = db.query(IPAMNetwork).filter(IPAMNetwork.id == network_id).first()
if net:
if net.network and '/' in net.network:
network_mask = net.network.split('/')[1]
if not gateway and net.gateway:
gateway = net.gateway
except Exception as e:
logger.warning(f"Could not get network info from IPAM: {e}")
# Configure VM
ip_config = None
if allocated_ip:
logger.info(f"Configuring VM {new_vmid} with IP {allocated_ip}/{network_mask}, gateway: {gateway}")
if gateway:
ip_config = f"ip={allocated_ip}/{network_mask},gw={gateway}"
else:
ip_config = f"ip={allocated_ip}/{network_mask}"
ip_config = f"ip={allocated_ip}/{network_mask},gw={gateway}" if gateway else f"ip={allocated_ip}/{network_mask}"
else:
ip_config = "ip=dhcp"
logger.info(f"Configuring VM {new_vmid} with DHCP")
# Combine user's SSH key with deploy request SSH keys
# Configure VM
_update_deploy_task(task_id, 'running', 'Настройка VM (CPU, RAM, Cloud-Init)...', 72, vmid=new_vmid, node=deploy_node)
ssh_keys = deploy_data.ssh_keys or ""
if current_user.ssh_public_key:
if ssh_keys:
ssh_keys = f"{ssh_keys}\n{current_user.ssh_public_key}"
else:
ssh_keys = current_user.ssh_public_key
if user_ssh_key:
ssh_keys = f"{ssh_keys}\n{user_ssh_key}".strip()
config_success = client.configure_vm(
node=deploy_node,
vmid=new_vmid,
cores=cores,
memory=memory,
disk_size=disk, # configure_vm will compare with current disk size
disk_size=disk,
network_bridge=deploy_data.network_bridge,
cloud_init_user=deploy_data.cloud_init_user,
cloud_init_password=deploy_data.cloud_init_password,
@@ -868,31 +881,25 @@ async def deploy_vm_from_template(
ip_config=ip_config,
onboot=deploy_data.onboot
)
if not config_success:
logger.warning(f"VM {new_vmid} created but configuration may be incomplete")
# Start VM if requested
# Start
if deploy_data.start_after_create:
_update_deploy_task(task_id, 'running', 'Запуск VM...', 88, vmid=new_vmid, node=deploy_node)
client.start_vm(deploy_node, new_vmid)
# Enable High Availability if requested (cluster only)
ha_enabled = False
# HA
if deploy_data.enable_ha:
try:
if client.is_cluster():
ha_result = client.add_to_ha(vmid=new_vmid, vm_type='vm', max_restart=3, max_relocate=3)
if ha_result.get('success'):
ha_enabled = True
logger.info(f"HA enabled for VM {new_vmid}")
else:
logger.warning(f"Failed to enable HA for VM {new_vmid}: {ha_result.get('error')}")
else:
logger.warning(f"HA requested but server {server.name} is not in a cluster")
client.add_to_ha(vmid=new_vmid, vm_type='vm', max_restart=3, max_relocate=3)
except Exception as ha_error:
logger.error(f"Error enabling HA for VM {new_vmid}: {ha_error}")
# Save VM configuration to database with owner
logger.error(f"HA error for VM {new_vmid}: {ha_error}")
# Save to DB
_update_deploy_task(task_id, 'running', 'Сохранение в базе данных...', 95, vmid=new_vmid, node=deploy_node)
user = db.query(User).filter(User.id == user_id).first()
save_vm_instance(
db=db,
server_id=server.id,
@@ -904,76 +911,137 @@ async def deploy_vm_from_template(
memory=memory,
disk_size=disk,
ip_address=allocated_ip,
ip_prefix=24,
gateway=deploy_data.gateway,
ip_prefix=int(network_mask),
gateway=gateway,
cloud_init_user=deploy_data.cloud_init_user,
cloud_init_password=deploy_data.cloud_init_password,
ssh_keys=deploy_data.ssh_keys,
template_id=template.id,
template_name=template.name,
description=f"Deployed from template {template.name}",
owner_id=current_user.id # VPS-style user isolation
owner_id=user_id
)
logger.info(f"User {current_user.username} deployed VM {new_vmid} ({deploy_data.name}) from template {template.name}")
# Log VM deployment to audit
LoggingService.log_proxmox_action(
db=db,
action="create",
resource_type="vm",
resource_id=new_vmid,
username=current_user.username,
username=username,
resource_name=deploy_data.name,
server_id=server.id,
server_name=server.name,
node_name=deploy_node,
details={
"template_id": template.id,
"template_name": template.name,
"source_node": source_node if source_node != deploy_node else None,
"cores": cores,
"memory": memory,
"disk": disk,
"ip_address": allocated_ip,
"ha_enabled": ha_enabled
},
details={"template_id": template.id, "template_name": template.name, "cores": cores,
"memory": memory, "disk": disk, "ip_address": allocated_ip},
success=True
)
return VMDeployResponse(
success=True,
vmid=new_vmid,
name=deploy_data.name,
node=deploy_node,
server_id=server.id,
task_upid=clone_upid,
message=f"VM {deploy_data.name} успешно создана (VMID: {new_vmid})"
)
except HTTPException:
raise
_update_deploy_task(task_id, 'completed', 'VM успешно создана!', 100, vmid=new_vmid, node=deploy_node)
logger.info(f"[DEPLOY] Task #{task_id}: VM {deploy_data.name} (VMID {new_vmid}) deployed by {username}")
except Exception as e:
# Log failed deployment
LoggingService.log_proxmox_action(
db=db,
action="create",
resource_type="vm",
resource_id=deploy_data.vmid,
username=current_user.username,
resource_name=deploy_data.name,
server_id=server.id if server else None,
server_name=server.name if server else None,
node_name=deploy_node if 'deploy_node' in locals() else None,
details={
"template_id": template.id if template else None,
"template_name": template.name if template else None
},
success=False,
error_message=str(e)
)
logger.error(f"Error deploying VM from template {template.id}: {e}")
raise HTTPException(status_code=500, detail=f"Ошибка развертывания VM: {str(e)}")
err = str(e)
logger.error(f"[DEPLOY] Task #{task_id} failed: {err}")
# Log failed
try:
if server and template:
LoggingService.log_proxmox_action(
db=db,
action="create",
resource_type="vm",
resource_id=deploy_data.vmid,
username=username,
resource_name=deploy_data.name,
server_id=server.id if server else None,
server_name=server.name if server else None,
node_name=deploy_node,
details={"template_id": template.id if template else None},
success=False,
error_message=err
)
except Exception:
pass
_update_deploy_task(task_id, 'failed', f'Ошибка: {err[:150]}', 0, error=err)
finally:
db.close()
@router.post("/api/deploy", response_model=DeployTaskStartResponse, status_code=202)
async def deploy_vm_from_template(
deploy_data: VMDeployRequest,
db: Session = Depends(get_db),
current_user: User = Depends(PermissionChecker("vms.create"))
):
"""
Асинхронный деплой VM из шаблона.
Немедленно возвращает task_id, выполнение идёт в фоне.
Статус можно отслеживать через GET /api/deploy/{task_id}.
"""
# Fast validation before queuing
template = db.query(OSTemplate).filter(OSTemplate.id == deploy_data.template_id).first()
if not template:
raise HTTPException(status_code=404, detail="Шаблон не найден")
if not template.is_active:
raise HTTPException(status_code=400, detail="Шаблон неактивен")
server = db.query(ProxmoxServer).filter(ProxmoxServer.id == template.server_id).first()
if not server:
raise HTTPException(status_code=404, detail="Proxmox сервер не найден")
cores = deploy_data.cores or template.default_cores
memory = deploy_data.memory or template.default_memory
disk = deploy_data.disk or template.default_disk
if cores < template.min_cores:
raise HTTPException(status_code=400, detail=f"Минимум {template.min_cores} ядер CPU")
if memory < template.min_memory:
raise HTTPException(status_code=400, detail=f"Минимум {template.min_memory} MB памяти")
if disk < template.min_disk:
raise HTTPException(status_code=400, detail=f"Минимум {template.min_disk} GB диска")
# Create deploy task record
task = DeployTask(
status='pending',
step='В очереди...',
progress=0,
name=deploy_data.name,
template_id=template.id,
server_id=server.id,
user_id=current_user.id,
)
db.add(task)
db.commit()
db.refresh(task)
task_id = task.id
# Schedule blocking deploy in thread pool
loop = asyncio.get_event_loop()
loop.run_in_executor(
_deploy_executor,
_do_deploy_sync,
task_id,
deploy_data,
current_user.id,
current_user.username,
current_user.ssh_public_key or ''
)
logger.info(f"[DEPLOY] Task #{task_id} queued: {deploy_data.name} by {current_user.username}")
return DeployTaskStartResponse(task_id=task_id, status='pending', name=deploy_data.name)
@router.get("/api/deploy/{task_id}", response_model=DeployTaskStatusResponse)
async def get_deploy_task_status(
task_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Получить статус задачи деплоя."""
task = db.query(DeployTask).filter(DeployTask.id == task_id).first()
if not task:
raise HTTPException(status_code=404, detail="Задача не найдена")
if task.user_id != current_user.id and not current_user.is_admin:
raise HTTPException(status_code=403, detail="Нет доступа")
return DeployTaskStatusResponse(**task.to_dict())
# ==================== Quick Deploy (for UI) ====================

View File

@@ -193,7 +193,7 @@
"console": "Console",
"reinstall": "Reinstall",
"fill_all_fields": "Fill all required fields",
"select_node": "— Select node —",
"select_node": "Select Node",
"no_nodes": "No nodes available",
"login": "Login",
"username": "Username",
@@ -1190,7 +1190,6 @@
"nav_networks": "Networks",
"networks_management_desc": "Manage SDN zones, VNets, subnets and node network interfaces",
"node_networks": "Node Interfaces",
"network_interfaces": "Network Interfaces",
"iface_type": "Type",
"iface_bridge": "Bridge",
"iface_bond": "Bond",
@@ -1202,7 +1201,6 @@
"bond_slaves_hint": "Space-separated list of slave interfaces, e.g. ens3 ens4",
"bond_mode": "Bond Mode",
"autostart": "Autostart",
"active": "Active",
"apply_network_config": "Apply Network Config",
"apply_network_config_confirm": "Apply pending network changes to this node? This will activate the new configuration.",
"revert_network_config": "Revert Network Config",
@@ -1222,7 +1220,6 @@
"ipam_network_will_not_be_deleted": "The linked IPAM network (if any) will not be deleted.",
"synced_with_ipam": "IPAM",
"vlan_aware": "VLAN Aware",
"vlan_id": "VLAN ID",
"vlan_raw_device": "Raw Device",
"vxlan_peer": "Peer Addresses",
"vxlan_peer_hint": "Space-separated list of remote peer IP addresses",
@@ -1230,8 +1227,37 @@
"alias": "Alias",
"subnet": "Subnet",
"netmask": "Netmask",
"select_node": "Select Node",
"iface_bridge_name_hint": "e.g. vmbr0, vmbr1",
"iface_bond_name_hint": "e.g. bond0, bond1",
"iface_vlan_name_hint": "e.g. ens3.100"
"iface_vlan_name_hint": "e.g. ens3.100",
"vmid": "VMID",
"boot_order": "Boot Order",
"qemu_agent": "QEMU Agent",
"primary_ip": "Primary IP",
"os_type": "OS Type",
"vcpu": "vCPU",
"instance_info": "Instance Info",
"interfaces": "Interfaces",
"dns": "DNS",
"my_proxmox_server": "My Proxmox Server",
"qemu_virtual_machine": "QEMU Virtual Machine",
"root": "root",
"ubuntu": "ubuntu",
"my_vm": "my-vm-01",
"notes_saved": "Notes saved",
"before_update": "before-update",
"placeholder_root": "root",
"placeholder_ubuntu": "ubuntu",
"placeholder_vm_name": "my-vm-01",
"placeholder_ssh_key": "ssh-rsa AAAA...",
"placeholder_command": "ls -la /tmp",
"placeholder_disk_size": "e.g. 100",
"placeholder_snapshot_name": "e.g. before-update",
"placeholder_root_pam": "root@pam",
"placeholder_disk": "scsi0",
"placeholder_notes": "Описание / заметки об инстансе...",
"log_level_debug": "Debug",
"log_level_info": "Info",
"log_level_warning": "Warning",
"log_level_error": "Error"
}

View File

@@ -193,7 +193,7 @@
"console": "Консоль",
"reinstall": "Переустановка",
"fill_all_fields": "Заполните все обязательные поля",
"select_node": "— Выберите ноду —",
"select_node": "Выберите узел",
"no_nodes": "Нет доступных нод",
"login": "Войти",
"username": "Пользователь",
@@ -1190,7 +1190,6 @@
"nav_networks": "Сети",
"networks_management_desc": "Управление SDN-зонами, VNet, подсетями и сетевыми интерфейсами узлов",
"node_networks": "Интерфейсы узлов",
"network_interfaces": "Сетевые интерфейсы",
"iface_type": "Тип",
"iface_bridge": "Бридж",
"iface_bond": "Бонд",
@@ -1202,7 +1201,6 @@
"bond_slaves_hint": "Список slave-интерфейсов через пробел, например ens3 ens4",
"bond_mode": "Режим бонда",
"autostart": "Автозапуск",
"active": "Активен",
"apply_network_config": "Применить конфигурацию",
"apply_network_config_confirm": "Применить ожидающие изменения сети на этом узле? Новая конфигурация будет активирована.",
"revert_network_config": "Откатить конфигурацию",
@@ -1222,7 +1220,6 @@
"ipam_network_will_not_be_deleted": "Связанная сеть IPAM (при наличии) удалена не будет.",
"synced_with_ipam": "IPAM",
"vlan_aware": "VLAN Aware",
"vlan_id": "VLAN ID",
"vlan_raw_device": "Raw-устройство",
"vxlan_peer": "Адреса пиров",
"vxlan_peer_hint": "Список IP-адресов удалённых пиров через пробел",
@@ -1230,8 +1227,37 @@
"alias": "Псевдоним",
"subnet": "Подсеть",
"netmask": "Маска",
"select_node": "Выберите узел",
"iface_bridge_name_hint": "например vmbr0, vmbr1",
"iface_bond_name_hint": "например bond0, bond1",
"iface_vlan_name_hint": "например ens3.100"
"iface_vlan_name_hint": "например ens3.100",
"vmid": "VMID",
"boot_order": "Порядок загрузки",
"qemu_agent": "QEMU агент",
"primary_ip": "Основной IP",
"os_type": "Тип ОС",
"vcpu": "vCPU",
"instance_info": "Информация об инстансе",
"interfaces": "Интерфейсы",
"dns": "DNS",
"my_proxmox_server": "Мой Proxmox сервер",
"qemu_virtual_machine": "Виртуальная машина QEMU",
"root": "root",
"ubuntu": "ubuntu",
"my_vm": "my-vm-01",
"notes_saved": "Заметки сохранены",
"before_update": "перед-обновлением",
"placeholder_root": "root",
"placeholder_ubuntu": "ubuntu",
"placeholder_vm_name": "my-vm-01",
"placeholder_ssh_key": "ssh-rsa AAAA...",
"placeholder_command": "ls -la /tmp",
"placeholder_disk_size": "например 100",
"placeholder_snapshot_name": "например перед-обновлением",
"placeholder_root_pam": "root@pam",
"placeholder_disk": "scsi0",
"placeholder_notes": "Описание / заметки об инстансе...",
"log_level_debug": "Отладка",
"log_level_info": "Информация",
"log_level_warning": "Предупреждение",
"log_level_error": "Ошибка"
}

View File

@@ -148,6 +148,14 @@ async def lifespan(app: FastAPI):
except Exception as e:
logger.error(f"Error running admin initialization: {e}")
# Register the main event loop for cross-thread WebSocket broadcasts
try:
import asyncio
from .websocket_manager import set_main_loop
set_main_loop(asyncio.get_event_loop())
except Exception as e:
logger.warning(f"Failed to register main event loop: {e}")
# Start background monitoring worker
try:
from .workers.monitoring_worker import start_monitoring_worker

View File

@@ -1057,3 +1057,63 @@ class WorkspaceUser(Base):
Index('idx_ws_users_user', 'user_id'),
Index('idx_ws_users_unique', 'workspace_id', 'user_id', unique=True),
)
# ==================== Deploy Task ====================
class DeployTask(Base):
"""Tracks async VM deployment tasks from templates"""
__tablename__ = "deploy_tasks"
id = Column(Integer, primary_key=True, index=True)
status = Column(String(20), nullable=False, default='pending', index=True)
# pending / running / completed / failed
# Human-readable current step
step = Column(String(200), nullable=True)
# Progress 0–100
progress = Column(Integer, nullable=False, default=0)
# Input info
name = Column(String(100), nullable=False)
template_id = Column(Integer, nullable=False)
server_id = Column(Integer, nullable=True)
user_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True)
# Output info (filled on completion)
vmid = Column(Integer, nullable=True)
node = Column(String(100), nullable=True)
error_message = Column(Text, nullable=True)
created_at = Column(DateTime(timezone=True), default=utcnow, nullable=False, index=True)
started_at = Column(DateTime(timezone=True), nullable=True)
completed_at = Column(DateTime(timezone=True), nullable=True)
__table_args__ = (
Index('idx_deploy_task_user_status', 'user_id', 'status'),
)
def to_dict(self) -> dict:
return {
'id': self.id,
'kind': 'deploy',
'status': self.status,
'step': self.step,
'current_step': self.step,
'progress': self.progress,
'progress_percent': self.progress,
'description': f"Создание {self.name}",
'name': self.name,
'template_id': self.template_id,
'server_id': self.server_id,
'vmid': self.vmid,
'node': self.node,
'error_message': self.error_message,
'created_at': self.created_at.isoformat() if self.created_at else None,
'started_at': self.started_at.isoformat() if self.started_at else None,
'completed_at': self.completed_at.isoformat() if self.completed_at else None,
}
def __repr__(self):
return f"<DeployTask(id={self.id}, name='{self.name}', status='{self.status}')>"

View File

@@ -342,6 +342,184 @@ class ProxmoxClient:
except Exception as e:
logger.error(f"Ошибка перезапуска LXC {vmid} на {node}: {e}")
return None
def shutdown_vm(self, node: str, vmid: int, timeout: int = 60) -> Optional[str]:
"""Корректно выключить VM (ACPI). Возвращает UPID задачи."""
if not self.proxmox:
return None
try:
upid = self.proxmox.nodes(node).qemu(vmid).status.shutdown.post(timeout=timeout)
return upid if isinstance(upid, str) else None
except Exception as e:
logger.error(f"Ошибка корректного выключения VM {vmid} на {node}: {e}")
return None
def shutdown_container(self, node: str, vmid: int, timeout: int = 60) -> Optional[str]:
"""Корректно выключить LXC. Возвращает UPID задачи."""
if not self.proxmox:
return None
try:
upid = self.proxmox.nodes(node).lxc(vmid).status.shutdown.post(timeout=timeout)
return upid if isinstance(upid, str) else None
except Exception as e:
logger.error(f"Ошибка корректного выключения LXC {vmid} на {node}: {e}")
return None
def clone_vm(self, node: str, vmid: int, new_vmid: int, name: str,
full: bool = True, target_node: Optional[str] = None,
target_storage: Optional[str] = None,
description: Optional[str] = None) -> Optional[str]:
"""Клонировать VM (qemu). Работает и для шаблонов, и для обычных VM."""
if not self.proxmox:
return None
try:
params = {'newid': new_vmid, 'name': name, 'full': 1 if full else 0}
if target_storage:
params['storage'] = target_storage
if target_node and target_node != node:
params['target'] = target_node
if description:
params['description'] = description
upid = self.proxmox.nodes(node).qemu(vmid).clone.post(**params)
return upid if isinstance(upid, str) else None
except Exception as e:
logger.error(f"Ошибка клонирования VM {vmid} -> {new_vmid}: {e}")
return None
def clone_container(self, node: str, vmid: int, new_vmid: int, hostname: str,
full: bool = True, target_node: Optional[str] = None,
target_storage: Optional[str] = None,
description: Optional[str] = None) -> Optional[str]:
"""Клонировать LXC контейнер."""
if not self.proxmox:
return None
try:
params = {'newid': new_vmid, 'hostname': hostname, 'full': 1 if full else 0}
if target_storage:
params['storage'] = target_storage
if target_node and target_node != node:
params['target'] = target_node
if description:
params['description'] = description
upid = self.proxmox.nodes(node).lxc(vmid).clone.post(**params)
return upid if isinstance(upid, str) else None
except Exception as e:
logger.error(f"Ошибка клонирования LXC {vmid} -> {new_vmid}: {e}")
return None
def change_vm_password(self, node: str, vmid: int, username: str, password: str) -> Dict:
"""
Сменить пароль пользователя на VM через QEMU guest agent.
Требует установленный qemu-guest-agent в VM.
"""
if not self.proxmox:
return {'success': False, 'error': 'Proxmox connection not initialized'}
try:
# Proxmox API: POST /nodes/{node}/qemu/{vmid}/agent/set-user-password
self.proxmox.nodes(node).qemu(vmid).agent('set-user-password').post(
username=username,
password=password,
)
return {'success': True}
except Exception as e:
logger.error(f"Не удалось сменить пароль на VM {vmid}: {e}")
return {'success': False, 'error': str(e)}
def change_container_password(self, node: str, vmid: int, username: str, password: str) -> Dict:
"""
Сменить пароль пользователя в LXC через `pct exec` (chpasswd).
Контейнер должен быть запущен.
"""
if not self.proxmox:
return {'success': False, 'error': 'Proxmox connection not initialized'}
try:
# Используем `chpasswd` в формате "user:password"
# Запускаем bash -c "echo 'user:pass' | chpasswd"
cmd_str = f"echo '{username}:{password}' | chpasswd"
result = self.proxmox.nodes(node).lxc(vmid).status.current.get()
if result.get('status') != 'running':
return {'success': False, 'error': 'Container is not running'}
res = self.proxmox.nodes(node).lxc(vmid).exec.post(
command=['bash', '-c', cmd_str]
)
return {'success': True, 'result': res}
except Exception as e:
logger.error(f"Не удалось сменить пароль в LXC {vmid}: {e}")
return {'success': False, 'error': str(e)}
def set_vm_notes(self, node: str, vmid: int, description: str) -> bool:
"""Установить description (заметки) у VM."""
if not self.proxmox:
return False
try:
self.proxmox.nodes(node).qemu(vmid).config.put(description=description or '')
return True
except Exception as e:
logger.error(f"Ошибка обновления заметок VM {vmid}: {e}")
return False
def set_container_notes(self, node: str, vmid: int, description: str) -> bool:
"""Установить description (заметки) у LXC."""
if not self.proxmox:
return False
try:
self.proxmox.nodes(node).lxc(vmid).config.put(description=description or '')
return True
except Exception as e:
logger.error(f"Ошибка обновления заметок LXC {vmid}: {e}")
return False
def attach_iso(self, node: str, vmid: int, iso_volid: str, device: str = 'ide2') -> bool:
"""Подключить ISO образ к VM (только KVM)."""
if not self.proxmox:
return False
try:
value = f"{iso_volid},media=cdrom"
self.proxmox.nodes(node).qemu(vmid).config.put(**{device: value})
return True
except Exception as e:
logger.error(f"Ошибка подключения ISO {iso_volid} к VM {vmid}: {e}")
return False
def detach_iso(self, node: str, vmid: int, device: str = 'ide2') -> bool:
"""Отключить ISO образ (CD-ROM) у VM."""
if not self.proxmox:
return False
try:
self.proxmox.nodes(node).qemu(vmid).config.put(**{device: 'none,media=cdrom'})
return True
except Exception as e:
logger.error(f"Ошибка отключения ISO у VM {vmid}: {e}")
return False
def get_node_isos(self, node: str) -> List[Dict]:
"""Получить список ISO-образов со всех ISO-хранилищ ноды."""
if not self.proxmox:
return []
result: List[Dict] = []
try:
storages = self.proxmox.nodes(node).storage.get()
for s in storages:
content = s.get('content', '')
if 'iso' not in content:
continue
storage_id = s.get('storage')
try:
items = self.proxmox.nodes(node).storage(storage_id).content.get(content='iso')
for it in items:
result.append({
'volid': it.get('volid'),
'storage': storage_id,
'size': it.get('size'),
'format': it.get('format'),
'name': (it.get('volid') or '').split('/')[-1],
})
except Exception as e:
logger.debug(f"Не удалось получить ISO из {storage_id}: {e}")
except Exception as e:
logger.error(f"Ошибка получения списка ISO для ноды {node}: {e}")
return result
def force_stop_vm(self, node: str, vmid: int) -> bool:
"""Принудительно остановить ВМ (через SSH - аналог kill -9)"""

View File

@@ -126,12 +126,29 @@ class ConnectionManager:
ws_manager = ConnectionManager()
# Stores the main asyncio event loop set at application startup.
# Background threads (APScheduler) use this to schedule coroutines safely.
_main_loop: asyncio.AbstractEventLoop | None = None
def set_main_loop(loop: asyncio.AbstractEventLoop) -> None:
"""Call this once at application startup to register the running event loop."""
global _main_loop
_main_loop = loop
def run_async_safe(coro) -> None:
"""
Schedule an async coroutine from a synchronous background thread.
Used by APScheduler jobs to push WS updates without blocking.
"""
global _main_loop
try:
# Prefer the stored main loop (set at startup) for cross-thread scheduling
if _main_loop is not None and _main_loop.is_running():
asyncio.run_coroutine_threadsafe(coro, _main_loop)
return
# Fallback: try to get/run the current thread's loop
loop = asyncio.get_event_loop()
if loop.is_running():
asyncio.run_coroutine_threadsafe(coro, loop)

View File

@@ -688,6 +688,8 @@ class MonitoringWorker:
# Track all seen (server_id, vmid) pairs for cleanup
all_seen = set()
# Track status changes for WebSocket broadcast
status_changes = [] # list of (server_id, vmid, node, old_status, new_status)
# Upsert all VMs to database
for server_id, vm_data, vm_type in all_vms_data:
@@ -713,6 +715,8 @@ class MonitoringWorker:
if os_type:
os_type = os_type.capitalize()
new_status = vm_data.get('status', 'unknown')
# Find existing by (server_id, vmid)
existing = db.query(VMInstance).filter(
VMInstance.server_id == server_id,
@@ -720,10 +724,13 @@ class MonitoringWorker:
).first()
if existing:
# Detect status change for WS broadcast
if existing.status != new_status:
status_changes.append((server_id, vmid, node_name, existing.status, new_status))
# Update existing entry
existing.name = vm_name
existing.node = node_name
existing.status = vm_data.get('status', 'unknown')
existing.status = new_status
existing.cores = vm_data.get('cpus', vm_data.get('maxcpu'))
existing.memory = vm_data.get('maxmem')
existing.disk_size = vm_data.get('maxdisk')
@@ -741,7 +748,7 @@ class MonitoringWorker:
node=node_name,
vm_type=vm_type,
name=vm_name,
status=vm_data.get('status', 'unknown'),
status=new_status,
cores=vm_data.get('cpus', vm_data.get('maxcpu')),
memory=vm_data.get('maxmem'),
disk_size=vm_data.get('maxdisk'),
@@ -803,6 +810,22 @@ class MonitoringWorker:
db.commit()
logger.debug(f"[VM SYNC] Cache sync completed. Processed {len(all_vms_data)} VMs/containers, marked {deleted_count} as deleted, released {released_ips} IPs")
# Broadcast status changes to all connected WebSocket clients
if status_changes:
try:
from app.websocket_manager import ws_manager, run_async_safe
for srv_id, vmid, node_name, old_status, new_status in status_changes:
logger.info(f"[VM SYNC] Status change: server={srv_id} vmid={vmid} {old_status} -> {new_status}")
run_async_safe(ws_manager.broadcast({
"type": "vm_status_update",
"server_id": srv_id,
"vmid": vmid,
"node": node_name,
"status": new_status,
}))
except Exception as _we:
logger.warning(f"[VM SYNC] Failed to broadcast status changes: {_we}")
except Exception as e:
logger.error(f"[VM SYNC] Critical error: {e}", exc_info=True)
db.rollback()
@@ -1034,10 +1057,10 @@ def start_monitoring_worker():
replace_existing=True
)
# VM cache sync - every 30 seconds
# VM cache sync - every 10 seconds
scheduler.add_job(
monitoring_worker.sync_vm_cache,
trigger=IntervalTrigger(seconds=30),
trigger=IntervalTrigger(seconds=10),
id='vm_cache_sync',
name='Sync VM cache from Proxmox',
replace_existing=True

View File

@@ -1256,6 +1256,37 @@ def migrate_user_role_remove_dashboard(conn):
logger.info("✓ user role: dashboard:view already False, skipping")
def migrate_deploy_tasks(conn):
"""Migration 22: Create deploy_tasks table for async VM deployment tracking"""
conn.execute(text("""
CREATE TABLE IF NOT EXISTS deploy_tasks (
id SERIAL PRIMARY KEY,
status VARCHAR(20) NOT NULL DEFAULT 'pending',
step VARCHAR(200),
progress INTEGER NOT NULL DEFAULT 0,
name VARCHAR(100) NOT NULL,
template_id INTEGER NOT NULL,
server_id INTEGER,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
vmid INTEGER,
node VARCHAR(100),
error_message TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
started_at TIMESTAMPTZ,
completed_at TIMESTAMPTZ
)
"""))
conn.execute(text("""
CREATE INDEX IF NOT EXISTS idx_deploy_task_user_status
ON deploy_tasks (user_id, status)
"""))
conn.execute(text("""
CREATE INDEX IF NOT EXISTS idx_deploy_tasks_created
ON deploy_tasks (created_at)
"""))
logger.info("✓ deploy_tasks table created")
def run_all_migrations(engine, db_session=None):
"""
Run all migrations in order.
@@ -1438,6 +1469,14 @@ def run_all_migrations(engine, db_session=None):
logger.warning(f"User role dashboard migration: {e}")
conn.rollback()
# Migration 22: Deploy Tasks table
try:
migrate_deploy_tasks(conn)
conn.commit()
except Exception as e:
logger.warning(f"Deploy tasks migration: {e}")
conn.rollback()
logger.info("=" * 50)
logger.info("✓ All migrations completed successfully")
logger.info("=" * 50)

17
backend/pytest.ini Normal file
View File

@@ -0,0 +1,17 @@
[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
addopts =
-ra
--strict-markers
--tb=short
--color=yes
filterwarnings =
ignore::DeprecationWarning:passlib.*
ignore::DeprecationWarning:pkg_resources.*
markers =
smoke: quick smoke tests
unit: pure unit tests
integration: integration tests touching multiple modules

View File

@@ -0,0 +1,5 @@
-r requirements.txt
pytest==8.3.3
pytest-cov==5.0.0
pytest-asyncio==0.24.0
freezegun==1.5.1

87
backend/tests/README.md Normal file
View File

@@ -0,0 +1,87 @@
# Backend Tests
Smoke / unit tests for the FastAPI backend, focused on the security-critical
modules: `app.crypto`, `app.auth`, `app.rbac`.
## Layout
```
backend/
├── pytest.ini # pytest config (testpaths=tests, markers, addopts)
├── requirements-dev.txt # test deps (-r requirements.txt + pytest …)
└── tests/
├── conftest.py # env setup + reusable fixtures
├── test_crypto.py # Fernet field-encryption helpers
├── test_auth.py # password hashing, JWT, IP parsing, perm helpers
└── test_rbac.py # PermissionEngine + registry
```
## How `conftest.py` keeps tests hermetic
The application reads configuration at import time:
- `app.config.Settings` is a `pydantic-settings` model that auto-loads `.env`
from the current working directory.
- `app.db` constructs the SQLAlchemy engine on import.
`tests/conftest.py` therefore, **before any `app.*` import**:
1. Sets `DATABASE_URL`, `SECRET_KEY`, `ADMIN_PASSWORD`, `LOG_LEVEL` to safe
test values via `os.environ.setdefault`.
2. Generates a fresh `FERNET_KEY` so encrypt/decrypt round-trips succeed.
3. Adds `backend/` to `sys.path`.
4. `chdir`s to a temp directory so the developer's local `.env` (which contains
keys not declared on `Settings`) is **not** picked up. Without this step
pydantic raises `extra_forbidden`.
No real Postgres connection is opened — SQLAlchemy connects lazily, and these
tests never issue a query.
## Running locally
There is no system-wide Python on this host, so use the same Python image as CI:
```bash
docker run --rm -v "$PWD/backend:/app" -w /app python:3.12-slim bash -c '
pip install -r requirements-dev.txt &&
python -m pytest -p no:warnings
'
```
Run a single test:
```bash
python -m pytest tests/test_auth.py::TestJWT -v
```
## CI
GitHub Actions runs the suite on every push / PR to `main` that touches
`backend/**` — see [.github/workflows/backend-tests.yml](../.github/workflows/backend-tests.yml).
Coverage is reported to the job log and uploaded as an artifact.
## Adding more tests
- **Pure / domain logic** (no DB): use the `fake_user_factory` /
`fake_role_factory` fixtures in `conftest.py`. They produce
`SimpleNamespace` objects that satisfy duck-typed access in `auth.py` and
`rbac/engine.py`.
- **Touches the DB**: build out a SQLite-backed fixture (override
`app.db.engine` with `sqlite:///:memory:` and `Base.metadata.create_all`).
Not required for the current smoke layer.
- **API endpoints**: spin up `TestClient(app)` once `get_db` and any external
Proxmox / SSH dependencies are dependency-overridden.
## Markers
Defined in `pytest.ini`:
- `@pytest.mark.smoke` — quickest sanity checks.
- `@pytest.mark.unit` — pure unit tests.
- `@pytest.mark.integration` — multi-module integration.
Run only smoke tests:
```bash
python -m pytest -m smoke
```

View File

92
backend/tests/conftest.py Normal file
View File

@@ -0,0 +1,92 @@
"""
Shared pytest fixtures.
The DB engine is constructed at import time in ``app.db``. To keep tests
hermetic we set environment variables BEFORE importing the application.
The engine itself is created lazily-connected, so as long as no test issues
a real query the configured URL is irrelevant.
"""
from __future__ import annotations
import os
import sys
import tempfile
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock
# --- Environment must be set BEFORE importing app.* ---------------------------
os.environ.setdefault("DATABASE_URL", "postgresql://test:test@localhost:5432/test")
os.environ.setdefault("SECRET_KEY", "test-secret-key-do-not-use-in-prod")
os.environ.setdefault("ADMIN_PASSWORD", "test-admin-password")
os.environ.setdefault("LOG_LEVEL", "WARNING")
# Generate a deterministic Fernet key for tests so encrypt/decrypt are stable.
from cryptography.fernet import Fernet # noqa: E402
os.environ.setdefault("FERNET_KEY", Fernet.generate_key().decode())
# Ensure the backend root is on sys.path so ``import app`` works regardless of
# the working directory pytest is launched from.
BACKEND_ROOT = Path(__file__).resolve().parent.parent
if str(BACKEND_ROOT) not in sys.path:
sys.path.insert(0, str(BACKEND_ROOT))
# pydantic-settings auto-loads ``.env`` from the current working directory.
# A developer .env may contain operational keys (POSTGRES_PASSWORD, ALLOWED_HOSTS,
# ...) that are not declared on the Settings model and would raise
# ``extra_forbidden``. Run from a clean temp dir so only env vars are honoured.
_TEST_CWD = Path(tempfile.mkdtemp(prefix="pvemanager-tests-"))
os.chdir(_TEST_CWD)
import pytest # noqa: E402
# --- Reusable fixtures --------------------------------------------------------
@pytest.fixture
def fake_role_factory():
"""Build a stand-in Role object with a permissions dict."""
def _make(permissions: dict | None = None, name: str = "tester"):
return SimpleNamespace(
name=name,
permissions=permissions or {},
has_permission=lambda perm: bool((permissions or {}).get(perm)),
)
return _make
@pytest.fixture
def fake_user_factory(fake_role_factory):
"""Build a stand-in User object compatible with auth/rbac code paths."""
def _make(
username: str = "alice",
is_admin: bool = False,
is_active: bool = True,
role=None,
permissions: dict | None = None,
):
if role is None and permissions is not None:
role = fake_role_factory(permissions=permissions)
return SimpleNamespace(
id=1,
username=username,
is_admin=is_admin,
is_active=is_active,
role=role,
locked_until=None,
)
return _make
@pytest.fixture
def mock_db():
"""Bare MagicMock standing in for a SQLAlchemy session."""
return MagicMock(name="db_session")

113
backend/tests/test_auth.py Normal file
View File

@@ -0,0 +1,113 @@
"""Smoke tests for ``app.auth`` — pure functions only (no DB)."""
from __future__ import annotations
import importlib
from datetime import timedelta
import jwt
import pytest
from fastapi import HTTPException
@pytest.fixture(scope="module")
def auth():
return importlib.import_module("app.auth")
@pytest.fixture(scope="module")
def settings():
from app.config import settings as s
return s
class TestPasswordHashing:
@pytest.mark.smoke
def test_hash_then_verify(self, auth):
pw = "Correct-Horse-Battery-Staple-1!"
hashed = auth.get_password_hash(pw)
assert hashed != pw
assert auth.verify_password(pw, hashed) is True
def test_verify_rejects_wrong_password(self, auth):
hashed = auth.get_password_hash("right-password")
assert auth.verify_password("wrong-password", hashed) is False
def test_hash_is_salted(self, auth):
# Two hashes of the same password must differ (random salt).
a = auth.get_password_hash("same")
b = auth.get_password_hash("same")
assert a != b
class TestJWT:
@pytest.mark.smoke
def test_create_then_decode(self, auth):
token = auth.create_access_token({"sub": "alice"})
payload = auth.decode_access_token(token)
assert payload["sub"] == "alice"
assert "exp" in payload and "iat" in payload
def test_session_binding_round_trips(self, auth):
token = auth.create_access_token({"sub": "bob"}, session_token="sess-xyz")
payload = auth.decode_access_token(token)
assert payload["session"] == "sess-xyz"
def test_expired_token_raises_401(self, auth):
token = auth.create_access_token({"sub": "x"}, expires_delta=timedelta(seconds=-1))
with pytest.raises(HTTPException) as exc:
auth.decode_access_token(token)
assert exc.value.status_code == 401
def test_tampered_token_raises_401(self, auth):
token = auth.create_access_token({"sub": "x"})
tampered = token + "tamper"
with pytest.raises(HTTPException) as exc:
auth.decode_access_token(tampered)
assert exc.value.status_code == 401
def test_token_signed_with_wrong_secret_raises_401(self, auth, settings):
bogus = jwt.encode({"sub": "x"}, "different-secret", algorithm=settings.ALGORITHM)
with pytest.raises(HTTPException) as exc:
auth.decode_access_token(bogus)
assert exc.value.status_code == 401
class TestPermissionHelpers:
def test_check_permission_true_for_admin(self, auth, fake_user_factory):
admin = fake_user_factory(is_admin=True)
assert auth.check_permission(admin, "vm:delete") is True
def test_check_permission_false_when_role_lacks_it(self, auth, fake_user_factory):
user = fake_user_factory(permissions={"vm:view": True})
assert auth.check_permission(user, "vm:delete") is False
def test_require_permission_raises_403(self, auth, fake_user_factory):
user = fake_user_factory(permissions={})
with pytest.raises(HTTPException) as exc:
auth.require_permission(user, "vm:delete")
assert exc.value.status_code == 403
class TestClientIPParsing:
def _request_with_headers(self, headers: dict, client_host: str = "10.0.0.1"):
# Minimal stand-in mimicking starlette.Request surface used by get_client_ip.
from types import SimpleNamespace
# headers must be lower-cased keys; .get is what get_client_ip uses.
return SimpleNamespace(
headers={k.lower(): v for k, v in headers.items()},
client=SimpleNamespace(host=client_host),
)
def test_returns_first_valid_xff(self, auth):
req = self._request_with_headers({"X-Forwarded-For": "203.0.113.5, 10.0.0.1"})
assert auth.get_client_ip(req) == "203.0.113.5"
def test_rejects_spoofed_xff(self, auth):
# Non-IP value must be ignored — fall back to client.host.
req = self._request_with_headers(
{"X-Forwarded-For": "not-an-ip; rm -rf /"},
client_host="10.0.0.99",
)
assert auth.get_client_ip(req) == "10.0.0.99"

View File

@@ -0,0 +1,69 @@
"""Smoke tests for ``app.crypto`` field-level encryption helpers."""
from __future__ import annotations
import importlib
import pytest
@pytest.fixture(scope="module")
def crypto():
return importlib.import_module("app.crypto")
class TestEncryptDecryptRoundtrip:
@pytest.mark.smoke
def test_roundtrip_simple_string(self, crypto):
token = crypto.encrypt_value("hunter2")
assert token != "hunter2"
assert crypto.decrypt_value(token) == "hunter2"
@pytest.mark.smoke
def test_roundtrip_unicode(self, crypto):
secret = "пароль-密码-🔐"
assert crypto.decrypt_value(crypto.encrypt_value(secret)) == secret
def test_empty_value_passthrough(self, crypto):
assert crypto.encrypt_value("") == ""
assert crypto.decrypt_value("") == ""
assert crypto.encrypt_value(None) is None
assert crypto.decrypt_value(None) is None
def test_each_call_produces_unique_ciphertext(self, crypto):
# Fernet embeds a random IV — same plaintext should encrypt differently.
a = crypto.encrypt_value("same")
b = crypto.encrypt_value("same")
assert a != b
assert crypto.decrypt_value(a) == crypto.decrypt_value(b) == "same"
class TestLegacyPlaintextFallback:
def test_decrypt_invalid_token_returns_input(self, crypto):
# Legacy/un-encrypted rows must not blow up — fall through to plaintext.
assert crypto.decrypt_value("plain-value-not-fernet") == "plain-value-not-fernet"
class TestNoFernetKey:
def test_encrypt_passthrough_without_key(self, crypto, monkeypatch):
# ``settings`` is imported lazily inside ``_get_fernet`` — patch the live
# singleton on the config module instead.
from app import config as app_config
monkeypatch.setattr(app_config.settings, "FERNET_KEY", None, raising=False)
monkeypatch.setattr(crypto, "_warned_no_key", False, raising=False)
assert crypto.encrypt_value("plain") == "plain"
assert crypto.decrypt_value("plain") == "plain"
class TestEncryptedStringColumn:
def test_bind_and_result_processors(self, crypto):
col = crypto.EncryptedString(255)
bound = col.process_bind_param("topsecret", dialect=None)
assert bound != "topsecret"
assert col.process_result_value(bound, dialect=None) == "topsecret"
def test_bind_and_result_handle_none(self, crypto):
col = crypto.EncryptedString(255)
assert col.process_bind_param(None, dialect=None) is None
assert col.process_result_value(None, dialect=None) is None

120
backend/tests/test_rbac.py Normal file
View File

@@ -0,0 +1,120 @@
"""Smoke tests for the RBAC permission engine."""
from __future__ import annotations
import importlib
import pytest
from fastapi import HTTPException
@pytest.fixture(scope="module")
def rbac():
return importlib.import_module("app.rbac")
class TestRegistry:
@pytest.mark.smoke
def test_well_known_permissions_registered(self, rbac):
for code in ("vm:view", "vm:create", "vm:delete", "user:create", "role:manage"):
assert rbac.PERMISSIONS.get(code) is not None, f"missing {code}"
def test_legacy_resolution(self, rbac):
assert rbac.resolve_permission("vms.create") == "vm:create"
assert rbac.resolve_permission("vm:create") == "vm:create"
# Unknown codes are returned untouched.
assert rbac.resolve_permission("totally:unknown") == "totally:unknown"
def test_validate_partitions_codes(self, rbac):
# The registry's internal legacy map uses ``resource.action`` form (e.g.
# ``vm.create``). The richer ``LEGACY_PERMISSION_MAP`` (e.g. ``vms.create``)
# is consumed by ``resolve_permission`` / the engine, not by ``validate``.
valid, invalid = rbac.PERMISSIONS.validate(["vm:view", "vm.create", "bogus:perm"])
assert "vm:view" in valid
assert "vm:create" in valid # registry legacy resolved
assert invalid == ["bogus:perm"]
class TestEngineHasPermission:
@pytest.mark.smoke
def test_admin_bypass(self, rbac, fake_user_factory):
admin = fake_user_factory(is_admin=True)
assert rbac.PermissionEngine.has_permission(admin, "vm:delete") is True
assert rbac.PermissionEngine.has_permission(admin, "anything:at:all") is True
def test_direct_grant(self, rbac, fake_user_factory):
user = fake_user_factory(permissions={"vm:view": True})
assert rbac.PermissionEngine.has_permission(user, "vm:view") is True
assert rbac.PermissionEngine.has_permission(user, "vm:delete") is False
def test_legacy_grant_resolves_to_new(self, rbac, fake_user_factory):
user = fake_user_factory(permissions={"vms.create": True})
assert rbac.PermissionEngine.has_permission(user, "vm:create") is True
def test_wildcard_grants_all_actions_on_resource(self, rbac, fake_user_factory):
user = fake_user_factory(permissions={"vm:*": True})
assert rbac.PermissionEngine.has_permission(user, "vm:view") is True
assert rbac.PermissionEngine.has_permission(user, "vm:delete") is True
# Other resources untouched.
assert rbac.PermissionEngine.has_permission(user, "user:delete") is False
def test_manage_implies_common_actions(self, rbac, fake_user_factory):
user = fake_user_factory(permissions={"vm:manage": True})
assert rbac.PermissionEngine.has_permission(user, "vm:view") is True
assert rbac.PermissionEngine.has_permission(user, "vm:start") is True
# delete is NOT in the implied set.
assert rbac.PermissionEngine.has_permission(user, "vm:delete") is False
def test_user_without_role_has_no_permissions(self, rbac, fake_user_factory):
user = fake_user_factory(role=None)
assert rbac.PermissionEngine.has_permission(user, "vm:view") is False
def test_disabled_permission_not_granted(self, rbac, fake_user_factory):
user = fake_user_factory(permissions={"vm:view": False})
assert rbac.PermissionEngine.has_permission(user, "vm:view") is False
class TestEngineCheckPermission:
def test_check_permission_raises_403(self, rbac, fake_user_factory):
user = fake_user_factory(permissions={})
with pytest.raises(HTTPException) as exc:
rbac.PermissionEngine.check_permission(user, "vm:delete")
assert exc.value.status_code == 403
assert "vm:delete" in exc.value.detail
def test_check_permission_passes_for_admin(self, rbac, fake_user_factory):
admin = fake_user_factory(is_admin=True)
rbac.PermissionEngine.check_permission(admin, "vm:delete") # no raise
class TestAggregateHelpers:
def test_has_any_permission(self, rbac, fake_user_factory):
user = fake_user_factory(permissions={"vm:view": True})
assert rbac.PermissionEngine.has_any_permission(user, ["vm:delete", "vm:view"]) is True
assert rbac.PermissionEngine.has_any_permission(user, ["vm:delete", "vm:create"]) is False
def test_has_all_permissions(self, rbac, fake_user_factory):
user = fake_user_factory(permissions={"vm:view": True, "vm:create": True})
assert rbac.PermissionEngine.has_all_permissions(user, ["vm:view", "vm:create"]) is True
assert rbac.PermissionEngine.has_all_permissions(user, ["vm:view", "vm:delete"]) is False
def test_filter_permissions(self, rbac, fake_user_factory):
user = fake_user_factory(permissions={"vm:view": True})
kept = rbac.PermissionEngine.filter_permissions(user, ["vm:view", "vm:delete"])
assert kept == ["vm:view"]
class TestUserModelIntegration:
"""``User.has_permission`` should defer to PermissionEngine."""
def test_user_has_permission_delegates(self, rbac, fake_user_factory):
# Use the SimpleNamespace user — exercises the same call path as the
# real model thanks to duck typing in PermissionEngine.
user = fake_user_factory(permissions={"vm:view": True})
assert rbac.PermissionEngine.has_permission(user, "vm:view") is True
def test_get_user_permissions_admin_returns_full_set(self, rbac, fake_user_factory):
admin = fake_user_factory(is_admin=True)
perms = rbac.PermissionEngine.get_user_permissions(admin)
# Admin shortcut returns every registered permission code.
assert {"vm:view", "vm:delete", "user:create"}.issubset(perms)