PapaTramp
d3a337992c
Path traversal fix in SPA fallback, admin-only host delete, login rate limit with 3 attempts and Telegram alert, JWT validated against active users in DB, and DOMPurify for agent report HTML. Co-authored-by: Cursor <cursoragent@cursor.com>
126 lines
3.6 KiB
Python
126 lines
3.6 KiB
Python
import logging
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from sqlalchemy import select
|
|
|
|
from app.api.v1.router import api_router
|
|
from app.auth.api_key import hash_api_key
|
|
from app.config import get_settings
|
|
from app.database import SessionLocal
|
|
from app.models import ApiKey
|
|
from app.services.user_auth import bootstrap_admin_user
|
|
from app.version import APP_VERSION, APP_VERSION_LABEL
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
logger = logging.getLogger("sac")
|
|
|
|
|
|
def bootstrap_api_key() -> None:
|
|
settings = get_settings()
|
|
if not settings.sac_bootstrap_api_key:
|
|
return
|
|
db = SessionLocal()
|
|
try:
|
|
key_hash = hash_api_key(settings.sac_bootstrap_api_key)
|
|
exists = db.scalar(select(ApiKey).where(ApiKey.key_hash == key_hash))
|
|
if exists:
|
|
return
|
|
prefix = settings.sac_bootstrap_api_key[:12]
|
|
db.add(
|
|
ApiKey(
|
|
name="bootstrap",
|
|
key_prefix=prefix,
|
|
key_hash=key_hash,
|
|
is_active=True,
|
|
)
|
|
)
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def bootstrap_users() -> None:
|
|
db = SessionLocal()
|
|
try:
|
|
bootstrap_admin_user(db)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_app: FastAPI):
|
|
logger.info("%s — application startup (version %s)", APP_VERSION_LABEL, APP_VERSION)
|
|
bootstrap_api_key()
|
|
bootstrap_users()
|
|
yield
|
|
logger.info("%s — application shutdown", APP_VERSION_LABEL)
|
|
|
|
|
|
def configure_logging() -> None:
|
|
if logging.getLogger().handlers:
|
|
return
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(levelname)s: %(message)s",
|
|
)
|
|
|
|
|
|
def create_app() -> FastAPI:
|
|
configure_logging()
|
|
settings = get_settings()
|
|
app = FastAPI(
|
|
title="Security Alert Center",
|
|
version=APP_VERSION,
|
|
lifespan=lifespan,
|
|
)
|
|
origins = [o.strip() for o in settings.cors_origins.split(",") if o.strip()]
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=origins if origins != ["*"] else ["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
from app.api.v1.health import router as health_router
|
|
|
|
app.include_router(api_router, prefix="/api/v1")
|
|
app.include_router(health_router)
|
|
|
|
frontend_dist = ROOT / "frontend" / "dist"
|
|
if frontend_dist.is_dir():
|
|
assets_dir = frontend_dist / "assets"
|
|
if assets_dir.is_dir():
|
|
app.mount(
|
|
"/assets",
|
|
StaticFiles(directory=str(assets_dir)),
|
|
name="ui-assets",
|
|
)
|
|
|
|
@app.get("/{spa_path:path}")
|
|
def spa_fallback(spa_path: str) -> FileResponse:
|
|
if spa_path.startswith("api/"):
|
|
raise HTTPException(status_code=404, detail="Not Found")
|
|
dist_root = frontend_dist.resolve()
|
|
if spa_path:
|
|
candidate = (frontend_dist / spa_path).resolve()
|
|
try:
|
|
candidate.relative_to(dist_root)
|
|
except ValueError:
|
|
raise HTTPException(status_code=404, detail="Not Found") from None
|
|
if candidate.is_file():
|
|
return FileResponse(candidate)
|
|
index = frontend_dist / "index.html"
|
|
if not index.is_file():
|
|
raise HTTPException(status_code=404, detail="UI not built")
|
|
return FileResponse(index)
|
|
|
|
return app
|
|
|
|
|
|
app = create_app()
|