NAS-132659 / 24.10.2 / Don't alow to enable ssh password login for user where 2fa secret is not available (by sonicaj)

src/middlewared/middlewared/api/v25_04_0/user.py

@@ -25,16 +25,16 @@ class UserEntry(BaseModel):
     builtin: bool
     smb: bool = True
     group: dict
-    groups: list[int] = []
+    groups: list[int] = Field(default_factory=list)
     """Specifies whether the user should be allowed access to SMB shares. User will also automatically be added to
     the `builtin_users` group."""
     password_disabled: bool = False
     ssh_password_enabled: bool = False
     "Required if `password_disabled` is false."
     sshpubkey: LongString | None = None
     locked: bool = False
-    sudo_commands: list[NonEmptyString] = []
-    sudo_commands_nopasswd: list[NonEmptyString] = []
+    sudo_commands: list[NonEmptyString] = Field(default_factory=list)
+    sudo_commands_nopasswd: list[NonEmptyString] = Field(default_factory=list)
     email: EmailStr | None = None
     id_type_both: bool
     local: bool

src/middlewared/middlewared/plugins/account.py

@@ -1437,6 +1437,25 @@ async def common_validation(self, verrors, data, schema, group_ids, old=None):
                 await self.middleware.run_in_thread(validate_sudo_commands, data['sudo_commands_nopasswd']),
             )
 
+        two_factor_config = await self.middleware.call('auth.twofactor.config')
+        if (
+            data.get(
+                'ssh_password_enabled', False
+            ) and two_factor_config['enabled'] and two_factor_config['services']['ssh']
+        ):
+            error = [
+                f'{schema}.ssh_password_enabled',
+                '2FA for this user needs to be explicitly configured before password based SSH access is enabled.'
+            ]
+            if old is None:
+                error[1] += (' User will be created with SSH password access disabled and after 2FA has been '
+                             'configured for this user, SSH password access can be enabled.')
+                verrors.add(*error)
+            elif (
+                await self.middleware.call('user.translate_username', old['username'])
+            )['twofactor_auth_configured'] is False:
+                verrors.add(*error)
+
     def __set_password(self, data):
         if 'password' not in data:
             return

src/middlewared/middlewared/pytest/unit/plugins/test_user_ssh_enabled_validation.py

@@ -0,0 +1,128 @@
+import pytest
+
+from middlewared.plugins.account import UserService
+from middlewared.pytest.unit.middleware import Middleware
+from middlewared.service import ValidationErrors
+
+
+@pytest.mark.parametrize('data,old_data,twofactor_enabled,twofactor_config,expected_error', [
+    (
+        {
+            'ssh_password_enabled': True
+        },
+        None,
+        False,
+        {
+            'services': {
+                'ssh': False
+            },
+            'enabled': False,
+        },
+        ''
+    ),
+    (
+        {
+            'ssh_password_enabled': True
+        },
+        None,
+        False,
+        {
+            'services': {
+                'ssh': True
+            },
+            'enabled': True,
+        },
+        '[EINVAL] test_schema.ssh_password_enabled:'
+        ' 2FA for this user needs to be explicitly configured before password based SSH access is enabled.'
+        ' User will be created with SSH password access disabled and after 2FA has been'
+        ' configured for this user, SSH password access can be enabled.'
+    ),
+    (
+        {
+            'ssh_password_enabled': True
+        },
+        {},
+        False,
+        {
+            'services': {
+                'ssh': True
+            },
+            'enabled': True,
+        },
+        '[EINVAL] test_schema.ssh_password_enabled:'
+        ' 2FA for this user needs to be explicitly configured before password based SSH access is enabled.'
+    ),
+    (
+        {
+            'ssh_password_enabled': True
+        },
+        {},
+        True,
+        {
+            'services': {
+                'ssh': True
+            },
+            'enabled': True,
+        },
+        ''
+    ),
+    (
+        {
+            'ssh_password_enabled': True
+        },
+        {},
+        False,
+        {
+            'services': {
+                'ssh': False
+            },
+            'enabled': True,
+        },
+        ''
+    ),
+    (
+        {
+            'ssh_password_enabled': True
+        },
+        None,
+        False,
+        {
+            'services': {
+                'ssh': False
+            },
+            'enabled': True,
+        },
+        ''
+    ),
+
+])
+@pytest.mark.asyncio
+async def test_use_ssh_enabled_validation(data, old_data, twofactor_enabled, twofactor_config, expected_error):
+    m = Middleware()
+    m['datastore.query'] = lambda *arg: []
+    m['smb.is_configured'] = lambda *arg: False
+    m['auth.twofactor.config'] = lambda *arg: twofactor_config
+    m['user.translate_username'] = lambda *args: {'twofactor_auth_configured': twofactor_enabled}
+    data['smb'] = False
+    data.update(
+        {
+            'smb': False,
+            'password_disabled': True
+        }
+    )
+    if old_data is not None:
+        old_data.update(
+            {
+                'username': '',
+                'id': 0
+            }
+        )
+    verrors = ValidationErrors()
+    if expected_error:
+        with pytest.raises(ValidationErrors) as ve:
+            await UserService(m).common_validation(verrors, data, 'test_schema', [], old_data,)
+            verrors.check()
+        assert str(ve.value.errors[0]) == expected_error
+    else:
+        await UserService(m).common_validation(verrors, data, 'test_schema', [], old_data, )
+        assert list(verrors) == []