truetype
b2ee220783
Use python3/venv consistently for local test execution and make package discovery deterministic for pytest. Also migrate FastAPI shutdown handling to lifespan to remove deprecation warnings while preserving cleanup behavior.
114 lines
3.2 KiB
Python
114 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import dataclasses
|
|
from contextlib import asynccontextmanager
|
|
|
|
from fastapi import FastAPI, HTTPException, Request
|
|
from fastapi.responses import HTMLResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
from pydantic import BaseModel
|
|
|
|
from app.policy import TunnelPolicy
|
|
from app.services import PortScanner
|
|
from app.storage import Storage
|
|
from app.tunnel_manager import TunnelManager
|
|
|
|
policy = TunnelPolicy.load()
|
|
storage = Storage.from_env()
|
|
scanner = PortScanner(policy=policy, storage=storage)
|
|
manager = TunnelManager(storage=storage)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_: FastAPI):
|
|
try:
|
|
yield
|
|
finally:
|
|
manager.shutdown()
|
|
storage.close()
|
|
|
|
|
|
app = FastAPI(title="Gradio Tunnel Dashboard", lifespan=lifespan)
|
|
|
|
app.mount("/static", StaticFiles(directory="app/static"), name="static")
|
|
templates = Jinja2Templates(directory="app/templates")
|
|
|
|
|
|
class TunnelRequest(BaseModel):
|
|
port: int
|
|
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
async def index(request: Request) -> HTMLResponse:
|
|
return templates.TemplateResponse("index.html", {"request": request})
|
|
|
|
|
|
@app.get("/api/ports")
|
|
async def list_ports() -> list[dict[str, str | int | None]]:
|
|
records = scanner.scan()
|
|
return [dataclasses.asdict(record) for record in sorted(records, key=lambda item: item.port)]
|
|
|
|
|
|
@app.get("/api/tunnels")
|
|
async def list_tunnels() -> list[dict[str, str | int | None]]:
|
|
tunnels = manager.list_tunnels()
|
|
return [
|
|
{
|
|
"port": info.port,
|
|
"command": info.command,
|
|
"pid": info.pid,
|
|
"url": info.url,
|
|
"status": info.status,
|
|
"exit_code": info.exit_code,
|
|
"last_output": info.last_output,
|
|
}
|
|
for info in tunnels.values()
|
|
]
|
|
|
|
|
|
@app.get("/api/history/events")
|
|
async def history_events() -> list[dict[str, object | None]]:
|
|
return storage.recent_events()
|
|
|
|
|
|
@app.get("/api/history/services")
|
|
async def history_services() -> list[dict[str, object | None]]:
|
|
return storage.recent_services()
|
|
|
|
|
|
@app.get("/api/tunnels/{port}/logs")
|
|
async def tunnel_logs(port: int) -> dict[str, list[str]]:
|
|
logs = manager.get_logs(port)
|
|
if logs is None:
|
|
raise HTTPException(status_code=404, detail="Tunnel not running.")
|
|
return {"logs": logs}
|
|
|
|
|
|
@app.post("/api/tunnels")
|
|
async def start_tunnel(payload: TunnelRequest) -> dict[str, str | int | None]:
|
|
if payload.port <= 0 or payload.port > 65535:
|
|
raise HTTPException(status_code=400, detail="Invalid port.")
|
|
allowed, reason = policy.is_port_allowed(payload.port)
|
|
if not allowed:
|
|
raise HTTPException(status_code=403, detail=reason or "Porta non autorizzata.")
|
|
info = manager.start_tunnel(payload.port)
|
|
return {
|
|
"port": info.port,
|
|
"command": info.command,
|
|
"pid": info.pid,
|
|
"url": info.url,
|
|
"status": info.status,
|
|
"exit_code": info.exit_code,
|
|
"last_output": info.last_output,
|
|
}
|
|
|
|
|
|
@app.delete("/api/tunnels/{port}")
|
|
async def stop_tunnel(port: int) -> dict[str, bool]:
|
|
stopped = manager.stop_tunnel(port)
|
|
if not stopped:
|
|
raise HTTPException(status_code=404, detail="Tunnel not running.")
|
|
return {"stopped": True}
|
|
|