Security: request-size caps, concurrency limits, model-pull allowlist #3
@@ -18,6 +18,7 @@ export function Chatbot({ conversationId, setConversationId, onConversationChang
|
||||
const [images, setImages] = useState([]); // {name, b64} for vision models
|
||||
const [activeTool, setActiveTool] = useState(null); // playbook tool currently running
|
||||
const [pendingApproval, setPendingApproval] = useState(null); // [{name, arguments}] awaiting yes/no
|
||||
const [approvalToken, setApprovalToken] = useState(null); // single-use token authorizing /chat/approve
|
||||
const [editingIdx, setEditingIdx] = useState(null); // user message being edited
|
||||
const [editText, setEditText] = useState("");
|
||||
const [listening, setListening] = useState(false); // mic dictation active
|
||||
@@ -286,7 +287,11 @@ export function Chatbot({ conversationId, setConversationId, onConversationChang
|
||||
continue;
|
||||
}
|
||||
if (pendingEventType === "tool_request") {
|
||||
try { setPendingApproval(JSON.parse(payload)); } catch { /* ignore */ }
|
||||
try {
|
||||
const parsed = JSON.parse(payload);
|
||||
setPendingApproval(parsed.actions || []);
|
||||
setApprovalToken(parsed.token || null);
|
||||
} catch { /* ignore */ }
|
||||
pendingEventType = null;
|
||||
continue;
|
||||
}
|
||||
@@ -401,13 +406,15 @@ export function Chatbot({ conversationId, setConversationId, onConversationChang
|
||||
// Approve or deny the pending action tool(s); the open chat stream resumes.
|
||||
const resolveApproval = async (approve) => {
|
||||
const req = pendingApproval || [];
|
||||
const token = approvalToken;
|
||||
setPendingApproval(null);
|
||||
setApprovalToken(null);
|
||||
const decisions = {};
|
||||
req.forEach(a => { decisions[a.name] = approve; });
|
||||
try {
|
||||
await fetch(`${API_BASE}/chat/approve`, {
|
||||
method: "POST", headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ conversation_id: conversationId, decisions }),
|
||||
body: JSON.stringify({ conversation_id: conversationId, token, decisions }),
|
||||
});
|
||||
} catch { /* ignore */ }
|
||||
};
|
||||
|
||||
+7
-1
@@ -112,6 +112,12 @@ class Service:
|
||||
return self._argv() if callable(self._argv) else self._argv
|
||||
|
||||
|
||||
# Bind loopback by default: the backend/memory REST APIs are unauthenticated, so
|
||||
# binding 0.0.0.0 handed the full admin+data plane to any host on the LAN. Set
|
||||
# NEXUS_BIND_HOST=0.0.0.0 to opt into LAN exposure once real auth is in place.
|
||||
BIND_HOST = os.environ.get("NEXUS_BIND_HOST", "127.0.0.1")
|
||||
|
||||
|
||||
def _uvicorn(app: str, port: int):
|
||||
# No --reload. It is a dev-loop flag: uvicorn's reloader runs a supervisor
|
||||
# that spawns the real server as a CHILD, so every service became two
|
||||
@@ -120,7 +126,7 @@ def _uvicorn(app: str, port: int):
|
||||
# terminals for what should have been a silent start. launch_nexus.sh still
|
||||
# passes --reload for the Linux dev loop, where a visible console is the
|
||||
# point; this launcher is the one users run.
|
||||
return [str(PYTHON), "-m", "uvicorn", app, "--host", "0.0.0.0",
|
||||
return [str(PYTHON), "-m", "uvicorn", app, "--host", BIND_HOST,
|
||||
"--port", str(port)]
|
||||
|
||||
|
||||
|
||||
+15
-6
@@ -3,6 +3,7 @@ from __future__ import annotations
|
||||
import asyncio
|
||||
import json as _json
|
||||
import logging
|
||||
import secrets
|
||||
import threading
|
||||
from typing import AsyncGenerator, Dict, List, Optional, Any
|
||||
|
||||
@@ -171,12 +172,20 @@ async def _run_tool_loop(manager, messages, model, tool_schemas, temperature, nu
|
||||
action_calls = [c for c in calls if _tools.is_action(c.get("function", {}).get("name", ""))]
|
||||
if policy == "ask" and action_calls:
|
||||
event = asyncio.Event()
|
||||
pending_approvals[conversation_id] = {"event": event, "decisions": {}}
|
||||
yield "__approve__" + _json.dumps([
|
||||
{"name": c.get("function", {}).get("name", ""),
|
||||
"arguments": c.get("function", {}).get("arguments")}
|
||||
for c in action_calls
|
||||
])
|
||||
# Single-use capability token, delivered only to the client that owns
|
||||
# this stream. /chat/approve requires it, so knowing the (guessable,
|
||||
# enumerable) conversation_id is no longer enough to approve someone
|
||||
# else's pending action.
|
||||
token = secrets.token_urlsafe(32)
|
||||
pending_approvals[conversation_id] = {"event": event, "decisions": {}, "token": token}
|
||||
yield "__approve__" + _json.dumps({
|
||||
"token": token,
|
||||
"actions": [
|
||||
{"name": c.get("function", {}).get("name", ""),
|
||||
"arguments": c.get("function", {}).get("arguments")}
|
||||
for c in action_calls
|
||||
],
|
||||
})
|
||||
try:
|
||||
await asyncio.wait_for(event.wait(), timeout=_APPROVAL_TIMEOUT)
|
||||
decisions = pending_approvals[conversation_id]["decisions"]
|
||||
|
||||
+143
-19
@@ -2,7 +2,9 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio as _asyncio
|
||||
import contextlib as _contextlib
|
||||
import json as _json
|
||||
import secrets as _secrets
|
||||
import uuid as _uuid
|
||||
from collections import Counter as _Counter
|
||||
from typing import Any, AsyncGenerator, Dict, List, Optional, Tuple
|
||||
@@ -12,11 +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 fastapi.responses import StreamingResponse, FileResponse
|
||||
from starlette.middleware.trustedhost import TrustedHostMiddleware
|
||||
from fastapi.responses import StreamingResponse, FileResponse, JSONResponse
|
||||
|
||||
from .nexus_config import settings, VERSION, DEFAULT_CHAT_MODEL
|
||||
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
|
||||
@@ -179,18 +186,82 @@ app = FastAPI(title="Synapse Backend", version=VERSION)
|
||||
# Alias for startup scripts
|
||||
sio_app = app
|
||||
|
||||
# --- CORS ---
|
||||
# allow_credentials=True is incompatible with allow_origins=["*"] — browsers
|
||||
# reject such responses. Since this is a local-only service with no cookies/auth,
|
||||
# wildcard origins without credentials is correct.
|
||||
# --- Local-access guard ---
|
||||
# These APIs are unauthenticated and meant for this machine only. Two layers:
|
||||
# 1. TrustedHostMiddleware rejects a foreign Host header, which is what defeats
|
||||
# DNS-rebinding — the browser treats a rebound attacker domain as
|
||||
# same-origin, so CORS alone can't stop it.
|
||||
# 2. CORS is scoped to known local origins (not "*"), so a malicious page can't
|
||||
# read responses cross-origin from the victim's browser.
|
||||
# NEXUS_ALLOWED_HOSTS=* / a custom NEXUS_ALLOWED_ORIGINS opts into LAN exposure,
|
||||
# and should only be used once real authentication is in front of these apps.
|
||||
if "*" not in ALLOWED_HOSTS:
|
||||
app.add_middleware(TrustedHostMiddleware, allowed_hosts=ALLOWED_HOSTS)
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["*"],
|
||||
allow_origins=ALLOWED_ORIGINS,
|
||||
allow_credentials=False,
|
||||
allow_methods=["*"],
|
||||
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
|
||||
@@ -236,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()
|
||||
@@ -489,23 +565,48 @@ 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")
|
||||
async def chat_approve(payload: Dict[str, Any] = Body(...)):
|
||||
"""Resolve a pending per-call tool approval. `decisions` maps tool name ->
|
||||
bool; the awaiting chat stream resumes and runs the approved actions."""
|
||||
bool; the awaiting chat stream resumes and runs the approved actions.
|
||||
|
||||
The `token` (issued in the stream's tool_request event) is required: without
|
||||
it, anyone who can guess/enumerate a conversation_id could approve another
|
||||
client's pending action. Compared in constant time."""
|
||||
conversation_id = payload.get("conversation_id") or ""
|
||||
token = payload.get("token") or ""
|
||||
decisions = payload.get("decisions") or {}
|
||||
waiter = _chat.pending_approvals.get(conversation_id)
|
||||
if not waiter:
|
||||
raise HTTPException(status_code=404, detail="no pending approval for this conversation")
|
||||
expected = waiter.get("token") or ""
|
||||
if not token or not _secrets.compare_digest(str(token), str(expected)):
|
||||
raise HTTPException(status_code=403, detail="invalid or missing approval token")
|
||||
waiter["decisions"] = {k: bool(v) for k, v in decisions.items()}
|
||||
waiter["event"].set()
|
||||
return {"status": "resumed"}
|
||||
@@ -667,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:
|
||||
@@ -1103,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))
|
||||
@@ -1125,14 +1237,26 @@ 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")
|
||||
try:
|
||||
text = _extract_text(filename, raw).strip()
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=400, detail=f"could not read {filename}: {e}")
|
||||
if not text:
|
||||
raise HTTPException(status_code=400, detail="no extractable text in file")
|
||||
title = _os.path.splitext(_os.path.basename(filename))[0] or filename
|
||||
return await store.add_document(title, text, get_ollama_manager().embed, _active_project())
|
||||
# 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:
|
||||
raise HTTPException(status_code=400, detail="no extractable text in file")
|
||||
title = _os.path.splitext(_os.path.basename(filename))[0] or filename
|
||||
return await store.add_document(title, text, get_ollama_manager().embed, _active_project())
|
||||
|
||||
|
||||
@app.get("/documents/{doc_id}")
|
||||
|
||||
@@ -19,17 +19,24 @@ from typing import Any, Dict, List, Optional
|
||||
|
||||
from fastapi import FastAPI, HTTPException, Body
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from starlette.middleware.trustedhost import TrustedHostMiddleware
|
||||
from pydantic import BaseModel
|
||||
|
||||
from .store import store, MemoryItem
|
||||
from .extractor import extract_memory
|
||||
from ..ollama_manager import get_ollama_manager
|
||||
from ..nexus_config import ALLOWED_HOSTS, ALLOWED_ORIGINS
|
||||
|
||||
app = FastAPI(title="Nexus Memory Service", version="1.0")
|
||||
|
||||
# Same unauthenticated-local-only posture as the main backend: reject foreign
|
||||
# Host headers (anti DNS-rebind) and scope CORS to known local origins rather
|
||||
# than "*". See nexus_config.ALLOWED_HOSTS / ALLOWED_ORIGINS for env overrides.
|
||||
if "*" not in ALLOWED_HOSTS:
|
||||
app.add_middleware(TrustedHostMiddleware, allowed_hosts=ALLOWED_HOSTS)
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["*"],
|
||||
allow_origins=ALLOWED_ORIGINS,
|
||||
allow_credentials=False,
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
|
||||
+64
-1
@@ -143,6 +143,65 @@ class Settings:
|
||||
"ollama_timeout": self.ollama_timeout,
|
||||
}
|
||||
|
||||
# --- local-access allowlists (shared by the backend + memory FastAPI apps) ---
|
||||
# The REST APIs are unauthenticated, so they are meant to be reached only from
|
||||
# this machine. Two independent browser-side defenses depend on these lists:
|
||||
# * ALLOWED_ORIGINS drives CORS — blocks a malicious page from *reading*
|
||||
# responses cross-origin (was previously "*", which let any site read them).
|
||||
# * ALLOWED_HOSTS drives TrustedHostMiddleware — rejects a foreign Host header,
|
||||
# which is what stops DNS-rebinding (same-origin from the browser's view, so
|
||||
# CORS can't help there).
|
||||
# Both accept a comma-separated env override for the intentional-LAN case, to be
|
||||
# paired with real auth. NEXUS_ALLOWED_HOSTS=* disables the Host check.
|
||||
def _csv_env(name: str, default: list) -> list:
|
||||
raw = os.getenv(name, "").strip()
|
||||
return [p.strip() for p in raw.split(",") if p.strip()] if raw else list(default)
|
||||
|
||||
_LOCAL_HOSTS = ["localhost", "127.0.0.1", "[::1]", "::1", "testserver"]
|
||||
_LOCAL_ORIGINS = [
|
||||
f"http://{h}:{p}"
|
||||
for h in ("localhost", "127.0.0.1")
|
||||
for p in (8000, 8001, 5173)
|
||||
]
|
||||
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()
|
||||
|
||||
@@ -152,7 +211,11 @@ __all__ = ["Settings", "settings", "path", "VERSION",
|
||||
"PROJECT_ROOT", "DATA_DIR", "MODELS_DIR", "RUNTIME_DIR",
|
||||
"MEMORY_DIR", "LOGS_DIR", "PLAYBOOK_DIR", "UPLOADS_DIR",
|
||||
"EXPORTS_DIR", "MEMORY_DB",
|
||||
"BACKEND_LOG", "OLLAMA_LOG", "CHAT_LOG"]
|
||||
"BACKEND_LOG", "OLLAMA_LOG", "CHAT_LOG",
|
||||
"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__":
|
||||
|
||||
+70
-7
@@ -60,20 +60,83 @@ async def _web_search(query: str = "", **_) -> str:
|
||||
return res or "(no results)"
|
||||
|
||||
|
||||
_FETCH_MAX_REDIRECTS = 5
|
||||
|
||||
|
||||
def _ip_is_blocked(ip: str) -> bool:
|
||||
"""True if an address is one an outbound fetch has no business reaching:
|
||||
loopback, RFC1918/ULA private, link-local (incl. 169.254.169.254 cloud
|
||||
metadata), multicast, reserved, or unspecified. IPv4-mapped IPv6 is unwrapped
|
||||
first so ::ffff:127.0.0.1 can't sneak a loopback past the check."""
|
||||
import ipaddress
|
||||
try:
|
||||
addr = ipaddress.ip_address(ip.split("%")[0]) # drop any IPv6 zone id
|
||||
except ValueError:
|
||||
return True # unparseable -> refuse rather than guess
|
||||
mapped = getattr(addr, "ipv4_mapped", None)
|
||||
if mapped is not None:
|
||||
addr = mapped
|
||||
return (
|
||||
addr.is_loopback or addr.is_private or addr.is_link_local
|
||||
or addr.is_multicast or addr.is_reserved or addr.is_unspecified
|
||||
)
|
||||
|
||||
|
||||
def _ssrf_guard(host: str) -> str | None:
|
||||
"""Resolve a hostname and return an error string if ANY of its A/AAAA
|
||||
records is a blocked address, else None. Checking every answer stops a name
|
||||
from smuggling one private record alongside a public one.
|
||||
|
||||
ponytail: this validates then httpx re-resolves on connect, so a sub-second
|
||||
DNS-rebind could still slip a private address through the TOCTOU gap. That's
|
||||
an advanced attack against a playbook-gated, single-user tool; pin the
|
||||
connection to the resolved IP if this ever faces untrusted callers."""
|
||||
import socket
|
||||
if not host:
|
||||
return "missing host"
|
||||
try:
|
||||
infos = socket.getaddrinfo(host, None)
|
||||
except socket.gaierror as e:
|
||||
return f"cannot resolve host: {e}"
|
||||
ips = {info[4][0] for info in infos}
|
||||
if not ips:
|
||||
return "host did not resolve"
|
||||
blocked = [ip for ip in ips if _ip_is_blocked(ip)]
|
||||
if blocked:
|
||||
return f"refusing to fetch a private/loopback/link-local address ({', '.join(sorted(blocked))})"
|
||||
return None
|
||||
|
||||
|
||||
async def _fetch_url(url: str = "", **_) -> str:
|
||||
import re
|
||||
import httpx
|
||||
from urllib.parse import urlparse, urljoin
|
||||
url = (url or "").strip()
|
||||
if not url.startswith(("http://", "https://")):
|
||||
return json.dumps({"error": "url must start with http:// or https://"})
|
||||
# ponytail: no SSRF allow/deny-list — local single-user assistant, and the
|
||||
# tool only runs when a playbook explicitly grants fetch_url. Add host
|
||||
# filtering if this ever serves multiple/untrusted users.
|
||||
# SSRF guard: validate the host of the initial URL AND every redirect hop
|
||||
# against the private/loopback/link-local block-list before connecting, so a
|
||||
# granted fetch_url can't be steered at 127.0.0.1:11434, cloud metadata, or
|
||||
# LAN hosts — and a public URL can't 302 its way there either.
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as c:
|
||||
r = await c.get(url, headers={"User-Agent": "NexusOS/1.0"})
|
||||
r.raise_for_status()
|
||||
html = r.text
|
||||
async with httpx.AsyncClient(timeout=15.0, follow_redirects=False) as c:
|
||||
for _ in range(_FETCH_MAX_REDIRECTS + 1):
|
||||
parsed = urlparse(url)
|
||||
if parsed.scheme not in ("http", "https"):
|
||||
return json.dumps({"error": "only http(s) URLs are allowed"})
|
||||
err = _ssrf_guard(parsed.hostname or "")
|
||||
if err:
|
||||
return json.dumps({"error": f"blocked: {err}"})
|
||||
r = await c.get(url, headers={"User-Agent": "NexusOS/1.0"})
|
||||
location = r.headers.get("location")
|
||||
if r.is_redirect and location:
|
||||
url = urljoin(url, location)
|
||||
continue
|
||||
r.raise_for_status()
|
||||
html = r.text
|
||||
break
|
||||
else:
|
||||
return json.dumps({"error": "too many redirects"})
|
||||
except Exception as e:
|
||||
return json.dumps({"error": f"fetch failed: {e}"})
|
||||
text = re.sub(r"(?is)<(script|style).*?</\1>", " ", html)
|
||||
|
||||
Reference in New Issue
Block a user