前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >如何使用Python连接ldap

如何使用Python连接ldap

作者头像
Ryan-Miao
发布2019-08-20 15:51:01
2.6K0
发布2019-08-20 15:51:01
举报
文章被收录于专栏:Ryan MiaoRyan Miao

如何使用Python连接ldap

好多使用ldap认证的软件都是Python的,比如superset和airflow, 好吧,他们都是airbnb家的。在配置ldap的时候可能会出现认证失败,你不知道是因为什么导致配置失败的。所以,就要

跟踪源码,看看内部怎么认证实现的。

ldap介绍和使用安装参见: https://cloud.tencent.com/developer/article/1490857

登录的源码参见: https://github.com/apache/airflow/blob/70e937a8d8ff308a9fb9055ceb7ef2c034200b36/airflow/contrib/auth/backends/ldap_auth.py#L191

具体来实现如下:

为了模拟环境,我们使用docker-python。基于Debian Python3: https://github.com/Ryan-Miao/docker-china-source/tree/master/docker-python

启动

docker run -it ryan/python:3 /bin/bash

下载ldap3

pip install ldap3

测试连接

root@5edee218d962:/# python
Python 3.7.4 (default, Jul 13 2019, 14:20:24) 
[GCC 6.3.0 20170516] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> from ldap3 import Server, Connection, ALL, NTLM, ALL_ATTRIBUTES, LEVEL, SUBTREE
>>> server = Server('172.17.0.2', get_info=ALL)
>>> conn = Connection(server, 'cn=admin,dc=demo,dc=com', 'admin', auto_bind=True)
>>> conn.extend.standard.who_am_i()
'dn:cn=admin,dc=demo,dc=com'

测试登录部分

登录源码如下:

@staticmethod
    def try_login(username, password):
        conn = get_ldap_connection(configuration.conf.get("ldap", "bind_user"),
                                   configuration.conf.get("ldap", "bind_password"))

        search_filter = "(&({0})({1}={2}))".format(
            configuration.conf.get("ldap", "user_filter"),
            configuration.conf.get("ldap", "user_name_attr"),
            username
        )

        search_scope = LEVEL
        if configuration.conf.has_option("ldap", "search_scope"):
            if configuration.conf.get("ldap", "search_scope") == "SUBTREE":
                search_scope = SUBTREE
            else:
                search_scope = LEVEL

        # todo: BASE or ONELEVEL?

        res = conn.search(configuration.conf.get("ldap", "basedn"), search_filter, search_scope=search_scope)

        # todo: use list or result?
        if not res:
            log.info("Cannot find user %s", username)
            raise AuthenticationError("Invalid username or password")

        entry = conn.response[0]

        conn.unbind()

        if 'dn' not in entry:
            # The search filter for the user did not return any values, so an
            # invalid user was used for credentials.
            raise AuthenticationError("Invalid username or password")

        try:
            conn = get_ldap_connection(entry['dn'], password)
        except KeyError:
            log.error("""
            Unable to parse LDAP structure. If you're using Active Directory
            and not specifying an OU, you must set search_scope=SUBTREE in airflow.cfg.
            %s
            """, traceback.format_exc())
            raise LdapException(
                "Could not parse LDAP structure. "
                "Try setting search_scope in airflow.cfg, or check logs"
            )

        if not conn:
            log.info("Password incorrect for user %s", username)
            raise AuthenticationError("Invalid username or password")

第一步: 获取连接

from ldap3 import Server, Connection, ALL, NTLM, ALL_ATTRIBUTES, LEVEL, SUBTREE


server = Server('172.17.0.2:389', get_info=ALL)
conn = Connection(server, 'cn=admin,dc=demo,dc=com', 'admin', auto_bind=True)
conn.extend.standard.who_am_i()

第二步: 根据filter search用户。 这里我们的配置文件如下:

[ldap]
# set this to ldaps://<your.ldap.server>:<port>
uri = ldap://172.17.0.2:389
user_filter = objectClass=inetOrgPerson
user_name_attr = sn
group_member_attr = memberOf
superuser_filter =
data_profiler_filter =
bind_user = cn=admin,dc=demo,dc=com
bind_password = admin
basedn = dc=demo,dc=com
cacert =
search_scope = SUBTREE

源码就是拼接filter, 最后变成(&(objectClass=inetOrgPerson)(sn=ryanmiao)), 然后search scope.

>>> with Connection(server, 'cn=admin,dc=demo,dc=com', 'admin') as conn:
...         conn.search('dc=demo,dc=com', '(&(objectClass=inetOrgPerson)(cn=hr-ryan))', search_scope=SUBTREE)
...         entry = conn.entries[0]
... 
True
>>> entry
DN: cn=hr-ryan,ou=HR,ou=People,dc=demo,dc=com - STATUS: Read - READ TIME: 2019-08-19T12:58:34.966181

第三步: 从上一步得到dn,然后根据用户输入的密码,再次连接, 不抛异常就证明密码正确

>>> conn = Connection(server, 'cn=hr-ryan,ou=HR,ou=People,dc=demo,dc=com', '123456', auto_bind=True)
>>> 
>>> conn = Connection(server, 'cn=hr-ryan,ou=HR,ou=People,dc=demo,dc=com', '1234567', auto_bind=True)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/local/lib/python3.7/site-packages/ldap3/core/connection.py", line 325, in __init__
    self.do_auto_bind()
  File "/usr/local/lib/python3.7/site-packages/ldap3/core/connection.py", line 353, in do_auto_bind
    raise LDAPBindError(self.last_error)
ldap3.core.exceptions.LDAPBindError: automatic bind not successful - invalidCredentials

测试分组

ldap提供了分组,配置见前文。我们采用ldap统一登录之后,还要把用户放入不同的group里来区分权限。比如splunk-users, splunk-admin, gitlab-admin, gitlab-user等。

根据用户名和密码,我们实现了用户登录密码验证,接下来需要取到用户所属于的group。

源码如下:

def groups_user(conn, search_base, user_filter, user_name_att, username):
    search_filter = "(&({0})({1}={2}))".format(user_filter, user_name_att, username)
    try:
        memberof_attr = configuration.conf.get("ldap", "group_member_attr")
    except Exception:
        memberof_attr = "memberOf"
    res = conn.search(search_base, search_filter, attributes=[memberof_attr])
    if not res:
        log.info("Cannot find user %s", username)
        raise AuthenticationError("Invalid username or password")

    if conn.response and memberof_attr not in conn.response[0]["attributes"]:
        log.warning("""Missing attribute "%s" when looked-up in Ldap database.
        The user does not seem to be a member of a group and therefore won't see any dag
        if the option filter_by_owner=True and owner_mode=ldapgroup are set""",
                    memberof_attr)
        return []

    user_groups = conn.response[0]["attributes"][memberof_attr]

    regex = re.compile("cn=([^,]*).*", re.IGNORECASE)
    groups_list = []
    try:
        groups_list = [regex.search(i).group(1) for i in user_groups]
    except IndexError:
        log.warning("Parsing error when retrieving the user's group(s)."
                    " Check if the user belongs to at least one group"
                    " or if the user's groups name do not contain special characters")

    return groups_list

同样,第一步,连接,第二步search,但需要返回字段memberof。注意我们前面配置了group_member_attr=memberof

>>> with Connection(server, 'cn=admin,dc=demo,dc=com', 'admin') as conn:
...         conn.search('dc=demo,dc=com', '(&(objectClass=inetOrgPerson)(sn=hr-ryan))', attributes=['memberof'])
...         entry = conn.entries[0]
...         entry
... 
True
DN: cn=hr-ryan,ou=HR,ou=People,dc=demo,dc=com - STATUS: Read - READ TIME: 2019-08-19T13:05:20.592805
    memberOf: cn=g-admin,ou=Group,dc=demo,dc=com
              cn=g-users,ou=Group,dc=demo,dc=com
              cn=g-a,ou=Group,dc=demo,dc=com

如果取不到group会报错。

以上就差不多是airflow的ldap配置原理了。其他雷同,不一样的地方也许是在filter的地方,我们找对应软件的源码look一下就ok了。

本文参与 腾讯云自媒体分享计划,分享自作者个人站点/博客。
原始发表:2019-08-20 ,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 如何使用Python连接ldap
    • 测试连接
      • 测试登录部分
        • 测试分组
        相关产品与服务
        容器镜像服务
        容器镜像服务(Tencent Container Registry,TCR)为您提供安全独享、高性能的容器镜像托管分发服务。您可同时在全球多个地域创建独享实例,以实现容器镜像的就近拉取,降低拉取时间,节约带宽成本。TCR 提供细颗粒度的权限管理及访问控制,保障您的数据安全。
        领券
        问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档