使用LDAP搜索过滤器查询Active Directory结果不全问题求助

环境与代码信息 #

依赖版本 #

• ldap3: 2.9.1
• Python: 3.9.16

连接配置 #

'ldap_server': 'ldap://IP:389',
'ldap_user': 'user',
'ldap_password': 'password',
'base_dn': 'DC=ORG,DC=COM'

连接代码 #

server = Server(config['ldap_server'], get_info=ALL)
conn = Connection(server, user=config['ldap_user'], password=config['ldap_password'], auto_bind=True)

用户查询函数 #

def get_domain_users(conn):
    search_filter = '(&(objectClass=user)(objectCategory=person)(!(sAMAccountName=*$))(!(userAccountControl:1.2.840.113556.1.4.803:=2)))'
    attributes = ['userAccountControl','displayName', 'sAMAccountName', 'mail', 'whenCreated', 'whenChanged', 'lastLogontimeStamp',
        'sAMAccountType', 'memberOf']

    conn.search(config['base_dn'], search_filter, SUBTREE, attributes=attributes)
    domain_users = []
    for entry in conn.entries:
        user_info = {
            'source.uac': get_account_type(entry['userAccountControl'].value)[1],
            'source.display_name': entry.displayName.value,
            'source.mail': entry.mail.value,
            'ad.memberOf': entry.memberOf.value,
            'ad.SamAccountName': entry.sAMAccountName.value,
            'ad.SamAccountType': get_sam_acc_type(entry.sAMAccountType.value)[1],
            'ad.whenCreated': str(entry.whenCreated.value),
            'ad.whenChanged': str(entry.whenChanged.value),
            'ad.lastLogonTime': str(entry.lastLogontimeStamp.value)
        }
        domain_users.append(user_info)
    return domain_users

问题描述 #

指定搜索范围内存在符合条件的用户，但当前搜索无法返回全部记录。过滤器已设置为筛选启用状态、sAMAccountName不以$结尾的用户，结果仍存在部分用户遗漏。

1. LDAP分页限制导致结果截断 #

Active Directory默认限制单页搜索结果数量（通常为1000条），ldap3默认未开启分页，当结果超过阈值时，仅返回第一页数据，后续用户会被遗漏。

修改建议：
在搜索时开启分页，可通过两种方式实现：

# 方式1：直接添加分页参数
conn.search(config['base_dn'], search_filter, SUBTREE, attributes=attributes, paged_size=1000)

# 方式2：使用PagedSearch类（更稳定）
from ldap3 import PagedSearch
domain_users = []
with PagedSearch(conn, config['base_dn'], search_filter, SUBTREE, attributes=attributes, paged_size=1000) as search:
    for entry in search:
        # 后续用户信息处理逻辑保持不变（需结合下文属性判断优化）
        user_info = {
            # ... 字段处理
        }
        domain_users.append(user_info)

2. 属性缺失引发循环中断 #

部分用户可能存在属性为空的情况（如mail、lastLogontimeStamp未配置），直接调用.value会抛出AttributeError，导致循环提前终止，后续用户无法被处理。

修改建议：
添加属性存在性判断，避免异常：

user_info = {
    'source.uac': get_account_type(entry['userAccountControl'].value)[1],
    'source.display_name': entry.displayName.value if 'displayName' in entry else None,
    'source.mail': entry.mail.value if 'mail' in entry else None,
    'ad.memberOf': entry.memberOf.value if 'memberOf' in entry else [],
    'ad.SamAccountName': entry.sAMAccountName.value,
    'ad.SamAccountType': get_sam_acc_type(entry.sAMAccountType.value)[1],
    'ad.whenCreated': str(entry.whenCreated.value) if 'whenCreated' in entry else None,
    'ad.whenChanged': str(entry.whenChanged.value) if 'whenChanged' in entry else None,
    'ad.lastLogonTime': str(entry.lastLogontimeStamp.value) if 'lastLogontimeStamp' in entry else None
}

注：memberOf为多值属性，空值时建议设为空列表，避免类型不一致问题。

3. 搜索过滤器存在转义错误 #

原过滤器中的&是HTML转义后的&，AD无法正确解析该字符，会导致过滤器逻辑失效，返回结果不符合预期。

修改建议：
修正过滤器的转义字符：

search_filter = '(&(objectClass=user)(objectCategory=person)(!(sAMAccountName=*$))(!(userAccountControl:1.2.840.113556.1.4.803:=2)))'

4. 绑定用户权限不足 #

如果绑定的LDAP用户没有读取AD中部分OU的权限，这些OU下的用户不会被返回。例如部分OU设置了独立权限，绑定用户无访问权限。

排查建议：

• 通过AD用户和计算机工具，检查绑定用户对所有目标OU的读取权限。
• 尝试使用域管理员账号测试，确认是否为权限问题。

内容的提问来源于stack exchange，提问作者Daniel