xfvsdf/qqbot

1#!/usr/bin/env python32"""3OAuth Web 服务器 - 独立的OAuth认证服务4提供简化的OAuth认证界面，只包含验证功能，不包含上传和管理功能5"""6 7from log import log8import asyncio9from contextlib import asynccontextmanager10from fastapi import FastAPI, HTTPException, Depends11from fastapi.responses import HTMLResponse, JSONResponse12from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials13from pydantic import BaseModel14 15from src.auth import (16    create_auth_url, 17    verify_password, 18    generate_auth_token, 19    verify_auth_token,20    asyncio_complete_auth_flow,21    complete_auth_flow_from_callback_url,22    CALLBACK_HOST,23)24 25# 创建FastAPI应用26app = FastAPI(27    title="Google OAuth 认证服务",28    description="独立的OAuth认证服务，用于获取Google Cloud认证文件",29)30 31# HTTP Bearer认证32security = HTTPBearer()33 34# 请求模型35class LoginRequest(BaseModel):36    password: str37 38class AuthStartRequest(BaseModel):39    project_id: str = None  # 现在是可选的，支持自动检测40 41class AuthCallbackRequest(BaseModel):42    project_id: str = None  # 现在是可选的，支持自动检测43 44class AuthCallbackUrlRequest(BaseModel):45    callback_url: str  # OAuth回调完整URL46    project_id: str = None  # 可选的项目ID47 48def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):49    """验证认证令牌"""50    if not verify_auth_token(credentials.credentials):51        raise HTTPException(status_code=401, detail="无效的认证令牌")52    return credentials.credentials53 54 55@app.get("/", response_class=HTMLResponse)56async def serve_oauth_page():57    """提供OAuth认证页面"""58    try:59        # 读取HTML文件60        html_file_path = "./front/multi_user_auth_web.html"61        62        with open(html_file_path, "r", encoding="utf-8") as f:63            html_content = f.read()64        65        return HTMLResponse(content=html_content)66    except FileNotFoundError:67        raise HTTPException(status_code=404, detail="认证页面不存在")68    except Exception as e:69        log.error(f"加载认证页面失败: {e}")70        raise HTTPException(status_code=500, detail="服务器内部错误")71 72@app.post("/auth/login")73async def login(request: LoginRequest):74    """用户登录"""75    try:76        if await verify_password(request.password):77            token = generate_auth_token()78            return JSONResponse(content={"token": token, "message": "登录成功"})79        else:80            raise HTTPException(status_code=401, detail="密码错误")81    except HTTPException:82        raise83    except Exception as e:84        log.error(f"登录失败: {e}")85        raise HTTPException(status_code=500, detail=str(e))86 87 88@app.post("/auth/start")89async def start_auth(request: AuthStartRequest, token: str = Depends(verify_token)):90    """开始认证流程，支持自动检测项目ID"""91    try:92        # 如果没有提供项目ID，尝试自动检测93        project_id = request.project_id94        if not project_id:95            log.info("未提供项目ID，后续将尝试自动检测...")96        97        # 使用认证令牌作为用户会话标识98        user_session = token if token else None99        result = await create_auth_url(project_id, user_session)100        101        if result['success']:102            # 构建动态回调URL103            callback_port = result.get('callback_port')104            callback_url = f"http://{CALLBACK_HOST}:{callback_port}" if callback_port else None105            106            response_data = {107                "auth_url": result['auth_url'],108                "state": result['state'],109                "auto_project_detection": result.get('auto_project_detection', False),110                "detected_project_id": result.get('detected_project_id')111            }112            113            # 如果有回调端口信息，添加到响应中114            if callback_port:115                response_data["callback_port"] = callback_port116                response_data["callback_url"] = callback_url117            118            return JSONResponse(content=response_data)119        else:120            raise HTTPException(status_code=500, detail=result['error'])121            122    except HTTPException:123        raise124    except Exception as e:125        log.error(f"开始认证流程失败: {e}")126        raise HTTPException(status_code=500, detail=str(e))127 128 129@app.post("/auth/callback")130async def auth_callback(request: AuthCallbackRequest, token: str = Depends(verify_token)):131    """处理认证回调（异步等待），支持自动检测项目ID"""132    try:133        # 项目ID现在是可选的，在回调处理中进行自动检测134        project_id = request.project_id135        136        # 使用认证令牌作为用户会话标识137        user_session = token if token else None138        # 异步等待OAuth回调完成139        result = await asyncio_complete_auth_flow(project_id, user_session)140        141        if result['success']:142            return JSONResponse(content={143                "credentials": result['credentials'],144                "file_path": result['file_path'],145                "message": "认证成功，凭证已保存",146                "auto_detected_project": result.get('auto_detected_project', False)147            })148        else:149            # 如果需要手动项目ID或项目选择，在响应中标明150            if result.get('requires_manual_project_id'):151                # 使用JSON响应152                return JSONResponse(153                    status_code=400,154                    content={155                        "error": result['error'],156                        "requires_manual_project_id": True157                    }158                )159            elif result.get('requires_project_selection'):160                # 返回项目列表供用户选择161                return JSONResponse(162                    status_code=400,163                    content={164                        "error": result['error'],165                        "requires_project_selection": True,166                        "available_projects": result['available_projects']167                    }168                )169            else:170                raise HTTPException(status_code=400, detail=result['error'])171            172    except HTTPException:173        raise174    except Exception as e:175        log.error(f"处理认证回调失败: {e}")176        raise HTTPException(status_code=500, detail=str(e))177 178 179@app.post("/auth/callback-url")180async def auth_callback_url(request: AuthCallbackUrlRequest, token: str = Depends(verify_token)):181    """从回调URL直接完成认证，无需启动本地服务器"""182    try:183        # 验证URL格式184        if not request.callback_url or not request.callback_url.startswith(('http://', 'https://')):185            raise HTTPException(status_code=400, detail="请提供有效的回调URL")186        187        # 从回调URL完成认证188        result = await complete_auth_flow_from_callback_url(request.callback_url, request.project_id)189        190        if result['success']:191            return JSONResponse(content={192                "credentials": result['credentials'],193                "file_path": result['file_path'],194                "message": "从回调URL认证成功，凭证已保存",195                "auto_detected_project": result.get('auto_detected_project', False)196            })197        else:198            # 处理各种错误情况199            if result.get('requires_manual_project_id'):200                return JSONResponse(201                    status_code=400,202                    content={203                        "error": result['error'],204                        "requires_manual_project_id": True205                    }206                )207            elif result.get('requires_project_selection'):208                return JSONResponse(209                    status_code=400,210                    content={211                        "error": result['error'],212                        "requires_project_selection": True,213                        "available_projects": result['available_projects']214                    }215                )216            else:217                raise HTTPException(status_code=400, detail=result['error'])218            219    except HTTPException:220        raise221    except Exception as e:222        log.error(f"从回调URL处理认证失败: {e}")223        raise HTTPException(status_code=500, detail=str(e))224 225 226@asynccontextmanager227async def lifespan(app: FastAPI):228    log.info("OAuth认证服务启动中...")229 230    # OAuth回调服务器现在动态按需启动，每个认证流程使用独立端口231    log.info("OAuth回调服务器将为每个认证流程动态分配端口")232 233    # 从配置获取密码和端口234    from config import get_panel_password, get_server_port235    password = await get_panel_password()236    port = await get_server_port()237 238    log.info("Web服务已由 ASGI 服务器启动")239    240    print("\n" + "="*60)241    print("🚀 Google OAuth 认证服务已启动")242    print("="*60)243    print(f"📱 Web界面: http://localhost:{port}")244    print(f"🔐 默认密码: {'已设置' if password else 'pwd (请设置PASSWORD环境变量)'}")245    print(f"🔄 多用户并发: 支持多用户同时认证（动态端口分配）")246    print("="*60 + "\n")247 248    try:249        yield250    finally:251        log.info("OAuth认证服务关闭中...")252        # OAuth服务器由认证流程自动管理，无需手动清理253        log.info("OAuth认证服务已关闭")254 255# 注册 lifespan 处理器256app.router.lifespan_context = lifespan257 258if __name__ == "__main__":259    from hypercorn.asyncio import serve260    from hypercorn.config import Config261 262    async def main():263        # 从配置获取端口264        from config import get_server_port265        PORT = await get_server_port()266        267        config = Config()268        config.bind = [f"0.0.0.0:{PORT}"]269        config.accesslog = "-"270        config.errorlog = "-"271        config.loglevel = "INFO"272        273        await serve(app, config)274    275    asyncio.run(main())