如何让Flask AppBuilder接受服务账号的JWT Bearer令牌？

要同时让终端用户的OIDC浏览器认证和服务账号的客户端凭证流JWT认证都能通过FAB的权限校验（兼容@protect注解），核心是扩展FAB的SecurityManager，让它在认证流程中优先检查Bearer Token，再回退到原OIDC会话逻辑。

步骤1：复用OIDC配置并添加服务账号参数 #

确保你的FAB应用已配置好OIDC（不管用flask-oidc还是fab-oidc），额外添加服务账号所需的验证参数到配置中：

# app.config 或 superset_config.py
OIDC_ISSUER = "https://your-idp-domain.com"
OIDC_CLIENT_ID = "frontend-user-client-id"  # 原终端用户的客户端ID
OIDC_CLIENT_SECRET = "frontend-user-client-secret"
# 服务账号相关配置
OIDC_SERVICE_AUDIENCE = "your-api-audience"  # IDP中配置的API受众，或服务账号的client_id
OIDC_JWKS_URL = f"{OIDC_ISSUER}/.well-known/jwks.json"

步骤2：自定义SecurityManager #

继承FAB的OIDCSecurityManager，重写认证逻辑，添加Bearer Token的校验逻辑：

from flask import request
from flask_appbuilder.security.manager import OIDCSecurityManager
import jwt
from jwt.exceptions import InvalidTokenError
from jwt import PyJWKClient
import logging

logger = logging.getLogger(__name__)

class DualAuthSecurityManager(OIDCSecurityManager):
    def __init__(self, appbuilder):
        super().__init__(appbuilder)
        # 复用应用已有的OIDC实例（如果用fab-oidc）
        self.oidc = self.appbuilder.app.extensions.get("oidc")
        # 从配置读取服务账号验证参数
        self.jwt_issuer = self.appbuilder.app.config["OIDC_ISSUER"]
        self.jwt_audience = self.appbuilder.app.config["OIDC_SERVICE_AUDIENCE"]
        self.jwks_url = self.appbuilder.app.config["OIDC_JWKS_URL"]
        # 缓存JWKS客户端，避免重复请求
        self.jwks_client = PyJWKClient(self.jwks_url)

    def _validate_bearer_token(self):
        """验证请求头中的JWT Bearer Token，返回对应的FAB用户"""
        auth_header = request.headers.get("Authorization", "")
        if not auth_header.startswith("Bearer "):
            return None
        
        token = auth_header.split(" ")[1]
        try:
            # 获取IDP的签名密钥
            signing_key = self.jwks_client.get_signing_key_from_jwt(token)
            # 解码并验证JWT的有效性
            payload = jwt.decode(
                token,
                signing_key.key,
                algorithms=["RS256"],  # 根据IDP实际算法调整
                issuer=self.jwt_issuer,
                audience=self.jwt_audience
            )
            # 从JWT payload中提取用户标识（通常是sub字段，或自定义username）
            username = payload.get("sub", payload.get("username"))
            if not username:
                logger.error("JWT payload missing user identifier")
                return None
            
            # 查找或自动创建FAB用户（根据业务需求选择是否自动创建）
            user = self.find_user(username=username)
            if not user:
                # 自动创建服务账号用户，分配最小权限角色
                user = self.add_user(
                    username=username,
                    first_name="Service",
                    last_name="Account",
                    email=f"{username}@service.local",
                    role=self.find_role("API_User")  # 需提前创建对应角色
                )
            return user
        except InvalidTokenError as e:
            logger.error(f"Invalid JWT token: {str(e)}")
            return None

    def is_authenticated(self):
        """优先检查Bearer Token，无则走原OIDC会话认证"""
        token_user = self._validate_bearer_token()
        if token_user:
            self.set_current_user(token_user)
            return True
        # 回退到终端用户的OIDC会话认证逻辑
        return super().is_authenticated()

步骤3：注册自定义SecurityManager #

在FAB应用初始化时指定使用自定义的SecurityManager：

# Flask AppBuilder 基础应用示例
from flask import Flask
from flask_appbuilder import AppBuilder

app = Flask(__name__)
app.config.from_object("your_config_module")
# 使用自定义SecurityManager
appbuilder = AppBuilder(app, security_manager_class=DualAuthSecurityManager)

# Superset 中配置（superset_config.py）
CUSTOM_SECURITY_MANAGER = "your_module.DualAuthSecurityManager"

关键说明 #

1. 兼容@protect注解：FAB的@protect注解会调用is_authenticated方法，我们重写后的方法已覆盖两种认证场景，无需额外添加@oidc.accept_token注解。
2. 用户映射逻辑：确保JWT中的用户标识（如sub）与FAB中的用户用户名对应，自动创建用户的逻辑需根据权限需求调整，避免赋予过高权限。
3. JWT验证细节：必须匹配IDP的签名算法（如RS256）、issuer和audience，否则会验证失败。
4. 缓存优化：JWKS客户端会自动缓存密钥，避免每次请求都去IDP拉取，减少延迟。

测试服务账号认证 #

1. 从IDP获取服务账号的JWT令牌（客户端凭证流）：

curl -X POST "${OIDC_ISSUER}/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "grant_type=client_credentials&client_id=your-service-client-id&client_secret=your-service-client-secret&scope=api-access"

2. 使用令牌调用FAB受保护的API：

curl -X GET "https://your-fab-app/api/v1/your-resource" \
  -H "Authorization: Bearer your-access-token"

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