Maas-Model-platform/backend/app/middleware/auth_log_middleware.py

"""
全局权限+日志管理中间件

提供接口鉴权装饰器和请求日志记录
功能：
- 角色权限验证（admin/user）
- 请求日志记录（路径、用户ID、时间、响应状态、响应结果）
- 本地环境默认管理员权限
"""
import os
import time
import logging
import functools
from enum import Enum
from typing import Optional, List, Callable
from datetime import datetime

from fastapi import Request, HTTPException, status, Depends
from fastapi.responses import JSONResponse
from sqlalchemy.orm import Session

from app.database import get_db, SessionLocal
from app.models.user import User
from app.services.auth_service import AuthService
from app.services.user_service import UserService
from app.services.token_revocation_service import token_revocation_service

logger = logging.getLogger(__name__)


class UserRole(str, Enum):
    """用户角色枚举"""
    ADMIN = "admin"
    USER = "user"


def get_debug_mode() -> bool:
    """获取调试模式状态"""
    return os.getenv("DEBUG", "False").lower() == "true"


def get_default_admin_user(db: Session) -> Optional[User]:
    """获取默认管理员用户（本地环境使用）"""
    return db.query(User).filter(User.username == "admin").first()


def extract_token_from_request(request: Request) -> Optional[str]:
    """从请求中提取JWT令牌"""
    auth_header = request.headers.get("Authorization")
    if auth_header and auth_header.startswith("Bearer "):
        return auth_header[7:]
    return None


def get_user_from_token(token: str, db: Session) -> Optional[User]:
    """从令牌获取用户（支持 JWT 和平台 API Key）"""
    # 优先尝试 platform API Key（sk-aigc- 前缀）
    if token.startswith("sk-aigc-"):
        try:
            from app.services.platform_api_key_service import PlatformApiKeyService
            result = PlatformApiKeyService(db).verify_api_key(token)
            if result:
                user_id, _ = result
                user_service = UserService(db)
                return user_service.get_user_by_id(user_id)
        except Exception:
            pass
        return None

    # 否则按 JWT 处理
    try:
        payload = AuthService.verify_token(token)
        if token_revocation_service.is_payload_revoked(payload):
            return None
        user_id = payload.get("user_id")
        if user_id:
            user_service = UserService(db)
            return user_service.get_user_by_id(user_id)
    except Exception:
        pass
    return None


def get_user_role(user: User) -> UserRole:
    """获取用户角色"""
    if user.username == "admin":
        return UserRole.ADMIN
    return UserRole.USER


class RequestLogger:
    """请求日志记录器"""
    
    @staticmethod
    def log_request(
        request: Request,
        user_id: Optional[str],
        status_code: int,
        response_data: Optional[dict],
        duration_ms: float
    ):
        """记录请求日志"""
        log_data = {
            "timestamp": datetime.now().isoformat(),
            "method": request.method,
            "path": str(request.url.path),
            "query_params": str(request.query_params),
            "user_id": user_id or "anonymous",
            "status_code": status_code,
            "duration_ms": round(duration_ms, 2),
            "response": response_data
        }
        
        if status_code >= 400:
            logger.warning(f"Request failed: {log_data}")
        else:
            logger.info(f"Request completed: {log_data}")


def require_auth(
    roles: Optional[List[UserRole]] = None,
    allow_anonymous: bool = False
):
    """
    权限验证装饰器
    
    Args:
        roles: 允许的角色列表，None表示只需登录
        allow_anonymous: 是否允许匿名访问（仅在DEBUG模式下生效）
    """
    def decorator(func: Callable):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            request: Request = kwargs.get('request')
            db: Session = kwargs.get('db')
            
            if not request or not db:
                for arg in args:
                    if isinstance(arg, Request):
                        request = arg
                    elif isinstance(arg, Session):
                        db = arg
            
            start_time = time.time()
            user = None
            user_id = None
            
            try:
                is_debug = get_debug_mode()
                token = extract_token_from_request(request)
                
                if token:
                    user = get_user_from_token(token, db)
                
                # 本地环境且无token时，使用默认管理员
                if not user and is_debug:
                    user = get_default_admin_user(db)
                    if user:
                        logger.debug(f"DEBUG模式：使用默认管理员用户 {user.username}")
                
                # 非匿名接口必须有用户
                if not user and not allow_anonymous:
                    raise HTTPException(
                        status_code=status.HTTP_401_UNAUTHORIZED,
                        detail="Authentication required"
                    )
                
                if user:
                    user_id = user.id
                    # 角色验证
                    if roles:
                        user_role = get_user_role(user)
                        if user_role not in roles:
                            raise HTTPException(
                                status_code=status.HTTP_403_FORBIDDEN,
                                detail=f"Permission denied. Required roles: {[r.value for r in roles]}"
                            )
                    
                    # 将用户注入到kwargs
                    kwargs['current_user'] = user
                
                result = await func(*args, **kwargs) if asyncio.iscoroutinefunction(func) else func(*args, **kwargs)
                
                duration_ms = (time.time() - start_time) * 1000
                response_data = None
                status_code = 200
                
                if hasattr(result, 'status_code'):
                    status_code = result.status_code
                if hasattr(result, 'body'):
                    try:
                        import json
                        response_data = json.loads(result.body)
                    except:
                        response_data = {"type": "non-json"}
                
                RequestLogger.log_request(request, user_id, status_code, response_data, duration_ms)
                return result
                
            except HTTPException as e:
                duration_ms = (time.time() - start_time) * 1000
                RequestLogger.log_request(
                    request, user_id, e.status_code, 
                    {"error": e.detail}, duration_ms
                )
                raise
            except Exception as e:
                duration_ms = (time.time() - start_time) * 1000
                RequestLogger.log_request(
                    request, user_id, 500,
                    {"error": str(e)}, duration_ms
                )
                raise
        
        return wrapper
    return decorator


import asyncio


def get_current_user_from_request(
    request: Request,
    db: Session = Depends(get_db)
) -> User:
    """
    FastAPI依赖：从请求获取当前用户

    安全：DEBUG 模式下的默认管理员仅在本地回环地址生效，
    防止生产环境误开 DEBUG 导致任意请求获得管理员权限。
    """
    token = extract_token_from_request(request)
    user = None
    
    if token:
        user = get_user_from_token(token, db)
    
    if not user and get_debug_mode():
        # 仅本地回环地址允许 DEBUG 默认管理员
        client_ip = request.client.host if request.client else ""
        if client_ip in ("127.0.0.1", "::1", "localhost"):
            user = get_default_admin_user(db)
    
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Authentication required",
            headers={"WWW-Authenticate": "Bearer"}
        )
    
    return user


def require_role(roles: List[UserRole]):
    """
    角色验证依赖
    
    用法: Depends(require_role([UserRole.ADMIN]))
    """
    def role_checker(
        request: Request,
        db: Session = Depends(get_db)
    ) -> User:
        user = get_current_user_from_request(request, db)
        user_role = get_user_role(user)
        
        if user_role not in roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Permission denied. Required roles: {[r.value for r in roles]}"
            )
        return user
    
    return role_checker