diff --git a/src/middlewared/middlewared/main.py b/src/middlewared/middlewared/main.py
index 1d1bf2abe2ebb..f6de5297212fd 100644
--- a/src/middlewared/middlewared/main.py
+++ b/src/middlewared/middlewared/main.py
@@ -796,7 +796,7 @@ async def run(self, ws, request, conndata):
                 try:
                     user = await self.middleware.call(
                         'user.query',
-                        [['username', '=', token['username']]],
+                        [['username', '=', token['username']], ['local', '=', True]],
                         {'get': True},
                     )
                 except MatchNotFound:
@@ -806,7 +806,9 @@ async def run(self, ws, request, conndata):
                     if 'ALL' in user['sudo_commands'] or 'ALL' in user['sudo_commands_nopasswd']:
                         as_root = True
                     else:
-                        for group in await self.middleware.call('group.query', [['id', 'in', user['groups']]]):
+                        for group in await self.middleware.call('group.query', [
+                            ['id', 'in', user['groups']], ['local', '=', True]
+                        ]):
                             if 'ALL' in group['sudo_commands'] or 'ALL' in group['sudo_commands_nopasswd']:
                                 as_root = True
                                 break
diff --git a/src/middlewared/middlewared/plugins/account.py b/src/middlewared/middlewared/plugins/account.py
index c24eb19490f58..4f1af4090d86c 100644
--- a/src/middlewared/middlewared/plugins/account.py
+++ b/src/middlewared/middlewared/plugins/account.py
@@ -24,6 +24,7 @@
 import middlewared.sqlalchemy as sa
 from middlewared.utils import run, filter_list
 from middlewared.utils.crypto import sha512_crypt
+from middlewared.utils.directoryservices.constants import DSType, DSStatus
 from middlewared.utils.nss import pwd, grp
 from middlewared.utils.nss.nss_common import NssModule
 from middlewared.utils.privilege import credential_has_full_admin, privileges_group_mapping
@@ -123,6 +124,30 @@ def validate_sudo_commands(commands):
     return verrors
 
 
+def filters_include_ds_accounts(filters):
+    """ Check for filters limiting to local accounts """
+    for f in filters:
+        if len(f) < 3:
+            # OR -- assume evaluation for this will result in including DS
+            continue
+
+        # Directory services do not provide builtin accounts
+        # local explicitly denotes not directory service
+        if f[0] in ('local', 'builtin'):
+            match f[1]:
+                case '=':
+                    if f[2] is True:
+                        return False
+                case '!=':
+                    if f[2] is False:
+                        return False
+
+                case _:
+                    pass
+
+    return True
+
+
 class UserModel(sa.Model):
     __tablename__ = 'account_bsdusers'
 
@@ -167,7 +192,7 @@ async def user_extend_context(self, rows, extra):
             [], {'prefix': 'bsdgrpmember_'}
         )
 
-        group_roles = await self.middleware.call('group.query', [], {'select': ['id', 'roles']})
+        group_roles = await self.middleware.call('group.query', [['local', '=', True]], {'select': ['id', 'roles']})
 
         for i in res:
             uid = i['user']['id']
@@ -251,9 +276,6 @@ async def query(self, filters, options):
         The following `additional_information` options are supported:
         `SMB` - include Windows SID and NT Name for user. If this option is not specified, then these
             keys will have `null` value.
-        `DS` - include users from Directory Service (LDAP or Active Directory) in results
-
-        `"extra": {"search_dscache": true}` is a legacy method of querying for directory services users.
         """
         ds_users = []
         options = options or {}
@@ -269,13 +291,8 @@ async def query(self, filters, options):
         datastore_options.pop('select', None)
 
         extra = options.get('extra', {})
-        dssearch = extra.pop('search_dscache', False)
         additional_information = extra.get('additional_information', [])
 
-        if 'DS' in additional_information:
-            dssearch = True
-            additional_information.remove('DS')
-
         username_sid = {}
         if 'SMB' in additional_information:
             try:
@@ -289,16 +306,23 @@ async def query(self, filters, options):
                 # broken
                 self.logger.error('Failed to retrieve passdb information', exc_info=True)
 
-        if dssearch:
-            ds_state = await self.middleware.call('directoryservices.get_state')
-            if ds_state['activedirectory'] == 'HEALTHY' or ds_state['ldap'] == 'HEALTHY':
-                ds_users = await self.middleware.call('directoryservices.cache.query', 'USER', filters, options.copy())
-                # For AD users, we will not have 2FA attribute normalized so let's do that
-                ad_users_2fa_mapping = await self.middleware.call('auth.twofactor.get_ad_users')
-                for index, user in enumerate(filter(
-                    lambda u: not u['local'] and 'twofactor_auth_configured' not in u, ds_users)
-                ):
-                    ds_users[index]['twofactor_auth_configured'] = bool(ad_users_2fa_mapping.get(user['sid']))
+        if filters_include_ds_accounts(filters):
+            ds = await self.middleware.call('directoryservices.status')
+            if ds['type'] is not None and ds['status'] == DSStatus.HEALTHY.name:
+                ds_users = await self.middleware.call(
+                   'directoryservices.cache.query', 'USER', filters, options.copy()
+                )
+
+                match DSType(ds['type']):
+                    case DSType.AD:
+                        ad_users_2fa_mapping = await self.middleware.call('auth.twofactor.get_ad_users')
+                        for index, user in enumerate(filter(
+                            lambda u: not u['local'] and 'twofactor_auth_configured' not in u, ds_users)
+                        ):
+                            ds_users[index]['twofactor_auth_configured'] = bool(ad_users_2fa_mapping.get(user['sid']))
+                    case _:
+                        # FIXME - map twofactor_auth_configured hint for LDAP users
+                        pass
 
         result = await self.middleware.call(
             'datastore.query', self._config.datastore, [], datastore_options
@@ -521,7 +545,10 @@ def do_create(self, data):
         group_created = False
 
         if create:
-            group = self.middleware.call_sync('group.query', [('group', '=', data['username'])])
+            group = self.middleware.call_sync('group.query', [
+                ('group', '=', data['username']),
+                ('local', '=', True)
+            ])
             if group:
                 group = group[0]
             else:
@@ -532,7 +559,9 @@ def do_create(self, data):
                     'sudo_commands_nopasswd': [],
                     'allow_duplicate_gid': False
                 }, False)
-                group = self.middleware.call_sync('group.query', [('id', '=', group)])[0]
+                group = self.middleware.call_sync('group.query', [
+                    ('id', '=', group), ('local', '=', True)
+                ])[0]
                 group_created = True
 
             data['group'] = group['id']
@@ -544,7 +573,7 @@ def do_create(self, data):
 
         if data['smb']:
             groups.append((self.middleware.call_sync(
-                'group.query', [('group', '=', 'builtin_users')], {'get': True},
+                'group.query', [('group', '=', 'builtin_users'), ('local', '=', True)], {'get': True},
             ))['id'])
 
         if data.get('uid') is None:
@@ -1196,21 +1225,64 @@ async def setup_local_administrator(self, app, username, password, options):
             raise CallError('Local administrator is already set up', errno.EEXIST)
 
         if username == 'admin':
-            if await self.middleware.call('user.query', [['uid', '=', ADMIN_UID]]):
+            # first check based on NSS to catch collisions with AD / LDAP users
+            try:
+                pwd_obj = await self.middleware.call('user.get_user_obj', {'uid': ADMIN_UID})
+                raise CallError(
+                    f'A {pwd_obj["source"].lower()} user with uid={ADMIN_UID} already exists, '
+                    'setting up local administrator is not possible',
+                    errno.EEXIST,
+                )
+            except KeyError:
+                pass
+
+            try:
+                pwd_obj = await self.middleware.call('user.get_user_obj', {'username': 'admin'})
+                raise CallError(f'"admin" {pwd_obj["source"].lower()} user already exists, '
+                                'setting up local administrator is not possible',
+                                errno.EEXIST)
+            except KeyError:
+                pass
+
+            try:
+                grp_obj = await self.middleware.call('group.get_group_obj', {'gid': ADMIN_GID})
+                raise CallError(
+                    f'A {grp_obj["source"].lower()} group with gid={ADMIN_GID} already exists, '
+                    'setting up local administrator is not possible',
+                    errno.EEXIST,
+                )
+            except KeyError:
+                pass
+
+            try:
+                grp_obj = await self.middleware.call('get.get_group_obj', {'groupname': 'admin'})
+                raise CallError(f'"admin" {grp_obj["source"].lower()} group already exists, '
+                                'setting up local administrator is not possible',
+                                errno.EEXIST)
+            except KeyError:
+                pass
+
+            # double-check our database in case we have for some reason failed to write to passwd
+            local_users = await self.middleware.call('user.query', [['local', '=', True]])
+            local_groups = await self.middleware.call('group.query', [['local', '=', True]])
+
+            if filter_list(local_users, [['uid', '=', ADMIN_UID]]):
                 raise CallError(
                     f'A user with uid={ADMIN_UID} already exists, setting up local administrator is not possible',
                     errno.EEXIST,
                 )
-            if await self.middleware.call('user.query', [['username', '=', 'admin']]):
+
+            if filter_list(local_users, [['username', '=', 'admin']]):
                 raise CallError('"admin" user already exists, setting up local administrator is not possible',
                                 errno.EEXIST)
 
-            if await self.middleware.call('group.query', [['gid', '=', ADMIN_GID]]):
+            if filter_list(local_groups, [['gid', '=', ADMIN_GID]]):
                 raise CallError(
                     f'A group with gid={ADMIN_GID} already exists, setting up local administrator is not possible',
                     errno.EEXIST,
                 )
-            if await self.middleware.call('group.query', [['group', '=', 'admin']]):
+
+            if filter_list(local_groups, [['group', '=', 'admin']]):
                 raise CallError('"admin" group already exists, setting up local administrator is not possible',
                                 errno.EEXIST)
 
@@ -1713,9 +1785,6 @@ async def query(self, filters, options):
         The following `additional_information` options are supported:
         `SMB` - include Windows SID and NT Name for group. If this option is not specified, then these
             keys will have `null` value.
-        `DS` - include groups from Directory Service (LDAP or Active Directory) in results
-
-        `"extra": {"search_dscache": true}` is a legacy method of querying for directory services groups.
         """
         ds_groups = []
         options = options or {}
@@ -1731,16 +1800,11 @@ async def query(self, filters, options):
         datastore_options.pop('select', None)
 
         extra = options.get('extra', {})
-        dssearch = extra.pop('search_dscache', False)
         additional_information = extra.get('additional_information', [])
 
-        if 'DS' in additional_information:
-            dssearch = True
-            additional_information.remove('DS')
-
-        if dssearch:
-            ds_state = await self.middleware.call('directoryservices.get_state')
-            if ds_state['activedirectory'] == 'HEALTHY' or ds_state['ldap'] == 'HEALTHY':
+        if filters_include_ds_accounts(filters):
+            ds = await self.middleware.call('directoryservices.status')
+            if ds['type'] is not None and ds['status'] == DSStatus.HEALTHY.name:
                 ds_groups = await self.middleware.call('directoryservices.cache.query', 'GROUP', filters, options)
 
         if 'SMB' in additional_information:
diff --git a/src/middlewared/middlewared/plugins/account_/2fa.py b/src/middlewared/middlewared/plugins/account_/2fa.py
index b61f6e7b8e596..2d1f0c2f3bf36 100644
--- a/src/middlewared/middlewared/plugins/account_/2fa.py
+++ b/src/middlewared/middlewared/plugins/account_/2fa.py
@@ -103,7 +103,7 @@ async def translate_username(self, username):
         return await self.middleware.call(
             'user.query', [['username', '=', user['pw_name']]], {
                 'get': True,
-                'extra': {'additional_information': ['DS', 'SMB']},
+                'extra': {'additional_information': ['SMB']},
             }
         )
 
diff --git a/src/middlewared/middlewared/plugins/account_/privilege.py b/src/middlewared/middlewared/plugins/account_/privilege.py
index 4efc7c077b4da..cd14ea606ca8e 100644
--- a/src/middlewared/middlewared/plugins/account_/privilege.py
+++ b/src/middlewared/middlewared/plugins/account_/privilege.py
@@ -430,9 +430,9 @@ async def full_privilege(self):
     @private
     async def always_has_root_password_enabled(self, users=None, groups=None):
         if users is None:
-            users = await self.middleware.call('user.query')
+            users = await self.middleware.call('user.query', [['local', '=', True]])
         if groups is None:
-            groups = await self.middleware.call('group.query')
+            groups = await self.middleware.call('group.query', [['local', '=', True]])
 
         root_user = filter_list(
             users,
@@ -458,9 +458,9 @@ async def always_has_root_password_enabled(self, users=None, groups=None):
     async def local_administrators(self, exclude_user_ids=None, users=None, groups=None):
         exclude_user_ids = exclude_user_ids or []
         if users is None:
-            users = await self.middleware.call('user.query')
+            users = await self.middleware.call('user.query', [['local', '=', True]])
         if groups is None:
-            groups = await self.middleware.call('group.query')
+            groups = await self.middleware.call('group.query', [['local', '=', True]])
 
         local_administrator_privilege = await self.middleware.call(
             'datastore.query',
diff --git a/src/middlewared/middlewared/plugins/etc.py b/src/middlewared/middlewared/plugins/etc.py
index 1ecb27b258206..fc6ecc99d430e 100644
--- a/src/middlewared/middlewared/plugins/etc.py
+++ b/src/middlewared/middlewared/plugins/etc.py
@@ -76,7 +76,7 @@ class EtcService(Service):
         ],
         'shadow': {
             'ctx': [
-                {'method': 'user.query'},
+                {'method': 'user.query', 'args': [[['local', '=', True]]]},
             ],
             'entries': [
                 {'type': 'mako', 'path': 'shadow', 'group': 'shadow', 'mode': 0o0640},
@@ -84,8 +84,8 @@ class EtcService(Service):
         },
         'user': {
             'ctx': [
-                {'method': 'user.query'},
-                {'method': 'group.query'},
+                {'method': 'user.query', 'args': [[['local', '=', True]]]},
+                {'method': 'group.query', 'args': [[['local', '=', True]]]},
             ],
             'entries': [
                 {'type': 'mako', 'path': 'group'},
diff --git a/src/middlewared/middlewared/plugins/idmap.py b/src/middlewared/middlewared/plugins/idmap.py
index 88de2c04b4c63..fa63b9b5b49e4 100644
--- a/src/middlewared/middlewared/plugins/idmap.py
+++ b/src/middlewared/middlewared/plugins/idmap.py
@@ -1102,7 +1102,6 @@ async def id_to_name(self, xid, id_type):
         """
         idtype = IDType[id_type]
         idmap_timeout = 5.0
-        options = {'extra': {'additional_information': ['DS']}, 'get': True}
 
         match idtype:
             # IDType.BOTH is possible return by nss_winbind / nss_sss
@@ -1122,7 +1121,7 @@ async def id_to_name(self, xid, id_type):
 
         try:
             ret = await asyncio.wait_for(
-                self.middleware.create_task(self.middleware.call(method, filters, options)),
+                self.middleware.create_task(self.middleware.call(method, filters, {'get': True})),
                 timeout=idmap_timeout
             )
             name = ret[key]
diff --git a/src/middlewared/middlewared/plugins/keychain.py b/src/middlewared/middlewared/plugins/keychain.py
index 711164024931b..2d62d0a547bec 100644
--- a/src/middlewared/middlewared/plugins/keychain.py
+++ b/src/middlewared/middlewared/plugins/keychain.py
@@ -637,7 +637,7 @@ def remote_ssh_semiautomatic_setup(self, data):
             except Exception as e:
                 raise CallError(f"Semi-automatic SSH connection setup failed: {e!r}")
 
-            user = c.call("user.query", [["username", "=", data["username"]]], {"get": True})
+            user = c.call("user.query", [["username", "=", data["username"]], ['local', '=', True]], {"get": True})
             user_update = {}
             if user["shell"] == "/usr/sbin/nologin":
                 user_update["shell"] = "/usr/bin/bash"
@@ -679,7 +679,7 @@ def ssh_pair(self, data):
         service = self.middleware.call_sync("service.query", [("service", "=", "ssh")], {"get": True})
         ssh = self.middleware.call_sync("ssh.config")
         try:
-            user = self.middleware.call_sync("user.query", [("username", "=", data["username"])], {"get": True})
+            user = self.middleware.call_sync("user.query", [("username", "=", data["username"]), ("local", "=", True)], {"get": True})
         except MatchNotFound:
             raise CallError(f"User {data['username']} does not exist")
 
diff --git a/src/middlewared/middlewared/plugins/smb.py b/src/middlewared/middlewared/plugins/smb.py
index c94b67307f132..e8afea2a71350 100644
--- a/src/middlewared/middlewared/plugins/smb.py
+++ b/src/middlewared/middlewared/plugins/smb.py
@@ -1437,7 +1437,7 @@ async def share_precheck(self, data):
         if not ad_enabled:
             local_smb_user_cnt = await self.middleware.call(
                 'user.query',
-                [['smb', '=', True]],
+                [['smb', '=', True], ['local', '=', True]],
                 {'count': True}
             )
             if local_smb_user_cnt == 0:
diff --git a/src/middlewared/middlewared/plugins/smb_/groupmap.py b/src/middlewared/middlewared/plugins/smb_/groupmap.py
index 1ce8937d907a9..09d6fe806d16b 100644
--- a/src/middlewared/middlewared/plugins/smb_/groupmap.py
+++ b/src/middlewared/middlewared/plugins/smb_/groupmap.py
@@ -150,7 +150,7 @@ async def sync_foreign_groups(self):
             admin_sid = None
             grp_obj = await self.middleware.call(
                 'group.query',
-                [('group', '=', admin_group)],
+                [('group', '=', admin_group), ('local', '=', True)],
                 {'extra': {'additional_information': ['SMB', 'DS']}}
             )
             if grp_obj:
@@ -431,9 +431,9 @@ async def synchronize_group_mappings(self, job, bypass_sentinel_check=False):
         groupmap = await self.groupmap_list()
         must_remove_cache = False
 
-        groups = await self.middleware.call('group.query', [('builtin', '=', False), ('smb', '=', True)])
+        groups = await self.middleware.call('group.query', [('builtin', '=', False), ('local', '=', True), ('smb', '=', True)])
         g_dict = {x["gid"]: x for x in groups}
-        g_dict[545] = await self.middleware.call('group.query', [('gid', '=', 545)], {'get': True})
+        g_dict[545] = await self.middleware.call('group.query', [('gid', '=', 545), ('local', '=', True)], {'get': True})
 
         intersect = set(g_dict.keys()).intersection(set(groupmap["local"].keys()))
 
diff --git a/src/middlewared/middlewared/plugins/smb_/passdb.py b/src/middlewared/middlewared/plugins/smb_/passdb.py
index 74818c68f1bed..68c55111bc6b0 100644
--- a/src/middlewared/middlewared/plugins/smb_/passdb.py
+++ b/src/middlewared/middlewared/plugins/smb_/passdb.py
@@ -238,5 +238,5 @@ async def synchronize_passdb(self, job, bypass_sentinel_check=False):
                 "This may indicate system dataset setup failure."
             )
 
-        conf_users = await self.middleware.call('user.query', [("smb", "=", True)])
+        conf_users = await self.middleware.call('user.query', [("smb", "=", True), ('local', '=', True)])
         await self.passdb_sync_impl(conf_users)
diff --git a/src/middlewared/middlewared/plugins/usage.py b/src/middlewared/middlewared/plugins/usage.py
index f0282ca49303f..07800728fe420 100644
--- a/src/middlewared/middlewared/plugins/usage.py
+++ b/src/middlewared/middlewared/plugins/usage.py
@@ -266,7 +266,7 @@ async def gather_system(self, context):
             'system_hash': await self.middleware.call('system.host_id'),
             'usage_version': 1,
             'system': [{
-                'users': await self.middleware.call('user.query', [], {'count': True}),
+                'users': await self.middleware.call('user.query', [['local', '=', True]], {'count': True}),
                 'snapshots': context['total_snapshots'],
                 'zvols': context['total_zvols'],
                 'datasets': context['total_datasets'],
diff --git a/src/middlewared/middlewared/test/integration/assets/ftp.py b/src/middlewared/middlewared/test/integration/assets/ftp.py
index 4633525a87b67..ceb5a9f085dab 100644
--- a/src/middlewared/middlewared/test/integration/assets/ftp.py
+++ b/src/middlewared/middlewared/test/integration/assets/ftp.py
@@ -36,7 +36,7 @@ def anonymous_ftp_server(config=None, dataset_name="anonftp"):
 @contextlib.contextmanager
 def ftp_server_with_user_account(config=None):
     config = config or {}
-    ftp_id = call("group.query", [["name", "=", "ftp"]], {"get": True})["id"]
+    ftp_id = call("group.query", [["name", "=", "ftp"], ['local', '=', True]], {"get": True})["id"]
 
     with dataset("ftptest") as ds:
         with user({