NAS-133758 / 25.04 / Add support for randomizing user passwords

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

@@ -70,6 +70,12 @@ class UserEntry(BaseModel):
     api_keys: list[int]
 
 
+class UserCreateUpdateResult(UserEntry):
+    password: NonEmptyString | None
+    """Password if it was specified in create or update payload. If random_password
+    was specified then this will be a 20 character random string."""
+
+
 class UserCreate(UserEntry):
     id: Excluded = excluded_field()
     unixhash: Excluded = excluded_field()
@@ -93,6 +99,8 @@ class UserCreate(UserEntry):
     home_create: bool = False
     home_mode: str = "700"
     password: Secret[str | None] = None
+    random_password: bool = False
+    "Generate a random 20 character password for the user"
 
 
 class UserUpdate(UserCreate, metaclass=ForUpdateMetaclass):
@@ -105,7 +113,7 @@ class UserCreateArgs(BaseModel):
 
 
 class UserCreateResult(BaseModel):
-    result: int
+    result: UserCreateUpdateResult
 
 
 class UserUpdateArgs(BaseModel):
@@ -114,7 +122,7 @@ class UserUpdateArgs(BaseModel):
 
 
 class UserUpdateResult(BaseModel):
-    result: int
+    result: UserCreateUpdateResult
 
 
 class UserDeleteOptions(BaseModel):

src/middlewared/middlewared/plugins/account.py

@@ -51,7 +51,7 @@
 from middlewared.service_exception import MatchNotFound
 import middlewared.sqlalchemy as sa
 from middlewared.utils import run, filter_list
-from middlewared.utils.crypto import generate_nt_hash, sha512_crypt
+from middlewared.utils.crypto import generate_nt_hash, sha512_crypt, generate_string
 from middlewared.utils.directoryservices.constants import DSType, DSStatus
 from middlewared.utils.filesystem.copy import copytree, CopyTreeConfig
 from middlewared.utils.nss import pwd, grp
@@ -305,6 +305,7 @@ def user_compress(self, user):
             'immutable',
             'home_create',
             'roles',
+            'random_password',
             'twofactor_auth_configured',
         ]
 
@@ -610,6 +611,7 @@ def do_create(self, data):
                 raise
 
         pk = None  # Make sure pk exists to rollback in case of an error
+        password = data['password']
         data = self.user_compress(data)
         try:
             self.__set_password(data)
@@ -660,7 +662,7 @@ def do_create(self, data):
                 self.logger.warning('Failed to update authorized keys', exc_info=True)
                 raise CallError(f'Failed to update authorized keys: {e}')
 
-        return pk
+        return self.middleware.call_sync('user.query', [['id', '=', pk]], {'get': True}) | {'password': password}
 
     @api_method(UserUpdateArgs, UserUpdateResult, audit='Update user', audit_callback=True)
     @pass_app()
@@ -851,6 +853,8 @@ def do_update(self, app, audit_callback, pk, data):
             perm_job.wait_sync()
 
         user.pop('sshpubkey', None)
+        password = user.get('password')
+
         self.__set_password(user)
 
         user = self.user_compress(user)
@@ -861,7 +865,7 @@ def do_update(self, app, audit_callback, pk, data):
         if user['smb'] and must_change_pdb_entry:
             self.middleware.call_sync('smb.update_passdb_user', user)
 
-        return pk
+        return self.middleware.call_sync('user.query', [['id', '=', pk]], {'get': True}) | {'password': password}
 
     @private
     def recreate_homedir_if_not_exists(self, user, group, mode):
@@ -1269,6 +1273,15 @@ async def common_validation(self, verrors, data, schema, group_ids, old=None):
             {'prefix': 'bsdusr_'}
         )
 
+        if data.get('random_password') and data.get('password'):
+            verrors.add(
+                f'{schema}.random_password',
+                'Requesting a randomized password while simultaneously supplying '
+                'an explicit password is not permitted.'
+            )
+        elif data.get('random_password'):
+            data['password'] = generate_string(string_size=20)
+
         if data.get('uid') is not None:
             try:
                 existing_user = await self.middleware.call(

src/middlewared/middlewared/test/integration/assets/account.py

@@ -14,17 +14,18 @@ def user(data, *, get_instance=True):
     data.setdefault('home_create', True)  # create user homedir by default
 
     user = call("user.create", data)
+    pk = user['id']
 
     try:
         value = None
 
         if get_instance:
-            value = call("user.get_instance", user)
+            value = user
 
         yield value
     finally:
         try:
-            call("user.delete", user)
+            call("user.delete", pk)
         except InstanceNotFound:
             pass
 

tests/api2/test_011_user.py

@@ -132,8 +132,9 @@ def create_user_with_dataset(ds_info, user_info):
 
         user_id = None
         try:
-            user_id = call('user.create', user_info['payload'])
-            yield call('user.query', [['id', '=', user_id]], {'get': True})
+            user = call('user.create', user_info['payload'])
+            user_id = user['id']
+            yield user
         finally:
             if user_id is not None:
                 call('user.delete', user_id, {"delete_group": True})

tests/api2/test_426_smb_vss.py

@@ -121,7 +121,7 @@ def test_002_creating_shareuser_to_test_acls(request):
         "group_create": True,
         "password": SMB_PWD,
         "uid": next_uid,
-    })
+    })['id']
 
 
 def test_003_changing_dataset_owner(request):

tests/api2/test_account.py

@@ -33,7 +33,7 @@ def test_create_account_audit():
                 "home": "/nonexistent",
                 "password": "password",
             }
-            user_id = call("user.create", payload)
+            user_id = call("user.create", payload)['id']
     finally:
         if user_id is not None:
             call("user.delete", user_id)
@@ -180,3 +180,41 @@ def test_create_account_invalid_gid():
             pass
 
     assert "This group does not exist." in str(ve)
+
+
+def test_create_user_random_password_with_specified_password_fail():
+    with pytest.raises(ValidationErrors, match='Requesting a randomized password while') as ve:
+        with user({
+            "username": "bobshouldnotexist",
+            "full_name": "bob",
+            "group_create": True,
+            "password": "test1234",
+            "random_password": True
+        }):
+            pass
+
+
+def test_create_user_with_random_password():
+    with user({
+        "username": "bobrandom",
+        "full_name": "bob",
+        "group_create": True,
+        "random_password": True
+    }, get_instance=True) as u:
+        assert u['password']
+
+
+def test_update_user_with_random_password():
+    with user({
+        "username": "bobrandom",
+        "full_name": "bob",
+        "group_create": True,
+        "password": "canary"
+    }, get_instance=True) as u:
+        assert u['password'] == 'canary'
+
+        new = call('user.update', u['id'], {'random_password': True})
+        assert new['password'] != 'canary'
+
+        new = call('user.update', u['id'], {'full_name': 'bob2'})
+        assert not new['password']

tests/api2/test_audit_rest.py

@@ -126,7 +126,7 @@ def test_authenticated_call():
                 "password": "password",
             })
             assert r.status_code == 200
-            user_id = r.json()
+            user_id = r.json()['id']
     finally:
         if user_id is not None:
             call("user.delete", user_id)