fix(instances): assign the owner on create and inject the owner's SSH keys

Newly created VMs/LXCs ended up with owner_id = NULL, and a VM created by an
admin on behalf of a user received the admin's SSH key instead of the user's.

Ownership: save_vm_instance() only set owner_id when inserting a new row. The
row practically always existed by then — sync_vm_cache runs every 10s and
registers VMs it sees on Proxmox with owner_id = NULL while the clone is still
running, whereas the deploy saves the owner minutes later at 95% progress. So
every deploy landed in the update branch and silently dropped the owner. Set
owner_id there too, but only when it is passed explicitly: callers such as
save-before-delete do not know the owner and must not blank it out.

SSH keys: the profile key was taken from whoever pressed "Create" rather than
from the instance owner, and the LXC path injected no profile key at all. Add
build_deploy_ssh_keys(), which merges pasted keys, keys picked from the library
(same authorization rules as before) and the owner's profile key, deduplicated.
Both the template and LXC deploy endpoints now use it, so the key material is
resolved in one place instead of inside the deploy worker.

ISO VMs are left alone on purpose: they are blank VMs without cloud-init, so
there is nowhere to inject keys.

Covered by backend/tests/test_instance_ownership.py, including the sync race.

backend/app/api/proxmox/_helpers.py

@@ -465,6 +465,15 @@ def save_vm_instance(
         existing.template_name = template_name
         existing.description = description
         existing.extra_config = extra_config
+        # Владельца проставляем только когда он передан явно: часть вызовов
+        # (например, сохранение конфига перед удалением) владельца не знает,
+        # и None не должен обезличивать инстанс.
+        # Без этой строки владелец, выбранный при создании, терялся: фоновый
+        # sync_vm_cache идёт каждые 10с и заводит запись с owner_id=NULL ещё
+        # во время клонирования, поэтому деплой почти всегда попадает сюда,
+        # а не в ветку создания новой записи.
+        if owner_id is not None:
+            existing.owner_id = owner_id
         existing.updated_at = func.now()
         db.commit()
         db.refresh(existing)
@@ -496,6 +505,7 @@ def save_vm_instance(
         deleted_existing.template_name = template_name
         deleted_existing.description = description
         deleted_existing.extra_config = extra_config
+        if owner_id is not None:
             deleted_existing.owner_id = owner_id
         deleted_existing.status = 'unknown'
         deleted_existing.deleted_at = None  # Восстанавливаем запись

backend/app/api/proxmox/lxc.py

@@ -463,11 +463,10 @@ async def deploy_lxc_from_template(
     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)
-        existing = (req.ssh_keys or "").strip()
-        req.ssh_keys = (existing + "\n" + resolved).strip() if existing else resolved
+    from ..ssh_keys import build_deploy_ssh_keys
+    req.ssh_keys = build_deploy_ssh_keys(
+        db, current_user, instance_owner_id, req.ssh_key_ids, req.ssh_keys,
+    ) or None
 
     # Create task record
     task = DeployTask(

backend/app/api/ssh_keys.py

@@ -451,3 +451,48 @@ def resolve_ssh_keys_for_deploy(
             )
 
     return "\n".join(k.public_key.strip() for k in keys if k.public_key).strip()
+
+
+def build_deploy_ssh_keys(
+    db: Session,
+    requester: User,
+    instance_owner_id: int,
+    key_ids: Optional[List[int]] = None,
+    raw_keys: Optional[str] = None,
+) -> str:
+    """Собрать итоговый блок public-ключей для cloud-init / LXC.
+
+    Объединяет вставленные вручную ключи, выбранные из библиотеки ключи
+    (с проверкой прав) и профильный ключ **владельца** инстанса.
+
+    Профильный ключ берётся именно у владельца, а не у того, кто нажал
+    «Создать»: иначе VM, созданная админом для пользователя, получала ключ
+    админа, а ключ самого пользователя на неё не попадал.
+    """
+    parts: List[str] = []
+
+    if raw_keys:
+        parts.append(raw_keys)
+
+    if key_ids:
+        parts.append(
+            resolve_ssh_keys_for_deploy(db, requester, key_ids, instance_owner_id)
+        )
+
+    owner = (
+        requester
+        if instance_owner_id == requester.id
+        else db.query(User).filter(User.id == instance_owner_id).first()
+    )
+    if owner and owner.ssh_public_key:
+        parts.append(owner.ssh_public_key)
+
+    seen: set = set()
+    uniq: List[str] = []
+    for block in parts:
+        for line in (block or "").splitlines():
+            line = line.strip()
+            if line and line not in seen:
+                seen.add(line)
+                uniq.append(line)
+    return "\n".join(uniq)

backend/app/api/templates.py

@@ -754,7 +754,7 @@ 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,
                     owner_id: int = None):
     """Blocking deploy logic — runs in a thread pool."""
     db = SessionLocal()
@@ -918,18 +918,10 @@ def _do_deploy_sync(task_id: int, deploy_data: VMDeployRequest,
 
         # Configure VM
         _update_deploy_task(task_id, 'running', 'Настройка VM (CPU, RAM, Cloud-Init)...', 72, vmid=new_vmid, node=deploy_node)
+        # Ключи уже собраны и дедуплицированы в эндпоинте
+        # (build_deploy_ssh_keys): вставленные вручную + выбранные из
+        # библиотеки + профильный ключ владельца инстанса.
         ssh_keys = deploy_data.ssh_keys or ""
-        if user_ssh_key:
-            ssh_keys = f"{ssh_keys}\n{user_ssh_key}".strip()
-        # Deduplicate SSH keys while preserving order
-        _seen_ssh: set = set()
-        _uniq_ssh = []
-        for _k in ssh_keys.splitlines():
-            _k = _k.strip()
-            if _k and _k not in _seen_ssh:
-                _seen_ssh.add(_k)
-                _uniq_ssh.append(_k)
-        ssh_keys = "\n".join(_uniq_ssh)
 
         config_result = client.configure_vm(
             node=deploy_node,
@@ -1084,13 +1076,11 @@ async def deploy_vm_from_template(
     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(
-            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
+    from .ssh_keys import build_deploy_ssh_keys
+    deploy_data.ssh_keys = build_deploy_ssh_keys(
+        db, current_user, instance_owner_id,
+        deploy_data.ssh_key_ids, deploy_data.ssh_keys,
+    ) or None
 
     # Create deploy task record
     task = DeployTask(
@@ -1116,7 +1106,6 @@ async def deploy_vm_from_template(
         deploy_data,
         current_user.id,
         current_user.username,
-        current_user.ssh_public_key or '',
         instance_owner_id,
     )
 

backend/tests/test_instance_ownership.py

@@ -0,0 +1,185 @@
+"""Tests for instance ownership and SSH key assignment on deploy.
+
+Covers two regressions:
+
+* ``save_vm_instance`` used to drop ``owner_id`` whenever a row for the
+  (server, vmid) pair already existed. Since ``sync_vm_cache`` runs every 10s
+  and registers VMs it sees on Proxmox with ``owner_id=NULL``, a deploy that
+  takes longer than one sync tick (i.e. every real deploy) always landed in
+  that branch and produced ownerless instances.
+* SSH keys were resolved against the *requester's* profile, so a VM created by
+  an admin on behalf of a user received the admin's key instead of the owner's.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from app.api.proxmox._helpers import save_vm_instance
+from app.api.ssh_keys import build_deploy_ssh_keys
+from app.models import ProxmoxServer, VMInstance
+
+
+KEY_ADMIN = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIadmin admin@panel"
+KEY_OWNER = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIowner owner@laptop"
+KEY_PASTED = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIpasted pasted@manual"
+
+
+@pytest.fixture
+def server(db_session):
+    srv = ProxmoxServer(
+        name="pve-test",
+        hostname="pve-test.local",
+        ip_address="10.0.0.1",
+        port=8006,
+        api_user="root@pam",
+        password="secret",
+        use_password=True,
+        verify_ssl=False,
+    )
+    db_session.add(srv)
+    db_session.commit()
+    return srv
+
+
+def _sync_worker_registers_vm(db_session, server_id: int, vmid: int) -> VMInstance:
+    """Mimic sync_vm_cache picking the VM up mid-deploy (no owner known)."""
+    row = VMInstance(
+        server_id=server_id, vmid=vmid, node="pve1",
+        vm_type="qemu", name="web-01", status="running",
+    )
+    db_session.add(row)
+    db_session.commit()
+    return row
+
+
+# ── Ownership ───────────────────────────────────────────────────────────────
+
+def test_owner_assigned_when_sync_registered_vm_first(db_session, server, seed_users):
+    """The deploy must claim the row the background sync already created."""
+    owner = seed_users["user"]
+    _sync_worker_registers_vm(db_session, server.id, 100)
+
+    save_vm_instance(
+        db=db_session, server_id=server.id, vmid=100, node="pve1",
+        vm_type="qemu", name="web-01", owner_id=owner.id,
+    )
+
+    row = db_session.query(VMInstance).filter_by(server_id=server.id, vmid=100).one()
+    assert row.owner_id == owner.id
+
+
+def test_owner_assigned_when_no_prior_row(db_session, server, seed_users):
+    owner = seed_users["user"]
+    save_vm_instance(
+        db=db_session, server_id=server.id, vmid=101, node="pve1",
+        vm_type="qemu", name="web-02", owner_id=owner.id,
+    )
+
+    row = db_session.query(VMInstance).filter_by(server_id=server.id, vmid=101).one()
+    assert row.owner_id == owner.id
+
+
+def test_owner_preserved_when_caller_omits_owner(db_session, server, seed_users):
+    """Callers that don't know the owner (e.g. save-before-delete) must not wipe it."""
+    owner = seed_users["user"]
+    save_vm_instance(
+        db=db_session, server_id=server.id, vmid=102, node="pve1",
+        vm_type="qemu", name="web-03", owner_id=owner.id,
+    )
+
+    save_vm_instance(
+        db=db_session, server_id=server.id, vmid=102, node="pve1",
+        vm_type="qemu", name="web-03",
+    )
+
+    row = db_session.query(VMInstance).filter_by(server_id=server.id, vmid=102).one()
+    assert row.owner_id == owner.id
+
+
+def test_owner_reassigned_on_restore_of_soft_deleted_row(db_session, server, seed_users):
+    admin, owner = seed_users["admin"], seed_users["user"]
+    save_vm_instance(
+        db=db_session, server_id=server.id, vmid=103, node="pve1",
+        vm_type="qemu", name="web-04", owner_id=admin.id,
+    )
+    row = db_session.query(VMInstance).filter_by(server_id=server.id, vmid=103).one()
+    from sqlalchemy import func
+    row.deleted_at = func.now()
+    db_session.commit()
+
+    save_vm_instance(
+        db=db_session, server_id=server.id, vmid=103, node="pve1",
+        vm_type="qemu", name="web-04", owner_id=owner.id,
+    )
+
+    row = db_session.query(VMInstance).filter_by(server_id=server.id, vmid=103).one()
+    assert row.deleted_at is None
+    assert row.owner_id == owner.id
+
+
+# ── SSH keys ────────────────────────────────────────────────────────────────
+
+def test_admin_deploy_uses_owner_profile_key_not_own(db_session, seed_users):
+    admin, owner = seed_users["admin"], seed_users["user"]
+    admin.ssh_public_key = KEY_ADMIN
+    owner.ssh_public_key = KEY_OWNER
+    db_session.commit()
+
+    keys = build_deploy_ssh_keys(db_session, admin, owner.id)
+
+    assert KEY_OWNER in keys
+    assert KEY_ADMIN not in keys
+
+
+def test_self_deploy_uses_own_profile_key(db_session, seed_users):
+    user = seed_users["user"]
+    user.ssh_public_key = KEY_OWNER
+    db_session.commit()
+
+    keys = build_deploy_ssh_keys(db_session, user, user.id)
+
+    assert keys.strip() == KEY_OWNER
+
+
+def test_pasted_keys_and_profile_key_are_merged_and_deduped(db_session, seed_users):
+    user = seed_users["user"]
+    user.ssh_public_key = KEY_OWNER
+    db_session.commit()
+
+    keys = build_deploy_ssh_keys(
+        db_session, user, user.id, raw_keys=f"{KEY_PASTED}\n{KEY_OWNER}"
+    )
+
+    assert keys.splitlines() == [KEY_PASTED, KEY_OWNER]
+
+
+def test_saved_library_keys_are_included(db_session, seed_users):
+    from app.models import UserSSHKey
+
+    user = seed_users["user"]
+    saved = UserSSHKey(user_id=user.id, name="laptop", public_key=KEY_PASTED)
+    db_session.add(saved)
+    db_session.commit()
+
+    keys = build_deploy_ssh_keys(db_session, user, user.id, key_ids=[saved.id])
+
+    assert KEY_PASTED in keys
+
+
+def test_user_cannot_use_another_users_saved_key(db_session, seed_users):
+    from fastapi import HTTPException
+    from app.models import UserSSHKey
+
+    admin, user = seed_users["admin"], seed_users["user"]
+    admin_key = UserSSHKey(user_id=admin.id, name="admin-key", public_key=KEY_ADMIN)
+    db_session.add(admin_key)
+    db_session.commit()
+
+    with pytest.raises(HTTPException) as exc:
+        build_deploy_ssh_keys(db_session, user, user.id, key_ids=[admin_key.id])
+    assert exc.value.status_code == 403
+
+
+def test_no_keys_anywhere_yields_empty_string(db_session, seed_users):
+    assert build_deploy_ssh_keys(db_session, seed_users["user"], seed_users["user"].id) == ""