DF 前端                                Superset
  │                                      │
  │  ① 用户在 DF 数据面板点击"Login via Superset"
  │  window.open 打开弹窗                │
  │ ───────────────────────────────────> │  /df-sso-bridge/?df_origin=...
  │                                      │
  │                                      │  ② 未登录 → 重定向到 /login/ → SSO 认证
  │                                      │     已登录 → 直接跳到 ④
  │                                      │
  │                                      │  ③ SSO 认证成功 → Superset 创建 Session
  │                                      │     重定向回 /df-sso-bridge/
  │                                      │
  │                                      │  ④ 为当前用户签发 JWT
  │                                      │     通过 postMessage 发送给 DF
  │                                      │     弹窗自动关闭
  │  ⑤ DF 收到 JWT                       │
  │ <─────────────────────────────────── │
  │  ⑥ Token 存入 TokenStore（session 级） │
  │  ⑦ 后续 API 调用使用 JWT              │
  │ ───────────────────────────────────> │  Authorization: Bearer <jwt>

DF 后端                                Superset
  │                                      │
  │  ① 用户在 DF 完成 SSO 登录           │
  │  ② DF 后端自动发起换票请求            │
  │ ───────────────────────────────────> │  POST /api/v1/df-token-exchange/
  │    body: {sso_access_token: "..."}   │    body: {"sso_access_token": "..."}
  │                                      │
  │                                      │  ③ 用 SSO token 调 IdP userinfo 验证
  │                                      │  ④ 查找 Superset 用户 → 签发 JWT
  │                                      │
  │  ⑤ DF 收到 Superset JWT              │
  │ <─────────────────────────────────── │  {"access_token": "...", ...}
  │  ⑥ Token 存入 TokenStore（自动）      │
  │  ⑦ 后续 API 调用使用 JWT              │
  │ ───────────────────────────────────> │  Authorization: Bearer <jwt>

# 用一个有效的 SSO access_token 测试换票端点
curl -X POST http://SUPERSET地址:8088/api/v1/df-token-exchange/ \
  -H "Content-Type: application/json" \
  -d '{"sso_access_token": "<your_sso_access_token>"}'

# 预期返回：
# {"access_token": "eyJ...", "refresh_token": "eyJ...", "expires_in": 3600, "user": {...}}

# Superset 端
DF_EXCHANGE_SHARED_SECRET=your-random-secret-here

# DF 端 .env
PLG_SUPERSET_EXCHANGE_SECRET=your-random-secret-here

superset_config.py      ← Superset 主配置（导入 oauth_config）
oauth_config.py         ← SSO 认证 + DF 桥接（独立文件，便于维护）

import logging
import os
import secrets
import json
import requests
from urllib.parse import quote, urlparse

from flask_appbuilder.security.manager import AUTH_OAUTH
from flask_appbuilder import BaseView, expose
from flask import g, request, Response, redirect, session, render_template_string
from flask_login import current_user
from superset.security import SupersetSecurityManager
from superset import db

logger = logging.getLogger(__name__)


class SsoHandler:
    """从 SSO userinfo 端点解析用户详情。每接一个新 IdP 只需继承此类并覆盖两个字段。"""
    handler_name = 'SSO系统'
    userinfo_url = '<YOUR_SSO_USERINFO_URL>'     # ← 替换为你的 SSO userinfo 端点
    role_mapping = {                              # ← 替换为你的 SSO 角色 → Superset 角色映射
        'admin': 'Admin',
        'analyst': 'Alpha',                       # Alpha 有 SQL Lab 权限，可通过 DF 拉取数据
        'viewer': 'Gamma',                        # ⚠️ Gamma 没有 SQL Lab 权限，见下方说明
    }

    @classmethod
    def parse_user_details(cls, access_token):
        """使用 SSO access_token 调用 userinfo 端点获取用户信息。"""
        resp = requests.get(
            cls.userinfo_url,
            headers={'Authorization': f'Bearer {access_token}'},
            verify=False,    # 如果 SSO 使用自签名证书则设为 False
            timeout=10,
        )
        resp.raise_for_status()
        data = resp.json()
        logger.info(f"{cls.handler_name}-用户信息: {data}")

        username = data.get('preferred_username') or data.get('sub')
        if not username:
            logger.error("用户名 username 为空！")
            return None

        full_name = data.get('name') or username
        email = data.get('email') or f"{username}@example.com"
        name_parts = full_name.split(' ', 1)

        return {
            'username': username,
            'email': email,
            'first_name': name_parts[0],
            'last_name': name_parts[1] if len(name_parts) > 1 else '',
            'sub': str(data.get('sub')) if data.get('sub') is not None else None,
            'sso_roles': data.get('sso_roles', []),
            'user_code': data.get('user_code', None),
        }

class TestSsoHandler(SsoHandler):
    handler_name = '测试系统'
    userinfo_url = 'https://test-sso.example.com/api/v1/oauth2/userinfo'

# Provider → Handler 映射
PROVIDER_HANDLERS = {
    'prod-sso': SsoHandler,
    'test-sso': TestSsoHandler,
}

class CustomSsoSecurityManager(SupersetSecurityManager):

    def oauth_user_info(self, provider, response=None):
        """从 OAuth 响应中解析用户信息。"""
        access_token = response.get('access_token') if response else None
        handler = PROVIDER_HANDLERS.get(provider)

        if not handler or not access_token:
            logger.error(f"缺少 provider handler 或 token: {provider}")
            return super().oauth_user_info(provider, response)

        try:
            user_details = handler.parse_user_details(access_token)
            user_details['active_provider'] = provider
            return user_details
        except Exception as e:
            logger.error(f"OAuth 用户信息同步失败: {str(e)}", exc_info=True)
            return None

    def auth_user_oauth(self, userinfo):
        """创建或更新用户，并同步 SSO 角色。"""
        username = userinfo.get('username')
        provider = userinfo.get('active_provider')
        handler = PROVIDER_HANDLERS.get(provider)
        sso_roles = userinfo.get('sso_roles', [])

        # 检查用户是否有任何已映射的角色
        has_mapped_role = any(role in handler.role_mapping for role in sso_roles)
        if not has_mapped_role:
            from flask import flash
            flash("登录失败：您的账号尚未被分配 Superset 访问权限，请联系管理员。", "danger")
            return None

        user = self.find_user(username=username)
        if not user:
            if self.auth_user_registration:
                user = super().auth_user_oauth(userinfo)
            else:
                logger.warning(f"用户 {username} 不存在且未开启自动注册")
                return None
        else:
            # 更新已有用户的基础信息
            user.first_name = userinfo.get('first_name', user.first_name)
            user.last_name = userinfo.get('last_name', user.last_name)
            user.email = userinfo.get('email', user.email)

        # 同步角色
        if sso_roles:
            self._sync_roles(user, sso_roles, handler.role_mapping)

        return user

    def _sync_roles(self, user, sso_roles, mapping):
        """将 SSO 角色映射为 Superset 角色。本地 Admin 身份始终保留。"""
        target_names = set()
        for sr in sso_roles:
            mapped = mapping.get(sr)
            if mapped:
                target_names.add(mapped)
            else:
                logger.info(f"SSO 角色 '{sr}' 未在映射表中定义，跳过")

        # 保护：如果用户已是 Admin，强制保留
        if any(r.name == 'Admin' for r in user.roles):
            target_names.add('Admin')

        new_roles = []
        for name in target_names:
            role = self.find_role(name)
            if role:
                new_roles.append(role)
            else:
                logger.warning(f"Superset 中找不到角色: {name}")

        if not new_roles:
            return

        try:
            user.roles = new_roles
            db.session.merge(user)
            db.session.commit()
            logger.info(f"用户 {user.username} 角色已同步: {[r.name for r in new_roles]}")
        except Exception as e:
            db.session.rollback()
            logger.error(f"角色同步失败: {e}")

_SSO_BRIDGE_TEMPLATE = """<!DOCTYPE html>
<html><head><meta charset="utf-8"><title>SSO Bridge</title></head>
<body>
<p>正在同步登录状态...</p>
<script nonce="{{ csp_nonce }}">
(function() {
    var payload = {{ payload | tojson }};
    var targetOrigin = {{ target_origin | tojson }};
    if (window.opener) {
        window.opener.postMessage(payload, targetOrigin);
        setTimeout(function() { window.close(); }, 500);
    } else {
        document.body.textContent = '登录成功，请关闭此窗口并返回 Data Formulator。';
    }
})();
</script>
</body></html>"""

# 允许接收 JWT 的 DF 前端 origin 白名单。
# 可通过环境变量 DF_ALLOWED_ORIGINS 追加（逗号分隔）。
_DEFAULT_ALLOWED_ORIGINS = {
    "http://localhost:5567",
    "http://127.0.0.1:5567",
    "http://localhost:5173",
    "http://127.0.0.1:5173",
    # 生产环境在此添加或通过环境变量设置
    # "https://your-df-domain.com",
}


def _normalise_origin(raw):
    """规范化浏览器 origin；非法或非 origin 格式返回空字符串。"""
    raw = (raw or "").strip()
    parsed = urlparse(raw)
    if parsed.scheme not in ("http", "https") or not parsed.netloc:
        return ""
    if parsed.username or parsed.password:
        return ""
    if parsed.path not in ("", "/") or parsed.params or parsed.query or parsed.fragment:
        return ""
    return f"{parsed.scheme.lower()}://{parsed.netloc.lower()}"


def _get_allowed_origins():
    origins = {_normalise_origin(o) for o in _DEFAULT_ALLOWED_ORIGINS}
    env_origins = os.environ.get("DF_ALLOWED_ORIGINS", "")
    for raw in env_origins.split(","):
        origin = _normalise_origin(raw)
        if origin:
            origins.add(origin)
    origins.discard("")
    return origins


class SSOBridgeView(BaseView):
    route_base = "/df-sso-bridge"

    @staticmethod
    def _is_real_logged_in_user():
        """
        判断当前请求是否来自一个真正登录过的用户。
        当 Superset 启用 PUBLIC_ROLE_LIKE 时，current_user.is_authenticated
        对匿名用户也可能返回 True，必须检查 session._user_id。
        """
        if not session.get("_user_id"):
            return False
        if getattr(current_user, "is_anonymous", True):
            return False
        if not getattr(current_user, "is_authenticated", False):
            return False
        if not getattr(current_user, "id", None):
            return False
        username = getattr(current_user, "username", "") or ""
        if username.lower() in ("public", "anonymous", "guest", ""):
            return False
        return True

    @staticmethod
    def _validate_origin(raw_origin):
        """校验 df_origin 是否在白名单中。返回规范化的合法 origin 或空字符串。"""
        origin = _normalise_origin(raw_origin)
        return origin if origin in _get_allowed_origins() else ""

    @staticmethod
    def _safe_next_path(raw_path):
        """只允许站内相对路径进入 login next 参数，防止 open redirect。"""
        next_path = (raw_path or "/").rstrip("?").replace("\\", "/")
        parsed = urlparse(next_path)
        if (
            not next_path.startswith("/")
            or next_path.startswith("//")
            or parsed.scheme
            or parsed.netloc
        ):
            return "/df-sso-bridge/"
        return next_path

    @expose("/", methods=["GET"])
    def df_sso_bridge(self):
        logger.info(
            "进入 df-sso-bridge. is_anonymous=%s, session._user_id=%s, username=%s",
            getattr(current_user, "is_anonymous", "N/A"),
            session.get("_user_id"),
            getattr(current_user, "username", "N/A"),
        )

        # ── 未登录：重定向到 Superset 登录页 ──
        if not self._is_real_logged_in_user():
            next_url = self._safe_next_path(request.full_path)
            login_redirect = f"/login/?next={quote(next_url)}"
            logger.info("用户未真实登录，重定向到: %s", login_redirect)
            return redirect(login_redirect)

        logger.info("已验证真实用户: id=%s, username=%s", current_user.id, current_user.username)

        # ── 校验 df_origin（白名单）──
        df_origin = self._validate_origin(request.args.get("df_origin"))
        if not df_origin:
            logger.warning("df_origin 校验失败: %s", request.args.get("df_origin"))
            return Response(
                "Invalid df_origin. 请确认 DF 前端 origin 已加入白名单。",
                status=400,
                mimetype="text/plain",
            )

        # ── 签发 JWT ──
        from flask_jwt_extended import create_access_token, create_refresh_token

        user_id_str = str(current_user.id)
        additional_claims = {
            "user": {
                "id": current_user.id,
                "username": current_user.username,
                "first_name": getattr(current_user, "first_name", "") or "",
                "last_name": getattr(current_user, "last_name", "") or "",
            }
        }

        access_token = create_access_token(
            identity=user_id_str, fresh=True, additional_claims=additional_claims,
        )
        refresh_token = create_refresh_token(
            identity=user_id_str, additional_claims=additional_claims,
        )

        logger.info("JWT 已为用户 %s (id=%s) 生成", current_user.username, current_user.id)

        # ── 使用 Jinja2 模板渲染 postMessage 页面 ──
        payload = {
            "type": "df-sso-auth",
            "access_token": access_token,
            "refresh_token": refresh_token,
            "user": additional_claims["user"],
        }
        csp_nonce = getattr(g, "csp_nonce", "") or secrets.token_urlsafe(16)

        html = render_template_string(
            _SSO_BRIDGE_TEMPLATE,
            payload=payload,
            target_origin=df_origin,
            csp_nonce=csp_nonce,
        )
        return Response(html, mimetype="text/html")

OAUTH_CONFIG = {
    'AUTH_TYPE': AUTH_OAUTH,
    'AUTH_USER_REGISTRATION': True,
    'AUTH_USER_REGISTRATION_ROLE': "Public",
    'AUTH_OAUTH_ROLES_SYNC': True,
    'AUTH_OAUTH_ROLES_UPDATE': True,
    'CUSTOM_SECURITY_MANAGER': CustomSsoSecurityManager,
    'OAUTH_PROVIDERS': [
        {
            'name': '<YOUR_PROVIDER_NAME>',       # ← provider 名称，对应 PROVIDER_HANDLERS 的 key
            'token_key': 'access_token',
            'icon': 'fa-key',
            'remote_app': {
                'client_id': '<YOUR_CLIENT_ID>',
                'client_secret': '<YOUR_CLIENT_SECRET>',
                'server_metadata_url': '<YOUR_DISCOVERY_URL>',
                'client_kwargs': {
                    'scope': 'openid profile email',
                    'verify': False,              # 自签名证书时设为 False
                },
            },
        }
    ]
}

{
    "type": "df-sso-auth",
    "access_token": "eyJhbGci...",
    "refresh_token": "eyJhbGci...",
    "user": {
        "id": 1,
        "username": "zhangsan",
        "first_name": "三",
        "last_name": "张"
    }
}

try:
    from oauth_config import OAUTH_CONFIG
    globals().update(OAUTH_CONFIG)
    print("OAuth 配置加载成功")
except ImportError as e:
    print(f"OAuth 配置加载失败: {e}")

import os
import secrets
import requests
from flask_appbuilder import BaseView, expose
from flask import request, jsonify, current_app
from flask_jwt_extended import create_access_token, create_refresh_token


def _verify_exchange_secret():
    """校验共享密钥。未配置 DF_EXCHANGE_SHARED_SECRET 时放行（开发模式）。"""
    expected = os.environ.get("DF_EXCHANGE_SHARED_SECRET", "").strip()
    if not expected:
        return True
    provided = (request.headers.get("X-DF-Exchange-Secret") or "").strip()
    return secrets.compare_digest(expected, provided)


class TokenExchangeView(BaseView):
    route_base = "/api/v1/df-token-exchange"

    @expose("/", methods=["POST"])
    def exchange(self):
        """
        接收 DF 后端传来的 SSO access_token，验证后签发 Superset JWT。
        请求体: {"sso_access_token": "<IdP access token>"}
        """
        # ⓪ 验证调用方密钥（生产环境必须设置 DF_EXCHANGE_SHARED_SECRET）
        if not _verify_exchange_secret():
            return jsonify({"error": "unauthorized", "message": "Invalid exchange secret"}), 403

        data = request.get_json(force=True)
        sso_token = data.get("sso_access_token")
        if not sso_token:
            return jsonify({"error": "missing_token"}), 400

        # ① 用 SSO token 向 IdP userinfo 端点验证身份（复用 SsoHandler 的 URL）
        try:
            from oauth_config import ExampleSsoHandler  # 或你的 SsoHandler 类名
            resp = requests.get(
                ExampleSsoHandler.userinfo_url,
                headers={"Authorization": f"Bearer {sso_token}"},
                verify=False,
                timeout=5,
            )
            resp.raise_for_status()
            user_info = resp.json()
        except Exception:
            return jsonify({"error": "sso_token_invalid"}), 401

        # ② 查找 Superset 中对应的用户
        username = user_info.get("preferred_username") or user_info.get("username")
        if not username:
            return jsonify({"error": "no_username"}), 401

        sm = current_app.appbuilder.sm
        user = sm.find_user(username=username)
        if not user or not user.is_active:
            return jsonify({"error": "user_not_in_superset"}), 403

        # ③ 签发 Superset JWT（含用户信息和过期时间）
        user_id_str = str(user.id)
        additional_claims = {
            "user": {
                "id": user.id,
                "username": user.username,
                "first_name": getattr(user, "first_name", "") or "",
                "last_name": getattr(user, "last_name", "") or "",
            }
        }

        return jsonify({
            "access_token": create_access_token(
                identity=user_id_str, fresh=True, additional_claims=additional_claims,
            ),
            "refresh_token": create_refresh_token(
                identity=user_id_str, additional_claims=additional_claims,
            ),
            "expires_in": 3600,
            "user": additional_claims["user"],
        })

# Superset 端
DF_EXCHANGE_SHARED_SECRET=your-random-secret-here

# DF 端 .env
PLG_SUPERSET_EXCHANGE_SECRET=your-random-secret-here

TALISMAN_ENABLED = False  # Bridge 端点需要内联 <script>，必须关闭 Talisman


def FLASK_APP_MUTATOR(app):
    import logging
    _jwt_logger = logging.getLogger(__name__)

    # ──── 1. 注册 SSO Bridge 视图 ────
    try:
        from oauth_config import SSOBridgeView
        app.extensions["appbuilder"].add_view_no_menu(SSOBridgeView())
        print("SSO Bridge View registered successfully!")
    except Exception as e:
        print(f"SSO Bridge View registration failed: {e}")

    # ──── 2. 注册 Token Exchange 视图（可选，不需要可删除此段）────
    try:
        app.extensions["appbuilder"].add_view_no_menu(TokenExchangeView())
        print("Token Exchange View registered successfully!")
    except Exception as e:
        print(f"Token Exchange View registration failed: {e}")

    # ──── 3. JWT → g.user 同步中间件 ────
    @app.before_request
    def _ensure_user_from_jwt():
        from flask import g, request as req
        from flask_login import current_user, login_user

        # 如果已经有有效的 session 用户，直接同步到 g.user
        if getattr(current_user, "is_authenticated", False) \
           and not getattr(current_user, "is_anonymous", True):
            if not hasattr(g, "user") or g.user is None \
               or getattr(g.user, "is_anonymous", True):
                g.user = current_user._get_current_object()
            return

        # 尝试从 Authorization: Bearer <jwt> 头解析用户
        auth_header = req.headers.get("Authorization", "")
        if not auth_header.lower().startswith("bearer "):
            return

        try:
            from flask_jwt_extended import verify_jwt_in_request, get_jwt_identity
            verify_jwt_in_request(optional=True)
            identity = get_jwt_identity()
            if not identity:
                return

            sm = app.appbuilder.sm
            user = sm.get_user_by_id(int(identity))
            if user and getattr(user, "is_active", False):
                login_user(user)
                g.user = user
                _jwt_logger.debug(
                    "JWT->g.user sync OK: user_id=%s, username=%s",
                    user.id, user.username,
                )
        except Exception as exc:
            _jwt_logger.debug("JWT->g.user sync skipped: %s", exc)

    print("JWT->g.user middleware registered successfully!")

# Cookie 安全配置（弹窗模式需要 Lax）
SESSION_COOKIE_SAMESITE = "Lax"   # 不能为 Strict，否则弹窗无法携带 cookie
SESSION_COOKIE_SECURE = False      # HTTP 环境设为 False，HTTPS 环境设为 True
SESSION_COOKIE_HTTPONLY = True

# 如果启用了 PUBLIC_ROLE_LIKE，Bridge 需要区分匿名和真实用户
PUBLIC_ROLE_LIKE = "Gamma"         # 可选：让未登录用户拥有只读权限

# CORS 配置（DF 需要跨域调用 Superset API）
ENABLE_CORS = True
CORS_OPTIONS = {
    'supports_credentials': True,
    'allow_headers': ['*'],
    'resources': ['*'],
    'origins': ['*']               # 生产环境请限制为 DF 的实际域名
}

# Superset 地址（必填）
PLG_SUPERSET_URL=http://你的SUPERSET地址:8088/

PLG_SUPERSET_SSO_LOGIN_URL=https://superset.example.com/df-sso-bridge/

# ❌ 错误示例 — 已登录用户被 /login/ 重定向到 welcome 页，DF 无法获取 token
PLG_SUPERSET_SSO_LOGIN_URL=http://superset:8088/login/?next=/df-sso-bridge/

# ✅ 正确 — 直接指向 Bridge 端点
PLG_SUPERSET_SSO_LOGIN_URL=http://superset:8088/df-sso-bridge/

AUTH_PROVIDER=oidc
OIDC_ISSUER_URL=https://your-idp/oauth2
OIDC_CLIENT_ID=df-client-id
OIDC_CLIENT_SECRET=df-client-secret

http://SUPERSET地址:8088/df-sso-bridge/?df_origin=http://test

curl -H "Authorization: Bearer <access_token>" http://SUPERSET地址:8088/api/v1/me/

curl -X POST http://SUPERSET地址:8088/api/v1/df-token-exchange/ \
  -H "Content-Type: application/json" \
  -d '{"sso_access_token": "<valid_sso_access_token>"}'