feat(quotas): per-user resource limits for VM/LXC deployment

Add per-user quotas (max instances, vCPU, RAM, disk) enforced at deploy
time and surfaced to users self-service.

Backend:
- UserQuota model (one-to-one, NULL = unlimited) + migration #28
- quota_service: usage from vm_instances cache, check_quota raises 429
- enforce in VM (templates.py) and LXC (lxc.py) deploy paths
- RBAC quota:view / quota:manage, granted to admin
- admin API GET/PUT /admin/api/users/{id}/quota
- self-service GET /settings/api/quota
- tighten VM list isolation via can_view_all_instances

Frontend:
- admin quota dialog in UsersPage with usage bars
- quota card on Settings page (own limits)
- remaining-quota hint in CreateInstanceWizard config step
- useUserQuota/useSetUserQuota, useMyQuota hooks; RU/EN locales

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backend/app/api/proxmox/_helpers.py

@@ -20,43 +20,46 @@ from ...ipam_service import IPAMService
 
 # ==================== Helper Functions for User Isolation ====================
 
+def can_view_all_instances(current_user: User) -> bool:
+    """
+    Whether the user may see/manage *every* instance, not just their own.
+
+    Only administrators — or roles that explicitly grant ``vm:manage`` — get the
+    global view. Every other authenticated user (including users with no role,
+    custom roles, or the stock ``user`` role) is scoped to instances they own.
+
+    Note: ``vm:view`` is deliberately NOT treated as a global view permission.
+    The stock ``user`` role carries ``vm:view`` simply so the instances feature
+    is usable at all; granting it would let any tenant see every VM.
+    """
+    if getattr(current_user, 'is_admin', False):
+        return True
+    role = current_user.role
+    if role and role.name == 'admin':
+        return True
+    perms = (role.permissions if role else None) or {}
+    return bool(perms.get('vm:manage', False))
+
+
 def check_vm_access(db: Session, current_user: User, server_id: int, vmid: int) -> bool:
     """
     Check if user has access to a specific VM instance.
 
-    Admins and users with 'vms:view' permission can access all VMs.
-    Users with 'user' role can only access VMs where they are the owner.
+    Admins (and roles granting ``vm:manage``) can access all VMs. Everyone else
+    can only access VMs where they are the owner.
 
     Returns True if access is allowed, False otherwise.
     """
-    # Admin role has full access
-    if current_user.role and current_user.role.name == 'admin':
+    if can_view_all_instances(current_user):
         return True
 
-    # Check permissions
-    if current_user.role and current_user.role.permissions:
-        perms = current_user.role.permissions
-        # If user has full vms:view, they can access all VMs
-        has_full_view = perms.get('vm:view', False) or perms.get('vms:view', False)
-        has_own_only = perms.get('vms:view:own', False) or perms.get('vms.view.own', False)
+    instance = db.query(VMInstance).filter(
+        VMInstance.server_id == server_id,
+        VMInstance.vmid == vmid,
+        VMInstance.deleted_at.is_(None)
+    ).first()
 
-        if has_full_view and not has_own_only:
-            return True
-
-    # For limited users, check ownership
-    if current_user.role and current_user.role.name == 'user':
-        instance = db.query(VMInstance).filter(
-            VMInstance.server_id == server_id,
-            VMInstance.vmid == vmid,
-            VMInstance.deleted_at.is_(None)
-        ).first()
-
-        if instance and instance.owner_id == current_user.id:
-            return True
-        return False
-
-    # Default: allow access (for backwards compatibility)
-    return True
+    return bool(instance and instance.owner_id == current_user.id)
 
 
 def require_vm_access(db: Session, current_user: User, server_id: int, vmid: int):

backend/app/api/proxmox/lxc.py

@@ -448,6 +448,10 @@ async def deploy_lxc_from_template(
             raise HTTPException(status_code=404, detail="Owner user not found")
         instance_owner_id = owner.id
 
+    # Enforce per-user resource quota (no-op if owner has no quota set)
+    from ...services.quota_service import check_quota
+    check_quota(db, instance_owner_id, add_cores=req.cores, add_memory_mb=req.memory, add_disk_gb=req.disk)
+
     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)

backend/app/api/proxmox/vms.py

@@ -20,7 +20,8 @@ from ...logging_service import LoggingService
 from ...ipam_service import IPAMService
 from ._helpers import (check_vm_access, require_vm_access, _get_proxmox_client,
                         get_next_vmid, archive_and_delete_snapshots,
-                        save_vm_instance, get_vm_instance, soft_delete_vm_instance)
+                        save_vm_instance, get_vm_instance, soft_delete_vm_instance,
+                        can_view_all_instances)
 from ...services.metrics_history import query_instance_metrics, timeframe_to_range
 
 router = APIRouter()
@@ -146,21 +147,11 @@ def get_all_virtual_machines(
     if workspace_server_ids is not None:
         query = query.filter(VMInstance.server_id.in_(workspace_server_ids))
     
-    # VPS-style user isolation: users with 'user' role see only their own instances
-    # Check if user has 'vm:view' but not 'vms.view:all' (or is user role)
-    user_role = current_user.role.name if current_user.role else None
-    is_limited_user = user_role == 'user'
-    
-    # Also check permission-based isolation
-    if hasattr(current_user, 'role') and current_user.role:
-        perms = current_user.role.permissions or {}
-        # If user has vms:view:own but not vms:view (full), filter by owner
-        has_view_own = perms.get('vms:view:own', False) or perms.get('vms.view.own', False)
-        has_view_all = perms.get('vms:view', False) or perms.get('vm:view', False)
-        
-        # Limited users can only see their own VMs
-        if is_limited_user or (has_view_own and not has_view_all):
-            query = query.filter(VMInstance.owner_id == current_user.id)
+    # VPS-style user isolation: only admins (or roles granting vm:manage) see
+    # every instance. Everyone else — including users with no role or a custom
+    # role — is restricted to instances they own. See can_view_all_instances.
+    if not can_view_all_instances(current_user):
+        query = query.filter(VMInstance.owner_id == current_user.id)
     
     cached_vms = query.order_by(VMInstance.name).all()
 

backend/app/api/settings.py

@@ -123,6 +123,30 @@ async def get_profile(
     }
 
 
+@router.get("/api/quota")
+async def get_my_quota(
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user)
+):
+    """Get the current user's own resource quota and usage (self-service)."""
+    from ..models import UserQuota
+    from ..services.quota_service import get_user_usage
+
+    quota = db.query(UserQuota).filter(UserQuota.user_id == current_user.id).first()
+    usage = get_user_usage(db, current_user.id)
+    return {
+        "user_id": current_user.id,
+        "max_instances": quota.max_instances if quota else None,
+        "max_cores": quota.max_cores if quota else None,
+        "max_memory_mb": quota.max_memory_mb if quota else None,
+        "max_disk_gb": quota.max_disk_gb if quota else None,
+        "used_instances": usage["instances"],
+        "used_cores": usage["cores"],
+        "used_memory_mb": usage["memory_mb"],
+        "used_disk_gb": usage["disk_gb"],
+    }
+
+
 @router.put("/api/profile")
 async def update_profile(
     data: UpdateProfileRequest,

backend/app/api/templates.py

@@ -1044,6 +1044,10 @@ async def deploy_vm_from_template(
             raise HTTPException(status_code=404, detail="Owner user not found")
         instance_owner_id = owner.id
 
+    # Enforce per-user resource quota (no-op if owner has no quota set)
+    from ..services.quota_service import check_quota
+    check_quota(db, instance_owner_id, add_cores=cores, add_memory_mb=memory, add_disk_gb=disk)
+
     if deploy_data.ssh_key_ids:
         from .ssh_keys import resolve_ssh_keys_for_deploy
         resolved = resolve_ssh_keys_for_deploy(

backend/app/api/users.py

@@ -10,7 +10,7 @@ from pydantic import BaseModel, Field, EmailStr
 from loguru import logger
 
 from ..db import get_db
-from ..models import User, Role, ProxmoxServer, Workspace, WorkspaceServer, WorkspaceUser
+from ..models import User, Role, ProxmoxServer, Workspace, WorkspaceServer, WorkspaceUser, UserQuota
 from ..auth import (
     get_password_hash,
     get_current_user,
@@ -115,6 +115,27 @@ class SecuritySettingsUpdate(BaseModel):
     password_require_special: Optional[bool] = None
 
 
+class QuotaUpdate(BaseModel):
+    """All fields optional; None / omitted clears the limit (unlimited)."""
+    max_instances: Optional[int] = Field(None, ge=0)
+    max_cores: Optional[int] = Field(None, ge=0)
+    max_memory_mb: Optional[int] = Field(None, ge=0)
+    max_disk_gb: Optional[int] = Field(None, ge=0)
+
+
+class QuotaResponse(BaseModel):
+    user_id: int
+    max_instances: Optional[int] = None
+    max_cores: Optional[int] = None
+    max_memory_mb: Optional[int] = None
+    max_disk_gb: Optional[int] = None
+    # Current usage from the vm_instances cache
+    used_instances: int = 0
+    used_cores: int = 0
+    used_memory_mb: int = 0
+    used_disk_gb: int = 0
+
+
 # ==================== Role API ====================
 
 @router.get("/api/roles", response_model=List[RoleResponse])
@@ -717,6 +738,98 @@ async def set_user_servers(
     return {"message": "Servers assigned successfully", "server_ids": [s.id for s in servers]}
 
 
+# ==================== User Quota API ====================
+
+@router.get("/api/users/{user_id}/quota", response_model=QuotaResponse)
+async def get_user_quota(
+    user_id: int,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(PermissionChecker("quota:view"))
+):
+    """Get a user's resource quota plus current usage.
+
+    If the user has no quota row, all limits are returned as null (unlimited).
+    """
+    from ..services.quota_service import get_user_usage
+
+    user = db.query(User).filter(User.id == user_id).first()
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    quota = db.query(UserQuota).filter(UserQuota.user_id == user_id).first()
+    usage = get_user_usage(db, user_id)
+
+    return QuotaResponse(
+        user_id=user_id,
+        max_instances=quota.max_instances if quota else None,
+        max_cores=quota.max_cores if quota else None,
+        max_memory_mb=quota.max_memory_mb if quota else None,
+        max_disk_gb=quota.max_disk_gb if quota else None,
+        used_instances=usage["instances"],
+        used_cores=usage["cores"],
+        used_memory_mb=usage["memory_mb"],
+        used_disk_gb=usage["disk_gb"],
+    )
+
+
+@router.put("/api/users/{user_id}/quota", response_model=QuotaResponse)
+async def set_user_quota(
+    user_id: int,
+    data: QuotaUpdate,
+    request: Request,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(PermissionChecker("quota:manage"))
+):
+    """Upsert a user's resource quota. A null/omitted field clears that limit."""
+    from ..services.quota_service import get_user_usage
+
+    user = db.query(User).filter(User.id == user_id).first()
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    quota = db.query(UserQuota).filter(UserQuota.user_id == user_id).first()
+    if quota is None:
+        quota = UserQuota(user_id=user_id)
+        db.add(quota)
+
+    quota.max_instances = data.max_instances
+    quota.max_cores = data.max_cores
+    quota.max_memory_mb = data.max_memory_mb
+    quota.max_disk_gb = data.max_disk_gb
+    db.commit()
+
+    LoggingService.log(
+        db=db,
+        level=LoggingService.INFO,
+        category=LoggingService.AUTH,
+        action="user_quota_updated",
+        message=(
+            f"Quota set for '{user.username}' by {current_user.username}: "
+            f"instances={data.max_instances}, cores={data.max_cores}, "
+            f"memory_mb={data.max_memory_mb}, disk_gb={data.max_disk_gb}"
+        ),
+        username=current_user.username,
+        user_id=current_user.id,
+        ip_address=get_client_ip(request),
+        resource_type="user",
+        resource_id=str(user.id),
+        resource_name=user.username
+    )
+
+    usage = get_user_usage(db, user_id)
+    return QuotaResponse(
+        user_id=user_id,
+        max_instances=quota.max_instances,
+        max_cores=quota.max_cores,
+        max_memory_mb=quota.max_memory_mb,
+        max_disk_gb=quota.max_disk_gb,
+        used_instances=usage["instances"],
+        used_cores=usage["cores"],
+        used_memory_mb=usage["memory_mb"],
+        used_disk_gb=usage["disk_gb"],
+    )
+
+
 @router.post("/api/users/{user_id}/reset-password")
 async def reset_user_password(
     user_id: int,