feat(security): request-size caps, concurrency limits, model-pull allowlist
DoS/quota guardrails for the unauthenticated local APIs: * Body-size middleware rejects oversized requests (Content-Length) before they are buffered/base64-decoded (NEXUS_MAX_REQUEST_MB, default 32). * Document upload enforces a decoded-byte cap (NEXUS_MAX_UPLOAD_MB, default 20) and a PDF page-count cap (NEXUS_MAX_PDF_PAGES, default 500) as backstops for chunked bodies and pathological files. * A counter-based in-flight limiter bounds concurrent chats and document ingests (NEXUS_MAX_CONCURRENT_CHATS/UPLOADS), returning 429 when saturated; the chat slot is held for the whole SSE stream and released on completion or client disconnect. * /models/pull gains an opt-in allowlist (NEXUS_MODEL_ALLOWLIST); empty by default so behaviour is unchanged, otherwise a bare repo name permits all its tags. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Cursor
+112
-5
@@ -2,6 +2,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio as _asyncio
|
||||
import contextlib as _contextlib
|
||||
import json as _json
|
||||
import secrets as _secrets
|
||||
import uuid as _uuid
|
||||
@@ -13,12 +14,16 @@ import httpx
|
||||
import os as _os
|
||||
import platform as _platform
|
||||
from pathlib import Path
|
||||
from fastapi import FastAPI, HTTPException, Body
|
||||
from fastapi import FastAPI, HTTPException, Body, Request
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from starlette.middleware.trustedhost import TrustedHostMiddleware
|
||||
from fastapi.responses import StreamingResponse, FileResponse
|
||||
from fastapi.responses import StreamingResponse, FileResponse, JSONResponse
|
||||
|
||||
from .nexus_config import settings, VERSION, DEFAULT_CHAT_MODEL, ALLOWED_HOSTS, ALLOWED_ORIGINS
|
||||
from .nexus_config import (
|
||||
settings, VERSION, DEFAULT_CHAT_MODEL, ALLOWED_HOSTS, ALLOWED_ORIGINS,
|
||||
MAX_REQUEST_BYTES, MAX_UPLOAD_BYTES, MAX_PDF_PAGES,
|
||||
MAX_CONCURRENT_CHATS, MAX_CONCURRENT_UPLOADS, model_pull_allowed,
|
||||
)
|
||||
from .chat import generate_chat_response, stream_chat_response, _synapse_trace
|
||||
from . import chat as _chat
|
||||
from .ollama_manager import initialize_ollama, initialize_ollama_async, get_ollama_manager
|
||||
@@ -200,6 +205,63 @@ app.add_middleware(
|
||||
allow_headers=["*"],
|
||||
)
|
||||
|
||||
|
||||
# --- Request-size cap ---
|
||||
# Reject oversized bodies before they are buffered/decoded (a base64 upload or a
|
||||
# giant chat payload could otherwise exhaust memory). Header-based: covers the
|
||||
# realistic clients (browsers/httpx always send Content-Length for JSON). A
|
||||
# chunked request with no length still reaches the handler, where the upload path
|
||||
# enforces its own decoded-byte cap as a backstop.
|
||||
@app.middleware("http")
|
||||
async def _limit_request_body(request: Request, call_next):
|
||||
cl = request.headers.get("content-length")
|
||||
if cl:
|
||||
try:
|
||||
if int(cl) > MAX_REQUEST_BYTES:
|
||||
return JSONResponse(
|
||||
status_code=413,
|
||||
content={"detail": f"request body too large (max {MAX_REQUEST_BYTES} bytes)"},
|
||||
)
|
||||
except ValueError:
|
||||
pass
|
||||
return await call_next(request)
|
||||
|
||||
|
||||
class _InFlightLimiter:
|
||||
"""Bounds concurrent in-flight ops for one expensive entry point so a single
|
||||
caller can't fan out unlimited inference/embedding work. The server runs on
|
||||
one asyncio loop, so a plain counter is safe — there's no await between the
|
||||
check and the increment. Over the limit -> HTTP 429."""
|
||||
|
||||
def __init__(self, limit: int, label: str):
|
||||
self._limit = max(1, int(limit))
|
||||
self._label = label
|
||||
self._n = 0
|
||||
|
||||
def acquire(self) -> None:
|
||||
if self._n >= self._limit:
|
||||
raise HTTPException(
|
||||
status_code=429,
|
||||
detail=f"{self._label} is busy ({self._limit} concurrent max) — retry shortly",
|
||||
)
|
||||
self._n += 1
|
||||
|
||||
def release(self) -> None:
|
||||
if self._n > 0:
|
||||
self._n -= 1
|
||||
|
||||
@_contextlib.contextmanager
|
||||
def slot(self):
|
||||
self.acquire()
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
self.release()
|
||||
|
||||
|
||||
_CHAT_INFLIGHT = _InFlightLimiter(MAX_CONCURRENT_CHATS, "chat")
|
||||
_UPLOAD_INFLIGHT = _InFlightLimiter(MAX_CONCURRENT_UPLOADS, "document ingest")
|
||||
|
||||
# --- GLOBALS ---
|
||||
playbooks = PlaybookManager()
|
||||
ollama = None
|
||||
@@ -245,6 +307,11 @@ async def root():
|
||||
# -------------------------
|
||||
@app.post("/chat/stream")
|
||||
async def chat_stream_endpoint(payload: Dict[str, Any]):
|
||||
# Bound concurrent chats so a flood can't fan out unlimited model inference.
|
||||
# Acquired here (covers the pre-stream RAG/embedding work too) and released
|
||||
# when the SSE stream finishes — see the guarded wrapper below.
|
||||
_CHAT_INFLIGHT.acquire()
|
||||
_chat_slot_held = True
|
||||
try:
|
||||
message = payload.get("message", "")
|
||||
app_settings = store.get_settings()
|
||||
@@ -498,12 +565,29 @@ async def chat_stream_endpoint(payload: Dict[str, Any]):
|
||||
except Exception as e:
|
||||
_synapse_trace(f"\n⚠ memory extraction call failed: {e}\n")
|
||||
|
||||
return StreamingResponse(event_stream(), media_type="text/event-stream")
|
||||
# Hold the concurrency slot for the stream's lifetime, then release it
|
||||
# exactly once when the generator is exhausted or closed (client
|
||||
# disconnect). Ownership passes to the wrapper, so the endpoint's own
|
||||
# finally must not also release.
|
||||
_inner = event_stream()
|
||||
|
||||
async def _guarded_stream() -> AsyncGenerator[str, None]:
|
||||
try:
|
||||
async for _chunk in _inner:
|
||||
yield _chunk
|
||||
finally:
|
||||
_CHAT_INFLIGHT.release()
|
||||
|
||||
_chat_slot_held = False
|
||||
return StreamingResponse(_guarded_stream(), media_type="text/event-stream")
|
||||
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
finally:
|
||||
if _chat_slot_held:
|
||||
_CHAT_INFLIGHT.release()
|
||||
|
||||
|
||||
@app.post("/chat/approve")
|
||||
@@ -684,6 +768,11 @@ async def pull_model(payload: Dict[str, Any] = Body(...)):
|
||||
name = payload.get("name", "").strip()
|
||||
if not name:
|
||||
raise HTTPException(status_code=400, detail="Missing model name")
|
||||
if not model_pull_allowed(name):
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail=f"model '{name}' is not in NEXUS_MODEL_ALLOWLIST",
|
||||
)
|
||||
|
||||
async def _stream():
|
||||
try:
|
||||
@@ -1120,7 +1209,13 @@ def _extract_text(filename: str, data: bytes) -> str:
|
||||
if name.endswith(".pdf"):
|
||||
from pypdf import PdfReader
|
||||
reader = PdfReader(io.BytesIO(data))
|
||||
return "\n\n".join((page.extract_text() or "") for page in reader.pages)
|
||||
pages = reader.pages
|
||||
if len(pages) > MAX_PDF_PAGES:
|
||||
raise HTTPException(
|
||||
status_code=413,
|
||||
detail=f"PDF has {len(pages)} pages (max {MAX_PDF_PAGES})",
|
||||
)
|
||||
return "\n\n".join((page.extract_text() or "") for page in pages)
|
||||
if name.endswith(".docx"):
|
||||
import docx
|
||||
doc = docx.Document(io.BytesIO(data))
|
||||
@@ -1142,8 +1237,20 @@ async def upload_document(payload: Dict[str, Any] = Body(...)):
|
||||
raw = base64.b64decode(b64)
|
||||
except Exception:
|
||||
raise HTTPException(status_code=400, detail="data must be base64")
|
||||
# Decoded-byte backstop: the body-size middleware caps the encoded payload,
|
||||
# but base64 inflates ~33% and a chunked request has no Content-Length, so
|
||||
# enforce the real decoded ceiling here too.
|
||||
if len(raw) > MAX_UPLOAD_BYTES:
|
||||
raise HTTPException(
|
||||
status_code=413,
|
||||
detail=f"file too large: {len(raw)} bytes (max {MAX_UPLOAD_BYTES})",
|
||||
)
|
||||
# Bound concurrent ingests: extract + chunk + embed is CPU/model-heavy.
|
||||
with _UPLOAD_INFLIGHT.slot():
|
||||
try:
|
||||
text = _extract_text(filename, raw).strip()
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=400, detail=f"could not read {filename}: {e}")
|
||||
if not text:
|
||||
|
||||
+40
-1
@@ -166,6 +166,42 @@ _LOCAL_ORIGINS = [
|
||||
ALLOWED_HOSTS = _csv_env("NEXUS_ALLOWED_HOSTS", _LOCAL_HOSTS)
|
||||
ALLOWED_ORIGINS = _csv_env("NEXUS_ALLOWED_ORIGINS", _LOCAL_ORIGINS)
|
||||
|
||||
|
||||
# --- resource limits (DoS guardrails for the unauthenticated local APIs) ---
|
||||
# Even local-only, an unbounded base64 upload or a flood of concurrent inference
|
||||
# requests can exhaust RAM/CPU. These are generous defaults for single-user use,
|
||||
# all env-overridable.
|
||||
def _int_env(name: str, default: int) -> int:
|
||||
try:
|
||||
return int((os.getenv(name) or "").strip() or default)
|
||||
except ValueError:
|
||||
return default
|
||||
|
||||
MAX_REQUEST_BYTES = _int_env("NEXUS_MAX_REQUEST_MB", 32) * 1024 * 1024
|
||||
MAX_UPLOAD_BYTES = _int_env("NEXUS_MAX_UPLOAD_MB", 20) * 1024 * 1024
|
||||
MAX_PDF_PAGES = _int_env("NEXUS_MAX_PDF_PAGES", 500)
|
||||
MAX_CONCURRENT_CHATS = _int_env("NEXUS_MAX_CONCURRENT_CHATS", 4)
|
||||
MAX_CONCURRENT_UPLOADS = _int_env("NEXUS_MAX_CONCURRENT_UPLOADS", 2)
|
||||
|
||||
# Opt-in allowlist for /models/pull. Empty (default) = unrestricted, preserving
|
||||
# current behaviour. Set NEXUS_MODEL_ALLOWLIST=mistral,llama3 to bound which
|
||||
# models can be downloaded; a bare repo name (before the ':tag') matches all of
|
||||
# its tags, so "mistral" permits "mistral:latest", "mistral:7b", etc.
|
||||
MODEL_ALLOWLIST = _csv_env("NEXUS_MODEL_ALLOWLIST", [])
|
||||
|
||||
|
||||
def model_pull_allowed(name: str) -> bool:
|
||||
"""True if `name` may be pulled: always when no allowlist is configured,
|
||||
otherwise when the full name or its repo part (before the first ':') is
|
||||
listed. Case-insensitive."""
|
||||
if not MODEL_ALLOWLIST:
|
||||
return True
|
||||
n = (name or "").strip().lower()
|
||||
if not n:
|
||||
return False
|
||||
allow = {a.lower() for a in MODEL_ALLOWLIST}
|
||||
return n in allow or n.split(":", 1)[0] in allow
|
||||
|
||||
# exported instance
|
||||
settings = Settings()
|
||||
|
||||
@@ -176,7 +212,10 @@ __all__ = ["Settings", "settings", "path", "VERSION",
|
||||
"MEMORY_DIR", "LOGS_DIR", "PLAYBOOK_DIR", "UPLOADS_DIR",
|
||||
"EXPORTS_DIR", "MEMORY_DB",
|
||||
"BACKEND_LOG", "OLLAMA_LOG", "CHAT_LOG",
|
||||
"ALLOWED_HOSTS", "ALLOWED_ORIGINS"]
|
||||
"ALLOWED_HOSTS", "ALLOWED_ORIGINS",
|
||||
"MAX_REQUEST_BYTES", "MAX_UPLOAD_BYTES", "MAX_PDF_PAGES",
|
||||
"MAX_CONCURRENT_CHATS", "MAX_CONCURRENT_UPLOADS",
|
||||
"MODEL_ALLOWLIST", "model_pull_allowed"]
|
||||
|
||||
# --- quick runtime sanity check when run directly (no side effects on import) ---
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user
GitNexus · part of the Nexus stack