diff --git a/src/middlewared/middlewared/plugins/account.py b/src/middlewared/middlewared/plugins/account.py
index fd118ea149ad1..1a3fa313d4f3c 100644
--- a/src/middlewared/middlewared/plugins/account.py
+++ b/src/middlewared/middlewared/plugins/account.py
@@ -10,10 +10,11 @@
 import subprocess
 import time
 import warnings
+import wbclient
 from pathlib import Path
 from contextlib import suppress
 
-from middlewared.schema import accepts, Bool, Dict, Int, List, Password, Patch, returns, Str, LocalUsername
+from middlewared.schema import accepts, Bool, Dict, Int, List, Password, Patch, returns, SID, Str, LocalUsername
 from middlewared.service import (
     CallError, CRUDService, ValidationErrors, no_auth_required, no_authz_required, pass_app, private, filterable, job
 )
@@ -22,15 +23,18 @@
 from middlewared.utils import run, filter_list
 from middlewared.utils.crypto import sha512_crypt
 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
 from middlewared.validators import Email, Range
 from middlewared.async_validators import check_path_resides_within_volume
 from middlewared.plugins.smb_.constants import SMBBuiltin
-from middlewared.plugins.idmap_.utils import (
+from middlewared.plugins.idmap_.idmap_constants import (
     TRUENAS_IDMAP_DEFAULT_LOW,
     SID_LOCAL_USER_PREFIX,
     SID_LOCAL_GROUP_PREFIX
 )
+from middlewared.plugins.idmap_ import idmap_winbind
+from middlewared.plugins.idmap_ import idmap_sss
 
 ADMIN_UID = 950  # When googled, does not conflict with anything
 ADMIN_GID = 950
@@ -1053,7 +1057,8 @@ def shell_choices(self, group_ids):
         Int('pw_uid'),
         Int('pw_gid'),
         List('grouplist'),
-        Dict('sid_info'),
+        SID('sid', null=True),
+        Str('source', enum=['LOCAL', 'ACTIVEDIRECTORY', 'LDAP']),
         Bool('local'),
         register=True,
     ))
@@ -1089,14 +1094,9 @@ def get_user_obj(self, data):
         `grouplist` - optional list of group ids for groups of which this account is a member. If `get_groups`
         is not specified, this value will be null.
 
-        `sid_info - optional dictionary object containing details of SID and domain information. If `sid_info`
-        is not specified, this value will be null.
+        `sid` - optional SID value for the accoun that is present if `sid_info` is specified in payload.
 
-        NOTE: in some pathological scenarios this may make the operation hang until
-        the winbindd request timeout has been reached if the winbindd connection manager
-        has not yet marked the domain as offline. The TrueNAS middleware is more aggressive
-        about marking AD domains as FAULTED and so it may be advisable to first check the
-        Active Directory service state prior to batch operations using this option.
+        `source` - the source for the user account.
         """
         verrors = ValidationErrors()
         if not data['username'] and data['uid'] is None:
@@ -1123,40 +1123,70 @@ def get_user_obj(self, data):
             except KeyError:
                 raise KeyError(f'{data["uid"]}: user with this id does not exist') from None
 
-        source = user_obj.pop('source')
-        user_obj['local'] = source == 'FILES'
+        match user_obj['source']:
+            case NssModule.FILES.name:
+                user_obj['source'] = 'LOCAL'
+            case NssModule.WINBIND.name:
+                user_obj['source'] = 'ACTIVEDIRECTORY'
+            case NssModule.SSS.name:
+                user_obj['source'] = 'LDAP'
+            case _:
+                raise ValueError(f'{user_obj["source"]}: unknown ID source. Please file a bug report.')
+
+        user_obj['local'] = user_obj['source'] == 'LOCAL'
 
         if data['get_groups']:
             user_obj['grouplist'] = os.getgrouplist(user_obj['pw_name'], user_obj['pw_gid'])
 
         if data['sid_info']:
-            try:
-                if (idmap := self.middleware.call_sync('idmap.convert_unixids', [{
-                    'id_type': 'USER',
-                    'id': user_obj['pw_uid'],
-                }])['mapped']):
-                    sid = idmap[f'UID:{user_obj["pw_uid"]}']['sid']
-                else:
-                    sid = SID_LOCAL_USER_PREFIX + str(user_obj['pw_uid'])
-            except CallError as e:
-                # ENOENT means no winbindd entry for user
-                # ENOTCONN means winbindd is stopped / can't be started
-                # EAGAIN means the system dataset is hosed and needs to be fixed,
-                # but we need to let it through so that it's very clear in logs
-                if e.errno not in (errno.ENOENT, errno.ENOTCONN):
-                    self.logger.error('Failed to retrieve SID for uid: %d', user_obj['pw_uid'], exc_info=True)
-                sid = None
-            except Exception:
-                self.logger.error('Failed to retrieve SID for uid: %d', user_obj['pw_uid'], exc_info=True)
-                sid = None
+            match user_obj['source']:
+                case 'LOCAL' | 'ACTIVEDIRECTORY':
+                    # winbind provides idmapping for local and AD users
+                    try:
+                        idmap_ctx = idmap_winbind.WBClient()
+                    except wbclient.WBCError as e:
+                        if e.error_code != wbclient.WBC_ERR_WINBIND_NOT_AVAILABLE:
+                            self.logger.error('Failed to retrieve SID for uid: %d',
+                                              user_obj['pw_uid'], exc_info=True)
+
+                        idmap_ctx = None
+                case 'LDAP':
+                    # SSSD provides ID mapping for IPA domains
+                    idmap_ctx = idmap_sss.SSSClient()
+                case _:
+                    # We're not raising an exception here since it
+                    # can be a critical areai
+                    self.logger.error(
+                        '%s: unknown ID source. Please file a bug report.',
+                        user_obj['source']
+                    )
+                    idmap_ctx = None
 
-            if sid:
-                user_obj['sid_info'] = {
-                    'sid': sid,
-                    'domain_information': self.middleware.call_sync('idmap.parse_domain_info', sid)
-                }
+            if idmap_ctx is not None:
+                try:
+                    sid = idmap_ctx.uidgid_to_idmap_entry({
+                        'id_type': 'USER',
+                        'id': user_obj['pw_uid']
+                    })['sid']
+                except MatchNotFound:
+                    if user_obj['source'] == 'LOCAL':
+                        # Local user that doesn't have passdb entry
+                        # we can simply apply default prefix
+                        sid = SID_LOCAL_USER_PREFIX + str(user_obj['pw_uid'])
+                    else:
+                        # This is a more odd situation. The user accout exists
+                        # in IPA but doesn't have a SID assigned to it.
+                        sid = None
             else:
-                user_obj['sid_info'] = None
+                # We were unable to establish an idmap client context even
+                # though we were able to retrieve the user account info. This
+                # most likely means that we're dealing with a local account and
+                # winbindd is not running.
+                sid = None
+
+            user_obj['sid'] = sid
+        else:
+            user_obj['sid'] = None
 
         return user_obj
 
@@ -2052,7 +2082,8 @@ async def get_next_gid(self):
         Str('gr_name'),
         Int('gr_gid'),
         List('gr_mem'),
-        Dict('sid_info'),
+        SID('sid', null=True),
+        Str('source', enum=['LOCAL', 'ACTIVEDIRECTORY', 'LDAP']),
         Bool('local'),
     ))
     def get_group_obj(self, data):
@@ -2071,11 +2102,14 @@ def get_group_obj(self, data):
 
         `gr_mem` - list of gids that are members of the group
 
-        `sid_info` - optional SMB information if `sid_info` is specified specified, otherwise
-        this field will be null.
-
         `local` - boolean indicating whether this group is local to the NAS or provided by a
         directory service.
+
+        `sid` - optional SID value for the accoun that is present if `sid_info` is specified in payload.
+
+        `source` - the name server switch module that provided the user. Options are:
+        FILES - local user in passwd file of server, WINBIND - user provided by winbindd, SSS - user
+        provided by SSSD.
         """
         verrors = ValidationErrors()
         if not data['groupname'] and data['gid'] is None:
@@ -2101,33 +2135,66 @@ def get_group_obj(self, data):
             except KeyError:
                 raise KeyError(f'{data["gid"]}: group with this id does not exist') from None
 
-        source = grp_obj.pop('source')
-        grp_obj['local'] = source == 'FILES'
+        grp_obj['local'] = grp_obj['source'] == NssModule.FILES.name
+        match grp_obj['source']:
+            case NssModule.FILES.name:
+                grp_obj['source'] = 'LOCAL'
+            case NssModule.WINBIND.name:
+                grp_obj['source'] = 'ACTIVEDIRECTORY'
+            case NssModule.SSS.name:
+                grp_obj['source'] = 'LDAP'
+            case _:
+                raise ValueError(f'{grp_obj["source"]}: unknown ID source. Please file a bug report.')
+
+        grp_obj['local'] = grp_obj['source'] == 'LOCAL'
 
         if data['sid_info']:
-            try:
-                if (idmap := self.middleware.call_sync('idmap.convert_unixids', [{
-                    'id_type': 'GROUP',
-                    'id': grp_obj['gr_gid'],
-                }])['mapped']):
-                    sid = idmap[f'GID:{grp_obj["gr_gid"]}']['sid']
-                else:
-                    sid = SID_LOCAL_GROUP_PREFIX + str(grp_obj['gr_gid'])
-            except CallError as e:
-                if e.errno not in (errno.ENOENT, errno.ENOTCONN):
-                    self.logger.error('Failed to retrieve SID for gid: %d', grp_obj['gr_gid'], exc_info=True)
-                sid = None
-            except Exception:
-                self.logger.error('Failed to retrieve SID for gid: %d', grp['gr_gid'], exc_info=True)
-                sid = None
+            match grp_obj['source']:
+                case 'LOCAL' | 'ACTIVEDIRECTORY':
+                    # winbind provides idmapping for local and AD users
+                    try:
+                        idmap_ctx = idmap_winbind.WBClient()
+                    except wbclient.WBCError as e:
+                        # Library error from libwbclient.
+                        # Don't bother logging if winbind isn't running since
+                        # we have plenty of other places that are logging that
+                        # error condition
+                        if e.error_code != wbclient.WBC_ERR_WINBIND_NOT_AVAILABLE:
+                            self.logger.error('Failed to retrieve SID for gid: %d',
+                                              grp_obj['gr_gid'], exc_info=True)
+
+                        idmap_ctx = None
+                case 'LDAP':
+                    # SSSD provides ID mapping for IPA domains
+                    idmap_ctx = idmap_sss.SSSClient()
+                case _:
+                    # We're not raising an exception here since it
+                    # can be a critical areai
+                    self.logger.error(
+                        '%s: unknown ID source. Please file a bug report.',
+                        grp_obj['source']
+                    )
+                    idmap_ctx = None
 
-            if sid:
-                grp_obj['sid_info'] = {
-                    'sid': sid,
-                    'domain_information': self.middleware.call_sync('idmap.parse_domain_info', sid)
-                }
+            if idmap_ctx is not None:
+                try:
+                    sid = idmap_ctx.uidgid_to_idmap_entry({
+                        'id_type': 'GROUP',
+                        'id': grp_obj['gr_gid']
+                    })['sid']
+                except MatchNotFound:
+                    if grp_obj['source'] == 'LOCAL':
+                        # Local user that doesn't have groupmap entry
+                        # we can simply apply default prefix
+                        sid = SID_LOCAL_GROUP_PREFIX + str(grp_obj['gr_gid'])
+                    else:
+                        sid = None
             else:
-                grp_obj['sid_info'] = None
+                sid = None
+
+            grp_obj['sid'] = sid
+        else:
+            grp_obj['sid'] = None
 
         return grp_obj
 
diff --git a/src/middlewared/middlewared/plugins/activedirectory_/cache.py b/src/middlewared/middlewared/plugins/activedirectory_/cache.py
index 4994a8ff061d7..067caaad2eb09 100644
--- a/src/middlewared/middlewared/plugins/activedirectory_/cache.py
+++ b/src/middlewared/middlewared/plugins/activedirectory_/cache.py
@@ -1,4 +1,4 @@
-from middlewared.plugins.idmap_.utils import (
+from middlewared.plugins.idmap_.idmap_constants import (
     IDType,
     SID_LOCAL_USER_PREFIX,
     SID_LOCAL_GROUP_PREFIX,
diff --git a/src/middlewared/middlewared/plugins/activedirectory_/health.py b/src/middlewared/middlewared/plugins/activedirectory_/health.py
index d9f566e7925f0..00f66cd898d56 100644
--- a/src/middlewared/middlewared/plugins/activedirectory_/health.py
+++ b/src/middlewared/middlewared/plugins/activedirectory_/health.py
@@ -9,7 +9,7 @@
 from middlewared.service import private, Service, ValidationErrors
 from middlewared.service_exception import CallError, MatchNotFound
 from middlewared.plugins.directoryservices import DSStatus
-from middlewared.plugins.idmap_.utils import WBClient, WBCErr
+from middlewared.plugins.idmap_.idmap_winbind import WBClient, WBCErr
 from middlewared.utils import filter_list
 
 
diff --git a/src/middlewared/middlewared/plugins/auth_/authenticate.py b/src/middlewared/middlewared/plugins/auth_/authenticate.py
index 1c8e091c0113a..560a064f99c98 100644
--- a/src/middlewared/middlewared/plugins/auth_/authenticate.py
+++ b/src/middlewared/middlewared/plugins/auth_/authenticate.py
@@ -86,18 +86,39 @@ async def authenticate_user(self, query, user_info):
             )
             return None
 
-        if user_info['local']:
-            twofactor_id = user_info['id']
-            groups_key = 'local_groups'
-            account_flags = ['LOCAL']
-        else:
-            twofactor_id = user['sid_info']['sid']
-            groups_key = 'ds_groups'
-            account_flags = ['DIRECTORY_SERVICE']
-            if user['sid_info']['domain_information']['activedirectory']:
-                account_flags.append('ACTIVE_DIRECTORY')
-            else:
-                account_flags.append('LDAP')
+        match user['source']:
+            case 'LOCAL':
+                # Local user
+                twofactor_id = user_info['id']
+                groups_key = 'local_groups'
+                account_flags = ['LOCAL']
+            case 'ACTIVEDIRECTORY':
+                # Active directory user
+                twofactor_id = user['sid']
+                groups_key = 'ds_groups'
+                account_flags = ['DIRECTORY_SERVICE', 'ACTIVE_DIRECTORY']
+            case 'LDAP':
+                # This includes both OpenLDAP and IPA domains
+                # Since IPA domains may have cross-realm trusts with separate
+                # idmap configuration we will preferentially use the SID if it is
+                # available (since it should be static and universally unique)
+                twofactor_id = user['sid'] or user_info['id']
+                groups_key = 'ds_groups'
+                account_flags = ['DIRECTORY_SERVICE', 'LDAP']
+            case _:
+                self.logger.error('[%s]: unknown user source. Rejecting access.', user['source'])
+                return None
+
+        if user['local'] != user_info['local']:
+            # There is a disagreement between our expectation of user account source
+            # based on our database and what NSS _actually_ returned.
+            self.logger.error(
+                'Rejecting access due to potenital collision between local and directory '
+                'service user account. TrueNAS configuration expected a %s user account '
+                'but received an account provided by %s.',
+                'local' if user_info['local'] else 'non-local', user['source']
+            )
+            return None
 
         # Two-factor authentication token is keyed by SID for activedirectory
         # users.
@@ -112,6 +133,16 @@ async def authenticate_user(self, query, user_info):
             account_flags.append('2FA')
 
         if user['pw_uid'] in (0, 950):
+            if not user['local']:
+                # Although this should be covered in above check for mismatch in
+                # value of `local`, perform an extra explicit check for the case
+                # of root / root-equivalent accounts.
+                self.logger.error(
+                    'Rejecting admin account access due to collision with acccount provided '
+                    'by a directory service.'
+                )
+                return None
+
             account_flags.append('SYS_ADMIN')
 
         privileges = await self.middleware.call('privilege.privileges_for_groups', groups_key, groups)
diff --git a/src/middlewared/middlewared/plugins/directoryservices.py b/src/middlewared/middlewared/plugins/directoryservices.py
index abcae58f49d48..eaf11fae00363 100644
--- a/src/middlewared/middlewared/plugins/directoryservices.py
+++ b/src/middlewared/middlewared/plugins/directoryservices.py
@@ -9,23 +9,13 @@
 from middlewared.service import no_authz_required, Service, private, job
 from middlewared.plugins.smb_.constants import SMBCmd, SMBPath
 from middlewared.service_exception import CallError, MatchNotFound
+from middlewared.utils.directoryservices.constants import (
+    DSStatus, DSType, NSS_Info
+)
 
 DEPENDENT_SERVICES = ['smb', 'nfs', 'ssh']
 
 
-class DSStatus(enum.Enum):
-    DISABLED = enum.auto()
-    FAULTED = 1028  # MSG_WINBIND_OFFLINE
-    LEAVING = enum.auto()
-    JOINING = enum.auto()
-    HEALTHY = 1027  # MSG_WINBIND_ONLINE
-
-
-class DSType(enum.Enum):
-    AD = 'activedirectory'
-    LDAP = 'ldap'
-
-
 class SSL(enum.Enum):
     NOSSL = 'OFF'
     USESSL = 'ON'
@@ -38,19 +28,39 @@ class SASL_Wrapping(enum.Enum):
     SEAL = 'SEAL'
 
 
-class NSS_Info(enum.Enum):
-    SFU = ('SFU', [DSType.AD])
-    SFU20 = ('SFU20', [DSType.AD])
-    RFC2307 = ('RFC2307', [DSType.AD, DSType.LDAP])
-    RFC2307BIS = ('RFC2307BIS', [DSType.LDAP])
-    TEMPLATE = ('TEMPLATE', [DSType.AD])
-
-
 class DirectoryServices(Service):
     class Config:
         service = "directoryservices"
         cli_namespace = "directory_service"
 
+    @no_authz_required
+    @accepts()
+    @returns(Dict(
+        'directoryservices_status',
+        Str('type', enum=[x.value.upper() for x in DSType], null=True),
+        Ref('directoryservice_state', 'status')
+    ))
+    async def status(self):
+        """
+        Provide the type and status of the currently-enabled directory service
+        """
+        # Currently wrap around `get_state`. In upcoming PR we will get
+        # status more directly and deprecate the `get_state` method.
+        state = await self.get_state()
+        for ds in DSType:
+            if (status := state.get(ds.value, 'DISABLED')) == DSStatus.DISABLED.name:
+                continue
+
+            return {
+                'type': ds.value.upper(),
+                'status': status
+            }
+
+        return {
+            'type': None,
+            'status': None
+        }
+
     @no_authz_required
     @accepts()
     @returns(Dict(
@@ -76,6 +86,9 @@ async def get_state(self):
         except KeyError:
             ds_state = {}
             for srv in DSType:
+                if srv is DSType.IPA:
+                    # TODO: IPA join not implemented yet
+                    continue
                 try:
                     res = await self.middleware.call(f'{srv.value}.started')
                     ds_state[srv.value] = DSStatus.HEALTHY.name if res else DSStatus.DISABLED.name
diff --git a/src/middlewared/middlewared/plugins/idmap.py b/src/middlewared/middlewared/plugins/idmap.py
index 67ecbc87bb5ae..32f0196252abf 100644
--- a/src/middlewared/middlewared/plugins/idmap.py
+++ b/src/middlewared/middlewared/plugins/idmap.py
@@ -6,13 +6,14 @@
 
 from middlewared.schema import accepts, Bool, Dict, Int, Password, Patch, Ref, Str, LDAP_DN, OROperator
 from middlewared.service import CallError, CRUDService, job, private, ValidationErrors, filterable
-from middlewared.service_exception import MatchNotFound
-from middlewared.plugins.directoryservices import SSL
-from middlewared.plugins.idmap_.utils import (
-    IDType, SID_LOCAL_USER_PREFIX, SID_LOCAL_GROUP_PREFIX, TRUENAS_IDMAP_MAX, WBClient, WBCErr
+from middlewared.utils.directoryservices.constants import SSL
+from middlewared.plugins.idmap_.idmap_constants import (
+    IDType, SID_LOCAL_USER_PREFIX, SID_LOCAL_GROUP_PREFIX, TRUENAS_IDMAP_MAX
 )
+from middlewared.plugins.idmap_.idmap_winbind import (WBClient, WBCErr)
+from middlewared.plugins.idmap_.idmap_sss import SSSClient
 import middlewared.sqlalchemy as sa
-from middlewared.utils import run, filter_list
+from middlewared.utils import filter_list
 from middlewared.validators import Range
 from middlewared.plugins.smb import SMBPath
 try:
@@ -235,7 +236,6 @@ class Config:
         cli_namespace = 'directory_service.idmap'
         role_prefix = 'DIRECTORY_SERVICE'
 
-
     def __wbclient_ctx(self, retry=True):
         """
         Wrapper around setting up a temporary winbindd client context
@@ -480,9 +480,19 @@ async def validate(self, schema_name, data, verrors):
             verrors.add(f'{schema_name}.certificate', 'Please specify a valid certificate.')
 
         configured_domains = await self.query()
-        ds_state = await self.middleware.call("directoryservices.get_state")
-        ldap_enabled = True if ds_state['ldap'] in ['HEALTHY', 'JOINING'] else False
-        ad_enabled = True if ds_state['activedirectory'] in ['HEALTHY', 'JOINING'] else False
+        ds = await self.middleware.call("directoryservices.status")
+        match ds['status']:
+            case 'HEALTHY' | 'JOINING':
+                if ds['type'] == 'ACTIVEDIRECTORY':
+                    ldap_enabled = False
+                    ad_enabled = True
+                else:
+                    ldap_enabled = True
+                    ad_enabled = False
+            case _:
+                ldap_enabled = False
+                ad_enabled = False
+
         new_range = range(data['range_low'], data['range_high'])
         idmap_backend = data.get('idmap_backend')
         for i in configured_domains:
@@ -765,7 +775,8 @@ async def do_create(self, data):
             verrors.add('idmap_domain_create.name', 'Domain names must be unique.')
 
         if data['options'].get('sssd_compat'):
-            if await self.middleware.call('activedirectory.get_state') != 'HEALTHY':
+            status = (await self.middleware.call('directoryservices.status'))['status']
+            if status != 'HEALTHY':
                 verrors.add('idmap_domain_create.options',
                             'AD service must be enabled and started to '
                             'generate an SSSD-compatible id range')
@@ -840,7 +851,8 @@ async def do_update(self, id_, data):
                         f'Changing name of default domain {old["name"]} is not permitted')
 
         if new['options'].get('sssd_compat') and not old['options'].get('sssd_compat'):
-            if await self.middleware.call('activedirectory.get_state') != 'HEALTHY':
+            ds_state = await self.middleware.call('directoryservices.get_state')
+            if ds_state['activedirectory'] != 'HEALTHY':
                 verrors.add('idmap_domain_update.options',
                             'AD service must be enabled and started to '
                             'generate an SSSD-compatible id range')
@@ -910,35 +922,6 @@ async def do_delete(self, id_):
         await self.middleware.call('etc.generate', 'smb')
         return ret
 
-    def _pyuidgid_to_dict(self, entry):
-        if entry.id_type == IDType.USER.wbc_const():
-            idtype = 'USER'
-        elif entry.id_type == IDType.GROUP.wbc_const():
-            idtype = 'GROUP'
-        else:
-            idtype = 'BOTH'
-        return {
-            'id_type': idtype,
-            'id': entry.id,
-            'sid': entry.sid
-        }
-
-    @private
-    async def name_to_sid(self, name):
-        try:
-            client = self.__wbclient_ctx()
-            entry = client.name_to_uidgid_entry(name)
-        except wbclient.WBCError as e:
-            raise CallError(str(e), WBCErr[e.error_code], e.error_code)
-
-        except MatchNotFound:
-            return {'sid': None, 'type': wbclient.SID_TYPE_NAME_INVALID}
-
-        return {
-            'sid': entry.sid,
-            'type': entry.sid_type['raw']
-        }
-
     @private
     def convert_sids(self, sidlist):
         """
@@ -961,34 +944,36 @@ def convert_sids(self, sidlist):
 
         for sid in sidlist:
             try:
-                entry = self.__unixsid_to_name(sid, client.ctx.separator.decode())
+                entry = self.__unixsid_to_name(sid, client.separator)
             except KeyError:
                 # This is a Unix Sid, but account doesn't exist
                 unmapped.update({sid: sid})
                 continue
 
             if entry:
-                mapped.update({sid: {
-                    'id_type': IDType.parse_wbc_id_type(entry['id_type']),
-                    'id': entry['id'],
-                    'name': entry['name']
-                }})
+                mapped[sid] = entry
                 continue
 
             to_check.append(sid)
 
+        # First try to retrieve SIDs via SSSD since SSSD and
+        # winbind are both running when we are joined to an IPA
+        # domain. Former provides authoritative SID<->XID resolution
+        # IPA accounts. The latter is authoritative for local accounts.
+        if self.middleware.call_sync('ldap.config')['enable']:
+            if to_check:
+                sss_ctx = SSSClient()
+                results = sss_ctx.sids_to_idmap_entries(to_check)
+                mapped |= results['mapped']
+                to_check = list(results['unmapped'].keys())
+
         if to_check:
             try:
-                results = client.sids_to_users_and_groups(to_check)
+                results = client.sids_to_idmap_entries(to_check)
             except wbclient.WBCError as e:
                 raise CallError(str(e), WBCErr[e.error_code], e.error_code)
 
-            mapped |= {sid: {
-                'id_type': IDType.parse_wbc_id_type(entry.id_type),
-                'id': entry.id,
-                'name': f'{entry.domain}{client.ctx.separator.decode()}{entry.name}',
-            } for sid, entry in results['mapped'].items()}
-
+            mapped |= results['mapped']
             unmapped |= results['unmapped']
 
         return {'mapped': mapped, 'unmapped': unmapped}
@@ -1001,37 +986,49 @@ def convert_unixids(self, id_list):
         from libwbclient (single winbindd request), and so it is the preferred
         method of batch conversion.
         """
-        payload = []
         output = {'mapped': {}, 'unmapped': {}}
 
         if not id_list:
             return output
 
-        for entry in id_list:
-            unixid = entry.get("id")
-            id_ = IDType[entry.get("id_type", "GROUP")]
-            payload.append({
-                'id_type': id_.wbc_str(),
-                'id': unixid
-            })
-
-        try:
-            client = self.__wbclient_ctx()
-            results = client.users_and_groups_to_sids(payload)
-        except wbclient.WBCError as e:
-            raise CallError(str(e), WBCErr[e.error_code], e.error_code)
-
-        output['mapped'] = {unixid: {
-            'id_type': entry.sid_type['parsed'][4:],
-            'id': entry.id,
-            'sid': entry.sid,
-            'name': f'{entry.domain}{client.ctx.separator.decode()}{entry.name}',
-        } for unixid, entry in results['mapped'].items()}
+        if self.middleware.call_sync('ldap.config')['enable']:
+            idmap = self.middleware.call_sync('idmap.query', [
+                ['name', '=', 'DS_TYPE_LDAP']
+            ], {'get': True})
+
+            sss_ctx = SSSClient()
+            results = sss_ctx.users_and_groups_to_idmap_entries(id_list)
+            if not results['unmapped']:
+                # short-circuit
+                return results
+
+            output['mapped'] = results['mapped']
+            id_list = []
+            for entry in results['unmapped'].keys():
+                id_type, xid = entry.split(':')
+                xid = int(xid)
+
+                if xid >= idmap['range_low'] and xid <= idmap['range_high']:
+                    # ID is provided by SSSD but does not have a SID allocated
+                    # do not include in list to look up via winbind since
+                    # we do not want to introduce potential for hanging for
+                    # the winbind request timeout.
+                    continue
+
+                id_list.append({
+                    'id_type': 'USER' if id_type == 'UID' else 'GROUP',
+                    'id': int(xid)
+                })
+
+        if id_list:
+            try:
+                client = self.__wbclient_ctx()
+                results = client.users_and_groups_to_idmap_entries(id_list)
+            except wbclient.WBCError as e:
+                raise CallError(str(e), WBCErr[e.error_code], e.error_code)
 
-        output['unmapped'] = {unixid: {
-            'id_type': 'GROUP' if unixid.startswith('GID') else 'USER',
-            'id': entry.id,
-        } for unixid, entry in results['unmapped'].items()}
+            output['mapped'] |= results['mapped']
+            output['unmapped'] = results['unmapped']
 
         return output
 
@@ -1045,7 +1042,8 @@ def __unixsid_to_name(self, sid, separator='\\'):
             return {
                 'name': f'Unix User{separator}{u["pw_name"]}',
                 'id': uid,
-                'id_type': IDType.USER.wbc_const()
+                'id_type': IDType.USER.name,
+                'sid': sid
             }
 
         gid = int(sid[len(SID_LOCAL_GROUP_PREFIX):])
@@ -1053,27 +1051,8 @@ def __unixsid_to_name(self, sid, separator='\\'):
         return {
             'name': f'Unix Group{separator}{g["gr_name"]}',
             'id': gid,
-            'id_type': IDType.GROUP.wbc_const()
-        }
-
-    @private
-    def sid_to_name(self, sid):
-        try:
-            client = self.__wbclient_ctx()
-        except wbclient.WBCError as e:
-            raise CallError(str(e), WBCErr[e.error_code], e.error_code)
-
-        if (entry := self.__unixsid_to_name, sid, client.ctx.separator.decode()):
-            return entry
-
-        try:
-            entry = client.sid_to_uidgid_entry(sid)
-        except wbclient.WBCError as e:
-            raise CallError(str(e), WBCErr[e.error_code], e.error_code)
-
-        return {
-            'name': f'{entry.domain}{client.ctx.separator.decode()}{entry.name}',
-            'type': entry.sid_type['raw']
+            'id_type': IDType.GROUP.name,
+            'sid': sid
         }
 
     @private
diff --git a/src/middlewared/middlewared/plugins/idmap_/gencache.py b/src/middlewared/middlewared/plugins/idmap_/gencache.py
index 70e4229c269e3..6a8f737bd1021 100644
--- a/src/middlewared/middlewared/plugins/idmap_/gencache.py
+++ b/src/middlewared/middlewared/plugins/idmap_/gencache.py
@@ -4,7 +4,6 @@
 from middlewared.service import Service
 from middlewared.service_exception import MatchNotFound
 from middlewared.plugins.tdb.utils import TDBError
-from .utils import SID_LOCAL_USER_PREFIX, SID_LOCAL_GROUP_PREFIX
 
 GENCACHE_FILE = '/var/run/samba-lock/gencache.tdb'
 
diff --git a/src/middlewared/middlewared/plugins/idmap_/idmap_constants.py b/src/middlewared/middlewared/plugins/idmap_/idmap_constants.py
new file mode 100644
index 0000000000000..580e2c67b565f
--- /dev/null
+++ b/src/middlewared/middlewared/plugins/idmap_/idmap_constants.py
@@ -0,0 +1,29 @@
+import enum
+import wbclient
+
+TRUENAS_IDMAP_MAX = 2147000000
+TRUENAS_IDMAP_DEFAULT_LOW = 90000001
+SID_LOCAL_USER_PREFIX = "S-1-22-1-"
+SID_LOCAL_GROUP_PREFIX = "S-1-22-2-"
+SID_BUILTIN_PREFIX = "S-1-5-32-"
+MAX_REQUEST_LENGTH = 100
+
+
+class IDType(enum.IntEnum):
+    """
+    SSSD and libwbclient use identical values for id types
+    """
+    USER = wbclient.ID_TYPE_UID
+    GROUP = wbclient.ID_TYPE_GID
+    BOTH = wbclient.ID_TYPE_BOTH
+
+    def wbc_str(self):
+        # py-libwbclient uses string repesentation of id type
+        if self == IDType.USER:
+            val = "UID"
+        elif self == IDType.GROUP:
+            val = "GID"
+        else:
+            val = "BOTH"
+
+        return val
diff --git a/src/middlewared/middlewared/plugins/idmap_/idmap_sss.py b/src/middlewared/middlewared/plugins/idmap_/idmap_sss.py
new file mode 100644
index 0000000000000..aeed5035defd9
--- /dev/null
+++ b/src/middlewared/middlewared/plugins/idmap_/idmap_sss.py
@@ -0,0 +1,160 @@
+import pysss_nss_idmap as sssclient
+
+from .idmap_constants import IDType
+from middlewared.utils.itertools import batched
+from middlewared.service_exception import MatchNotFound
+
+
+class SSSClient:
+
+    def _username_to_entry(self, username):
+        """
+        Sample entry returned by pysss_nss_idmap
+
+        `getsidbyusername`
+        {'smbuser': {'sid': 'S-1-5-21-3696504179-2855309571-923743039-1020', 'type': 1}}
+
+        `getidbysid`
+        {'S-1-5-21-3696504179-2855309571-923743039-1020': {'id': 565200020, 'type': 1}}
+
+        Sample of what we return:
+        {'id_type': 'USER', 'id': 565200020, 'name': 'smbuser'}
+        """
+        if not (sid_entry := sssclient.getsidbyusername(username)):
+            return None
+
+        sid = sid_entry[username]['sid']
+        id_type = sid_entry[username]['type']
+
+        if not (id_entry := sssclient.getidbysid(sid)):
+            return None
+
+        return {
+            'id_type': IDType(id_type).name,
+            'id': id_entry[sid]['id'],
+            'name': username,
+            'sid': sid
+        }
+
+    def _groupname_to_sid(self, groupname):
+        if not (sid_entry := sssclient.getsidbygroupname(groupname)):
+            return None
+
+        sid = sid_entry[groupname]['sid']
+        id_type = sid_entry[groupname]['type']
+
+        if not (id_entry := sssclient.getidbysid(sid)):
+            return None
+
+        return {
+            'id_type': IDType(id_type).name,
+            'id': id_entry[sid]['id'],
+            'name': groupname,
+            'sid': sid
+        }
+
+    def _gid_to_entry(self, gid):
+        if not (sid_entry := sssclient.getsidbygid(gid)):
+            return None
+
+        sid = sid_entry[gid]['sid']
+        id_type = sid_entry[gid]['type']
+
+        if not (name_entry := sssclient.getnamebysid(sid)):
+            return None
+
+        return {
+            'id_type': IDType(id_type).name,
+            'id': gid,
+            'name': name_entry[sid]['name'],
+            'sid': sid
+        }
+
+    def _uid_to_entry(self, uid):
+        if not (sid_entry := sssclient.getsidbyuid(uid)):
+            return None
+
+        sid = sid_entry[uid]['sid']
+        id_type = sid_entry[uid]['type']
+
+        if not (name_entry := sssclient.getnamebysid(sid)):
+            return None
+
+        return {
+            'id_type': IDType(id_type).name,
+            'id': uid,
+            'name': name_entry[sid]['name'],
+            'sid': sid
+        }
+
+    def _sid_to_entry(self, sid):
+        if not (id_entry := sssclient.getidbysid(sid)):
+            return None
+
+        if not (name_entry := sssclient.getnamebysid(sid)):
+            return None
+
+        return {
+            'id_type': IDType(id_entry[sid]['type']).name,
+            'id': id_entry[id_entry[sid]]['id'],
+            'name': name_entry[sid]['name'],
+            'sid': sid
+        }
+
+    def sids_to_idmap_entries(self, sidlist):
+        out = {'mapped': {}, 'unmapped': {}}
+        for sid in sidlist:
+            if not (entry := self._sid_to_entry(sid)):
+                out['unmapped'][sid] = sid
+                continue
+
+            out['mapped'][sid] = entry
+
+        return out
+
+    def users_and_groups_to_idmap_entries(self, uidgids):
+        out = {'mapped': {}, 'unmapped': {}}
+
+        for uidgid in uidgids:
+            match uidgid['id_type']:
+                case 'GROUP':
+                    entry = self._gid_to_entry(uidgid['id'])
+                case 'USER':
+                    entry = self._uid_to_entry(uidgid['id'])
+                case 'BOTH':
+                    if not (entry := self._gid_to_entry(uidgid['id'])):
+                        entry = self._uid_to_entry(uidgid['id'])
+                case _:
+                    raise ValueError(f'{uidgid["id_type"]}: Unknown id_type')
+
+            key = f'{IDType[uidgid["id_type"]].wbc_str()}:{uidgid["id"]}'
+            if not entry:
+                out['unmapped'][key] = entry
+                continue
+
+            out['mapped'][key] = entry
+
+        return out
+
+    def sid_to_idmap_entry(self, sid):
+        if not (entry := self._sid_to_entry(sid)):
+            raise MatchNotFound(sid)
+
+        return entry
+
+    def name_to_idmap_entry(self, name):
+        if entry := self._groupname_to_sid(name):
+            return entry
+
+        if entry := self._username_to_sid(name):
+            return entry
+
+        raise MatchNotFound(name)
+
+    def uidgid_to_idmap_entry(self, data):
+        mapped = self.users_and_groups_to_idmap_entries([data])['mapped']
+        if not mapped:
+            raise MatchNotFound(str(data))
+
+        key = f'{IDType[data["id_type"]].wbc_str()}:{data["id"]}'
+        return mapped[key]
diff --git a/src/middlewared/middlewared/plugins/idmap_/utils.py b/src/middlewared/middlewared/plugins/idmap_/idmap_winbind.py
similarity index 55%
rename from src/middlewared/middlewared/plugins/idmap_/utils.py
rename to src/middlewared/middlewared/plugins/idmap_/idmap_winbind.py
index 4a61a7133a3a8..60fbee090c985 100644
--- a/src/middlewared/middlewared/plugins/idmap_/utils.py
+++ b/src/middlewared/middlewared/plugins/idmap_/idmap_winbind.py
@@ -1,7 +1,7 @@
-import enum
 import errno
 import wbclient
 
+from .idmap_constants import IDType, MAX_REQUEST_LENGTH
 from middlewared.utils.itertools import batched
 from middlewared.service_exception import MatchNotFound
 
@@ -22,55 +22,30 @@
     wbclient.WBC_ERR_PWD_CHANGE_FAILED: errno.EFAULT
 }
 
-TRUENAS_IDMAP_MAX = 2147000000
-TRUENAS_IDMAP_DEFAULT_LOW = 90000001
-SID_LOCAL_USER_PREFIX = "S-1-22-1-"
-SID_LOCAL_GROUP_PREFIX = "S-1-22-2-"
-MAX_REQUEST_LENGTH = 100
-
-
-class IDType(enum.Enum):
-    USER = "USER"
-    GROUP = "GROUP"
-    BOTH = "BOTH"
-
-    def wbc_const(self):
-        if self == IDType.USER:
-            val = wbclient.ID_TYPE_UID
-        elif self == IDType.GROUP:
-            val = wbclient.ID_TYPE_GID
-        else:
-            val = wbclient.ID_TYPE_BOTH
-
-        return val
-
-    def wbc_str(self):
-        if self == IDType.USER:
-            val = "UID"
-        elif self == IDType.GROUP:
-            val = "GID"
-        else:
-            val = "BOTH"
-
-        return val
-
-    def parse_wbc_id_type(type_in):
-        if type_in == wbclient.ID_TYPE_UID:
-            return IDType.USER.value
-
-        if type_in == wbclient.ID_TYPE_GID:
-            return IDType.GROUP.value
-
-        if type_in == wbclient.ID_TYPE_BOTH:
-            return IDType.BOTH.value
-
-        raise ValueError(f'{type_in}: invalid winbind ID type')
-
-
 class WBClient:
     def __init__(self, **kwargs):
         self.ctx = wbclient.Ctx()
         self.dom = {}
+        self.separator = self.ctx.separator.decode()
+
+    def _pyuidgid_to_dict(self, entry):
+        return {
+            'id_type': IDType(entry.id_type).name,
+            'id': entry.id,
+            'name': f'{entry.domain}{self.separator}{entry.name}' if entry.name else None,
+            'sid': entry.sid
+        }
+
+    def _as_dict(self, results, do_unmapped=False):
+        for entry in list(results['mapped'].keys()):
+            new = self._pyuidgid_to_dict(results['mapped'][entry])
+            results['mapped'][entry] = new
+
+        for entry in list(results['unmapped'].keys()):
+            new = self._pyuidgid_to_dict(results['unmapped'][entry])
+            results['unmapped'][entry] = new
+
+        return results
 
     def init_domain(self, name='$thisdom'):
         domain = self.dom.get(name)
@@ -97,14 +72,6 @@ def domain_info(self, name='$thisdom'):
         dom = self.init_domain(name)
         return dom.domain_info()
 
-    def users(self, name='$thisdom'):
-        dom = self.init_domain(name)
-        return dom.users()
-
-    def groups(self, name='$thisdom'):
-        dom = self.init_domain(name)
-        return dom.groups()
-
     def _batch_request(self, request_fn, list_in):
         output = {'mapped': {}, 'unmapped': {}}
         for chunk in batched(list_in, MAX_REQUEST_LENGTH):
@@ -114,26 +81,52 @@ def _batch_request(self, request_fn, list_in):
 
         return output
 
-    def sids_to_users_and_groups(self, sidlist):
-        return self._batch_request(
+    def sids_to_idmap_entries(self, sidlist):
+        """
+        Bulk conversion of SIDs to idmap entries
+
+        Returns dictionary:
+        {"mapped": {}, "unmapped": {}
+
+        `mapped` contains entries keyed by SID
+
+        sid: {
+            'id': uid or gid,
+            'id_type': string ("USER", "GROUP", "BOTH"),
+            'name': string,
+            'sid': sid string
+        }
+
+        `unmapped` contains enries keyed by SID as well
+        but they only map to the sid itself. This is simply
+        to facilitate faster lookups of failures.
+        """
+        data = self._batch_request(
             self.ctx.uid_gid_objects_from_sids,
             sidlist
         )
+        return self._as_dict(data)
+
+    def users_and_groups_to_idmap_entries(self, uidgids):
+        payload = [{
+            'id_type': IDType[entry["id_type"]].wbc_str(),
+            'id': entry['id']
+        } for entry in uidgids]
 
-    def users_and_groups_to_sids(self, uidgids):
-        return self._batch_request(
+        data = self._batch_request(
             self.ctx.uid_gid_objects_from_unix_ids,
-            uidgids
+            payload
         )
+        return self._as_dict(data, True)
 
-    def sid_to_uidgid_entry(self, sid):
+    def sid_to_idmap_entry(self, sid):
         mapped = self.sids_to_users_and_groups([sid])['mapped']
         if not mapped:
             raise MatchNotFound(sid)
 
         return mapped[sid]
 
-    def name_to_uidgid_entry(self, name):
+    def name_to_idmap_entry(self, name):
         try:
             entry = self.ctx.uid_gid_object_from_name(name)
         except wbclient.WBCError as e:
@@ -142,14 +135,14 @@ def name_to_uidgid_entry(self, name):
 
             raise
 
-        return entry
+        return self._pyuidgid_to_dict(entry)
 
-    def uidgid_to_sid(self, data):
-        mapped = self.users_and_groups_to_sids([data])['mapped']
+    def uidgid_to_idmap_entry(self, data):
+        mapped = self.users_and_groups_to_idmap_entries([data])['mapped']
         if not mapped:
             raise MatchNotFound(str(data))
 
-        return mapped[f'{data["id_type"]}:{data["id"]}']
+        return mapped[f'{IDType[data["id_type"]].wbc_str()}:{data["id"]}']
 
     def all_domains(self):
         return self.ctx.all_domains()
diff --git a/src/middlewared/middlewared/plugins/smb.py b/src/middlewared/middlewared/plugins/smb.py
index e34e98c25470d..d92534d4282ab 100644
--- a/src/middlewared/middlewared/plugins/smb.py
+++ b/src/middlewared/middlewared/plugins/smb.py
@@ -38,7 +38,7 @@
 from middlewared.plugins.smb_.util_smbconf import generate_smb_conf_dict
 from middlewared.plugins.smb_.utils import apply_presets, is_time_machine_share, smb_strip_comments
 from middlewared.plugins.tdb.utils import TDBError
-from middlewared.plugins.idmap_.utils import IDType, SID_LOCAL_USER_PREFIX, SID_LOCAL_GROUP_PREFIX
+from middlewared.plugins.idmap_.idmap_constants import IDType, SID_LOCAL_USER_PREFIX, SID_LOCAL_GROUP_PREFIX
 from middlewared.utils import filter_list, run
 from middlewared.utils.mount import getmnttree
 from middlewared.utils.path import FSLocation, path_location, is_child_realpath
diff --git a/src/middlewared/middlewared/plugins/smb_/sharesec.py b/src/middlewared/middlewared/plugins/smb_/sharesec.py
index e0069b96ede19..7a71bca4922fe 100644
--- a/src/middlewared/middlewared/plugins/smb_/sharesec.py
+++ b/src/middlewared/middlewared/plugins/smb_/sharesec.py
@@ -162,12 +162,8 @@ async def _ae_to_string(self, ae):
         Convert aclentry in Securty Descriptor dictionary to string
         representation used by sharesec.
         """
-        if not ae['ae_who_sid'] and not ae['ae_who_name']:
-            raise CallError('ACL Entry must have ae_who_sid or ae_who_name.', errno.EINVAL)
-
         if not ae['ae_who_sid']:
-            name = f'{ae["ae_who_name"]["domain"]}\\{ae["ae_who_name"]["name"]}'
-            ae['ae_who_sid'] = (await self.middleware.call('idmap.name_to_sid', name))['sid']
+            raise CallError('ACL Entry must have ae_who_sid.', errno.EINVAL)
 
         return f'{ae["ae_who_sid"]}:{ae["ae_type"]}/0x0/{ae["ae_perm"]}'
 
diff --git a/src/middlewared/middlewared/plugins/smb_/util_smbconf.py b/src/middlewared/middlewared/plugins/smb_/util_smbconf.py
index 19d448b873ae2..0ba8536a76f22 100644
--- a/src/middlewared/middlewared/plugins/smb_/util_smbconf.py
+++ b/src/middlewared/middlewared/plugins/smb_/util_smbconf.py
@@ -3,7 +3,6 @@
 from logging import getLogger
 from middlewared.utils import filter_list
 from middlewared.plugins.account import DEFAULT_HOME_PATH
-from middlewared.plugins.idmap_.utils import TRUENAS_IDMAP_MAX
 from middlewared.plugins.smb_.constants import LOGLEVEL_MAP, SMBPath
 from middlewared.plugins.directoryservices import DSStatus, SSL
 
diff --git a/src/middlewared/middlewared/utils/directoryservices/constants.py b/src/middlewared/middlewared/utils/directoryservices/constants.py
new file mode 100644
index 0000000000000..1d2dcb9d24a1b
--- /dev/null
+++ b/src/middlewared/middlewared/utils/directoryservices/constants.py
@@ -0,0 +1,43 @@
+import enum
+
+
+class DSStatus(enum.Enum):
+    DISABLED = enum.auto()
+    FAULTED = enum.auto()
+    LEAVING = enum.auto()
+    JOINING = enum.auto()
+    HEALTHY = enum.auto()
+
+
+class DSType(enum.Enum):
+    AD = 'activedirectory'
+    IPA = 'ipa'
+    LDAP = 'ldap'
+
+
+class SASL_Wrapping(enum.Enum):
+    PLAIN = 'PLAIN'
+    SIGN = 'SIGN'
+    SEAL = 'SEAL'
+
+
+class SSL(enum.Enum):
+    NOSSL = 'OFF'
+    USESSL = 'ON'
+    USESTARTTLS = 'START_TLS'
+
+
+class NSS_Info(enum.Enum):
+    SFU = ('SFU', (DSType.AD,))
+    SFU20 = ('SFU20', (DSType.AD,))
+    RFC2307 = ('RFC2307', (DSType.AD, DSType.LDAP))
+    RFC2307BIS = ('RFC2307BIS', (DSType.LDAP, DSType.IPA))
+    TEMPLATE = ('TEMPLATE', (DSType.AD,))
+
+    @property
+    def nss_type(self):
+        return self.value[0]
+
+    @property
+    def valid_services(self):
+        return self.value[1]
diff --git a/tests/api2/test_011_user.py b/tests/api2/test_011_user.py
index 25eaa234513fe..67af2ca6336ad 100644
--- a/tests/api2/test_011_user.py
+++ b/tests/api2/test_011_user.py
@@ -213,9 +213,9 @@ def test_002_verify_user_exists_in_pwd(request):
     assert pw['pw_dir'] == VAR_EMPTY
 
     # At this point, we're not an SMB user
-    assert pw['sid_info'] is not None
-    assert pw['sid_info']['domain_information']['online']
-    assert pw['sid_info']['domain_information']['activedirectory'] is False
+    assert pw['sid'] is not None
+    assert pw['source'] == 'LOCAL'
+    assert pw['local'] is True
 
 
 def test_003_get_next_uid_again(request):
@@ -427,9 +427,9 @@ def test_031_create_user_with_homedir(request):
     assert pw['pw_uid'] == UserAssets.TestUser02['query_response']['uid']
     assert pw['pw_shell'] == UserAssets.TestUser02['query_response']['shell']
     assert pw['pw_gecos'] == UserAssets.TestUser02['query_response']['full_name']
-    assert pw['sid_info'] is not None
-    assert pw['sid_info']['domain_information']['online']
-    assert pw['sid_info']['domain_information']['activedirectory'] is False
+    assert pw['sid'] is not None
+    assert pw['source'] == 'LOCAL'
+    assert pw['local'] is True
 
     # verify smb user passdb entry
     assert qry['sid']
diff --git a/tests/api2/test_030_activedirectory.py b/tests/api2/test_030_activedirectory.py
index bea06f2b217c3..6995af5f7ed64 100644
--- a/tests/api2/test_030_activedirectory.py
+++ b/tests/api2/test_030_activedirectory.py
@@ -189,12 +189,10 @@ def test_07_enable_leave_activedirectory(request):
         pw = ad['user_obj']
 
         # Verify winbindd information
-        assert pw['sid_info'] is not None, str(ad)
-        assert not pw['sid_info']['sid'].startswith('S-1-22-1-'), str(ad)
-        assert pw['sid_info']['domain_information']['domain'] != 'LOCAL', str(ad)
-        assert pw['sid_info']['domain_information']['domain_sid'] is not None, str(ad)
-        assert pw['sid_info']['domain_information']['online'], str(ad)
-        assert pw['sid_info']['domain_information']['activedirectory'], str(ad)
+        assert pw['sid'] is not None, str(ad)
+        assert not pw['sid'].startswith('S-1-22-1-'), str(ad)
+        assert pw['local'] is False
+        assert pw['source'] == 'ACTIVEDIRECTORY'
 
         result = call('dnsclient.forward_lookup', {'names': [f'{hostname}.{AD_DOMAIN}']})
         assert len(result) != 0