Python FastAPI中Active Directory认证与JWT结合的实现问题

结论：完全可以实现「AD认证通过后签发JWT，后续用JWT访问受保护端点」的流程，你的思路是对的，当前代码的问题在于缺少实际的AD认证逻辑，以及JWT生成/验证的细节错误。

一、当前代码的核心问题 #

• /login接口没有执行任何AD认证逻辑，直接生成JWT，导致无论传入什么凭据都能通过。
• create_access_token函数错误地使用传入的access_token作为JWT签名密钥，而解码时用的是SECRET_KEY，这会导致后续令牌验证失败。
• 缺少AD组成员身份检查的逻辑。

二、完整实现方案 #

1. 依赖安装 #

先安装所需依赖库：

pip install fastapi uvicorn python-jose[cryptography] passlib bcrypt ldap3

2. 修正后的代码实现 #

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
import jwt
from jwt.exceptions import ExpiredSignatureError, InvalidSignatureError
import datetime
from ldap3 import Server, Connection, ALL, NTLM
from typing import Optional

app = FastAPI()

# 核心配置
SECRET_KEY = "C4460C71-7A98-4356-A6DC-CC9F3FBBB63E"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# AD服务器配置（根据实际环境修改）
AD_SERVER = "ldap://your-ad-server-domain-or-ip:389"  # 生产环境建议用ldaps://...:636加密连接
AD_DOMAIN = "YOUR_AD_DOMAIN"
AD_SEARCH_BASE = "OU=Users,DC=yourdomain,DC=com"
REQUIRED_AD_GROUP = "CN=AllowedUsers,OU=Groups,DC=yourdomain,DC=com"  # 需要校验的AD组


def authenticate_ad_user(username: str, password: str) -> Optional[dict]:
    """验证AD用户凭据，返回用户信息（验证通过则返回）"""
    try:
        ad_username = f"{AD_DOMAIN}\\{username}"
        server = Server(AD_SERVER, get_info=ALL)
        # 建立AD连接并验证用户
        conn = Connection(server, user=ad_username, password=password, authentication=NTLM, auto_bind=True)
        
        # 查询用户DN（用于后续组检查）及基础信息
        conn.search(
            search_base=AD_SEARCH_BASE,
            search_filter=f"(sAMAccountName={username})",
            attributes=["distinguishedName", "displayName"]
        )
        
        if not conn.entries:
            conn.unbind()
            return None
        
        user_info = {
            "username": username,
            "dn": conn.entries[0].distinguishedName.value,
            "display_name": conn.entries[0].displayName.value
        }
        conn.unbind()
        return user_info
    except Exception as e:
        print(f"AD认证失败: {str(e)}")
        return None


def check_ad_group_membership(user_dn: str) -> bool:
    """检查用户是否属于指定AD组"""
    try:
        # 生产环境建议使用专用AD服务账户，避免普通用户权限不足
        service_username = f"{AD_DOMAIN}\\your-service-account"
        service_password = "your-service-account-password"
        
        server = Server(AD_SERVER, get_info=ALL)
        conn = Connection(server, user=service_username, password=service_password, authentication=NTLM, auto_bind=True)
        
        # 查询指定组的成员列表，判断用户是否在其中
        conn.search(
            search_base=REQUIRED_AD_GROUP,
            search_filter=f"(member={user_dn})",
            attributes=["member"]
        )
        
        is_member = len(conn.entries) > 0
        conn.unbind()
        return is_member
    except Exception as e:
        print(f"组检查失败: {str(e)}")
        return False


def create_access_token(data: dict, expires_delta: Optional[datetime.timedelta] = None):
    """生成JWT令牌"""
    to_encode = data.copy()
    expire = datetime.datetime.utcnow() + (expires_delta or datetime.timedelta(minutes=15))
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(token: str = Depends(OAuth2PasswordBearer(tokenUrl="/login"))):
    """验证JWT令牌并返回当前用户信息"""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="令牌已过期",
            headers={"WWW-Authenticate": "Bearer"},
        )
    except InvalidSignatureError:
        raise credentials_exception
    
    # 可选：定期验证用户在AD中的状态（如是否被禁用）
    # user_info = authenticate_ad_user(username, "")  # 仅查询状态无需密码
    # if not user_info:
    #     raise credentials_exception
    
    return {"username": username, "display_name": payload.get("display_name")}


@app.post("/login")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    """登录流程：AD认证 → 组检查 → 签发JWT"""
    # 1. 验证AD用户凭据
    user_info = authenticate_ad_user(form_data.username, form_data.password)
    if not user_info:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    # 2. 检查用户是否属于指定AD组
    if not check_ad_group_membership(user_info["dn"]):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="用户无访问权限",
        )
    
    # 3. 生成并返回JWT令牌
    access_token_expires = datetime.timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user_info["username"], "display_name": user_info["display_name"]},
        expires_delta=access_token_expires
    )
    
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/protected_route")
async def protected_route(current_user: dict = Depends(get_current_user)):
    """受保护端点，仅持有有效JWT的用户可访问"""
    return {
        "message": "你已成功认证并访问受保护资源",
        "user": current_user
    }

3. 关键说明 #

• AD认证：使用ldap3库通过NTLM协议连接AD，验证用户凭据的同时获取用户的distinguishedName（DN）用于组检查。
• 组检查：通过查询指定AD组的member属性，确认用户是否具备访问权限，只有通过检查的用户才能获取JWT。
• JWT安全：统一使用SECRET_KEY进行签名和解码，确保令牌的有效性；令牌包含用户基础信息，方便后续业务使用。
• 生产优化：建议使用LDAPS加密连接AD，避免凭据明文传输；组检查使用专用AD服务账户，避免普通用户权限不足问题。

三、测试方式 #

1. 发送POST请求到/login，用form-data传递username和password，正确AD凭据且属于指定组时会返回JWT。
2. 访问/protected_route时，在请求头添加Authorization: Bearer <你的JWT令牌>，即可访问受保护资源；令牌无效/过期/用户无权限时会返回对应HTTP错误。

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