Ubuntu
320 lines
12 KiB
Python
320 lines
12 KiB
Python
"""SSO 单点登录中心 —— FastAPI"""
|
||
import os
|
||
from datetime import datetime, timedelta, timezone
|
||
from typing import Optional
|
||
|
||
from fastapi import FastAPI, Form, HTTPException, Request, Response
|
||
from fastapi.responses import HTMLResponse, RedirectResponse
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
from jose import jwt, JWTError
|
||
from passlib.context import CryptContext
|
||
|
||
import database
|
||
from database import SessionLocal
|
||
import models
|
||
from sso_config import (
|
||
SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES,
|
||
DEFAULT_USERS, COOKIE_DOMAIN, COOKIE_SAMESITE,
|
||
)
|
||
|
||
app = FastAPI(title="SSO Center")
|
||
|
||
# CORS：允许 blog 前端 8888 调用
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=[
|
||
"http://129.211.173.87:8888",
|
||
"http://129.211.173.87:8889",
|
||
"http://129.211.173.87:8892",
|
||
"http://129.211.173.87:8895", # happy-games
|
||
"http://129.211.173.87:9000",
|
||
"http://129.211.173.87",
|
||
],
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*"],
|
||
)
|
||
|
||
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
||
|
||
# ── 初始化数据库 ──────────────────────────
|
||
models.Base.metadata.create_all(bind=database.engine)
|
||
db = database.SessionLocal()
|
||
try:
|
||
if db.query(models.User).count() == 0:
|
||
for uname, udata in DEFAULT_USERS.items():
|
||
hashed = pwd_context.hash(udata["password"])
|
||
db.add(models.User(username=uname, password=hashed))
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
|
||
# ── 加载静态登录页 ────────────────────────
|
||
_static_dir = os.path.join(os.path.dirname(__file__), "static")
|
||
_login_html_path = os.path.join(_static_dir, "login.html")
|
||
with open(_login_html_path, "r", encoding="utf-8") as f:
|
||
LOGIN_HTML = f.read()
|
||
|
||
|
||
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
|
||
to_encode = data.copy()
|
||
expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
|
||
to_encode.update({"exp": expire})
|
||
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
||
|
||
|
||
def verify_user(username: str, password: str) -> bool:
|
||
db = database.SessionLocal()
|
||
try:
|
||
user = db.query(models.User).filter(models.User.username == username).first()
|
||
if not user:
|
||
return False
|
||
return pwd_context.verify(password, user.password)
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
# ── 登录页 ──────────────────────────────
|
||
|
||
@app.get("/", response_class=HTMLResponse)
|
||
def login_page(request: Request):
|
||
redirect = request.query_params.get("redirect", "")
|
||
# 如果已登录（有 Cookie），直接跳转回原系统；无 redirect 则展示用户中心
|
||
token = request.cookies.get("access_token")
|
||
if token:
|
||
try:
|
||
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
||
username = payload.get("sub")
|
||
if not username:
|
||
raise JWTError
|
||
if redirect:
|
||
return RedirectResponse(redirect)
|
||
html = LOGGED_IN_HTML.replace("{{username}}", username).replace("{{redirect}}", "")
|
||
return HTMLResponse(html)
|
||
except JWTError:
|
||
pass # token 无效，继续展示登录页
|
||
|
||
return HTMLResponse(LOGIN_HTML.replace("{{redirect}}", redirect))
|
||
|
||
|
||
# ── 已登录页面 ──────────────────────────
|
||
|
||
LOGGED_IN_HTML = '''<!DOCTYPE html>
|
||
<html lang="zh-CN">
|
||
<head>
|
||
<meta charset="UTF-8">
|
||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||
<title>SSO 用户中心</title>
|
||
<style>
|
||
*{box-sizing:border-box;margin:0;padding:0}
|
||
body{font-family:-apple-system,BlinkMacSystemFont,"Segoe UI",Roboto,sans-serif;background:linear-gradient(135deg,#667eea 0%,#764ba2 100%);min-height:100vh;display:flex;align-items:center;justify-content:center}
|
||
.card{background:#fff;padding:40px;border-radius:16px;box-shadow:0 20px 60px rgba(0,0,0,.3);width:100%;max-width:400px}
|
||
h1{text-align:center;margin-bottom:8px;color:#333;font-size:24px}
|
||
p.sub{text-align:center;color:#888;margin-bottom:28px;font-size:14px}
|
||
.input-group{margin-bottom:20px}
|
||
label{display:block;margin-bottom:6px;color:#555;font-size:14px;font-weight:500}
|
||
input{width:100%;padding:12px 14px;border:1px solid #ddd;border-radius:8px;font-size:15px;transition:border .2s}
|
||
input:focus{outline:none;border-color:#667eea}
|
||
button{width:100%;padding:14px;background:linear-gradient(135deg,#667eea 0%,#764ba2 100%);color:#fff;border:none;border-radius:8px;font-size:16px;cursor:pointer;transition:opacity .2s;margin-bottom:12px}
|
||
button:hover{opacity:.9}
|
||
button.secondary{background:#f3f4f6;color:#374151}
|
||
button.secondary:hover{background:#e5e7eb}
|
||
.error{background:#fee2e2;color:#991b1b;padding:12px;border-radius:8px;margin-bottom:16px;font-size:14px;display:none}
|
||
.success{background:#d1fae5;color:#065f46;padding:12px;border-radius:8px;margin-bottom:16px;font-size:14px;display:none}
|
||
.links{text-align:center;margin-top:16px}
|
||
.links a{color:#667eea;text-decoration:none;font-size:14px}
|
||
.links a:hover{text-decoration:underline}
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<div class="card">
|
||
<h1>🔐 SSO 用户中心</h1>
|
||
<p class="sub">欢迎，<strong>{{username}}</strong></p>
|
||
<div class="success" id="success"></div>
|
||
<div class="error" id="error"></div>
|
||
<h3 style="margin-bottom:16px;color:#333;font-size:16px">修改密码</h3>
|
||
<form id="changeForm">
|
||
<div class="input-group">
|
||
<label>旧密码</label>
|
||
<input type="password" id="old_password" required>
|
||
</div>
|
||
<div class="input-group">
|
||
<label>新密码</label>
|
||
<input type="password" id="new_password" required minlength="6">
|
||
</div>
|
||
<div class="input-group">
|
||
<label>确认新密码</label>
|
||
<input type="password" id="confirm_password" required minlength="6">
|
||
</div>
|
||
<button type="submit">修改密码</button>
|
||
</form>
|
||
<button class="secondary" onclick="logout()">退出登录</button>
|
||
<div class="links" id="redirectLink" style="display:none">
|
||
<a id="backLink" href="#">← 返回原系统</a>
|
||
</div>
|
||
</div>
|
||
<script>
|
||
const redirect = "{{redirect}}";
|
||
if (redirect) {
|
||
document.getElementById('redirectLink').style.display = 'block';
|
||
document.getElementById('backLink').href = redirect;
|
||
}
|
||
document.getElementById('changeForm').addEventListener('submit', async (e) => {
|
||
e.preventDefault();
|
||
const errorDiv = document.getElementById('error');
|
||
const successDiv = document.getElementById('success');
|
||
errorDiv.style.display = 'none';
|
||
successDiv.style.display = 'none';
|
||
const oldPassword = document.getElementById('old_password').value;
|
||
const newPassword = document.getElementById('new_password').value;
|
||
const confirmPassword = document.getElementById('confirm_password').value;
|
||
if (newPassword !== confirmPassword) {
|
||
errorDiv.textContent = '两次输入的新密码不一致';
|
||
errorDiv.style.display = 'block';
|
||
return;
|
||
}
|
||
const form = new URLSearchParams();
|
||
form.append('old_password', oldPassword);
|
||
form.append('new_password', newPassword);
|
||
try {
|
||
const res = await fetch('/api/change-password', {method:'POST', body:form, credentials:'include'});
|
||
const data = await res.json();
|
||
if (res.ok) {
|
||
successDiv.textContent = '密码修改成功，请重新登录';
|
||
successDiv.style.display = 'block';
|
||
document.getElementById('changeForm').reset();
|
||
setTimeout(() => { window.location.href = '/'; }, 1500);
|
||
} else {
|
||
errorDiv.textContent = data.detail || '修改失败';
|
||
errorDiv.style.display = 'block';
|
||
}
|
||
} catch (err) {
|
||
errorDiv.textContent = '网络错误';
|
||
errorDiv.style.display = 'block';
|
||
}
|
||
});
|
||
async function logout() {
|
||
await fetch('/api/logout', {method:'POST', credentials:'include'});
|
||
window.location.href = '/';
|
||
}
|
||
</script>
|
||
</body>
|
||
</html>'''
|
||
|
||
|
||
# ── API ─────────────────────────────────
|
||
|
||
@app.post("/api/login")
|
||
def api_login(response: Response, username: str = Form(...), password: str = Form(...), redirect: str = Form("")):
|
||
if not verify_user(username, password):
|
||
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
||
|
||
access_token = create_access_token(data={"sub": username})
|
||
|
||
# 种 Cookie（Domain=IP，SameSite=Lax，同 IP 不同端口共享）
|
||
cookie_opts = {
|
||
"key": "access_token",
|
||
"value": access_token,
|
||
"domain": COOKIE_DOMAIN,
|
||
"path": "/",
|
||
"httponly": True,
|
||
"samesite": COOKIE_SAMESITE,
|
||
"max_age": ACCESS_TOKEN_EXPIRE_MINUTES * 60,
|
||
}
|
||
response.set_cookie(**cookie_opts)
|
||
|
||
return {"success": True, "username": username, "redirect": redirect}
|
||
|
||
|
||
@app.post("/api/logout")
|
||
def api_logout(response: Response):
|
||
response.delete_cookie("access_token", domain=COOKIE_DOMAIN, path="/")
|
||
return {"success": True}
|
||
|
||
|
||
@app.post("/api/change-password")
|
||
def api_change_password(
|
||
request: Request,
|
||
response: Response,
|
||
username: str = Form(""),
|
||
old_password: str = Form(...),
|
||
new_password: str = Form(...)
|
||
):
|
||
"""修改密码 —— 支持两种身份验证方式：
|
||
1) Cookie 中的 token（已登录状态）
|
||
2) 提交 username + old_password（未登录状态，登录页直接修改）
|
||
成功后清除 Cookie 要求重新登录
|
||
"""
|
||
# 1) 确定用户名
|
||
target_username = None
|
||
|
||
# 先尝试 Cookie
|
||
token = request.cookies.get("access_token")
|
||
if token:
|
||
try:
|
||
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
||
target_username = payload.get("sub")
|
||
except JWTError:
|
||
pass
|
||
|
||
# 无 Cookie 则使用提交的 username + old_password 验证
|
||
if not target_username:
|
||
if not username:
|
||
raise HTTPException(status_code=401, detail="未登录，请提供用户名")
|
||
db = SessionLocal()
|
||
try:
|
||
user = db.query(models.User).filter(models.User.username == username).first()
|
||
if not user:
|
||
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
||
if not pwd_context.verify(old_password, user.password):
|
||
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
||
target_username = username
|
||
finally:
|
||
db.close()
|
||
|
||
# 2) 验证旧密码并更新
|
||
db = SessionLocal()
|
||
try:
|
||
user = db.query(models.User).filter(models.User.username == target_username).first()
|
||
if not user:
|
||
raise HTTPException(status_code=404, detail="用户不存在")
|
||
if not pwd_context.verify(old_password, user.password):
|
||
raise HTTPException(status_code=400, detail="旧密码错误")
|
||
user.password = pwd_context.hash(new_password)
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
|
||
# 3) 清除 Cookie，强制重新登录
|
||
response.delete_cookie("access_token", domain=COOKIE_DOMAIN, path="/")
|
||
return {"success": True, "message": "密码修改成功，请重新登录"}
|
||
|
||
|
||
@app.get("/api/userinfo")
|
||
def api_userinfo(request: Request):
|
||
token = request.cookies.get("access_token")
|
||
if not token:
|
||
raise HTTPException(status_code=401, detail="未登录")
|
||
try:
|
||
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
||
username = payload.get("sub")
|
||
if not username:
|
||
raise HTTPException(status_code=401, detail="无效令牌")
|
||
return {"username": username}
|
||
except JWTError:
|
||
raise HTTPException(status_code=401, detail="令牌已过期或无效")
|
||
|
||
|
||
@app.get("/api/validate")
|
||
def api_validate(request: Request):
|
||
"""供业务系统可选调用：验证 Cookie 中的 token"""
|
||
try:
|
||
return api_userinfo(request)
|
||
except HTTPException:
|
||
raise
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import uvicorn
|
||
uvicorn.run(app, host="0.0.0.0", port=9000)
|