Spaces:
Sleeping
Sleeping
File size: 2,778 Bytes
e084365 9432aa8 e084365 e9670f4 e084365 9432aa8 e084365 faae9a1 e084365 faae9a1 e084365 faaab61 e9670f4 e084365 e9670f4 faae9a1 e084365 d99260b 6df2efa e084365 29bd9ed 9432aa8 8e183da e9670f4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 |
# uvicorn app:app --host 0.0.0.0 --port 7860 --reload
from fastapi import FastAPI, Request, HTTPException, Depends, status
from fastapi.responses import Response # 导入Response
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials # 导入HTTPBearer和HTTPAuthorizationCredentials
import httpx # 使用httpx替代requests，因为requests是同步的，而FastAPI是异步的
import os, json
import logging # 导入logging模块
from dotenv import load_dotenv
from enum import Enum
from proxy import do_proxy
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# 加载环境变量
load_dotenv()
# 从环境变量获取代理自身的API密钥
proxy_api_key = os.getenv("PROXY_API_KEY")
# 定义HTTPBearer安全方案
security = HTTPBearer()
# 枚举校验协议类型
class ProtocolType(str, Enum):
HTTP = "http"
HTTPS = "https"
# 依赖函数：验证代理的API密钥
def verify_proxy_api_key(credentials: HTTPAuthorizationCredentials = Depends(security)):
if not proxy_api_key or credentials.credentials != proxy_api_key:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or missing proxy API key",
headers={"WWW-Authenticate": "Bearer"},
)
return credentials.credentials
app = FastAPI(title="Common Proxy API")
@app.get("/")
def greet_json():
return {"Hello": "World!"}
# 3. 健康检查接口（可选，用于验证服务是否正常）
@app.get("/health")
async def health_check():
return {"status": "ok", "message": "Proxy service is running."}
@app.api_route("/v1/{protocol}/{host}/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
# @app.api_route("/v1/{protocol}/{host}/{path:path}", methods=["POST"])
async def proxy(request: Request, protocol: ProtocolType, host:str, path: str, proxy_api_key: str = Depends(verify_proxy_api_key)): # 添加代理认证依赖
real_url = f"{protocol.value}://{host}/{path}"
# 提取客户端请求的headers和body
client_headers = dict(request.headers)
client_body = await request.body()
# 移除FastAPI自带的Host头，避免目标网站校验报错
client_headers.pop("host", None)
client_headers.pop("authorization", None)
# 将User-Agent改为curl/8.7.1，以模拟curl请求
client_headers["User-Agent"] = "curl/8.7.1"
logger.info(f"Proxying request to: {real_url}")
logger.debug(f"Request method: {request.method}, Headers: {client_headers}")
# 调用 do_proxy 并返回其结果，此处传的 headers 是不带 api_key 的数据
return await do_proxy(real_url, request.method, client_headers, client_body)
|