"""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 = '''
SSO 用户中心
🔐 SSO 用户中心
欢迎，{{username}}
修改密码
'''
# ── 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)