Python Flask对接AD域如何使用sAMAccountName实现用户登录验证

你当前代码的错误逻辑是直接将用户输入的sAMAccountName作为DN中的CN字段传入simple_bind_s方法，而AD域的simple_bind_s校验要求传入用户完整的可分辨名称（DN），sAMAccountName只是AD用户的独立属性，不属于DN的组成部分，因此无法直接用该值拼接DN完成校验。

实现步骤 #

正确的校验逻辑分两步走：

• 先使用拥有AD用户查询权限的服务账号绑定LDAP服务，查询输入的sAMAccountName对应的用户完整DN
• 用查询到的用户DN和用户输入的密码做绑定校验，绑定成功即身份验证通过

代码修改 #

1. 调整__init__.py配置 #

新增服务账号和查询根域配置：

app.config['SQLALCHEMY_DATABASE_URI'] = 'connection_to_db'
app.config['SECRET_KEY'] = 'secret_key'

# The Attribute you want users to authenticate to LDAP with.
app.config['LDAP_USER_LOGIN_ATTR'] = 'sAMAccountName'

app.config['LDAP_PROVIDER_URL'] = 'connection_to_ldap'
app.config['LDAP_PROTOCOL_VERSION'] = 3
# 新增AD查询专用服务账号配置，该账号仅需AD用户查询权限即可
app.config['LDAP_BIND_DN'] = 'CN=ldap_query_user,DC=example,DC=com'
app.config['LDAP_BIND_PASSWORD'] = 'your_service_account_password'
# 新增AD用户搜索根域
app.config['LDAP_USER_SEARCH_BASE'] = 'DC=example,DC=com'
db = SQLAlchemy(app)

2. 调整models.py的try_login方法 #

修改后的校验逻辑如下：

import ldap
from flask import current_app

def get_ldap_connection():
    conn = ldap.initialize(current_app.config['LDAP_PROVIDER_URL'])
    conn.set_option(ldap.OPT_PROTOCOL_VERSION, current_app.config['LDAP_PROTOCOL_VERSION'])
    # 可选：关闭SSL校验如果是自签名证书，生产环境不建议开启
    # conn.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_NEVER)
    return conn

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(100))
 
    def __init__(self, username, password):
        self.username = username
 
    @staticmethod
    def try_login(username, password):
        conn = get_ldap_connection()
        try:
            # 第一步：用服务账号绑定获取查询权限
            conn.simple_bind_s(
                current_app.config['LDAP_BIND_DN'],
                current_app.config['LDAP_BIND_PASSWORD']
            )
            # 第二步：用sAMAccountName搜索对应用户DN
            search_filter = f"(sAMAccountName={username})"
            result = conn.search_s(
                current_app.config['LDAP_USER_SEARCH_BASE'],
                ldap.SCOPE_SUBTREE,
                search_filter,
                ['dn']
            )
            # 校验搜索结果是否唯一
            if len(result) != 1:
                raise ldap.INVALID_CREDENTIALS("用户不存在或存在多个匹配用户")
            user_dn = result[0][0]
            # 第三步：用用户DN和输入密码做绑定校验
            conn.simple_bind_s(user_dn, password)
            return True
        except ldap.LDAPError as e:
            # 所有LDAP异常都视为校验失败
            return False
        finally:
            conn.unbind_s()
 
    def is_authenticated(self):
        return True
 
    def is_active(self):
        return True
 
    def is_anonymous(self):
        return False
 
    def get_id(self):
        return str(self.id)

注意事项 #

• 服务账号仅需分配AD域的普通用户查询权限即可，无需更高权限，避免安全风险
• 如果AD域开启了登录失败锁定策略，建议增加密码错误次数限制，避免恶意请求导致用户账号被锁
• 生产环境建议使用LDAPS协议对接AD，避免账号密码明文传输

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