# --------------------------------------------------------------------------- # # Helper: verify that the URL is allowed (whitelist, scheme, size limit, etc.) # --------------------------------------------------------------------------- # def _validate_url(url: str) -> urllib.parse.ParseResult: try: parsed = urllib.parse.urlparse(url) except Exception as exc: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid URL: exc", )
# Optional: disallow certain file extensions (example) @validator("url") def disallow_exe(cls, v: HttpUrl): if v.path.lower().endswith(".exe"): raise ValueError("Executable files are not allowed.") return v heretic webdl
class Settings: # ---- security / abuse limits ------------------------------------------------- MAX_CONTENT_LENGTH = int(os.getenv("MAX_CONTENT_LENGTH", "104857600")) # 100 MiB # Optional whitelist (comma‑separated). If empty → allow any domain. WHITELISTED_DOMAINS = set( filter(None, os.getenv("WHITELISTED_DOMAINS", "").split(",")) ) # Rate limiting – simple token bucket stored in memory (good enough for free tier) RATE_LIMIT = int(os.getenv("RATE_LIMIT", "10")) # requests per minute per IP # Using httpx HEAD request (fast, no body)
# --------------------------------------------------- # # Simple in‑memory rate limiter (IP → timestamps) # --------------------------------------------------- # _rate_limiter: Dict[str, list[datetime]] = defaultdict(list) # Using httpx HEAD request (fast
# We need to fetch the headers *once* to get the MIME type. # Using httpx HEAD request (fast, no body) – fallback to generic if unavailable. import httpx async with httpx.AsyncClient(follow_redirects=True, timeout=10) as client: try: head = await client.head(payload.url, follow_redirects=True) if head.status_code == 200: ct = head.headers.get("Content-Type") if ct: headers["Content-Type"] = ct except Exception: # ignore – we will stream anyway pass
return StreamingResponse( generator(), media_type=headers.get("Content-Type", "application/octet-stream"), headers=headers, )