如何通过Windows AD（非Azure AD）实现FastAPI的授权功能？

要通过本地Windows AD（非Azure AD）实现FastAPI的登录授权，核心是通过LDAP协议验证AD用户凭证，再结合JWT实现接口权限控制。以下是完整实现方案：

1. 安装依赖 #

首先安装所需的Python包：

pip install fastapi uvicorn ldap3 python-jose[cryptography] python-multipart

2. 配置AD连接参数 #

定义AD服务器和JWT相关配置，替换为你的实际环境参数：

# AD服务器配置
AD_SERVER = "ldap://your-ad-server-ip-or-hostname"  # 示例：ldap://ad.company.com
AD_DOMAIN = "company.com"
# 用于绑定AD的服务账号（需具备读取用户信息的权限）
AD_BIND_USER = "ad-service-account@company.com"
AD_BIND_PASSWORD = "your-service-account-password"

# JWT配置
SECRET_KEY = "your-strong-secret-key-here"  # 生产环境用随机强密钥
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

3. 实现AD用户验证逻辑 #

使用ldap3库连接AD并验证用户凭证，支持NTLM认证：

from ldap3 import Server, Connection, NTLM

def authenticate_ad_user(username: str, password: str) -> bool:
    # 处理用户名格式，兼容DOMAIN\username和username@domain.com
    if "\\" in username:
        domain_part, user_part = username.split("\\", 1)
        user_principal = f"{user_part}@{domain_part}"
    elif "@" in username:
        user_principal = username
    else:
        user_principal = f"{username}@{AD_DOMAIN}"

    try:
        # 连接AD服务器并尝试用用户凭证绑定
        server = Server(AD_SERVER, get_info=True)
        conn = Connection(
            server,
            user=user_principal,
            password=password,
            authentication=NTLM,
            auto_bind=True
        )
        conn.unbind()
        return True
    except Exception as e:
        print(f"AD验证失败: {str(e)}")
        return False

4. 登录接口与JWT令牌生成 #

创建登录接口，验证AD用户后返回JWT令牌供后续接口授权使用：

from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import OAuth2PasswordRequestForm
from jose import jwt
from datetime import datetime, timedelta

app = FastAPI()

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

@app.post("/login")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # 验证AD用户凭证
    if not authenticate_ad_user(form_data.username, form_data.password):
        raise HTTPException(status_code=401, detail="用户名或密码无效")
    
    # 生成并返回JWT令牌
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": form_data.username},
        expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

5. 保护API路由 #

创建JWT验证依赖，保护需要授权访问的接口：

from fastapi.security import OAuth2PasswordBearer
from jose import JWTError

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=401,
        detail="无法验证用户凭证",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        # 解析JWT令牌
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if not username:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    
    # 可选：实时校验用户是否仍在AD中（比如用户被禁用的情况）
    # if not authenticate_ad_user(username, ""):  # 可通过AD查询用户状态替代密码验证
    #     raise credentials_exception
    
    return username

# 示例受保护接口
@app.get("/profile")
async def get_user_profile(current_user: str = Depends(get_current_user)):
    return {"username": current_user, "status": "已授权访问"}

关键注意事项 #

• AD端口与SSL：默认用389端口（非SSL），若需加密连接，改用636端口并将AD_SERVER改为ldaps://your-ad-server，确保AD服务器开启LDAPS服务。
• 服务账号权限：绑定AD的服务账号必须拥有读取AD用户信息的权限，否则会出现验证失败。
• JWT密钥安全：生产环境中SECRET_KEY应使用随机生成的强字符串，可通过环境变量读取，避免硬编码。
• 用户状态校验：若需实时校验用户是否被禁用/删除，可在get_current_user中添加AD用户查询逻辑，而非仅依赖JWT。

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