feat: SSH keys management, backups fixes, instances/nodes/templates improvements, dev mode support
Some checks failed
Backend Tests / pytest (Python 3.12) (push) Has been cancelled

This commit is contained in:
markmorado
2026-05-07 00:30:21 +02:00
parent a92cb0df33
commit 323d1e57fa
40 changed files with 4112 additions and 288 deletions

View File

@@ -197,11 +197,11 @@ async def get_log_categories():
"""Get available log categories"""
return {
"categories": [
{"value": "auth", "label": "🔐 Авторизация"},
{"value": "proxmox", "label": "🖥️ Proxmox"},
{"value": "ipam", "label": "🌐 IPAM"},
{"value": "system", "label": "⚙️ Система"},
{"value": "api", "label": "🔌 API"},
{"value": "docker", "label": "🐳 Docker"}
{"value": "auth", "label": "Авторизация"},
{"value": "proxmox", "label": "Proxmox"},
{"value": "ipam", "label": "IPAM"},
{"value": "system", "label": "Система"},
{"value": "api", "label": "API"},
{"value": "docker", "label": "Docker"}
]
}

View File

@@ -137,20 +137,59 @@ def delete_storage(
@router.get("/api/backups/list/{server_id}")
def list_backups(
server_id: int,
node: str,
storage: str,
node: Optional[str] = None,
vmid: Optional[int] = None,
db: Session = Depends(get_db),
current_user: User = Depends(PermissionChecker("backup:view")),
):
"""List backup files stored in a given storage on a node"""
"""List backup files stored in a given storage.
If ``node`` is omitted (or empty), backup files are aggregated from all
cluster nodes that have the storage available. Duplicate ``volid`` entries
(which appear on shared storages) are de-duplicated.
"""
server = db.query(ProxmoxServer).filter(ProxmoxServer.id == server_id).first()
if not server:
raise HTTPException(status_code=404, detail="Server not found")
try:
client = _get_proxmox_client(server)
backups = client.list_backups(node, storage, vmid)
nodes_to_query: list[str] = []
if node:
nodes_to_query = [node]
else:
try:
cluster_nodes = client.get_nodes() or []
nodes_to_query = [
n.get("node") for n in cluster_nodes
if n.get("node") and (n.get("status") in (None, "online"))
]
except Exception as e:
logger.warning(f"Failed to enumerate cluster nodes for backup listing: {e}")
seen: set[str] = set()
backups: list[dict] = []
for n in nodes_to_query:
try:
items = client.list_backups(n, storage, vmid) or []
except Exception as e:
logger.warning(f"Skipping node {n} when listing backups: {e}")
continue
for item in items:
volid = item.get("volid")
if volid and volid in seen:
continue
if volid:
seen.add(volid)
# annotate with the node where it was discovered (helpful for
# restore / delete actions on non-shared storages)
item.setdefault("node", n)
backups.append(item)
return JSONResponse(content={"backups": backups})
except HTTPException:
raise
except Exception as e:
logger.error(f"Error listing backups: {e}")
raise HTTPException(status_code=500, detail=str(e))

View File

@@ -53,6 +53,8 @@ class LXCDeployRequest(BaseModel):
ipam_pool_id: Optional[int] = None
password: Optional[str] = None
ssh_keys: Optional[str] = None
ssh_key_ids: Optional[List[int]] = None
owner_id: Optional[int] = None
nameserver: Optional[str] = None
searchdomain: Optional[str] = None
unprivileged: bool = True
@@ -105,7 +107,8 @@ def _update_task(task_id: int, status: str, step: str, progress: int,
db.close()
def _do_lxc_deploy(task_id: int, req: LXCDeployRequest, user_id: int, username: str):
def _do_lxc_deploy(task_id: int, req: LXCDeployRequest, user_id: int, username: str,
owner_id: Optional[int] = None):
"""Blocking LXC deploy — runs in thread pool."""
db = SessionLocal()
server = None
@@ -286,7 +289,7 @@ def _do_lxc_deploy(task_id: int, req: LXCDeployRequest, user_id: int, username:
template_id=None,
template_name=req.ostemplate,
description=f"Created from CT template {req.ostemplate}",
owner_id=user_id,
owner_id=owner_id or user_id,
)
LoggingService.log_proxmox_action(
@@ -434,6 +437,23 @@ async def deploy_lxc_from_template(
if not req.name or len(req.name) < 1:
raise HTTPException(status_code=400, detail="Укажите имя контейнера")
# Resolve owner / SSH keys
is_admin = current_user.has_permission("users.manage") or current_user.is_admin
instance_owner_id = current_user.id
if req.owner_id and req.owner_id != current_user.id:
if not is_admin:
raise HTTPException(status_code=403, detail="Only admins can assign instance to another user")
owner = db.query(User).filter(User.id == req.owner_id).first()
if not owner:
raise HTTPException(status_code=404, detail="Owner user not found")
instance_owner_id = owner.id
if req.ssh_key_ids:
from ..ssh_keys import resolve_ssh_keys_for_deploy
resolved = resolve_ssh_keys_for_deploy(db, current_user, req.ssh_key_ids, instance_owner_id)
existing = (req.ssh_keys or "").strip()
req.ssh_keys = (existing + "\n" + resolved).strip() if existing else resolved
# Create task record
task = DeployTask(
status="pending",
@@ -456,6 +476,7 @@ async def deploy_lxc_from_template(
req,
current_user.id,
current_user.username,
instance_owner_id,
)
return LXCDeployResponse(task_id=task_id, status="pending", name=req.name)

View File

@@ -523,6 +523,37 @@ def delete_proxmox_server(
return None
@router.post("/api/servers/test", status_code=status.HTTP_200_OK)
def test_proxmox_credentials(
payload: ProxmoxServerCreate,
current_user: User = Depends(PermissionChecker("proxmox.view"))
):
"""Проверить произвольные учётные данные Proxmox без сохранения в БД."""
try:
if payload.use_password:
client = ProxmoxClient(
host=payload.ip_address,
user=payload.api_user,
password=payload.password,
verify_ssl=payload.verify_ssl,
)
else:
client = ProxmoxClient(
host=payload.ip_address,
user=payload.api_user,
token_name=payload.api_token_name,
token_value=payload.api_token_value,
verify_ssl=payload.verify_ssl,
)
if client.is_connected():
return {"success": True, "status": "success", "message": "Connection successful"}
return {"success": False, "status": "error", "message": "Failed to connect to Proxmox server"}
except Exception as e:
logger.error(f"Error testing Proxmox credentials for {payload.ip_address}: {e}")
return {"success": False, "status": "error", "message": str(e)}
@router.post("/api/servers/{server_id}/test", status_code=status.HTTP_200_OK)
def test_proxmox_connection(
server_id: int,
@@ -533,25 +564,25 @@ def test_proxmox_connection(
server = db.query(ProxmoxServer).filter(ProxmoxServer.id == server_id).first()
if not server:
raise HTTPException(status_code=404, detail="Proxmox server not found")
try:
# Determine auth method
client = _get_proxmox_client(server)
if client.is_connected():
server.update_status(True)
db.commit()
return {"status": "success", "message": "Connection successful"}
return {"success": True, "status": "success", "message": "Connection successful"}
else:
server.update_status(False, "Failed to connect")
db.commit()
raise HTTPException(status_code=503, detail="Failed to connect to Proxmox server")
return {"success": False, "status": "error", "message": "Failed to connect to Proxmox server"}
except Exception as e:
server.update_status(False, str(e))
db.commit()
logger.error(f"Error testing Proxmox connection to {server.name}: {e}")
raise HTTPException(status_code=500, detail=str(e))
return {"success": False, "status": "error", "message": str(e)}
# ==================== Proxmox Resources ====================

View File

@@ -152,7 +152,24 @@ def get_all_virtual_machines(
query = query.filter(VMInstance.owner_id == current_user.id)
cached_vms = query.order_by(VMInstance.name).all()
# ── Live metrics from Proxmox cluster/resources (CPU, RAM, disk, uptime) ──
# One HTTP call per server returns metrics for all VMs/LXC in that cluster.
# Map: (server_id_in_db, vmid) -> live data dict
live_metrics: dict[tuple[int, int], dict] = {}
for server in servers:
try:
client = _get_proxmox_client(server)
if not client.is_connected():
continue
for res in client.get_cluster_resources(type_='vm'):
vmid = res.get('vmid')
if vmid is None:
continue
live_metrics[(server.id, vmid)] = res
except Exception as e:
logger.debug(f"Could not fetch live metrics from {server.name}: {e}")
result = []
# Detect cluster servers (servers with hostnames pve1, pve2, pve3 pattern)
@@ -195,7 +212,10 @@ def get_all_virtual_machines(
os_template = vm.template_name or vm.os_type or ("QEMU/KVM" if vm.vm_type == "qemu" else "Linux")
if not vm.template_name and vm.vm_type == "lxc" and os_template:
os_template = os_template.capitalize()
# Live metrics (cpu/mem/disk/uptime/net) from cluster/resources
live = live_metrics.get((vm.server_id, vm.vmid)) or {}
result.append({
"server_id": vm.server_id,
"server_name": server_name,
@@ -204,18 +224,28 @@ def get_all_virtual_machines(
"name": vm.name,
"hostname": ip_hostname or f"{'vps' if vm.vm_type == 'qemu' else 'lxc'}{vm.vmid}.{server.hostname if server else 'local'}",
"type": vm.vm_type,
"status": vm.status or "unknown",
"status": live.get('status') or vm.status or "unknown",
"node": vm.node,
"cores": vm.cores or 0,
"memory": vm.memory or 0,
"disk": vm.disk_size or 0,
"disk": live.get('disk') or 0,
"ip": ip_address,
"ip_address": ip_address,
"ip_hostname": ip_network_name,
"os": os_template,
"os_template": os_template,
"owner": owner,
"owner_hostname": "",
"storage": "Storage1 (DIR)"
"storage": "Storage1 (DIR)",
"tags": vm.tags or "",
# Live runtime metrics
"cpu": live.get('cpu'),
"mem": live.get('mem'),
"maxmem": live.get('maxmem') or vm.memory or 0,
"maxdisk": live.get('maxdisk') or vm.disk_size or 0,
"uptime": live.get('uptime'),
"netin": live.get('netin'),
"netout": live.get('netout'),
})
return JSONResponse(content=result)

453
backend/app/api/ssh_keys.py Normal file
View File

@@ -0,0 +1,453 @@
"""
SSH Keys management API.
Each user owns a personal library of SSH keys (public + optionally encrypted
private key). Keys can be referenced by id when deploying a VM/LXC instance.
* Regular users can manage only their own keys and may pick only their own
keys when deploying.
* Admins (`users.manage` permission) can manage SSH keys of any user and may
pick any user's key (typically their own + the new instance owner's) when
deploying through the admin flow.
"""
from __future__ import annotations
import base64
import hashlib
from datetime import datetime
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Request
from loguru import logger
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from ..auth import get_client_ip, get_current_user, PermissionChecker
from ..db import get_db
from ..logging_service import LoggingService
from ..models import User, UserSSHKey
router = APIRouter()
# ==================== Schemas ====================
class SSHKeyBase(BaseModel):
name: str = Field(..., min_length=1, max_length=100)
public_key: str = Field(..., min_length=20, max_length=10000)
private_key: Optional[str] = Field(None, max_length=20000)
comment: Optional[str] = Field(None, max_length=255)
class SSHKeyCreate(SSHKeyBase):
pass
class SSHKeyUpdate(BaseModel):
name: Optional[str] = Field(None, min_length=1, max_length=100)
public_key: Optional[str] = Field(None, min_length=20, max_length=10000)
private_key: Optional[str] = Field(None, max_length=20000)
comment: Optional[str] = Field(None, max_length=255)
class SSHKeyResponse(BaseModel):
id: int
user_id: int
name: str
public_key: str
fingerprint: Optional[str] = None
comment: Optional[str] = None
has_private_key: bool = False
created_at: datetime
class Config:
from_attributes = True
# ==================== Helpers ====================
def _normalize_public_key(value: str) -> str:
return value.strip()
def _compute_fingerprint(public_key: str) -> Optional[str]:
"""SHA256 fingerprint in OpenSSH format (SHA256:base64...)."""
try:
parts = public_key.strip().split()
if len(parts) < 2:
return None
key_b64 = parts[1]
raw = base64.b64decode(key_b64)
digest = hashlib.sha256(raw).digest()
b64 = base64.b64encode(digest).decode().rstrip("=")
return f"SHA256:{b64}"
except Exception:
return None
def _to_response(key: UserSSHKey) -> dict:
return {
"id": key.id,
"user_id": key.user_id,
"name": key.name,
"public_key": key.public_key,
"fingerprint": key.fingerprint,
"comment": key.comment,
"has_private_key": bool(key.private_key),
"created_at": key.created_at,
}
def _ensure_target_user(db: Session, user_id: int) -> User:
user = db.query(User).filter(User.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
# ==================== Self-service endpoints ====================
@router.get("/api/ssh-keys/my", response_model=List[SSHKeyResponse])
async def list_my_ssh_keys(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""List current user's SSH keys."""
keys = (
db.query(UserSSHKey)
.filter(UserSSHKey.user_id == current_user.id)
.order_by(UserSSHKey.created_at.desc())
.all()
)
return [_to_response(k) for k in keys]
@router.post("/api/ssh-keys/my", response_model=SSHKeyResponse, status_code=201)
async def create_my_ssh_key(
data: SSHKeyCreate,
request: Request,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Add an SSH key to the current user's library."""
return _create_key_for_user(db, request, current_user, current_user.id, data)
@router.put("/api/ssh-keys/my/{key_id}", response_model=SSHKeyResponse)
async def update_my_ssh_key(
key_id: int,
data: SSHKeyUpdate,
request: Request,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
return _update_key(db, request, current_user, current_user.id, key_id, data)
@router.delete("/api/ssh-keys/my/{key_id}")
async def delete_my_ssh_key(
key_id: int,
request: Request,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
return _delete_key(db, request, current_user, current_user.id, key_id)
@router.get("/api/ssh-keys/my/{key_id}/private")
async def download_my_private_key(
key_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
return _read_private_key(db, current_user.id, key_id)
# ==================== Admin endpoints (per user) ====================
@router.get("/api/ssh-keys/users/{user_id}", response_model=List[SSHKeyResponse])
async def list_user_ssh_keys(
user_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(PermissionChecker("users.manage")),
):
"""Admin: list a user's SSH keys."""
_ensure_target_user(db, user_id)
keys = (
db.query(UserSSHKey)
.filter(UserSSHKey.user_id == user_id)
.order_by(UserSSHKey.created_at.desc())
.all()
)
return [_to_response(k) for k in keys]
@router.post(
"/api/ssh-keys/users/{user_id}",
response_model=SSHKeyResponse,
status_code=201,
)
async def admin_create_user_ssh_key(
user_id: int,
data: SSHKeyCreate,
request: Request,
db: Session = Depends(get_db),
current_user: User = Depends(PermissionChecker("users.manage")),
):
target = _ensure_target_user(db, user_id)
return _create_key_for_user(db, request, current_user, target.id, data)
@router.put(
"/api/ssh-keys/users/{user_id}/{key_id}",
response_model=SSHKeyResponse,
)
async def admin_update_user_ssh_key(
user_id: int,
key_id: int,
data: SSHKeyUpdate,
request: Request,
db: Session = Depends(get_db),
current_user: User = Depends(PermissionChecker("users.manage")),
):
_ensure_target_user(db, user_id)
return _update_key(db, request, current_user, user_id, key_id, data)
@router.delete("/api/ssh-keys/users/{user_id}/{key_id}")
async def admin_delete_user_ssh_key(
user_id: int,
key_id: int,
request: Request,
db: Session = Depends(get_db),
current_user: User = Depends(PermissionChecker("users.manage")),
):
_ensure_target_user(db, user_id)
return _delete_key(db, request, current_user, user_id, key_id)
@router.get("/api/ssh-keys/users/{user_id}/{key_id}/private")
async def admin_download_user_private_key(
user_id: int,
key_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(PermissionChecker("users.manage")),
):
_ensure_target_user(db, user_id)
return _read_private_key(db, user_id, key_id)
# ==================== Internal helpers ====================
def _create_key_for_user(
db: Session,
request: Request,
current_user: User,
target_user_id: int,
data: SSHKeyCreate,
) -> dict:
public_key = _normalize_public_key(data.public_key)
if not public_key.startswith(("ssh-", "ecdsa-", "sk-")):
raise HTTPException(
status_code=400,
detail="Invalid SSH public key format",
)
fingerprint = _compute_fingerprint(public_key)
private_key = data.private_key.strip() if data.private_key else None
if private_key and "PRIVATE KEY" not in private_key:
raise HTTPException(
status_code=400,
detail="Invalid SSH private key format (PEM expected)",
)
key = UserSSHKey(
user_id=target_user_id,
name=data.name.strip(),
public_key=public_key,
private_key=private_key,
fingerprint=fingerprint,
comment=(data.comment or None),
)
db.add(key)
db.commit()
db.refresh(key)
LoggingService.log(
db=db,
level=LoggingService.INFO,
category=LoggingService.AUTH,
action="ssh_key_added",
message=f"SSH key '{key.name}' added to user_id={target_user_id} by {current_user.username}",
username=current_user.username,
user_id=current_user.id,
ip_address=get_client_ip(request),
resource_type="ssh_key",
resource_id=str(key.id),
resource_name=key.name,
)
return _to_response(key)
def _update_key(
db: Session,
request: Request,
current_user: User,
target_user_id: int,
key_id: int,
data: SSHKeyUpdate,
) -> dict:
key = (
db.query(UserSSHKey)
.filter(UserSSHKey.id == key_id, UserSSHKey.user_id == target_user_id)
.first()
)
if not key:
raise HTTPException(status_code=404, detail="SSH key not found")
if data.name is not None:
key.name = data.name.strip()
if data.public_key is not None:
public_key = _normalize_public_key(data.public_key)
if not public_key.startswith(("ssh-", "ecdsa-", "sk-")):
raise HTTPException(status_code=400, detail="Invalid SSH public key format")
key.public_key = public_key
key.fingerprint = _compute_fingerprint(public_key)
if data.private_key is not None:
pk = data.private_key.strip()
if pk:
if "PRIVATE KEY" not in pk:
raise HTTPException(
status_code=400,
detail="Invalid SSH private key format (PEM expected)",
)
key.private_key = pk
else:
key.private_key = None
if data.comment is not None:
key.comment = data.comment or None
db.commit()
db.refresh(key)
LoggingService.log(
db=db,
level=LoggingService.INFO,
category=LoggingService.AUTH,
action="ssh_key_updated",
message=f"SSH key #{key.id} ('{key.name}') updated by {current_user.username}",
username=current_user.username,
user_id=current_user.id,
ip_address=get_client_ip(request),
resource_type="ssh_key",
resource_id=str(key.id),
resource_name=key.name,
)
return _to_response(key)
def _delete_key(
db: Session,
request: Request,
current_user: User,
target_user_id: int,
key_id: int,
) -> dict:
key = (
db.query(UserSSHKey)
.filter(UserSSHKey.id == key_id, UserSSHKey.user_id == target_user_id)
.first()
)
if not key:
raise HTTPException(status_code=404, detail="SSH key not found")
name = key.name
db.delete(key)
db.commit()
LoggingService.log(
db=db,
level=LoggingService.INFO,
category=LoggingService.AUTH,
action="ssh_key_deleted",
message=f"SSH key '{name}' (id={key_id}) deleted by {current_user.username}",
username=current_user.username,
user_id=current_user.id,
ip_address=get_client_ip(request),
resource_type="ssh_key",
resource_id=str(key_id),
resource_name=name,
)
return {"success": True, "message": "SSH key deleted"}
def _read_private_key(db: Session, target_user_id: int, key_id: int) -> dict:
key = (
db.query(UserSSHKey)
.filter(UserSSHKey.id == key_id, UserSSHKey.user_id == target_user_id)
.first()
)
if not key:
raise HTTPException(status_code=404, detail="SSH key not found")
if not key.private_key:
raise HTTPException(status_code=404, detail="No private key stored for this entry")
return {
"id": key.id,
"name": key.name,
"private_key": key.private_key,
}
# ==================== Internal API (used by deploy flow) ====================
def resolve_ssh_keys_for_deploy(
db: Session,
requester: User,
key_ids: List[int],
instance_owner_id: Optional[int] = None,
) -> str:
"""Resolve a list of SSH key ids into a newline-joined block of public keys.
Authorization rules:
* Regular users may use only their own keys.
* Admins (`users.manage` permission) may use their own keys and the keys
of the designated `instance_owner_id` user.
"""
if not key_ids:
return ""
allowed_owners = {requester.id}
is_admin = requester.has_permission("users.manage") or requester.is_admin
if is_admin and instance_owner_id:
allowed_owners.add(instance_owner_id)
keys = (
db.query(UserSSHKey)
.filter(UserSSHKey.id.in_(key_ids))
.all()
)
if len(keys) != len(set(key_ids)):
raise HTTPException(status_code=404, detail="One or more SSH keys not found")
for k in keys:
if k.user_id not in allowed_owners:
logger.warning(
f"User {requester.username} tried to use ssh key #{k.id} "
f"belonging to user_id={k.user_id}"
)
raise HTTPException(
status_code=403,
detail="You are not allowed to use one of the selected SSH keys",
)
return "\n".join(k.public_key.strip() for k in keys if k.public_key).strip()

View File

@@ -166,6 +166,7 @@ def list_os_templates(
current_user: User = Depends(PermissionChecker("templates.view")),
group_id: int = None,
server_id: int = None,
vm_type: Optional[str] = None,
active_only: bool = True
):
"""Получить список всех OS шаблонов с информацией о группе"""
@@ -175,6 +176,8 @@ def list_os_templates(
query = query.filter(OSTemplate.group_id == group_id)
if server_id:
query = query.filter(OSTemplate.server_id == server_id)
if vm_type in ('qemu', 'lxc'):
query = query.filter(OSTemplate.vm_type == vm_type)
if active_only:
query = query.filter(OSTemplate.is_active == True)
@@ -358,20 +361,26 @@ def discover_proxmox_templates(
proxmox_templates = client.get_templates(node=node_filter)
# Get already added templates для ЭТОГО сервера
existing_vmids = set(
t.vmid for t in db.query(OSTemplate).filter(OSTemplate.server_id == server_id).all()
)
existing = db.query(OSTemplate).filter(OSTemplate.server_id == server_id).all()
existing_vmids = set(t.vmid for t in existing if t.vmid is not None)
existing_volids = set(t.volid for t in existing if t.volid)
result = []
for tpl in proxmox_templates:
volid = tpl.get('volid')
vmid = tpl.get('vmid')
already = (volid is not None and volid in existing_volids) or (vmid is not None and vmid in existing_vmids)
result.append({
'vmid': tpl.get('vmid'),
'name': tpl.get('name', f"Template-{tpl.get('vmid')}"),
'vmid': vmid,
'volid': volid,
'name': tpl.get('name', f"Template-{vmid or volid}"),
'node': tpl.get('node'),
'status': tpl.get('status'),
'vm_type': tpl.get('type', 'qemu'),
'is_file_template': tpl.get('is_file_template', False),
'maxmem': tpl.get('maxmem', 0),
'maxdisk': tpl.get('maxdisk', 0),
'already_added': tpl.get('vmid') in existing_vmids
'already_added': already,
})
return JSONResponse(content={
@@ -461,26 +470,31 @@ def _do_auto_import(server_id: int, username: str):
proxmox_templates = client.get_templates(node=node_filter)
existing_vmids = set(
t.vmid for t in db.query(OSTemplate).filter(OSTemplate.server_id == server_id).all()
)
existing = db.query(OSTemplate).filter(OSTemplate.server_id == server_id).all()
existing_vmids = set(t.vmid for t in existing if t.vmid is not None)
existing_volids = set(t.volid for t in existing if t.volid)
groups_cache = {g.name: g for g in db.query(OSTemplateGroup).all()}
templates_by_group = {}
for tpl in proxmox_templates:
volid = tpl.get('volid')
vmid = tpl.get('vmid')
if vmid in existing_vmids:
if volid and volid in existing_volids:
continue
template_name = tpl.get('name', f"Template-{vmid}")
if vmid is not None and vmid in existing_vmids:
continue
template_name = tpl.get('name', f"Template-{vmid or volid}")
group_info = detect_os_group(template_name)
group_name = group_info['name']
if group_name not in templates_by_group:
templates_by_group[group_name] = {'info': group_info, 'templates': []}
templates_by_group[group_name]['templates'].append({
'vmid': vmid,
'volid': volid,
'name': template_name,
'node': tpl.get('node'),
'vm_type': tpl.get('type', 'qemu'),
'maxmem': tpl.get('maxmem', 0),
'maxdisk': tpl.get('maxdisk', 0),
})
@@ -516,7 +530,9 @@ def _do_auto_import(server_id: int, username: str):
group_id=group.id,
server_id=server_id,
name=tpl['name'],
vmid=tpl['vmid'],
vmid=tpl.get('vmid'),
volid=tpl.get('volid'),
vm_type=tpl.get('vm_type', 'qemu'),
node=tpl.get('node'),
source_node=tpl.get('node'),
default_cores=2,
@@ -529,7 +545,7 @@ def _do_auto_import(server_id: int, username: str):
sort_order=global_sort_order
)
db.add(template)
imported.append({'vmid': tpl['vmid'], 'name': tpl['name'], 'group': group_name})
imported.append({'vmid': tpl.get('vmid'), 'volid': tpl.get('volid'), 'name': tpl['name'], 'group': group_name})
global_sort_order += 1
db.commit()
@@ -581,25 +597,30 @@ def auto_import_templates(
proxmox_templates = client.get_templates(node=node_filter)
existing_vmids = set(
t.vmid for t in db.query(OSTemplate).filter(OSTemplate.server_id == server_id).all()
)
existing = db.query(OSTemplate).filter(OSTemplate.server_id == server_id).all()
existing_vmids = set(t.vmid for t in existing if t.vmid is not None)
existing_volids = set(t.volid for t in existing if t.volid)
groups_cache = {g.name: g for g in db.query(OSTemplateGroup).all()}
templates_by_group = {}
for tpl in proxmox_templates:
volid = tpl.get('volid')
vmid = tpl.get('vmid')
if vmid in existing_vmids:
if volid and volid in existing_volids:
continue
template_name = tpl.get('name', f"Template-{vmid}")
if vmid is not None and vmid in existing_vmids:
continue
template_name = tpl.get('name', f"Template-{vmid or volid}")
group_info = detect_os_group(template_name)
group_name = group_info['name']
if group_name not in templates_by_group:
templates_by_group[group_name] = {'info': group_info, 'templates': []}
templates_by_group[group_name]['templates'].append({
'vmid': vmid,
'volid': volid,
'name': template_name,
'node': tpl.get('node'),
'vm_type': tpl.get('type', 'qemu'),
'maxmem': tpl.get('maxmem', 0),
'maxdisk': tpl.get('maxdisk', 0),
})
@@ -635,7 +656,9 @@ def auto_import_templates(
group_id=group.id,
server_id=server_id,
name=tpl['name'],
vmid=tpl['vmid'],
vmid=tpl.get('vmid'),
volid=tpl.get('volid'),
vm_type=tpl.get('vm_type', 'qemu'),
node=tpl.get('node'),
source_node=tpl.get('node'),
default_cores=2,
@@ -649,7 +672,8 @@ def auto_import_templates(
)
db.add(template)
imported.append({
'vmid': tpl['vmid'],
'vmid': tpl.get('vmid'),
'volid': tpl.get('volid'),
'name': tpl['name'],
'group': group_name,
'sort_order': global_sort_order
@@ -706,7 +730,8 @@ def _update_deploy_task(task_id: int, status: str, step: str, progress: int,
def _do_deploy_sync(task_id: int, deploy_data: VMDeployRequest,
user_id: int, username: str, user_ssh_key: str):
user_id: int, username: str, user_ssh_key: str,
owner_id: int = None):
"""Blocking deploy logic — runs in a thread pool."""
db = SessionLocal()
server = None
@@ -920,7 +945,7 @@ def _do_deploy_sync(task_id: int, deploy_data: VMDeployRequest,
template_id=template.id,
template_name=template.name,
description=f"Deployed from template {template.name}",
owner_id=user_id
owner_id=owner_id or user_id
)
LoggingService.log_proxmox_action(
@@ -999,6 +1024,25 @@ async def deploy_vm_from_template(
if disk < template.min_disk:
raise HTTPException(status_code=400, detail=f"Минимум {template.min_disk} GB диска")
# Resolve owner / SSH keys (admin may target another user)
is_admin = current_user.has_permission("users.manage") or current_user.is_admin
instance_owner_id = current_user.id
if deploy_data.owner_id and deploy_data.owner_id != current_user.id:
if not is_admin:
raise HTTPException(status_code=403, detail="Only admins can assign instance to another user")
owner = db.query(User).filter(User.id == deploy_data.owner_id).first()
if not owner:
raise HTTPException(status_code=404, detail="Owner user not found")
instance_owner_id = owner.id
if deploy_data.ssh_key_ids:
from .ssh_keys import resolve_ssh_keys_for_deploy
resolved = resolve_ssh_keys_for_deploy(
db, current_user, deploy_data.ssh_key_ids, instance_owner_id
)
existing = (deploy_data.ssh_keys or "").strip()
deploy_data.ssh_keys = (existing + "\n" + resolved).strip() if existing else resolved
# Create deploy task record
task = DeployTask(
status='pending',
@@ -1023,7 +1067,8 @@ async def deploy_vm_from_template(
deploy_data,
current_user.id,
current_user.username,
current_user.ssh_public_key or ''
current_user.ssh_public_key or '',
instance_owner_id,
)
logger.info(f"[DEPLOY] Task #{task_id} queued: {deploy_data.name} by {current_user.username}")

View File

@@ -23,6 +23,7 @@ from .api import settings as settings_router
from .api import notifications as notifications_router
from .api import users as users_router
from .api import workspaces as workspaces_router
from .api import ssh_keys as ssh_keys_router
from .logging_middleware import RequestLoggingMiddleware
from .language_middleware import LanguageMiddleware
from .i18n import I18nService
@@ -408,6 +409,7 @@ def create_app() -> FastAPI:
app.include_router(notifications_router.router, tags=["notifications"])
app.include_router(users_router.router, prefix="/admin", tags=["users", "admin"])
app.include_router(workspaces_router.router, tags=["workspaces"])
app.include_router(ssh_keys_router.router, tags=["ssh-keys"])
return app

View File

@@ -205,6 +205,33 @@ class BlockedIP(Base):
return f"<BlockedIP(id={self.id}, ip={self.ip_address})>"
class UserSSHKey(Base):
"""SSH key (public + optional encrypted private) attached to a user account.
Used as a library that user can pick from when deploying VM/LXC instances.
Admins can manage keys for any user; regular users only their own.
"""
__tablename__ = "user_ssh_keys"
id = Column(Integer, primary_key=True, index=True)
user_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True)
name = Column(String(100), nullable=False)
public_key = Column(Text, nullable=False)
private_key = Column(EncryptedString(8000), nullable=True)
fingerprint = Column(String(100), nullable=True, index=True)
comment = Column(String(255), nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
user = relationship("User", backref="ssh_keys_pool")
__table_args__ = (
Index('idx_user_ssh_keys_user', 'user_id'),
)
def __repr__(self):
return f"<UserSSHKey(id={self.id}, user_id={self.user_id}, name='{self.name}')>"
class SecuritySetting(Base):
"""Security settings"""
__tablename__ = "security_settings"
@@ -322,7 +349,9 @@ class OSTemplate(Base):
server_id = Column(Integer, nullable=False, index=True) # FK to ProxmoxServer
name = Column(String(100), nullable=False) # Display name, e.g., "Ubuntu 22.04 LTS"
vmid = Column(Integer, nullable=False) # Proxmox template VMID on source node
vmid = Column(Integer, nullable=True) # Proxmox template VMID (qemu/lxc-container template). NULL for vztmpl files.
vm_type = Column(String(10), nullable=False, default="qemu", server_default="qemu") # 'qemu' or 'lxc'
volid = Column(String(500), nullable=True) # vztmpl volume id, e.g. 'local:vztmpl/ubuntu-24.04...tar.zst'
node = Column(String(100), nullable=True) # Primary/source node (optional for cross-node templates)
# Cross-node replication support

View File

@@ -1,4 +1,5 @@
import logging
import shlex
from typing import Dict, List, Optional, Union
from contextlib import asynccontextmanager
from proxmoxer import ProxmoxAPI
@@ -238,7 +239,25 @@ class ProxmoxClient:
'vms': self.get_vms(node),
'containers': self.get_containers(node)
}
def get_cluster_resources(self, type_: Optional[str] = None) -> List[Dict]:
"""Получить живые метрики со всех VM/LXC одним запросом через /cluster/resources.
Args:
type_: 'vm' для VM/LXC, 'node' для нод, None для всех
Returns:
Список ресурсов с полями cpu, mem, maxmem, disk, maxdisk, uptime, netin, netout, status и т.д.
"""
if not self.proxmox:
return []
try:
if type_:
return self.proxmox.cluster.resources.get(type=type_) or []
return self.proxmox.cluster.resources.get() or []
except Exception as e:
logger.error(f"Ошибка получения cluster/resources с {self.host}: {e}")
return []
def get_vm_status(self, node: str, vmid: int) -> Optional[Dict]:
"""Получить статус конкретной VM"""
if not self.proxmox:
@@ -433,15 +452,15 @@ class ProxmoxClient:
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]
)
# Используем printf + chpasswd; shlex.quote защищает от shell-injection
# Proxmox exec API ожидает отдельные поля: command + args[N]
entry = shlex.quote(f'{username}:{password}')
cmd_str = f"printf '%s\\n' {entry} | chpasswd"
data = {'command': 'bash', 'args[0]': '-c', 'args[1]': cmd_str}
res = self.proxmox.nodes(node).lxc(vmid).exec.post(**data)
return {'success': True, 'result': res}
except Exception as e:
logger.error(f"Не удалось сменить пароль в LXC {vmid}: {e}")
@@ -1156,6 +1175,56 @@ class ProxmoxClient:
templates.append(vm)
except Exception as e:
logger.error(f"Ошибка получения шаблонов с ноды {node_name}: {e}")
try:
lxc_cts = self.proxmox.nodes(node_name).lxc.get()
for ct in lxc_cts:
# LXC контейнер-шаблон (template=1)
if ct.get('template') in (1, '1', True):
ct['node'] = node_name
ct['type'] = 'lxc'
templates.append(ct)
except Exception as e:
logger.error(f"Ошибка получения LXC-шаблонов с ноды {node_name}: {e}")
# vztmpl файлы из storage (стандартные LXC шаблоны)
try:
storages = self.proxmox.nodes(node_name).storage.get()
seen_volids = set()
for stor in storages:
stor_name = stor.get('storage')
stor_content = stor.get('content', '') or ''
if 'vztmpl' not in stor_content:
continue
try:
items = self.proxmox.nodes(node_name).storage(stor_name).content.get(content='vztmpl')
except Exception as e:
logger.warning(f"Не удалось прочитать vztmpl из {stor_name} на {node_name}: {e}")
continue
for item in items:
volid = item.get('volid')
if not volid or volid in seen_volids:
continue
seen_volids.add(volid)
# Имя файла без 'storage:vztmpl/'
fname = volid.split('/', 1)[-1] if '/' in volid else volid
# Убираем расширения .tar.zst / .tar.gz / .tar.xz
display_name = fname
for ext in ('.tar.zst', '.tar.gz', '.tar.xz', '.tar'):
if display_name.endswith(ext):
display_name = display_name[: -len(ext)]
break
templates.append({
'vmid': None,
'volid': volid,
'name': display_name,
'node': node_name,
'type': 'lxc',
'is_file_template': True,
'maxmem': 0,
'maxdisk': int(item.get('size') or 0),
'storage': stor_name,
})
except Exception as e:
logger.error(f"Ошибка получения vztmpl с ноды {node_name}: {e}")
except Exception as e:
logger.error(f"Ошибка получения списка шаблонов {self.host}: {e}")

View File

@@ -254,8 +254,10 @@ class OSTemplateBase(BaseModel):
group_id: int = Field(..., description="Template group ID")
server_id: int = Field(..., description="Proxmox server ID")
name: str = Field(..., min_length=1, max_length=100, description="Display name")
vmid: int = Field(..., ge=100, description="Proxmox template VMID")
node: str = Field(..., min_length=1, max_length=100, description="Proxmox node name")
vmid: Optional[int] = Field(None, ge=100, description="Proxmox template VMID (None for vztmpl file templates)")
vm_type: str = Field(default="qemu", description="Template type: qemu (KVM) or lxc")
volid: Optional[str] = Field(None, max_length=500, description="vztmpl volume id (for LXC file templates)")
node: Optional[str] = Field(None, max_length=100, description="Proxmox node name")
default_cores: int = Field(default=1, ge=1, le=128, description="Default CPU cores")
default_memory: int = Field(default=1024, ge=128, description="Default memory in MB")
default_disk: int = Field(default=10, ge=1, description="Default disk size in GB")
@@ -278,6 +280,8 @@ class OSTemplateUpdate(BaseModel):
server_id: Optional[int] = None
name: Optional[str] = Field(None, min_length=1, max_length=100)
vmid: Optional[int] = Field(None, ge=100)
vm_type: Optional[str] = None
volid: Optional[str] = Field(None, max_length=500)
node: Optional[str] = Field(None, min_length=1, max_length=100)
default_cores: Optional[int] = Field(None, ge=1, le=128)
default_memory: Optional[int] = Field(None, ge=128)
@@ -333,6 +337,8 @@ class VMDeployRequest(BaseModel):
cloud_init_user: Optional[str] = Field(None, max_length=50, description="Cloud-init username")
cloud_init_password: Optional[str] = Field(None, max_length=100, description="Cloud-init password")
ssh_keys: Optional[str] = Field(None, description="SSH public keys")
ssh_key_ids: Optional[List[int]] = Field(default=None, description="IDs of saved SSH keys to inject")
owner_id: Optional[int] = Field(default=None, description="Admin-only: assign instance to this user (also enables using their saved SSH keys)")
class VMDeployResponse(BaseModel):

View File

@@ -355,6 +355,76 @@ async def instance_metrics_loop(db_factory) -> None:
await asyncio.gather(*[_push_instance(ch) for ch in channels], return_exceptions=True)
# ──────────────────────────────────────────────────────────────────────────────
# Instances list metrics loop (push every 5 s)
# Pushes a slim list of live VM/LXC metrics from cluster/resources for the
# instances table page. One channel for all clients: 'instances_list_metrics'.
# ──────────────────────────────────────────────────────────────────────────────
def _fetch_all_instances_metrics(db_factory) -> list:
"""Pull live metrics for all VMs/LXC across all online servers via cluster/resources."""
from ..models import ProxmoxServer
from ..api.proxmox._helpers import _get_proxmox_client
db = db_factory()
try:
servers = db.query(ProxmoxServer).filter(ProxmoxServer.is_online == True).all()
finally:
db.close()
out: list = []
for server in servers:
try:
client = _get_proxmox_client(server)
# Skip is_connected() check (makes extra HTTP call); just try and catch
for res in client.get_cluster_resources(type_='vm'):
vmid = res.get('vmid')
if vmid is None:
continue
out.append({
"server_id": server.id,
"vmid": vmid,
"node": res.get('node'),
"status": res.get('status'),
"cpu": res.get('cpu'),
"mem": res.get('mem'),
"maxmem": res.get('maxmem'),
"disk": res.get('disk'),
"maxdisk": res.get('maxdisk'),
"uptime": res.get('uptime'),
"netin": res.get('netin'),
"netout": res.get('netout'),
})
except Exception as exc:
logger.debug(f"[metrics_broadcaster] instances list metrics error for server {server.id}: {exc}")
return out
async def instances_list_metrics_loop(db_factory) -> None:
"""Push live VM/LXC metrics list to 'instances_list_metrics' subscribers (~1 s)."""
wsman = _get_ws_manager()
channel = "instances_list_metrics"
while True:
await asyncio.sleep(1)
if not wsman.has_channel_subscribers(channel):
continue
try:
items = await asyncio.to_thread(_fetch_all_instances_metrics, db_factory)
if not items:
continue
payload = {
"type": "metrics_update",
"channel": channel,
"data": {"items": items},
}
_metrics_cache[channel] = payload
await wsman.broadcast_to_channel(channel, payload)
except asyncio.CancelledError:
break
except Exception as exc:
logger.warning(f"[metrics_broadcaster] instances list loop error: {exc}")
# ──────────────────────────────────────────────────────────────────────────────
# On-demand RRD request handler
# ──────────────────────────────────────────────────────────────────────────────
@@ -407,4 +477,5 @@ def start_metrics_broadcaster(db_factory) -> None:
asyncio.create_task(dashboard_metrics_loop(db_factory), name="metrics:dashboard")
asyncio.create_task(server_metrics_loop(db_factory), name="metrics:servers")
asyncio.create_task(instance_metrics_loop(db_factory), name="metrics:instances")
asyncio.create_task(instances_list_metrics_loop(db_factory), name="metrics:instances_list")
logger.info("[metrics_broadcaster] Metrics broadcasting tasks started")

View File

@@ -627,6 +627,7 @@ class MonitoringWorker:
seen_standalone = {} # server_id -> set of vmid
all_vms_data = [] # List of (server_id, vm_data, vm_type)
server_clients = {} # server_id (or cluster_id) -> ProxmoxClient
sync_time = utcnow()
# Collect all VMs from all servers with proper dedup
@@ -650,11 +651,15 @@ class MonitoringWorker:
seen_in_cluster[cluster_id] = set()
seen_set = seen_in_cluster[cluster_id]
use_server_id = cluster_id # Use cluster_id for all VMs in cluster
# Map cluster_id -> client for IP detection later
if cluster_id not in server_clients:
server_clients[cluster_id] = client
else:
if server.id not in seen_standalone:
seen_standalone[server.id] = set()
seen_set = seen_standalone[server.id]
use_server_id = server.id
server_clients[server.id] = client
server_vm_count = 0
server_ct_count = 0
@@ -690,6 +695,8 @@ class MonitoringWorker:
all_seen = set()
# Track status changes for WebSocket broadcast
status_changes = [] # list of (server_id, vmid, node, old_status, new_status)
# Track newly-created VMs for WebSocket broadcast
created_vms = [] # list of (server_id, vmid, node, name, vm_type)
# Upsert all VMs to database
for server_id, vm_data, vm_type in all_vms_data:
@@ -706,6 +713,37 @@ class MonitoringWorker:
ipam_by_name.get(vm_name.lower())
)
ip_address = ipam_alloc.ip_address if ipam_alloc else None
# If no IPAM IP, try to detect from Proxmox (LXC config or QEMU guest agent)
if not ip_address:
try:
px_client = server_clients.get(server_id)
if px_client and node_name:
if vm_type == 'lxc':
config = px_client.proxmox.nodes(node_name).lxc(vmid).config.get()
for i in range(4):
ipconfig = config.get(f'ipconfig{i}', '')
if 'ip=' in str(ipconfig):
ip_part = str(ipconfig).split('ip=')[1].split(',')[0]
if ip_part and ip_part != 'dhcp':
ip_address = ip_part.split('/')[0] if '/' in ip_part else ip_part
break
elif vm_type == 'qemu' and vm_data.get('status') == 'running':
result = px_client.proxmox.nodes(node_name).qemu(vmid).agent('network-get-interfaces').get()
if result and 'result' in result:
for iface in result['result']:
if iface.get('name') == 'lo':
continue
for ip_info in (iface.get('ip-addresses') or []):
if ip_info.get('ip-address-type') == 'ipv4':
addr = ip_info.get('ip-address', '')
if addr and not addr.startswith('127.'):
ip_address = addr
break
if ip_address:
break
except Exception:
pass
# Get OS type
if vm_type == 'qemu':
@@ -727,6 +765,10 @@ class MonitoringWorker:
# Detect status change for WS broadcast
if existing.status != new_status:
status_changes.append((server_id, vmid, node_name, existing.status, new_status))
# If this VM was previously soft-deleted but reappeared,
# treat it as a new VM for broadcast purposes
if existing.deleted_at is not None:
created_vms.append((server_id, vmid, node_name, vm_name, vm_type))
# Update existing entry
existing.name = vm_name
existing.node = node_name
@@ -737,8 +779,8 @@ class MonitoringWorker:
existing.os_type = os_type
existing.vm_type = vm_type
existing.is_template = bool(vm_data.get('template'))
# Only overwrite ip_address if IPAM has a value;
# otherwise preserve user-assigned IP from deploy/wizard
# Only overwrite ip_address if we have a value;
# IPAM and detected IPs take priority; otherwise preserve existing
if ip_address:
existing.ip_address = ip_address
existing.last_sync_at = sync_time
@@ -761,6 +803,7 @@ class MonitoringWorker:
last_sync_at=sync_time
)
db.add(new_vm)
created_vms.append((server_id, vmid, node_name, vm_name, vm_type))
except Exception as e:
logger.error(f"[VM SYNC] Error processing VM {vmid}: {e}")
@@ -788,10 +831,12 @@ class MonitoringWorker:
deleted_count = 0
released_ips = 0
deleted_broadcasts = [] # list of (server_id, vmid, node, name, vm_type)
for vm in active_vms:
if (vm.server_id, vm.vmid) not in all_seen:
vm.deleted_at = sync_time
deleted_count += 1
deleted_broadcasts.append((vm.server_id, vm.vmid, vm.node, vm.name, vm.vm_type))
logger.debug(f"[VM SYNC] Marked VM {vm.name} (server={vm.server_id}, vmid={vm.vmid}) as deleted")
# Release IPAM allocation for deleted VM
@@ -828,6 +873,40 @@ class MonitoringWorker:
}))
except Exception as _we:
logger.warning(f"[VM SYNC] Failed to broadcast status changes: {_we}")
# Broadcast deletions so connected clients can update their lists in real time
if deleted_broadcasts:
try:
from app.websocket_manager import ws_manager, run_async_safe
for srv_id, vmid, node_name, vm_name, vm_type in deleted_broadcasts:
logger.info(f"[VM SYNC] VM deleted: server={srv_id} vmid={vmid} name={vm_name}")
run_async_safe(ws_manager.broadcast({
"type": "vm_deleted",
"server_id": srv_id,
"vmid": vmid,
"node": node_name,
"name": vm_name,
"vm_type": vm_type,
}))
except Exception as _we:
logger.warning(f"[VM SYNC] Failed to broadcast deletions: {_we}")
# Broadcast newly-created VMs so clients can refetch and show them
if created_vms:
try:
from app.websocket_manager import ws_manager, run_async_safe
for srv_id, vmid, node_name, vm_name, vm_type in created_vms:
logger.info(f"[VM SYNC] VM created: server={srv_id} vmid={vmid} name={vm_name}")
run_async_safe(ws_manager.broadcast({
"type": "vm_created",
"server_id": srv_id,
"vmid": vmid,
"node": node_name,
"name": vm_name,
"vm_type": vm_type,
}))
except Exception as _we:
logger.warning(f"[VM SYNC] Failed to broadcast creations: {_we}")
except Exception as e:
logger.error(f"[VM SYNC] Critical error: {e}", exc_info=True)