ldaptor

import os from twisted.internet import reactor, defer from ldaptor.protocols.ldap import ldapclient, ldapsyntax, ldapconnector @defer.inlineCallbacks def example(): # Note: For production, load sensitive data securely (e.g., from environment variables). # It is recommended to use byte strings for ldaptor objects. server_ip = os.environ.get('LDAP_SERVER_IP', '127.0.0.1').encode('utf-8') basedn = os.environ.get('LDAP_BASE_DN', 'dc=example,dc=com').encode('utf-8') binddn = os.environ.get('LDAP_BIND_DN', 'cn=admin,dc=example,dc=com').encode('utf-8') bindpw = os.environ.get('LDAP_BIND_PASSWORD', 'secret').encode('utf-8') query = os.environ.get('LDAP_QUERY', '(objectClass=*)').encode('utf-8') # Create an LDAP client creator c = ldapconnector.LDAPClientCreator(reactor, ldapclient.LDAPClient) # Define overrides for connecting to the LDAP server overrides = {basedn: (server_ip, 389)} # Connect to the LDAP server client = yield c.connect(basedn, overrides=overrides) print(f"Connected to LDAP server at {server_ip.decode('utf-8')}") # Bind to the LDAP server yield client.bind(binddn, bindpw) print(f"Bound as {binddn.decode('utf-8')}") # Perform a search o = ldapsyntax.LDAPEntry(client, basedn) results = yield o.search(filterText=query) print(f"Found {len(results)} entries for query '{query.decode('utf-8')}'") # Print LDIF for each result for entry in results: print(entry.getLDIF()) print("LDAP operations complete.") if __name__ == '__main__': df = example() df.addErrback(lambda err: err.printTraceback()) df.addCallback(lambda _: reactor.stop()) reactor.run()