Allow specifying VNC passwords for incus VMs

src/middlewared/middlewared/api/v25_04_0/virt_instance.py

@@ -1,6 +1,6 @@
 from typing import Annotated, Literal, TypeAlias
 
-from pydantic import Field, model_validator, StringConstraints
+from pydantic import Field, model_validator, Secret, StringConstraints
 
 from middlewared.api.base import BaseModel, ForUpdateMetaclass, NonEmptyString, single_argument_args
 
@@ -63,6 +63,7 @@ class VirtInstanceEntry(BaseModel):
     raw: dict | None
     vnc_enabled: bool
     vnc_port: int | None
+    vnc_password: Secret[NonEmptyString | None]
 
 
 # Lets require at least 32MiB of reserved memory
@@ -87,6 +88,7 @@ class VirtInstanceCreateArgs(BaseModel):
     memory: MemoryType | None = None
     enable_vnc: bool = False
     vnc_port: int | None = Field(ge=5900, le=65535, default=None)
+    vnc_password: Secret[NonEmptyString | None] = None
 
     @model_validator(mode='after')
     def validate_attrs(self):
@@ -99,6 +101,9 @@ def validate_attrs(self):
             if self.enable_vnc and self.vnc_port is None:
                 raise ValueError('VNC port must be set when VNC is enabled')
 
+            if self.vnc_password is not None and not self.enable_vnc:
+                raise ValueError('VNC password can only be set when VNC is enabled')
+
             if self.source_type == 'ISO' and self.iso_volume is None:
                 raise ValueError('ISO volume must be set when source type is "ISO"')
 
@@ -119,6 +124,8 @@ class VirtInstanceUpdate(BaseModel, metaclass=ForUpdateMetaclass):
     memory: MemoryType | None = None
     vnc_port: int | None = Field(ge=5900, le=65535)
     enable_vnc: bool
+    vnc_password: Secret[NonEmptyString | None]
+    '''Setting vnc_password to null will unset VNC password'''
 
 
 class VirtInstanceUpdateArgs(BaseModel):

src/middlewared/middlewared/plugins/virt/instance.py

@@ -158,15 +158,35 @@ async def validate(self, new, schema_name, verrors, old=None):
             enable_vnc = new.get('enable_vnc')
             if enable_vnc is False:
                 # User explicitly disabled VNC support, let's remove vnc port
-                new['vnc_port'] = None
+                new.update({
+                    'vnc_port': None,
+                    'vnc_password': None,
+                })
+            elif enable_vnc is True:
+                if not old['vnc_port'] and not new.get('vnc_port'):
+                    verrors.add(f'{schema_name}.vnc_port', 'VNC port is required when VNC is enabled')
+                elif not new.get('vnc_port'):
+                    new['vnc_port'] = old['vnc_port']
+
+                if 'vnc_password' not in new:
+                    new['vnc_password'] = old['vnc_password']
             elif enable_vnc is None:
-                if new.get('vnc_port'):
-                    verrors.add(f'{schema_name}.enable_vnc', 'Should be set when vnc_port is specified')
-                elif old['vnc_enabled'] and old['vnc_port']:
+                for k in ('vnc_port', 'vnc_password'):
+                    if new.get(k):
+                        verrors.add(f'{schema_name}.enable_vnc', f'Should be set when {k!r} is specified')
+
+                if old['vnc_enabled'] and old['vnc_port']:
                     # We want to handle the case where nothing has been changed on vnc attrs
                     new.update({
                         'enable_vnc': True,
                         'vnc_port': old['vnc_port'],
+                        'vnc_password': old['vnc_password'],
+                    })
+                else:
+                    new.update({
+                        'enable_vnc': False,
+                        'vnc_port': None,
+                        'vnc_password': None,
                     })
         else:
             # Creation case
@@ -217,11 +237,20 @@ def __data_to_config(self, data: dict, raw: dict = None, instance_type=None):
             config['boot.autostart'] = str(data['autostart']).lower()
 
         if instance_type == 'VM':
+            config['user.ix_old_raw_qemu_config'] = raw.get('raw.qemu', '') if raw else ''
+            config['user.ix_vnc_config'] = json.dumps({
+                'vnc_enabled': data['enable_vnc'],
+                'vnc_port': data['vnc_port'],
+                'vnc_password': data['vnc_password'],
+            })
+
             if data.get('enable_vnc') and data.get('vnc_port'):
-                config['user.ix_old_raw_qemu_config'] = raw.get('raw.qemu', '') if raw else ''
-                config['raw.qemu'] = f'-vnc :{data["vnc_port"] - VNC_BASE_PORT}'
+                vnc_config = f'-vnc :{data["vnc_port"] - VNC_BASE_PORT}'
+                if data.get('vnc_password'):
+                    vnc_config = f'-object secret,id=vnc0,data={data["vnc_password"]} {vnc_config},password-secret=vnc0'
+
+                config['raw.qemu'] = vnc_config
             if data.get('enable_vnc') is False:
-                config['user.ix_old_raw_qemu_config'] = raw.get('raw.qemu', '') if raw else ''
                 config['raw.qemu'] = ''
 
         return config

src/middlewared/middlewared/plugins/virt/utils.py

@@ -2,7 +2,7 @@
 import aiohttp
 import enum
 import httpx
-import re
+import json
 from collections.abc import Callable
 
 from .websocket import IncusWS
@@ -12,7 +12,6 @@
 
 SOCKET = '/var/lib/incus/unix.socket'
 HTTP_URI = 'http://unix.socket'
-RE_VNC_PORT = re.compile(r'vnc.*?:(\d+)\s*')
 VNC_BASE_PORT = 5900
 
 
@@ -98,15 +97,9 @@ def get_vnc_info_from_config(config: dict):
     vnc_config = {
         'vnc_enabled': False,
         'vnc_port': None,
+        'vnc_password': None,
     }
-    if not (raw_qemu_config := config.get('raw.qemu')) or 'vnc' not in raw_qemu_config:
+    if not (vnc_raw_config := config.get('user.ix_vnc_config')):
         return vnc_config
 
-    for flag in raw_qemu_config.split('-'):
-        if port := RE_VNC_PORT.findall(flag):
-            return {
-                'vnc_enabled': True,
-                'vnc_port': int(port[0]) + VNC_BASE_PORT,
-            }
-
-    return vnc_config
+    return json.loads(vnc_raw_config)

tests/api2/test_virt_vm.py

@@ -35,6 +35,7 @@ def vm(virt_pool):
         'vnc_port': VNC_PORT,
         'enable_vnc': True,
         'instance_type': 'VM',
+        'vnc_password': 'test123'
     }, job=True)
     call('virt.instance.stop', VM_NAME, {'force': True, 'timeout': 1}, job=True)
     try:
@@ -119,11 +120,13 @@ def test_vm_props(vm):
     # Testing VNC specific bits
     assert instance['vnc_enabled'] is True, instance
     assert instance['vnc_port'] == VNC_PORT, instance
+    assert instance['vnc_password'] == 'test123', instance
 
     # Going to unset VNC
     call('virt.instance.update', VM_NAME, {'enable_vnc': False}, job=True)
     instance = call('virt.instance.get_instance', VM_NAME, {'extra': {'raw': True}})
-    assert instance['raw']['config']['user.ix_old_raw_qemu_config'] == f'-vnc :{VNC_PORT - 5900}'
+    assert instance['raw']['config']['user.ix_old_raw_qemu_config'] == f'-object secret,id=vnc0,data=test123 ' \
+                                                                       f'-vnc :{VNC_PORT - 5900},password-secret=vnc0'
     assert instance['vnc_enabled'] is False, instance
     assert instance['vnc_port'] is None, instance
 
@@ -134,9 +137,18 @@ def test_vm_props(vm):
     assert instance['raw']['config']['raw.qemu'] == f'-vnc :{1001}'
     assert instance['vnc_port'] == 6901, instance
 
+    # Going to update password
+    call('virt.instance.update', VM_NAME, {'vnc_password': 'update_test123', 'enable_vnc': True}, job=True)
+    instance = call('virt.instance.get_instance', VM_NAME, {'extra': {'raw': True}})
+    assert instance['raw']['config'].get('user.ix_old_raw_qemu_config') == f'-vnc :{1001}'
+    assert instance['raw']['config']['raw.qemu'] == f'-object secret,id=vnc0,data=update_test123' \
+                                                    f' -vnc :{1001},password-secret=vnc0'
+    assert instance['vnc_port'] == 6901, instance
+
     # Changing nothing
     instance = call('virt.instance.update', VM_NAME, {}, job=True)
     assert instance['vnc_port'] == 6901, instance
+    assert instance['vnc_password'] == 'update_test123', instance
 
 
 def test_vm_iso_volume(vm, iso_volume):
@@ -184,18 +196,20 @@ def test_iso_param_validation_on_vm_create(virt_pool, iso_volume, error_msg):
     assert ve.value.errors[0].errmsg == error_msg
 
 
-@pytest.mark.parametrize('enable_vnc,vnc_port,error_msg', [
-    (True, None, 'Value error, VNC port must be set when VNC is enabled'),
-    (True, 6901, 'VNC port is already in use by another virt instance'),
-    (True, 23, 'Input should be greater than or equal to 5900'),
+@pytest.mark.parametrize('enable_vnc,vnc_password,vnc_port,error_msg', [
+    (True, None, None, 'Value error, VNC port must be set when VNC is enabled'),
+    (True, None, 6901, 'VNC port is already in use by another virt instance'),
+    (True, None, 23, 'Input should be greater than or equal to 5900'),
+    (False, 'test_123', None, 'Value error, VNC password can only be set when VNC is enabled'),
 ])
-def test_vnc_validation_on_vm_create(virt_pool, enable_vnc, vnc_port, error_msg):
+def test_vnc_validation_on_vm_create(virt_pool, enable_vnc, vnc_password, vnc_port, error_msg):
     with pytest.raises(ValidationErrors) as ve:
         call('virt.instance.create', {
              'name': 'test-vnc-vm',
              'instance_type': 'VM',
              'source_type': None,
              'vnc_port': vnc_port,
+             'vnc_password': vnc_password,
              'enable_vnc': enable_vnc,
         }, job=True)