AgentGraph / backend /routers /agentgraph.py
wu981526092's picture
Security: Fix critical vulnerabilities before public release
bcbd2ec
from fastapi import APIRouter, Request
from fastapi.responses import HTMLResponse, FileResponse, JSONResponse
from pathlib import Path
router = APIRouter()
# Security: Define base directories for path traversal protection
DIST_DIR = Path("frontend/dist").resolve()
ASSETS_DIR = Path("frontend/dist/assets").resolve()
def is_safe_path(base_dir: Path, requested_path: Path) -> bool:
"""Check if the requested path is within the allowed base directory"""
try:
resolved = requested_path.resolve()
return str(resolved).startswith(str(base_dir))
except (OSError, ValueError):
return False
@router.get("/agentgraph", response_class=HTMLResponse)
async def agentgraph_interface(request: Request):
"""Serve the React-based AgentGraph interface (requires authentication)"""
# Serve the built React app from the new location
dist_path = DIST_DIR / "index.html"
if dist_path.exists():
with open(dist_path, 'r') as f:
content = f.read()
return HTMLResponse(content=content)
else:
# Return error message if React app not built
return JSONResponse(
content={"error": "React app not built. Please run 'npm run build' in the frontend directory."},
status_code=503
)
@router.get("/agentgraph/{path:path}")
async def agentgraph_assets(path: str):
"""Serve static assets for the React app with path traversal protection"""
requested_path = (DIST_DIR / path).resolve()
# Security: Prevent path traversal attacks
if not is_safe_path(DIST_DIR, requested_path):
return JSONResponse(
content={"error": "Access denied"},
status_code=403
)
if requested_path.is_file():
return FileResponse(requested_path)
return JSONResponse(content={"error": "File not found"}, status_code=404)
@router.get("/assets/{path:path}")
async def serve_assets(path: str):
"""Serve React assets from /assets/ path with path traversal protection"""
requested_path = (ASSETS_DIR / path).resolve()
# Security: Prevent path traversal attacks
if not is_safe_path(ASSETS_DIR, requested_path):
return JSONResponse(
content={"error": "Access denied"},
status_code=403
)
if requested_path.is_file():
return FileResponse(requested_path)
return JSONResponse(content={"error": "Asset not found"}, status_code=404)
@router.get("/vite.svg")
async def serve_vite_svg():
"""Serve the vite.svg favicon"""
file_path = DIST_DIR / "vite.svg"
if file_path.exists():
return FileResponse(file_path)
return JSONResponse(content={"error": "Favicon not found"}, status_code=404)