容器化部署OpenLDAP并使用Python查询LDAP数据

version: '3'

services:
  ldap:
    image: osixia/openldap:1.5.0
    container_name: openldap
    environment:
      LDAP_ORGANISATION: "tendcode"
      LDAP_DOMAIN: "tendcode.com"
      LDAP_ADMIN_PASSWORD: "admin@123"
    ports:
      - "389:389"
      - "636:636"
    volumes:
      - ./ldap/data:/var/lib/ldap
      - ./ldap/config:/etc/ldap/slapd.d
    restart: unless-stopped

  phpldapadmin:
    image: osixia/phpldapadmin:0.9.0
    container_name: phpldapadmin
    environment:
      PHPLDAPADMIN_LDAP_HOSTS: ldap
      PHPLDAPADMIN_HTTPS: "false"  # 如果需要HTTPS, 可以设置为"true"
    ports:
      - "8081:80"
    restart: unless-stopped
    depends_on:
      - ldap

docker-compose up -d

dn: uid=xiaohong,ou=product,ou=shanghai,dc=tendcode,dc=com
uid: xiaohong
sn: 小红
cn: xiaohong
givenName: xiaohong
displayName: xiaohong
mail: xiaohong@163.com
objectClass: person
objectClass: organizationalPerson
objectClass: inetOrgPerson
userpassword: npc@123

from ldap3 import Server, Connection, ALL, SUBTREE

class LDAPQuery:
    def __init__(self, ldap_server, ldap_user, ldap_password, base_dn, port=389):
        self.ldap_server = ldap_server
        self.ldap_user = ldap_user
        self.ldap_password = ldap_password
        self.base_dn = base_dn
        self.port = port
        self.conn = None

    def __enter__(self):
        """进入上下文管理器时，自动连接LDAP服务器"""
        self.connect()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        """退出上下文管理器时，自动关闭LDAP连接"""
        self.close()

    def connect(self):
        """连接到LDAP服务器"""
        server = Server(self.ldap_server, self.port, get_info=ALL)
        self.conn = Connection(server, self.ldap_user, self.ldap_password)
        if not self.conn.bind():
            raise Exception("Failed to bind to server: {}".format(self.conn.result['description']))

    def search(self, search_base=None, search_filter='(objectClass=*)', search_scope=SUBTREE,
               attributes=None):
        """
        通用LDAP查询函数
        :param search_filter: 查询筛选条件，默认返回所有对象
        :param search_base: 查询起始的基准DN，默认为初始化时的base_dn
        :param search_scope: 查询范围，可以是SUBTREE（默认）、LEVEL或BASE
        :param attributes: 要返回的属性列表，默认返回所有属性
        :return: 查询结果列表
        """
        if self.conn is None or not self.conn.bound:
            raise Exception("Not connected or bound to LDAP server")

        # 使用默认的 base_dn 作为 search_base，如果未指定
        search_base = search_base or self.base_dn

        # 执行查询
        self.conn.search(search_base=search_base,
                         search_filter=search_filter,
                         search_scope=search_scope,
                         attributes=attributes)

        return self.conn.entries

    def close(self):
        """关闭LDAP连接"""
        if self.conn:
            self.conn.unbind()
            self.conn = None


def main():
    with LDAPQuery(LDAP_SERVER, LDAP_USER, LDAP_PASSWORD, BASE_DN) as ldap_query:
        users = ldap_query.search(search_base='dc=tendcode,dc=com',
                                  search_filter='(objectClass=person)',
                                  attributes=['cn', 'sn', 'mail'])
        for user in users:
            print(user.entry_dn)
            print(json.dumps(user.entry_attributes_as_dict))


if __name__ == "__main__":
    LDAP_SERVER = 'ldap://100.88.88.203:389'
    LDAP_USER = 'CN=admin,DC=tendcode,DC=com'
    LDAP_PASSWORD = 'admin@123'
    BASE_DN = 'dc=tendcode,dc=com'

    main()

uid=xiaoming,ou=beijing,dc=tendcode,dc=com
{"mail": ["xiaoming@abc.com"], "sn": ["\u5c0f\u660e"], "cn": ["xiaoming"]}
uid=xiaoqiang,ou=develop,ou=wuhan,dc=tendcode,dc=com
{"mail": ["xiaoqiang@163.com"], "sn": ["\u5c0f\u5f3a"], "cn": ["xiaoqiang"]}
uid=xiaohong,ou=product,ou=shanghai,dc=tendcode,dc=com
{"mail": ["xiaohong@163.com"], "sn": ["\u5c0f\u7ea2"], "cn": ["xiaohong"]}