Yuzhiran Dev
b2d043b231
- 用户管理：新增编辑弹窗（修改用户名/角色/密码），增加组织/创建时间/最后登录列 - 角色管理：新增 Role 模型 + CRUD API，admin.html 新增角色管理 tab - 菜单管理：新增 Menu 模型 + CRUD API，导航栏从 API 动态加载菜单项 - 个人中心：右上角下拉菜单（个人信息/修改密码/退出），新增修改密码 API - 种子数据：initial_data.py 自动创建默认角色（admin/editor）和默认菜单（7项） - 修复 research.py 缺少 enrich_topic_research 函数导致导入失败 - 修复 db_helper.py 中 generated_at 条件导致重创作不更新时间戳 - admin.html 操作列加宽防止按钮换行，平台配置增加删除按钮 - articles.html 预览弹窗加 lock-scroll=false 防止页面尺寸跳动
195 lines
7.3 KiB
Python
195 lines
7.3 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from sqlalchemy.orm import Session
|
|
from datetime import datetime, timedelta
|
|
import jwt
|
|
import bcrypt
|
|
import os
|
|
from dotenv import load_dotenv
|
|
|
|
from ..database import get_db
|
|
from ..models import User, Role
|
|
from ..schemas import LoginRequest, TokenResponse, UserResponse, ChangePasswordRequest
|
|
from ..core.audit_logger import audit_log
|
|
|
|
# 加载环境变量
|
|
load_dotenv()
|
|
|
|
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
|
|
|
# JWT 配置
|
|
SECRET_KEY = os.getenv("SECRET_KEY", "dev-secret-key-change-this")
|
|
ALGORITHM = os.getenv("ALGORITHM", "HS256")
|
|
ACCESS_TOKEN_EXPIRE_DAYS = int(os.getenv("ACCESS_TOKEN_EXPIRE_DAYS", "7"))
|
|
|
|
# 默认管理员配置
|
|
DEFAULT_ADMIN_USERNAME = os.getenv("DEFAULT_ADMIN_USERNAME", "admin")
|
|
DEFAULT_ADMIN_PASSWORD = os.getenv("DEFAULT_ADMIN_PASSWORD", "admin123")
|
|
|
|
def create_token(user: User) -> str:
|
|
"""生成 JWT token"""
|
|
expire = datetime.utcnow() + timedelta(days=ACCESS_TOKEN_EXPIRE_DAYS)
|
|
payload = {
|
|
"sub": str(user.id),
|
|
"username": user.username,
|
|
"role": user.role,
|
|
"org_id": user.org_id,
|
|
"exp": expire
|
|
}
|
|
return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
|
|
|
|
def verify_token(token: str, db: Session) -> User:
|
|
"""验证 JWT token 并返回用户"""
|
|
try:
|
|
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
|
user_id = int(payload["sub"])
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if not user:
|
|
raise HTTPException(status_code=401, detail="用户不存在")
|
|
return user
|
|
except jwt.ExpiredSignatureError:
|
|
raise HTTPException(status_code=401, detail="token 已过期")
|
|
except jwt.InvalidTokenError:
|
|
raise HTTPException(status_code=401, detail="无效的 token")
|
|
|
|
@router.post("/login", response_model=TokenResponse)
|
|
def login(login_data: LoginRequest, request: Request, db: Session = Depends(get_db)):
|
|
"""管理员登录（支持 bcrypt 密码验证）"""
|
|
client_ip = request.client.host if request.client else None
|
|
user_agent = request.headers.get("user-agent", "")
|
|
|
|
# 检查是否是默认管理员（bcrypt 验证）
|
|
if login_data.username == DEFAULT_ADMIN_USERNAME:
|
|
# 查询或创建管理员用户
|
|
user = db.query(User).filter(User.username == DEFAULT_ADMIN_USERNAME).first()
|
|
if not user:
|
|
# 首次创建，密码哈希存储
|
|
hashed = bcrypt.hashpw(DEFAULT_ADMIN_PASSWORD.encode('utf-8'), bcrypt.gensalt())
|
|
user = User(
|
|
username=DEFAULT_ADMIN_USERNAME,
|
|
password_hash=hashed.decode('utf-8'),
|
|
role="admin",
|
|
org_id="default"
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
else:
|
|
# 验证密码
|
|
if not user.password_hash:
|
|
# 旧数据可能没有密码，设置为默认密码
|
|
user.password_hash = bcrypt.hashpw(DEFAULT_ADMIN_PASSWORD.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
|
db.commit()
|
|
else:
|
|
# 验证密码是否匹配
|
|
if not bcrypt.checkpw(login_data.password.encode('utf-8'), user.password_hash.encode('utf-8')):
|
|
# 记录失败日志
|
|
audit_log(
|
|
action="login_failed",
|
|
username=login_data.username,
|
|
details={"reason": "invalid_password"},
|
|
ip_address=client_ip,
|
|
user_agent=user_agent,
|
|
db=db
|
|
)
|
|
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
|
token = create_token(user)
|
|
# 记录成功登录日志
|
|
audit_log(
|
|
action="login",
|
|
user=user,
|
|
details={"role": user.role},
|
|
ip_address=client_ip,
|
|
user_agent=user_agent,
|
|
db=db
|
|
)
|
|
return TokenResponse(token=token, role=user.role, user=UserResponse.model_validate(user))
|
|
|
|
# 从数据库查询其他用户
|
|
user = db.query(User).filter(User.username == login_data.username).first()
|
|
if not user:
|
|
audit_log(
|
|
action="login_failed",
|
|
username=login_data.username,
|
|
details={"reason": "user_not_found"},
|
|
ip_address=client_ip,
|
|
user_agent=user_agent,
|
|
db=db
|
|
)
|
|
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
|
|
|
# bcrypt 验证
|
|
if not user.password_hash or not bcrypt.checkpw(login_data.password.encode('utf-8'), user.password_hash.encode('utf-8')):
|
|
audit_log(
|
|
action="login_failed",
|
|
user=user,
|
|
details={"reason": "invalid_password"},
|
|
ip_address=client_ip,
|
|
user_agent=user_agent,
|
|
db=db
|
|
)
|
|
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
|
|
|
token = create_token(user)
|
|
audit_log(
|
|
action="login",
|
|
user=user,
|
|
details={"role": user.role},
|
|
ip_address=client_ip,
|
|
user_agent=user_agent,
|
|
db=db
|
|
)
|
|
return TokenResponse(token=token, role=user.role, user=UserResponse.model_validate(user))
|
|
|
|
def get_current_user(request: Request, db: Session = Depends(get_db)) -> User:
|
|
"""依赖项：验证用户登录"""
|
|
auth_header = request.headers.get("Authorization")
|
|
if not auth_header or not auth_header.startswith("Bearer "):
|
|
raise HTTPException(status_code=401, detail="未提供认证令牌")
|
|
token = auth_header.split(" ")[1]
|
|
user = verify_token(token, db)
|
|
return user
|
|
|
|
def org_filter(current_user: User, model):
|
|
"""返回 org_id 过滤条件。管理员看到全部数据，普通用户仅限本组织。"""
|
|
if current_user.role == "admin":
|
|
return True # no filter
|
|
return model.org_id == current_user.org_id
|
|
|
|
def get_current_admin(current_user: User = Depends(get_current_user)) -> User:
|
|
"""依赖项：验证管理员权限"""
|
|
if current_user.role != "admin":
|
|
raise HTTPException(status_code=403, detail="需要管理员权限")
|
|
return current_user
|
|
@router.get("/me")
|
|
def get_me(
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""获取当前登录用户信息"""
|
|
return {"user": UserResponse.model_validate(current_user)}
|
|
|
|
@router.put("/password")
|
|
def change_password(
|
|
data: ChangePasswordRequest,
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""修改当前用户密码"""
|
|
if not bcrypt.checkpw(data.old_password.encode('utf-8'), current_user.password_hash.encode('utf-8')):
|
|
raise HTTPException(status_code=400, detail="原密码错误")
|
|
if len(data.new_password) < 6:
|
|
raise HTTPException(status_code=400, detail="新密码至少6位")
|
|
current_user.password_hash = bcrypt.hashpw(data.new_password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
|
db.commit()
|
|
audit_log(
|
|
action="change_password",
|
|
user=current_user,
|
|
ip_address=request.client.host if request.client else None,
|
|
user_agent=request.headers.get("user-agent", ""),
|
|
db=db
|
|
)
|
|
return {"message": "密码修改成功"}
|
|
|