Merge pull request #10 from o3LL/fix/review-20260923-bound-outbound-and-draft-limits

fix(security): harden scholarly lookups and editor draft limits

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.
 

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:

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]:

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"))

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:

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 = "<feed xmlns='http://www.w3.org/2005/Atom'></feed>"
+
+            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="<feed/>", 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"}

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::<v4>``, 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:

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::<v4>. 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