diff --git a/THREAT_MODEL.md b/THREAT_MODEL.md
index ee656087c..d63d1335b 100644
--- a/THREAT_MODEL.md
+++ b/THREAT_MODEL.md
@@ -60,6 +60,19 @@ External content that reaches the LLM is treated as untrusted via `src/prompt_se
**Untrusted surfaces that must go through this wrapper:** web search results, fetched URLs, emails (read), saved memories, skill text, notes, and any tool output sourced from outside the server. Injecting untrusted content directly into the system role is a security bug.
+### Post-external-context tool approval gate — off by default
+
+`src/tool_capabilities.py` carries a second layer: once untrusted content has entered a run, `ToolRunSecurityContext.decision_for()` blocks tools that execute code, mutate state, or cause external side effects until the user authorises the action separately.
+
+**It is disabled unless `ODYSSEUS_TOOL_APPROVAL_GATE` is set** (`1`/`true`/`yes`/`on`). The default is off because the gate is conservative enough to interrupt ordinary agent work. That is a deliberate usability trade, and it means a default deployment relies on the wrapper above — not on the gate — to contain injected instructions.
+
+Operators who run the agent against untrusted web or email content with side-effecting tools enabled should turn it on. With the gate off, a successful injection can reach `bash`, `host_shell`, `send_email` and `delete_email` without a separate confirmation; with it on, each of those is refused until approved.
+
+Two exemptions apply even when the gate is on, both deliberate:
+
+- Sources in `_CONTROL_PLANE_CONTEXT_SOURCES` (skills, runtime descriptors, the open editor document, the open email, uploaded files) are treated as control-plane metadata and still permit read-only tools.
+- A TUI run that advertises a host shell bridge and declares `unattended_mode` exempts the local execution set in `TUI_CLIENT_TOOL_NAMES`. Personal, network and deployment-local tools are never exempted.
+
## Security Headers
`core/middleware.py:SecurityHeadersMiddleware` sets headers on every response:
@@ -72,7 +85,7 @@ External content that reaches the LLM is treated as untrusted via `src/prompt_se
These are open, acknowledged, and contributor help is welcome:
-1. **No shell/filesystem sandbox.** The agent `bash` and `read_file`/`write_file` tools run as the app process user with no network egress filtering or filesystem confinement. A successful prompt-injection reaching a shell-enabled admin session can make outbound requests to internal services. See #1058 for the sandbox proposal.
+1. **No shell/filesystem sandbox.** The agent `bash` and `read_file`/`write_file` tools run as the app process user with no network egress filtering or filesystem confinement. A successful prompt-injection reaching a shell-enabled admin session can make outbound requests to internal services. See #1058 for the sandbox proposal. The tool approval gate above is the compensating control, and it is off by default — so on a default deployment this gap is unmitigated beyond the untrusted-context wrapper.
2. **SSRF via `/api/v1/chat` `base_url` parameter.** A chat-scoped API token can supply an arbitrary `base_url`; the server forwards the LLM request to that host without validating the scheme or address. PR #1039 fixes this.
diff --git a/routes/editor_draft_routes.py b/routes/editor_draft_routes.py
index cf1e5b0d9..ece6a8d8a 100644
--- a/routes/editor_draft_routes.py
+++ b/routes/editor_draft_routes.py
@@ -19,9 +19,10 @@ Each draft carries:
import json
import logging
import uuid
-from typing import Any, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional
-from fastapi import APIRouter, HTTPException, Request
+from fastapi import APIRouter, HTTPException, Request, Response
+from fastapi.routing import APIRoute
from pydantic import BaseModel
from core.database import EditorDraft, SessionLocal
@@ -76,18 +77,56 @@ def _load_payload(raw: Optional[str]) -> Dict[str, Any]:
return payload if isinstance(payload, dict) else {}
+def _draft_too_large() -> HTTPException:
+ return HTTPException(
+ 413,
+ f"Editor draft exceeds the {EDITOR_DRAFT_MAX_BYTES // (1024 * 1024)} MB safety limit",
+ )
+
+
+def reject_oversized_draft_body(request: Request) -> None:
+ """Refuse an oversized draft on declared ``Content-Length``, before the body is read or parsed.
+
+ ``_dump_payload`` still owns the authoritative byte count, but it only runs
+ after the request has been parsed and re-serialised. Declaring a body past the
+ ceiling is enough to reject it early and cheaply. Requests with absent,
+ malformed, or chunked transfer encoding still hit the authoritative byte count
+ check further down.
+ """
+ raw_length = request.headers.get("content-length")
+ if not raw_length:
+ return
+ try:
+ declared = int(raw_length)
+ except (TypeError, ValueError):
+ return
+ if declared > EDITOR_DRAFT_MAX_BYTES:
+ raise _draft_too_large()
+
+
+class EditorDraftRoute(APIRoute):
+ """Route class that validates declared Content-Length before request body parsing."""
+
+ def get_route_handler(self) -> Callable:
+ original_route_handler = super().get_route_handler()
+
+ async def custom_route_handler(request: Request) -> Response:
+ if request.method in ("POST", "PUT", "PATCH"):
+ reject_oversized_draft_body(request)
+ return await original_route_handler(request)
+
+ return custom_route_handler
+
+
def _dump_payload(payload: Dict[str, Any]) -> str:
raw = json.dumps(payload or {}, separators=(",", ":"))
if len(raw.encode("utf-8")) > EDITOR_DRAFT_MAX_BYTES:
- raise HTTPException(
- 413,
- f"Editor draft exceeds the {EDITOR_DRAFT_MAX_BYTES // (1024 * 1024)} MB safety limit",
- )
+ raise _draft_too_large()
return raw
def setup_editor_draft_routes() -> APIRouter:
- router = APIRouter(tags=["editor-drafts"])
+ router = APIRouter(tags=["editor-drafts"], route_class=EditorDraftRoute)
@router.get("/api/editor-drafts")
async def list_drafts(request: Request) -> Dict[str, List[Dict[str, Any]]]:
@@ -120,7 +159,10 @@ def setup_editor_draft_routes() -> APIRouter:
db.close()
@router.post("/api/editor-drafts")
- async def create_draft(request: Request, body: DraftCreate) -> Dict[str, Any]:
+ async def create_draft(
+ request: Request,
+ body: DraftCreate,
+ ) -> Dict[str, Any]:
user = get_current_user(request)
db = SessionLocal()
try:
@@ -148,7 +190,11 @@ def setup_editor_draft_routes() -> APIRouter:
db.close()
@router.put("/api/editor-drafts/{draft_id}")
- async def update_draft(request: Request, draft_id: str, body: DraftUpdate) -> Dict[str, Any]:
+ async def update_draft(
+ request: Request,
+ draft_id: str,
+ body: DraftUpdate,
+ ) -> Dict[str, Any]:
user = get_current_user(request)
db = SessionLocal()
try:
diff --git a/services/search/core.py b/services/search/core.py
index dc97bfb61..bd7d934f0 100644
--- a/services/search/core.py
+++ b/services/search/core.py
@@ -3,11 +3,19 @@
import json
import logging
import re
+import time
import xml.etree.ElementTree as ET
from concurrent.futures import ThreadPoolExecutor, as_completed
+from contextlib import contextmanager
+from contextvars import ContextVar
from datetime import datetime, timedelta
from typing import Dict, Any, Optional, List, Set
from urllib.parse import urlparse
+from src.constants import (
+ ARXIV_API_URL,
+ OPENALEX_API_URL,
+ SCHOLARLY_LOOKUP_TOTAL_BUDGET,
+)
from src.search_passages import search_excerpt
import httpx
@@ -489,22 +497,106 @@ def _result_strongly_matches_title(title: str, result: dict) -> bool:
return overlap >= (1.0 if len(wanted) == 2 else 0.8)
+def _scholarly_user_agent() -> str:
+ """Identify this build to the scholarly APIs using the real app version."""
+ from src.constants import APP_VERSION
+
+ return f"Odysseus/{APP_VERSION} scholarly-title-resolver"
+
+
+_scholarly_deadline: ContextVar[Optional[float]] = ContextVar(
+ "scholarly_deadline", default=None
+)
+
+
+@contextmanager
+def _scholarly_budget():
+ """Open one wall-clock budget shared by every hop of a lookup chain."""
+ token = _scholarly_deadline.set(
+ time.monotonic() + SCHOLARLY_LOOKUP_TOTAL_BUDGET
+ )
+ try:
+ yield
+ finally:
+ _scholarly_deadline.reset(token)
+
+
+MAX_SCHOLARLY_REDIRECTS = 3
+
+
+def _scholarly_api_get(url: str, params: dict) -> Optional[httpx.Response]:
+ """GET a scholarly metadata API under the shared outbound policy.
+
+ Returns ``None`` when any destination URL fails the outbound check or the
+ caller's budget is already spent, so callers degrade to their next source
+ instead of raising. Bounded manual redirects ensure every hop passes
+ through ``check_outbound_url`` before the destination is contacted.
+ """
+ from src.constants import SCHOLARLY_LOOKUP_TIMEOUT
+ from src.url_safety import check_outbound_url
+
+ current_url = url
+ current_params: Optional[dict] = params
+
+ for _ in range(MAX_SCHOLARLY_REDIRECTS + 1):
+ ok, reason = check_outbound_url(current_url, block_private=True)
+ if not ok:
+ logger.warning("Scholarly lookup blocked for %s: %s", current_url, reason)
+ return None
+
+ timeout = SCHOLARLY_LOOKUP_TIMEOUT
+ deadline = _scholarly_deadline.get()
+ if deadline is not None:
+ remaining = deadline - time.monotonic()
+ if remaining <= 0:
+ logger.info("Scholarly lookup budget exhausted before %s", current_url)
+ return None
+ timeout = min(timeout, remaining)
+
+ response = httpx.get(
+ current_url,
+ params=current_params,
+ headers={"User-Agent": _scholarly_user_agent()},
+ timeout=timeout,
+ follow_redirects=False,
+ )
+
+ is_redirect = getattr(response, "is_redirect", False) or (
+ getattr(response, "status_code", None) in (301, 302, 303, 307, 308)
+ )
+ if is_redirect:
+ headers = getattr(response, "headers", {})
+ location = headers.get("location")
+ if not location:
+ logger.warning(
+ "Scholarly redirect missing Location header from %s", current_url
+ )
+ return None
+ current_url = str(httpx.URL(str(response.url)).join(location))
+ current_params = None
+ continue
+
+ response.raise_for_status()
+ return response
+
+ logger.warning("Scholarly lookup exceeded max redirects from %s", url)
+ return None
+
+
def _arxiv_title_results(title: str, count: int = 3) -> list[dict]:
"""Resolve a paper title through arXiv's public Atom API."""
try:
- response = httpx.get(
- "https://export.arxiv.org/api/query",
- params={
+ response = _scholarly_api_get(
+ ARXIV_API_URL,
+ {
"search_query": f'ti:"{title}"',
"start": 0,
"max_results": max(1, min(int(count), 5)),
},
- headers={"User-Agent": "Odysseus/0.20 scholarly-title-resolver"},
- timeout=12.0,
- follow_redirects=True,
)
- response.raise_for_status()
+ if response is None:
+ return []
root = ET.fromstring(response.text)
except Exception as exc:
logger.info("arXiv title lookup failed for %r: %s", title, exc)
@@ -541,20 +633,18 @@ def _openalex_title_results(title: str, count: int = 3) -> list[dict]:
# OpenAlex treats a literal question mark as query syntax and returns
# HTTP 400 for otherwise valid titles such as "How Far ... GPT-4V?".
search_title = re.sub(r"[?]+", " ", str(title or "")).strip()
- response = httpx.get(
- "https://api.openalex.org/works",
- params={
+ response = _scholarly_api_get(
+ OPENALEX_API_URL,
+ {
"search": search_title,
"per-page": max(1, min(int(count), 5)),
"select": (
"display_name,doi,primary_location,publication_year,type"
),
},
- headers={"User-Agent": "Odysseus/0.20 scholarly-title-resolver"},
- timeout=12.0,
- follow_redirects=True,
)
- response.raise_for_status()
+ if response is None:
+ return []
payload = response.json()
except Exception as exc:
logger.info("OpenAlex title lookup failed for %r: %s", title, exc)
@@ -597,8 +687,16 @@ def _openalex_title_results(title: str, count: int = 3) -> list[dict]:
def _scholarly_title_results(title: str, count: int = 3) -> list[dict]:
- """Retry a noisy scholarly query as a bare title, then use arXiv API."""
+ """Retry a noisy scholarly query as a bare title, then use arXiv API.
+ The three hops share one wall-clock budget so a slow upstream cannot hold a
+ user-facing search open for the sum of every per-request timeout.
+ """
+ with _scholarly_budget():
+ return _scholarly_title_results_inner(title, count)
+
+
+def _scholarly_title_results_inner(title: str, count: int) -> list[dict]:
try:
simplified = searxng_search_api(title, count=max(3, count))
except Exception as exc:
@@ -621,10 +719,11 @@ def _direct_scholarly_title_results(title: str, count: int = 3) -> list[dict]:
# OpenAlex typically resolves titles in under a second and often returns
# the official arXiv landing page. The arXiv API remains the fallback.
- openalex = _openalex_title_results(title, count)
- if openalex:
- return openalex
- return _arxiv_title_results(title, count)
+ with _scholarly_budget():
+ openalex = _openalex_title_results(title, count)
+ if openalex:
+ return openalex
+ return _arxiv_title_results(title, count)
def _augment_scholarly_results(query: str, results: list[dict], count: int) -> list[dict]:
diff --git a/src/constants.py b/src/constants.py
index 2b4919d3d..633141b22 100644
--- a/src/constants.py
+++ b/src/constants.py
@@ -140,6 +140,16 @@ LLM_HOSTS = [h.strip() for h in os.getenv("LLM_HOSTS", "").split(",") if h.strip
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
SEARXNG_INSTANCE = os.getenv("SEARXNG_INSTANCE", "http://localhost:8080")
+# Scholarly title resolution. These are the only third-party metadata APIs the
+# search path calls directly, so they get named constants rather than literals
+# repeated at each call site. The budget bounds the whole SearXNG -> OpenAlex ->
+# arXiv chain: each hop used to get its own full timeout, so one scholarly query
+# could stall a user-facing search for the sum of all three.
+ARXIV_API_URL = "https://export.arxiv.org/api/query"
+OPENALEX_API_URL = "https://api.openalex.org/works"
+SCHOLARLY_LOOKUP_TIMEOUT = 12.0
+SCHOLARLY_LOOKUP_TOTAL_BUDGET = 20.0
+
# Cleanup configuration
CLEANUP_ENABLED = os.getenv("CLEANUP_ENABLED", "True").lower() == "true"
CLEANUP_INTERVAL_HOURS = int(os.getenv("CLEANUP_INTERVAL_HOURS", "24"))
diff --git a/src/url_safety.py b/src/url_safety.py
index c8bf86327..e0f66cfeb 100644
--- a/src/url_safety.py
+++ b/src/url_safety.py
@@ -16,6 +16,12 @@ break the primary use case. What it *always* rejects:
For exposed multi-tenant deployments, set ``EMBEDDING_BLOCK_PRIVATE_IPS=true`` to
additionally reject all private and loopback targets (full SSRF lockdown).
+
+On a DNS64/NAT64 network an IPv4-only host resolves to the RFC 6052 Well-Known
+Prefix ``64:ff9b::/96``. Such an address is decoded to the IPv4 destination the
+translator will actually contact, and that destination is then judged under the
+strict policy — so the prefix reaches public IPv4 but never tunnels to loopback,
+private, shared or link-local space.
"""
import ipaddress
@@ -33,6 +39,39 @@ ALLOWED_SCHEMES = ("http", "https")
# versions for other special ranges.
_SHARED_ADDRESS_SPACE_V4 = ipaddress.ip_network("100.64.0.0/10")
+# RFC 6052 §2.1 Well-Known Prefix for IPv4/IPv6 address translation (NAT64).
+# An address inside exactly this /96 is not a destination in its own right: the
+# low 32 bits carry the IPv4 address the translator will actually contact. On a
+# DNS64/NAT64 network every public IPv4-only host resolves this way, so judging
+# the outer IPv6 (which CPython reports as ``is_reserved``) would reject the
+# whole public internet while telling us nothing about the real target.
+#
+# RFC 6052 §3.1 allows the Well-Known Prefix to represent *only* globally
+# routable IPv4. The embedded destination is therefore always evaluated under
+# the strict policy, whatever ``block_private`` the caller passed: the prefix
+# must never become a path to loopback, private, shared, link-local, multicast,
+# unspecified or otherwise non-global space.
+#
+# Deliberately exact. Network-specific prefixes carry locally assigned meaning
+# and are NOT decoded here — notably 64:ff9b:1::/48 (RFC 8215), which this /96
+# membership test excludes and which stays rejected as reserved.
+_NAT64_WELL_KNOWN_PREFIX_V6 = ipaddress.ip_network("64:ff9b::/96")
+
+
+def _nat64_well_known_embedded_ipv4(
+ ip: ipaddress._BaseAddress,
+) -> Optional[ipaddress.IPv4Address]:
+ """Return the IPv4 target embedded in an RFC 6052 Well-Known-Prefix address.
+
+ ``None`` when ``ip`` is not inside ``64:ff9b::/96``, i.e. when no IPv4
+ destination may be inferred from it.
+ """
+ if not isinstance(ip, ipaddress.IPv6Address):
+ return None
+ if ip not in _NAT64_WELL_KNOWN_PREFIX_V6:
+ return None
+ return ipaddress.IPv4Address(int(ip) & 0xFFFFFFFF)
+
def _default_resolver(host: str) -> List[str]:
"""Resolve a hostname to the list of IP strings it maps to (A + AAAA)."""
@@ -44,6 +83,15 @@ def _classify(ip: ipaddress._BaseAddress, *, block_private: bool) -> Optional[st
# IPv4-mapped IPv6 (e.g. ::ffff:169.254.169.254) — judge the embedded v4.
if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None:
ip = ip.ipv4_mapped
+ else:
+ # RFC 6052 Well-Known Prefix — judge the IPv4 destination the NAT64
+ # translator will contact, always under the strict policy.
+ translated = _nat64_well_known_embedded_ipv4(ip)
+ if translated is not None:
+ reason = _classify(translated, block_private=True)
+ if reason:
+ return f"NAT64 translated destination blocked: {reason}"
+ return None
if ip.is_link_local:
return f"link-local address blocked (SSRF metadata risk): {ip}"
if ip.is_multicast or ip.is_reserved or ip.is_unspecified:
diff --git a/tests/test_review_20260923_fixes.py b/tests/test_review_20260923_fixes.py
new file mode 100644
index 000000000..6d6cb1200
--- /dev/null
+++ b/tests/test_review_20260923_fixes.py
@@ -0,0 +1,522 @@
+"""Regressions for the 2026-09-23 review fixes.
+
+Two findings, both about bounding work the server does on someone else's behalf:
+
+* the scholarly metadata lookups called two hardcoded third-party endpoints with
+ a stale hand-written User-Agent, no outbound-URL policy, and a full timeout per
+ hop, so one query could hold a user-facing search open for the sum of all three;
+* editor drafts were only size-checked after the body had been parsed and
+ re-serialised, so the ceiling rejected an allocation it had already paid for.
+
+Each test fails on the pre-fix tree.
+"""
+
+import time
+
+import httpx
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from services.search import core as search_core
+from src.constants import (
+ APP_VERSION,
+ ARXIV_API_URL,
+ OPENALEX_API_URL,
+ SCHOLARLY_LOOKUP_TIMEOUT,
+)
+from src.upload_limits import EDITOR_DRAFT_MAX_BYTES
+
+
+# --------------------------------------------------------------------------
+# ODY-R09 — scholarly lookups
+# --------------------------------------------------------------------------
+
+
+def test_scholarly_endpoints_use_configured_constants(monkeypatch):
+ """Call sites must route through configured endpoints, not hardcoded URLs."""
+ requested_urls = []
+
+ def fake_scholarly_api_get(url: str, params: dict):
+ requested_urls.append(url)
+ if "arxiv" in url:
+ class FakeArxivResponse:
+ text = ""
+
+ return FakeArxivResponse()
+ elif "openalex" in url:
+ class FakeOpenAlexResponse:
+ def json(self):
+ return {"results": []}
+
+ return FakeOpenAlexResponse()
+ return None
+
+ monkeypatch.setattr(search_core, "_scholarly_api_get", fake_scholarly_api_get)
+
+ custom_arxiv = "https://custom.arxiv.test/api/query"
+ custom_openalex = "https://custom.openalex.test/works"
+ monkeypatch.setattr(search_core, "ARXIV_API_URL", custom_arxiv)
+ monkeypatch.setattr(search_core, "OPENALEX_API_URL", custom_openalex)
+
+ search_core._arxiv_title_results("Attention Is All You Need")
+ search_core._openalex_title_results("Attention Is All You Need")
+
+ assert custom_arxiv in requested_urls
+ assert custom_openalex in requested_urls
+
+
+def test_user_agent_tracks_app_version():
+ """A hand-written version string drifts; APP_VERSION cannot."""
+ agent = search_core._scholarly_user_agent()
+ assert APP_VERSION in agent
+ # The pre-fix tree hardcoded 0.20 while APP_VERSION was already 1.0.3.
+ assert "Odysseus/0.20 " not in agent
+
+
+def test_outbound_policy_rejection_skips_the_request(monkeypatch):
+ """A URL the outbound policy refuses must never reach httpx."""
+ calls = []
+ monkeypatch.setattr(
+ httpx, "get", lambda *a, **k: calls.append(a) or pytest.fail("request sent")
+ )
+ monkeypatch.setattr(
+ "src.url_safety.check_outbound_url", lambda url, **kw: (False, "blocked")
+ )
+
+ assert search_core._scholarly_api_get(ARXIV_API_URL, {}) is None
+ assert calls == []
+
+
+def test_redirect_to_prohibited_destination_is_blocked_and_never_requested(monkeypatch):
+ """An allowed initial URL must not be permitted to redirect into a prohibited destination."""
+ from src.url_safety import check_outbound_url as real_check
+
+ called_urls = []
+
+ def fake_get(url, **kwargs):
+ called_urls.append(url)
+ req = httpx.Request("GET", url)
+ return httpx.Response(
+ 302,
+ headers={"Location": "http://127.0.0.1:8080/internal-admin"},
+ request=req,
+ )
+
+ def mock_check(url, **kwargs):
+ if url == OPENALEX_API_URL:
+ return (True, "")
+ return real_check(url, **kwargs)
+
+ monkeypatch.setattr(httpx, "get", fake_get)
+ monkeypatch.setattr("src.url_safety.check_outbound_url", mock_check)
+
+ result = search_core._scholarly_api_get(OPENALEX_API_URL, {})
+
+ assert result is None
+ # Only the initial allowed URL was contacted; the prohibited redirect destination was never requested
+ assert called_urls == [OPENALEX_API_URL]
+
+
+def test_redirect_to_link_local_metadata_is_blocked_and_never_requested(monkeypatch):
+ """Redirects to cloud metadata or link-local addresses must be refused before connection."""
+ from src.url_safety import check_outbound_url as real_check
+
+ called_urls = []
+
+ def fake_get(url, **kwargs):
+ called_urls.append(url)
+ req = httpx.Request("GET", url)
+ return httpx.Response(
+ 301,
+ headers={"Location": "http://169.254.169.254/latest/meta-data"},
+ request=req,
+ )
+
+ def mock_check(url, **kwargs):
+ if url == ARXIV_API_URL:
+ return (True, "")
+ return real_check(url, **kwargs)
+
+ monkeypatch.setattr(httpx, "get", fake_get)
+ monkeypatch.setattr("src.url_safety.check_outbound_url", mock_check)
+
+ result = search_core._scholarly_api_get(ARXIV_API_URL, {})
+
+ assert result is None
+ assert called_urls == [ARXIV_API_URL]
+
+
+def test_allowed_redirect_is_followed_safely(monkeypatch):
+ """A safe redirect destination passing outbound checks is followed to completion."""
+ called_urls = []
+ canonical_url = "https://api.openalex.org/canonical-works"
+
+ def fake_get(url, **kwargs):
+ called_urls.append(url)
+ req = httpx.Request("GET", url)
+ if url == OPENALEX_API_URL:
+ return httpx.Response(301, headers={"Location": "/canonical-works"}, request=req)
+ return httpx.Response(200, json={"results": []}, request=req)
+
+ monkeypatch.setattr(httpx, "get", fake_get)
+ monkeypatch.setattr("src.url_safety.check_outbound_url", lambda url, **kw: (True, ""))
+
+ result = search_core._scholarly_api_get(OPENALEX_API_URL, {})
+
+ assert result is not None
+ assert result.status_code == 200
+ assert called_urls == [OPENALEX_API_URL, canonical_url]
+
+
+def test_redirect_limit_is_bounded(monkeypatch):
+ """Redirects exceeding MAX_SCHOLARLY_REDIRECTS must fail safely without looping."""
+ called_urls = []
+
+ def fake_get(url, **kwargs):
+ called_urls.append(url)
+ req = httpx.Request("GET", url)
+ return httpx.Response(302, headers={"Location": f"{url}/next"}, request=req)
+
+ monkeypatch.setattr(httpx, "get", fake_get)
+ monkeypatch.setattr("src.url_safety.check_outbound_url", lambda url, **kw: (True, ""))
+
+ result = search_core._scholarly_api_get("https://export.arxiv.org/api/query", {})
+
+ assert result is None
+ assert len(called_urls) == search_core.MAX_SCHOLARLY_REDIRECTS + 1
+
+
+def test_redirect_does_not_replay_query_parameters_to_the_new_destination(monkeypatch):
+ """The original query must not be re-sent to a redirect target.
+
+ The params belong to the endpoint that was asked for. Replaying them across
+ a redirect would hand the search terms to whatever host the redirect names.
+ """
+ calls = []
+
+ def fake_get(url, **kwargs):
+ params = kwargs.get("params")
+ calls.append((url, params))
+ # Mirror real httpx: response.url carries the query that was sent.
+ req = httpx.Request("GET", httpx.URL(url, params=params or {}))
+ if len(calls) == 1:
+ return httpx.Response(
+ 302, headers={"Location": "https://redirect.openalex.test/v2"}, request=req
+ )
+ return httpx.Response(200, json={"results": []}, request=req)
+
+ monkeypatch.setattr(httpx, "get", fake_get)
+ monkeypatch.setattr("src.url_safety.check_outbound_url", lambda url, **kw: (True, ""))
+
+ result = search_core._scholarly_api_get(
+ OPENALEX_API_URL, {"search": "confidential-title"}
+ )
+
+ assert result is not None
+ assert len(calls) == 2
+ second_url, second_params = calls[1]
+ assert second_url == "https://redirect.openalex.test/v2"
+ assert second_params is None
+ assert "confidential-title" not in second_url
+
+
+def test_relative_redirect_resolves_against_the_responding_url(monkeypatch):
+ """A relative Location resolves against the URL that answered, not the origin."""
+ calls = []
+
+ def fake_get(url, **kwargs):
+ calls.append(url)
+ req = httpx.Request("GET", httpx.URL(url, params=kwargs.get("params") or {}))
+ if len(calls) == 1:
+ return httpx.Response(
+ 302, headers={"Location": "https://mirror.arxiv.test/api/v1/query"}, request=req
+ )
+ if len(calls) == 2:
+ # Relative hop: must resolve against mirror.arxiv.test, not arxiv.
+ return httpx.Response(302, headers={"Location": "../v2/query"}, request=req)
+ return httpx.Response(200, text="", request=req)
+
+ monkeypatch.setattr(httpx, "get", fake_get)
+ monkeypatch.setattr("src.url_safety.check_outbound_url", lambda url, **kw: (True, ""))
+
+ result = search_core._scholarly_api_get(ARXIV_API_URL, {"search_query": "x"})
+
+ assert result is not None
+ assert calls[2] == "https://mirror.arxiv.test/api/v2/query"
+
+
+def test_http_error_status_degrades_to_no_results(monkeypatch):
+ """A 500 from a scholarly API must degrade to no results, not raise."""
+
+ def fake_get(url, **kwargs):
+ req = httpx.Request("GET", url)
+ return httpx.Response(500, text="upstream exploded", request=req)
+
+ monkeypatch.setattr(httpx, "get", fake_get)
+ monkeypatch.setattr("src.url_safety.check_outbound_url", lambda url, **kw: (True, ""))
+
+ assert search_core._arxiv_title_results("Attention Is All You Need") == []
+ assert search_core._openalex_title_results("Attention Is All You Need") == []
+
+
+def test_exhausted_budget_skips_the_request(monkeypatch):
+ """Once the chain's budget is spent, later hops are skipped, not retried."""
+ calls = []
+ monkeypatch.setattr(
+ httpx, "get", lambda *a, **k: calls.append(a) or pytest.fail("request sent")
+ )
+ monkeypatch.setattr(
+ "src.url_safety.check_outbound_url", lambda url, **kw: (True, "")
+ )
+
+ token = search_core._scholarly_deadline.set(time.monotonic() - 1)
+ try:
+ assert search_core._scholarly_api_get(OPENALEX_API_URL, {}) is None
+ finally:
+ search_core._scholarly_deadline.reset(token)
+ assert calls == []
+
+
+def test_remaining_budget_caps_the_per_request_timeout(monkeypatch):
+ """A hop cannot wait longer than the budget the chain has left."""
+ seen = {}
+
+ class _Response:
+ def raise_for_status(self):
+ return None
+
+ def fake_get(url, **kwargs):
+ seen["timeout"] = kwargs.get("timeout")
+ return _Response()
+
+ monkeypatch.setattr(httpx, "get", fake_get)
+ monkeypatch.setattr(
+ "src.url_safety.check_outbound_url", lambda url, **kw: (True, "")
+ )
+
+ token = search_core._scholarly_deadline.set(time.monotonic() + 2)
+ try:
+ search_core._scholarly_api_get(ARXIV_API_URL, {})
+ finally:
+ search_core._scholarly_deadline.reset(token)
+
+ assert seen["timeout"] <= 2.0
+ assert seen["timeout"] < SCHOLARLY_LOOKUP_TIMEOUT
+
+
+def test_budget_is_shared_across_the_whole_chain(monkeypatch):
+ """Both hops of one lookup draw on a single deadline."""
+ observed = []
+
+ monkeypatch.setattr(
+ "src.url_safety.check_outbound_url", lambda url, **kw: (True, "")
+ )
+ monkeypatch.setattr(
+ search_core,
+ "_openalex_title_results",
+ lambda title, count=3: observed.append(search_core._scholarly_deadline.get())
+ or [],
+ )
+ monkeypatch.setattr(
+ search_core,
+ "_arxiv_title_results",
+ lambda title, count=3: observed.append(search_core._scholarly_deadline.get())
+ or [],
+ )
+
+ search_core._direct_scholarly_title_results("a paper title")
+
+ assert len(observed) == 2
+ assert observed[0] is not None
+ assert observed[0] == observed[1]
+
+
+def test_budget_does_not_leak_out_of_the_chain():
+ """The deadline is scoped to the lookup, not left set on the context."""
+ assert search_core._scholarly_deadline.get() is None
+ with search_core._scholarly_budget():
+ assert search_core._scholarly_deadline.get() is not None
+ assert search_core._scholarly_deadline.get() is None
+
+
+# --------------------------------------------------------------------------
+# ODY-R11 — editor draft size ceiling
+# --------------------------------------------------------------------------
+
+
+def _draft_client() -> TestClient:
+ from routes.editor_draft_routes import setup_editor_draft_routes
+
+ app = FastAPI()
+ app.include_router(setup_editor_draft_routes())
+ return TestClient(app)
+
+
+def test_oversized_declared_body_is_refused_before_it_is_parsed():
+ """An over-ceiling Content-Length is rejected without reading the payload."""
+ client = _draft_client()
+ response = client.post(
+ "/api/editor-drafts",
+ content=b"{}",
+ headers={
+ "content-type": "application/json",
+ "content-length": str(EDITOR_DRAFT_MAX_BYTES + 1),
+ },
+ )
+ assert response.status_code == 413
+ assert "safety limit" in response.text
+
+
+def test_oversized_declared_body_guard_runs_before_body_consumption():
+ """An oversized Content-Length must reject the request without reading or consuming the body."""
+ body_consumed = False
+
+ def body_stream():
+ nonlocal body_consumed
+ body_consumed = True
+ yield b'{"layers": []}'
+
+ client = _draft_client()
+ response = client.post(
+ "/api/editor-drafts",
+ content=body_stream(),
+ headers={
+ "content-type": "application/json",
+ "content-length": str(EDITOR_DRAFT_MAX_BYTES + 1),
+ },
+ )
+ assert response.status_code == 413
+ assert "safety limit" in response.text
+ # Proves the body stream was never read or consumed before rejection
+ assert body_consumed is False
+
+
+def test_oversized_declared_body_guard_runs_before_body_consumption_on_put():
+ """Update route also rejects oversized Content-Length without consuming body."""
+ body_consumed = False
+
+ def body_stream():
+ nonlocal body_consumed
+ body_consumed = True
+ yield b'{"layers": []}'
+
+ client = _draft_client()
+ response = client.put(
+ "/api/editor-drafts/some-draft-id",
+ content=body_stream(),
+ headers={
+ "content-type": "application/json",
+ "content-length": str(EDITOR_DRAFT_MAX_BYTES + 1),
+ },
+ )
+ assert response.status_code == 413
+ assert body_consumed is False
+
+
+def test_oversized_declared_body_rejects_without_json_parsing():
+ """Even malformed or invalid JSON is rejected with 413 rather than 422 if Content-Length exceeds ceiling."""
+ client = _draft_client()
+ response = client.post(
+ "/api/editor-drafts",
+ content=b"this is completely invalid json {[[",
+ headers={
+ "content-type": "application/json",
+ "content-length": str(EDITOR_DRAFT_MAX_BYTES + 1),
+ },
+ )
+ assert response.status_code == 413
+
+
+def test_update_route_carries_the_same_guard():
+ client = _draft_client()
+ response = client.put(
+ "/api/editor-drafts/whatever",
+ content=b"{}",
+ headers={
+ "content-type": "application/json",
+ "content-length": str(EDITOR_DRAFT_MAX_BYTES + 1),
+ },
+ )
+ assert response.status_code == 413
+
+
+def test_absent_content_length_still_reaches_the_exact_check():
+ """The header guard is an optimisation; it must not become the only check."""
+ from routes.editor_draft_routes import _dump_payload, reject_oversized_draft_body
+
+ class _NoLengthRequest:
+ headers: dict = {}
+
+ # No header: the guard abstains rather than rejecting or accepting outright.
+ assert reject_oversized_draft_body(_NoLengthRequest()) is None
+
+ # The authoritative byte count still refuses an over-ceiling payload.
+ with pytest.raises(Exception) as excinfo:
+ _dump_payload({"blob": "x" * (EDITOR_DRAFT_MAX_BYTES + 1)})
+ assert getattr(excinfo.value, "status_code", None) == 413
+
+
+def test_malformed_content_length_does_not_crash_the_route():
+ from routes.editor_draft_routes import reject_oversized_draft_body
+
+ class _BadLengthRequest:
+ headers = {"content-length": "not-a-number"}
+
+ assert reject_oversized_draft_body(_BadLengthRequest()) is None
+
+
+def test_ordinary_draft_is_unaffected():
+ """The guard must not change behaviour for normal payloads."""
+ from routes.editor_draft_routes import _dump_payload
+
+ assert _dump_payload({"layers": []}) == '{"layers":[]}'
+
+
+def test_route_class_guard_applies_only_to_body_bearing_methods():
+ """The route class must not change GET/DELETE semantics.
+
+ ``EditorDraftRoute`` is attached to the whole editor-draft router, so the
+ bodyless routes run through it too. They must reach their handler
+ untouched — the early ceiling belongs to the methods that carry a draft.
+ """
+ from fastapi import APIRouter
+
+ from routes.editor_draft_routes import EditorDraftRoute
+
+ router = APIRouter(route_class=EditorDraftRoute)
+
+ @router.get("/probe")
+ async def _get_probe():
+ return {"reached": "get"}
+
+ @router.delete("/probe")
+ async def _delete_probe():
+ return {"reached": "delete"}
+
+ @router.post("/probe")
+ async def _post_probe():
+ return {"reached": "post"}
+
+ app = FastAPI()
+ app.include_router(router)
+ client = TestClient(app)
+
+ oversized = {
+ "content-type": "application/json",
+ "content-length": str(EDITOR_DRAFT_MAX_BYTES + 1),
+ }
+
+ # Bodyless methods are unaffected even when a bogus huge length is declared.
+ for method, expected in (("GET", "get"), ("DELETE", "delete")):
+ response = client.request(method, "/probe", content=b"{}", headers=oversized)
+ assert response.status_code == 200, (method, response.text)
+ assert response.json() == {"reached": expected}
+
+ # The same declaration on the body-bearing method is refused.
+ assert client.post("/probe", content=b"{}", headers=oversized).status_code == 413
+
+ # ...and an ordinary POST still reaches the handler.
+ ordinary = client.post("/probe", json={"layers": []})
+ assert ordinary.status_code == 200
+ assert ordinary.json() == {"reached": "post"}
diff --git a/tests/test_service_search_provider_guards.py b/tests/test_service_search_provider_guards.py
index 29b1b36c4..d7735ec32 100644
--- a/tests/test_service_search_provider_guards.py
+++ b/tests/test_service_search_provider_guards.py
@@ -4,6 +4,7 @@ The old src.search provider path aliases this module; these tests pin the
behavior at the single implementation point.
"""
+import ipaddress
import sys
import pytest
@@ -11,6 +12,34 @@ from services.search import core
from services.search import providers
+@pytest.fixture(autouse=True)
+def _deterministic_outbound_dns(monkeypatch):
+ """Keep the outbound-URL policy running, but take it off live DNS.
+
+ ``_scholarly_api_get`` validates each destination through
+ ``check_outbound_url`` before it reaches the mocked ``httpx`` transport, and
+ that check resolves the hostname. Without this stub these tests depend on
+ real DNS: on a DNS64/NAT64 network an IPv4-only host such as
+ export.arxiv.org resolves to ``64:ff9b::``, so the lookup is judged on a
+ synthesised address that has nothing to do with the title-resolution,
+ fallback, request-parameter and ordering behaviour asserted here.
+
+ Only name resolution is replaced. The policy itself still runs in full, and
+ a literal-IP host still resolves to itself, so a test that points at a
+ prohibited address is still genuinely rejected. This is deliberately not a
+ stand-in for SSRF/NAT64 coverage, which lives in tests/test_url_safety.py
+ and tests/test_review_20260923_fixes.py.
+ """
+
+ def _resolve(host: str):
+ try:
+ return [str(ipaddress.ip_address(host))]
+ except ValueError:
+ return ["93.184.216.34"] # public, policy-clean
+
+ monkeypatch.setattr("src.url_safety._default_resolver", _resolve)
+
+
def test_html_transport_fallback_preserves_query_constraints(monkeypatch):
seen = []
class Response:
diff --git a/tests/test_url_safety.py b/tests/test_url_safety.py
index d82d38878..efe0c7bb8 100644
--- a/tests/test_url_safety.py
+++ b/tests/test_url_safety.py
@@ -115,3 +115,146 @@ def test_resolver_skips_invalid_values_but_accepts_public_ip():
assert ok is True
assert reason == "ok"
+
+
+# --- RFC 6052 Well-Known Prefix (DNS64 / NAT64) ---------------------------
+#
+# On a DNS64/NAT64 network an IPv4-only host resolves to 64:ff9b::. The
+# effective destination is the embedded IPv4 address, so that is what the
+# policy must judge. RFC 6052 §3.1 permits the Well-Known Prefix to represent
+# only globally-routable IPv4, so the embedded target is always held to the
+# strict policy — the prefix must never tunnel past the SSRF guard.
+
+
+def _nat64(v4: str) -> str:
+ """Render the RFC 6052 Well-Known-Prefix form of an IPv4 address."""
+ import ipaddress
+
+ packed = int(ipaddress.IPv4Address(v4))
+ return str(ipaddress.IPv6Address(int(ipaddress.IPv6Address("64:ff9b::")) | packed))
+
+
+def test_nat64_well_known_prefix_allows_public_ipv4_destination():
+ # The reproduced failure: export.arxiv.org resolves to 64:ff9b::924b:5b2a
+ # under DNS64. The embedded 146.75.91.42 is public, so the URL is allowed.
+ res = _resolver({"export.arxiv.org": [_nat64("146.75.91.42")]})
+ ok, reason = check_outbound_url("https://export.arxiv.org/api/query", resolver=res)
+ assert ok is True, reason
+ # ...including under full lockdown, where scholarly lookups actually run.
+ ok, reason = check_outbound_url(
+ "https://export.arxiv.org/api/query", block_private=True, resolver=res
+ )
+ assert ok is True, reason
+
+
+def test_nat64_well_known_prefix_blocks_embedded_loopback():
+ res = _resolver({"evil.example": [_nat64("127.0.0.1")]})
+ for strict in (False, True):
+ ok, reason = check_outbound_url(
+ "http://evil.example/", block_private=strict, resolver=res
+ )
+ assert ok is False, strict
+ assert "NAT64" in reason and "127.0.0.1" in reason
+
+
+def test_nat64_well_known_prefix_blocks_embedded_metadata_address():
+ # The headline SSRF vector must not become reachable through DNS64.
+ res = _resolver({"evil.example": [_nat64("169.254.169.254")]})
+ for strict in (False, True):
+ ok, reason = check_outbound_url(
+ "http://evil.example/latest/meta-data/", block_private=strict, resolver=res
+ )
+ assert ok is False, strict
+ assert "link-local" in reason and "169.254.169.254" in reason
+
+
+def test_nat64_well_known_prefix_blocks_embedded_private_ipv4_under_strict():
+ res = _resolver({"evil.example": [_nat64("192.168.1.50")]})
+ ok, reason = check_outbound_url(
+ "http://evil.example/", block_private=True, resolver=res
+ )
+ assert ok is False
+ assert "NAT64" in reason and "192.168.1.50" in reason
+
+
+def test_nat64_well_known_prefix_blocks_embedded_shared_space_under_strict():
+ # RFC 6598 shared/CGNAT space is not globally routable, so the Well-Known
+ # Prefix may not represent it.
+ res = _resolver({"evil.example": [_nat64("100.64.0.1")]})
+ ok, reason = check_outbound_url(
+ "http://evil.example/", block_private=True, resolver=res
+ )
+ assert ok is False
+ assert "NAT64" in reason and "100.64.0.1" in reason
+
+
+def test_nat64_well_known_prefix_blocks_embedded_non_global_ipv4():
+ # Multicast, unspecified and other non-global space must not be reachable
+ # through the Well-Known Prefix, whatever block_private says.
+ for v4 in ("224.0.0.1", "0.0.0.0", "198.18.0.1", "192.0.2.10", "240.0.0.1"):
+ res = _resolver({"evil.example": [_nat64(v4)]})
+ for strict in (False, True):
+ ok, reason = check_outbound_url(
+ "http://evil.example/", block_private=strict, resolver=res
+ )
+ assert ok is False, (v4, strict)
+ assert v4 in reason, (v4, reason)
+
+
+def test_network_specific_nat64_prefixes_are_not_decoded():
+ # RFC 8215 64:ff9b:1::/48 and arbitrary network-specific prefixes carry
+ # locally assigned semantics, so no embedded IPv4 may be inferred from them.
+ # 64:ff9b:1::8ef:5b2a would decode to the *public* 8.239.91.42; it stays
+ # rejected on the outer IPv6 address, which proves no decoding happened.
+ for addr in ("64:ff9b:1::8ef:5b2a", "64:ff9b:1::7f00:1"):
+ res = _resolver({"svc.example": [addr]})
+ ok, reason = check_outbound_url("http://svc.example/", resolver=res)
+ assert ok is False, addr
+ assert "NAT64" not in reason, addr
+ assert addr in reason, addr
+
+ # A documentation-range prefix is judged as an ordinary IPv6 address under
+ # the existing policy (local-first by default, blocked under lockdown) and
+ # is never reinterpreted as carrying an IPv4 destination.
+ res = _resolver({"svc.example": ["2001:db8:1::8ef:5b2a"]})
+ ok, reason = check_outbound_url("http://svc.example/", resolver=res)
+ assert ok is True, reason
+ ok, reason = check_outbound_url(
+ "http://svc.example/", block_private=True, resolver=res
+ )
+ assert ok is False
+ assert "NAT64" not in reason and "8.239.91.42" not in reason
+
+
+def test_ordinary_ipv6_behaviour_is_unchanged():
+ # A global unicast IPv6 host is allowed in both modes.
+ GLOBAL6 = _resolver({"v6.example": ["2606:2800:220:1:248:1893:25c8:1946"]})
+ assert check_outbound_url("https://v6.example/", resolver=GLOBAL6)[0] is True
+ assert (
+ check_outbound_url("https://v6.example/", block_private=True, resolver=GLOBAL6)[0]
+ is True
+ )
+
+ # ULA is local-first allowed by default, blocked under lockdown.
+ ULA = _resolver({"ula.example": ["fd00::1"]})
+ assert check_outbound_url("http://ula.example/", resolver=ULA)[0] is True
+ ok, reason = check_outbound_url(
+ "http://ula.example/", block_private=True, resolver=ULA
+ )
+ assert ok is False and "private" in reason
+
+ # fe80::/10 is always blocked.
+ LL6 = _resolver({"ll.example": ["fe80::1"]})
+ ok, reason = check_outbound_url("http://ll.example/", resolver=LL6)
+ assert ok is False and "link-local" in reason
+
+ # Pre-existing behaviour, unchanged here: CPython reports ::1 as
+ # is_reserved, so IPv6 loopback is rejected in both modes (unlike 127.0.0.1,
+ # which the local-first default allows).
+ LOOP6 = _resolver({"loop6.example": ["::1"]})
+ for strict in (False, True):
+ ok, reason = check_outbound_url(
+ "http://loop6.example/", block_private=strict, resolver=LOOP6
+ )
+ assert ok is False, strict
+ assert "NAT64" not in reason