TERES_fastapi_backend/api/apps/models/auth_dependencies.py

#
#  Copyright 2024 The InfiniFlow Authors. All Rights Reserved.
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
#

from typing import Optional
from fastapi import Depends, Header, Security, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from api import settings
from api.utils.api_utils import get_json_result

# 创建 HTTPBearer 安全方案（auto_error=False 允许我们自定义错误处理）
http_bearer = HTTPBearer(auto_error=False)


def get_current_user(
    authorization: Optional[str] = Header(None, alias="Authorization"),
    credentials: Optional[HTTPAuthorizationCredentials] = Security(http_bearer)
):
    """FastAPI 依赖注入：获取当前用户（替代 Flask 的 login_required 和 current_user）
    
    支持两种格式的 Authorization 头：
    1. 标准格式：Bearer <token>
    2. 简化格式：<token>（不带 Bearer 前缀）
    
    使用 Security(http_bearer) 可以让 FastAPI 自动在 OpenAPI schema 中添加安全要求，
    这样 Swagger UI 就会显示授权输入框并自动在请求中添加 Authorization 头。
    """
    # 延迟导入以避免循环导入
    from api.apps.__init___fastapi import get_current_user_from_token
    
    token = None
    
    # 优先从 HTTPBearer 获取（标准格式：Bearer <token>）
    if credentials:
        token = credentials.credentials
    # 如果 HTTPBearer 没有获取到，尝试直接从 Header 获取（可能是简化格式）
    elif authorization:
        # 如果包含 "Bearer " 前缀，则去除它
        if authorization.startswith("Bearer "):
            token = authorization[7:]  # 去除 "Bearer " 前缀（7个字符）
        else:
            # 不带 Bearer 前缀，直接使用
            token = authorization
    
    if not token:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Authorization header is required"
        )
    
    user = get_current_user_from_token(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or expired token"
        )
    
    return user