yu-zhi-ran/platform/backend/app/api/auth.py

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
from ..schemas import LoginRequest, TokenResponse, UserResponse
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,
        "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"
            )
            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.from_orm(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.from_orm(user))

@router.get("/me")
def get_current_user(request: Request, db: Session = Depends(get_db)):
    """获取当前登录用户信息"""
    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": UserResponse.from_orm(user)}

def get_current_user(request: Request, db: Session = Depends(get_db)):
    """依赖项：验证用户登录"""
    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