| import uvicorn |
| from fastapi import FastAPI, Request, WebSocket |
| from fastapi.responses import HTMLResponse, StreamingResponse |
| import httpx |
| import asyncio |
| import websockets |
|
|
| app = FastAPI() |
| client = httpx.AsyncClient(timeout=None) |
|
|
| @app.get("/", response_class=HTMLResponse) |
| async def root(): |
| return "<body style='background:#000;color:#0f0;text-align:center;padding-top:100px;font-family:monospace;'><h1>[ SYSTEM ACTIVE ]</h1><p>Path: /terminal</p></body>" |
|
|
| |
| @app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"]) |
| async def proxy_http(request: Request, path: str): |
| if not path or path == "/": return await root() |
| target_url = f"http://127.0.0.1:8080/{path}" |
| headers = {k: v for k, v in request.headers.items() if k.lower() not in ["host", "origin"]} |
| headers["Host"] = "127.0.0.1:8080" |
| headers["Origin"] = "http://127.0.0.1:8080" |
| |
| try: |
| req = client.build_request(method=request.method, url=target_url, headers=headers, params=request.query_params, content=await request.body()) |
| resp = await client.send(req, stream=True) |
| return StreamingResponse(resp.aiter_raw(), status_code=resp.status_code, headers=dict(resp.headers)) |
| except: |
| return HTMLResponse("Backend Offline", status_code=503) |
|
|
| |
| @app.websocket("/{path:path}") |
| async def proxy_ws(websocket: WebSocket, path: str): |
| |
| subproto = websocket.headers.get("sec-websocket-protocol") |
| await websocket.accept(subprotocol=subproto) |
| |
| target_ws_url = f"ws://127.0.0.1:8080/{path}" |
|
|
| try: |
| |
| async with websockets.connect( |
| target_ws_url, |
| subprotocols=[subproto] if subproto else None, |
| ping_interval=None |
| ) as target_ws: |
| |
| async def forward(source, destination): |
| try: |
| while True: |
| if hasattr(source, 'receive_text'): |
| msg = await source.receive_text() |
| await destination.send(msg) |
| else: |
| msg = await source.recv() |
| await destination.send_text(msg) |
| except: pass |
|
|
| await asyncio.gather(forward(websocket, target_ws), forward(target_ws, websocket)) |
| except Exception as e: |
| print(f"WS Error: {str(e)}") |
| finally: |
| try: |
| await websocket.close() |
| except: pass |
|
|
| if __name__ == "__main__": |
| uvicorn.run(app, host="0.0.0.0", port=7860) |
| |
