import base64 from flask import ( Blueprint, make_response, redirect, request, jsonify, session, current_app, ) import jwt import requests oauth_bp = Blueprint("oauth", __name__) # 客户端 ID # CLIENT_ID = "A17336FC-A8D7-4CEA-8777-384BFEABABC1" # BASE64_CLIENT = "Basic QTE3MzM2RkMtQThENy00Q0VBLTg3NzctMzg0QkZFQUJBQkMxOlU3LUMxQlpRZEw4VjVjLnp+cUFfNjRTTzJsOUYzSjBv" # # SSO 授权服务器地址 # # "https://ssoalpha.dvb.corpinter.net/" # HOST = "https://ssoalpha.dvb.corpinter.net.cn/" # # 授权成功后，SSO 服务器重定向回你的应用的地址 (必须与 SSO 服务器上配置的一致) # REDIRECT_URL = "http://localhost:5221/report" # # 用户登录成功后，最终重定向到的前端页面地址 # DOMAIN = "http://localhost:3001" @oauth_bp.route("/error") def error(): """错误页面，用于显示认证失败信息""" msg = request.args.get("msg", "Unknown error") return f"Error: {msg}" # @oauth_bp.route("/report") @oauth_bp.route("/incident/authorized") def authorized(): """ SSO 回调接口。 接收 SSO 服务器返回的授权码 (code)，并使用它来兑换访问令牌 (access_token)。 """ CLIENT_ID = current_app.config["CLIENT_ID"] CLIENT_PASSWORD = current_app.config["CLIENT_PASSWORD"] HOST = current_app.config["SS0_HOST"] # 建议检查是否是笔误（SSO_HOST） REDIRECT_URL = current_app.config["REDIRECT_URL"] DOMAIN = current_app.config["DOMAIN"] code = request.args.get("code") state = request.args.get("state", "/dashboard") # 默认值设为/dashboard if not code: # return redirect(f"/error?msg=No authorization code provided.") return redirect(f"/incident/error?msg=No authorization code provided.") # 1. 准备请求体，用授权码兑换 Token body = { "grant_type": "authorization_code", "code": code, "redirect_uri": REDIRECT_URL, } auth_str = f"{CLIENT_ID}:{CLIENT_PASSWORD}" # 2. 准备请求头 headers = { "Authorization": "Basic "+base64.b64encode(auth_str.encode("utf-8")).decode("utf-8"), "Content-Type": "application/x-www-form-urlencoded", } try: # 3. 发送 POST 请求到 SSO 服务器的 Token 端点 response = requests.post(f"{HOST}v1/token", data=body, headers=headers) response.raise_for_status() response_data = response.json() access_token = response_data.get("access_token") # print('access_token', access_token) if not access_token: # return redirect(f"/error?msg=Failed to obtain access token. Response: {response_data}") return redirect(f"/incident/error?msg=Failed to obtain access token. Response: {response_data}") # 4. 解析 JWT Token decoded_jwt = jwt.decode(access_token, options={"verify_signature": False}) # 5. 验证和清理 state 参数 (防止开放重定向攻击) safe_state = "/callback" print(f"Redirecting to: {DOMAIN}{safe_state}") # 6. 设置 Cookie resp = make_response(redirect(f"{DOMAIN}{safe_state}")) # 关键：设置 Cookie 域名和属性 # domain = ".localhost" # 让所有 localhost 子域都能读取 domain = None # 上线时改为None # 设置 access_token resp.set_cookie( "authorization", access_token, max_age=3500, path="/", domain=domain, httponly=False, # 前端需要读取 samesite="Lax", # 允许跨站 secure=False, ) # 开发环境不需要HTTPS # 设置登录名 login_name = decoded_jwt.get("sub", "") resp.set_cookie( "loginName", login_name, max_age=3500, path="/", domain=domain, httponly=False, samesite="Lax", secure=False, ) # 设置用户名 (直接存储，不需要Base64编码) user_name = decoded_jwt.get("name", "") print("user_name", user_name) resp.set_cookie( "userName", user_name, # 直接存储，前端处理编码 max_age=3500, path="/", domain=domain, httponly=False, samesite="Lax", secure=False, ) return resp except requests.exceptions.RequestException as e: print(f"Error exchanging code for token: {e}") return redirect(f"/error?msg=Error communicating with SSO server.") except jwt.InvalidTokenError as e: print(f"Error decoding JWT: {e}") return redirect(f"/error?msg=Invalid access token received.") @oauth_bp.route("/logout", methods=["GET", "POST"]) def logout(): """登出接口""" # 清除 session session.clear() # print('logout66677') # 返回标准的 JSON 响应 return jsonify({"code": 200, "message": "退出登录成功", "data": {}}), 200 @oauth_bp.route("/login") def authorization(): """ 授权入口。 将用户浏览器重定向到 SSO 服务器的授权页面。 """ # print("login123") CLIENT_ID = current_app.config["CLIENT_ID"] HOST = current_app.config["SS0_HOST"] REDIRECT_URL = current_app.config["REDIRECT_URL"] # 1. 构造重定向到 SSO 授权页面的 URL auth_url = ( f"{HOST}v1/auth" f"?response_type=code" f"&client_id={CLIENT_ID}" f"&scope=groups+openid+email+profile" f"&redirect_uri={REDIRECT_URL}" ) # print(auth_url) # 2. 执行重定向 return redirect(auth_url) @oauth_bp.route("/report") def authorized_test(): """ SSO 回调接口。 接收 SSO 服务器返回的授权码 (code)，并使用它来兑换访问令牌 (access_token)。 """ CLIENT_ID = current_app.config["CLIENT_ID"] CLIENT_PASSWORD = current_app.config["CLIENT_PASSWORD"] HOST = current_app.config["SS0_HOST"] # 建议检查是否是笔误（SSO_HOST） REDIRECT_URL = current_app.config["REDIRECT_URL"] DOMAIN = current_app.config["DOMAIN"] code = request.args.get("code") state = request.args.get("state", "/dashboard") # 默认值设为/dashboard if not code: # return redirect(f"/error?msg=No authorization code provided.") return redirect(f"/incident/error?msg=No authorization code provided.") # 1. 准备请求体，用授权码兑换 Token body = { "grant_type": "authorization_code", "code": code, "redirect_uri": REDIRECT_URL, } auth_str = f"{CLIENT_ID}:{CLIENT_PASSWORD}" # 2. 准备请求头 headers = { "Authorization": "Basic "+base64.b64encode(auth_str.encode("utf-8")).decode("utf-8"), "Content-Type": "application/x-www-form-urlencoded", } try: # 3. 发送 POST 请求到 SSO 服务器的 Token 端点 response = requests.post(f"{HOST}v1/token", data=body, headers=headers) response.raise_for_status() response_data = response.json() access_token = response_data.get("access_token") # print('access_token', access_token) if not access_token: # return redirect(f"/error?msg=Failed to obtain access token. Response: {response_data}") return redirect(f"/incident/error?msg=Failed to obtain access token. Response: {response_data}") # 4. 解析 JWT Token decoded_jwt = jwt.decode(access_token, options={"verify_signature": False}) # 5. 验证和清理 state 参数 (防止开放重定向攻击) safe_state = "/callback" print(f"Redirecting to: {DOMAIN}{safe_state}") # 6. 设置 Cookie resp = make_response(redirect(f"{DOMAIN}{safe_state}")) # 关键：设置 Cookie 域名和属性 domain = ".localhost" # 让所有 localhost 子域都能读取 # 设置 access_token resp.set_cookie( "authorization", access_token, max_age=3500, path="/", domain=domain, httponly=False, # 前端需要读取 samesite="Lax", # 允许跨站 secure=False, ) # 开发环境不需要HTTPS # 设置登录名 login_name = decoded_jwt.get("sub", "") resp.set_cookie( "loginName", login_name, max_age=3500, path="/", domain=domain, httponly=False, samesite="Lax", secure=False, ) # 设置用户名 (直接存储，不需要Base64编码) user_name = decoded_jwt.get("name", "") print("user_name", user_name) resp.set_cookie( "userName", user_name, # 直接存储，前端处理编码 max_age=3500, path="/", domain=domain, httponly=False, samesite="Lax", secure=False, ) return resp except requests.exceptions.RequestException as e: print(f"Error exchanging code for token: {e}") return redirect(f"/error?msg=Error communicating with SSO server.") except jwt.InvalidTokenError as e: print(f"Error decoding JWT: {e}") return redirect(f"/error?msg=Invalid access token received.")