hf_app.py

# HF Space shim: serve the web player at / instead of a bare 404.
import os
import re
import time
from collections import deque

from fastapi import Request
from fastapi.responses import JSONResponse, PlainTextResponse, RedirectResponse

from api.src.main import app

MAX_BODY_BYTES = int(os.getenv("MAX_BODY_BYTES", 8192))
RATE_LIMIT_PER_MIN = int(os.getenv("RATE_LIMIT_PER_MIN", 20))
WEB_CACHE_SECONDS = int(os.getenv("WEB_CACHE_SECONDS", 600))

_hits: dict[str, deque] = {}

# Credential-scanner bait. Nothing here routes to a real endpoint.
_SCANNER = re.compile(
    r"\.env|\.git|\.aws|\.ssh|\.npmrc|\.netrc|\.azure|\.claude|\.gcloud|\.openai|\.anthropic"
    r"|credential|secrets?\.|service.?account|\.tfstate|terraform|wp-config|\.php"
    r"|actuator|telescope|debugbar|horizon|appsettings|application\.(properties|yml)"
    r"|docker-compose|web\.config|WEB-INF|\.(sql|bak|orig|swp)$|~$"
    r"|169\.254\.169\.254|metadata\.google\.internal|bash_history",
    re.I,
)

_HONEYPOT = """# nice try
AWS_ACCESS_KEY_ID=AKIA00000000GOTEEM
AWS_SECRET_ACCESS_KEY=touch/grass/and/get/a/real/job/friend
OPENAI_API_KEY=sk-proj-you-are-scanning-a-text-to-speech-demo
DATABASE_URL=postgres://nobody:nothing@localhost:5432/there_is_no_database
ADMIN_PASSWORD=hunter2
MOTD=this box only makes robot voices. go bother someone else.
"""


def _client_ip(request: Request) -> str:
    forwarded = request.headers.get("x-forwarded-for", "").split(",")[-1].strip()
    return forwarded or (request.client.host if request.client else "unknown")


@app.middleware("http")
async def throttle(request: Request, call_next):
    target = request.url.path + "?" + request.url.query
    if _SCANNER.search(target):
        return PlainTextResponse(_HONEYPOT, status_code=200)

    if request.method != "POST":
        response = await call_next(request)
        if request.url.path.startswith("/web/") and response.status_code == 200:
            response.headers["Cache-Control"] = f"public, max-age={WEB_CACHE_SECONDS}"
        return response

    length = request.headers.get("content-length")
    if length and int(length) > MAX_BODY_BYTES:
        return JSONResponse({"detail": "Request body too large"}, status_code=413)

    now = time.monotonic()
    for stale in [k for k, v in _hits.items() if not v or v[-1] < now - 60]:
        del _hits[stale]
    hits = _hits.setdefault(_client_ip(request), deque())
    while hits and hits[0] < now - 60:
        hits.popleft()
    if len(hits) >= RATE_LIMIT_PER_MIN:
        return JSONResponse(
            {"detail": "Rate limit exceeded"}, status_code=429, headers={"Retry-After": "60"}
        )
    hits.append(now)
    return await call_next(request)


@app.get("/", include_in_schema=False)
async def root():
    return RedirectResponse(url="/web/")