Spaces:
Running
Running
File size: 2,791 Bytes
071c2c6 662e3eb 071c2c6 662e3eb 071c2c6 8fbdd70 071c2c6 662e3eb 071c2c6 662e3eb 071c2c6 8fbdd70 071c2c6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 | # 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/")
|