285 lines
9.3 KiB
Python
285 lines
9.3 KiB
Python
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 = current_app.config["FRONTEND_CALLBACK_PATH"]
|
|
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 = current_app.config["FRONTEND_CALLBACK_PATH"]
|
|
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.") |