feat(tools): allow_private_network opt-in for private-address fetch/preview

turnstone's primary audience self-hosts it beside other lab services —
a web_fetch or open_preview aimed at Grafana, Home Assistant, or a dev
node on the local network is the operator using their own network, not
an attack. The hard SSRF refusal made those targets unreachable.

New runtime setting tools.allow_private_network (settings registry,
default off, rendered in console Settings → Tools; hot — read per tool
call, no restart). When enabled, a call NAMING a private address
becomes approvable: the approval prompt tags it "(private network)" so
the operator approves it as what it is, and the human gate stays.

The redirect side-door stays closed either way: a PUBLIC target that
302s into private address space is refused regardless of the opt-in —
that address never appeared on the approval card, so it is never
fetched. Only a chain whose approved origin was itself private skips
hop screening (its redirects are the operator's own network).

Refusals now teach the knob (mirrors the oidc opt-in hint): the error
names tools.allow_private_network and where to enable it. Surfaces
without a ConfigStore (bare CLI, eval) stay strict — there is no admin
surface to have opted in on.

(cherry picked from commit 09abc9d199)

docs/tools.md

@@ -289,7 +289,7 @@ Fetch a URL and extract specific information from it.
 | `url`      | string | yes      | The URL to fetch (must start with `http://` or `https://`). |
 | `question` | string | yes      | What to extract or answer from the page content. |
 
-- **What it does**: Fetches the URL, strips HTML to plain text, and uses the LLM to extract the answer to the question from the page content. Protected against SSRF (blocks private/internal IPs).
+- **What it does**: Fetches the URL, strips HTML to plain text, and uses the LLM to extract the answer to the question from the page content. Every redirect hop is SSRF-screened before it is requested. Private/internal addresses are refused by default; enable `tools.allow_private_network` (console Settings → Tools) to make them approvable for self-hosted setups whose services live on the local network — the approval prompt marks such requests, and a public site redirecting into private space is refused regardless.
 - **Auto-approve**: No -- requires user confirmation (makes network requests).
 - **Agent availability**: `task_agent`.
 
@@ -360,7 +360,8 @@ Show the user rich content in a preview pane beside the conversation.
 | `title`   | string | no       | Pane header title. Defaults to the page title, filename, or URL. |
 
 - **What it does**: Resolves the target to bytes (URLs fetch through the same
-  SSRF-guarded path as `web_fetch`, re-checked after redirects), classifies the
+  SSRF-guarded path as `web_fetch`, screened per redirect hop, honoring the
+  same `tools.allow_private_network` opt-in), classifies the
   content, stores it content-addressed against the workstream, and opens the
   frontend preview pane beside the conversation: web pages render in a fully
   sandboxed iframe (no scripts, opaque origin), PDFs in the browser viewer,

tests/test_open_preview_tool.py

@@ -590,3 +590,115 @@ class TestCancelledBatchPreservesPreview:
         # The in-memory synthesized turn carries the descriptor too.
         tool_turns = [t for t in s.messages if isinstance(t, Turn) and t.role is Role.TOOL]
         assert tool_turns and tool_turns[-1].meta.extra.get("preview") == descriptor
+
+
+# ---------------------------------------------------------------------------
+# tools.allow_private_network — the self-hoster opt-in (admin Settings → Tools)
+# ---------------------------------------------------------------------------
+
+
+class TestAllowPrivateNetwork:
+    def test_screen_public_url_passes(self):
+        from turnstone.core.session import _screen_tool_url
+
+        err, private = _screen_tool_url("https://example.com/x", False)
+        assert err is None and private is False
+
+    def test_screen_private_blocked_with_discoverable_hint(self):
+        from turnstone.core.session import _screen_tool_url
+
+        err, private = _screen_tool_url("http://10.0.0.7/grafana", False)
+        assert err is not None and private is False
+        # The refusal teaches the knob (mirrors the oidc opt-in hint pattern).
+        assert "tools.allow_private_network" in err
+        assert "Settings" in err
+
+    def test_screen_private_allowed_when_opted_in(self):
+        from turnstone.core.session import _screen_tool_url
+
+        err, private = _screen_tool_url("http://10.0.0.7/grafana", True)
+        assert err is None and private is True
+
+    def test_screen_invalid_url_never_hints(self):
+        from turnstone.core.session import _screen_tool_url
+
+        err, private = _screen_tool_url("http://", True)
+        assert err is not None and private is False
+        assert "allow_private_network" not in err
+
+    def test_bare_session_defaults_strict(self):
+        # No ConfigStore (CLI / eval surface) → no admin opted in → strict.
+        s = _make_session()
+        assert s._allow_private_network() is False
+
+    def test_prepare_web_fetch_private_opted_in(self, monkeypatch):
+        s = _make_session()
+        monkeypatch.setattr(ChatSession, "_allow_private_network", lambda self: True)
+        item = s._prepare_web_fetch(
+            "c1", {"url": "http://192.168.1.50:3000/d/home", "question": "what is shown?"}
+        )
+        assert "error" not in item
+        assert item["needs_approval"] is True  # the human gate stays
+        assert "(private network)" in item["header"]
+        assert item["allow_private_origin"] is True
+
+    def test_prepare_open_preview_private_opted_in(self, monkeypatch):
+        s = _make_session()
+        monkeypatch.setattr(ChatSession, "_allow_private_network", lambda self: True)
+        item = s._prepare_open_preview("c1", {"target": "http://192.168.1.50:3000/d/home"})
+        assert "error" not in item
+        assert item["needs_approval"] is True
+        assert "(private network)" in item["header"]
+        assert item["allow_private_origin"] is True
+
+    def test_prepare_private_still_blocked_by_default(self, monkeypatch):
+        s = _make_session()
+        monkeypatch.setattr(ChatSession, "_allow_private_network", lambda self: False)
+        for prepare, args in (
+            (s._prepare_web_fetch, {"url": "http://10.0.0.7/x", "question": "q"}),
+            (s._prepare_open_preview, {"target": "http://10.0.0.7/x"}),
+        ):
+            item = prepare("c1", args)
+            assert "error" in item
+            assert "tools.allow_private_network" in item["error"]
+
+    def test_executor_passes_private_origin_to_guard(self, monkeypatch):
+        s = _make_session()
+        monkeypatch.setattr(ChatSession, "_allow_private_network", lambda self: True)
+        seen = {}
+
+        def _capture(url, **kw):
+            seen.update(kw, url=url)
+            return _fake_response(url, b"<html><head></head><body>x</body></html>", "text/html")
+
+        monkeypatch.setattr("turnstone.core.session.fetch_with_ssrf_guard", _capture)
+        item = s._prepare_open_preview("c1", {"target": "http://10.0.0.7/status"})
+        s._exec_open_preview(item)
+        assert seen["allow_private_origin"] is True
+
+    def test_guard_skips_hop_screen_for_private_origin(self, monkeypatch):
+        from turnstone.core.web import fetch_with_ssrf_guard
+
+        _FakeClient.calls = []
+        _FakeClient.table = {
+            "http://10.0.0.7/a": _FakeHop(302, {"location": "http://10.0.0.8/b"}),
+            "http://10.0.0.8/b": _FakeHop(200, {}, url="http://10.0.0.8/b"),
+        }
+        monkeypatch.setattr("turnstone.core.web.httpx.Client", _FakeClient)
+
+        def _explode(url):
+            raise AssertionError("hop screening must be skipped for a private origin")
+
+        monkeypatch.setattr("turnstone.core.web.check_ssrf", _explode)
+        resp = fetch_with_ssrf_guard("http://10.0.0.7/a", timeout=5, allow_private_origin=True)
+        assert resp.status_code == 200
+        assert _FakeClient.calls == ["http://10.0.0.7/a", "http://10.0.0.8/b"]
+
+    def test_registry_entry_shape(self):
+        from turnstone.core.settings_registry import SETTINGS
+
+        d = SETTINGS["tools.allow_private_network"]
+        assert d.type == "bool"
+        assert d.default is False
+        assert d.section == "tools"
+        assert d.help  # the admin form renders this — it must explain the caveat

turnstone/core/session.py

@@ -1175,6 +1175,36 @@ def _notify_auth_headers() -> dict[str, str]:
     return header
 
 
+def _screen_tool_url(url: str, allow_private_network: bool) -> tuple[str | None, bool]:
+    """SSRF-screen a tool's target URL under the operator's private-network opt-in.
+
+    ``allow_private_network`` is the live ``tools.allow_private_network``
+    setting (admin Settings → Tools; DB-backed, hot-toggleable — the caller
+    reads it per prepare).  Returns ``(error, private_origin)``.  ``error`` is
+    the rejection text (``None`` = proceed); a private-address rejection names
+    the setting so a self-hosted operator learns the knob from the refusal
+    itself.  ``private_origin`` is True when the target NAMES a private
+    address the operator opted into: the approval header tags it, and the
+    guarded fetch skips per-hop screening for that chain — the gate approved a
+    private URL, so its redirects are the operator's own network.  Public
+    origins never set it, keeping the public→private redirect bounce blocked
+    regardless of the opt-in.
+    """
+    ssrf_err = check_ssrf(url)
+    if not ssrf_err:
+        return None, False
+    is_private_block = "private/internal address" in ssrf_err
+    if is_private_block and allow_private_network:
+        return None, True
+    hint = ""
+    if is_private_block:
+        hint = (
+            " Enable 'tools.allow_private_network' in the console"
+            " (Settings → Tools) to allow fetching private-network addresses."
+        )
+    return f"Error: {ssrf_err}.{hint}", False
+
+
 def _tool_turn_meta(
     status: EffectStatus | None, preview: dict[str, Any] | None = None
 ) -> str | None:
@@ -9684,6 +9714,19 @@ class ChatSession:
             "replace_all": replace_all,
         }
 
+    def _allow_private_network(self) -> bool:
+        """Live read of ``tools.allow_private_network`` (admin Settings → Tools).
+
+        Read per prepare — an admin flipping the toggle takes effect on the
+        next tool call, no restart or config push.  Surfaces without a
+        ConfigStore (bare CLI, eval) stay strict: there is no admin surface
+        to have opted in on.
+        """
+        cs = getattr(self, "_config_store", None)
+        if cs is None:
+            return False
+        return bool(cs.get("tools.allow_private_network"))
+
     def _prepare_web_fetch(self, call_id: str, args: dict[str, Any]) -> dict[str, Any]:
         url = args.get("url", "").strip()
         question = args.get("question", "").strip()
@@ -9714,29 +9757,35 @@ class ChatSession:
                 "needs_approval": False,
                 "error": f"Error: URL must start with http:// or https:// (got {url!r})",
             }
-        # SSRF protection: reject private/link-local/metadata IPs
-        ssrf_err = check_ssrf(url)
-        if ssrf_err:
+        # SSRF screen \u2014 a NAMED private address is approvable under the
+        # tools.allow_private_network opt-in (the header tags it so the
+        # operator approves it as what it is).
+        screen_err, private_origin = _screen_tool_url(url, self._allow_private_network())
+        if screen_err:
             return {
                 "call_id": call_id,
                 "func_name": "web_fetch",
-                "header": "\u2717 web_fetch: blocked (private network)",
+                "header": "\u2717 web_fetch: blocked (private network)"
+                if "private/internal" in screen_err
+                else "\u2717 web_fetch: blocked",
                 "preview": f"    {url}",
                 "needs_approval": False,
-                "error": f"Error: {ssrf_err}",
+                "error": screen_err,
             }
         q_preview = question[:200] + ("..." if len(question) > 200 else "")
         preview = f"    {url}\n    Q: {q_preview}"
+        private_tag = " (private network)" if private_origin else ""
         return {
             "call_id": call_id,
             "func_name": "web_fetch",
-            "header": f"\u2699 web_fetch: {url[:80]}",
+            "header": f"\u2699 web_fetch: {url[:80]}{private_tag}",
             "preview": preview,
             "needs_approval": True,
             "approval_label": "web_fetch",
             "execute": self._exec_web_fetch,
             "url": url,
             "question": question,
+            "allow_private_origin": private_origin,
         }
 
     def _prepare_open_preview(self, call_id: str, args: dict[str, Any]) -> dict[str, Any]:
@@ -9780,16 +9829,23 @@ class ChatSession:
             "title": str(title).strip() if title else None,
         }
         if target.startswith(("http://", "https://")):
-            ssrf_err = check_ssrf(target)
-            if ssrf_err:
+            # Same opt-in lane as web_fetch: a named private address is
+            # approvable under tools.allow_private_network, tagged so the
+            # operator approves it as what it is.
+            screen_err, private_origin = _screen_tool_url(target, self._allow_private_network())
+            if screen_err:
                 return {
                     "call_id": call_id,
                     "func_name": "open_preview",
-                    "header": "✗ open_preview: blocked (private network)",
+                    "header": "✗ open_preview: blocked (private network)"
+                    if "private/internal" in screen_err
+                    else "✗ open_preview: blocked",
                     "preview": f"    {target}",
                     "needs_approval": False,
-                    "error": f"Error: {ssrf_err}",
+                    "error": screen_err,
                 }
+            if private_origin:
+                item["header"] = f"⚙ open_preview: {target[:80]} (private network)"
             item.update(
                 {
                     "preview": f"    {target}",
@@ -9797,6 +9853,7 @@ class ChatSession:
                     "approval_label": "open_preview",
                     "target_kind": "url",
                     "url": target,
+                    "allow_private_origin": private_origin,
                 }
             )
             return item
@@ -15986,7 +16043,11 @@ class ChatSession:
         # redirect hop before requesting it (the prepare-time check covers
         # only the URL the model named, not where it 302s).
         try:
-            resp = fetch_with_ssrf_guard(url, timeout=self.tool_timeout)
+            resp = fetch_with_ssrf_guard(
+                url,
+                timeout=self.tool_timeout,
+                allow_private_origin=item.get("allow_private_origin", False),
+            )
             resp.raise_for_status()
             ct = resp.headers.get("content-type", "")
             text = resp.text
@@ -16099,7 +16160,11 @@ class ChatSession:
                 # Every redirect hop is SSRF-screened BEFORE its request goes
                 # out — the pre-approval screen covers only the URL the model
                 # named, not where it 302s.
-                resp = fetch_with_ssrf_guard(url, timeout=self.tool_timeout)
+                resp = fetch_with_ssrf_guard(
+                    url,
+                    timeout=self.tool_timeout,
+                    allow_private_origin=item.get("allow_private_origin", False),
+                )
                 resp.raise_for_status()
             except httpx.HTTPStatusError as e:
                 return _fail(f"Error: fetch failed: HTTP {e.response.status_code}")

turnstone/core/settings_registry.py

@@ -194,6 +194,19 @@ def _build_registry() -> dict[str, SettingDef]:
             "Use with caution \u2014 the model will be able to run commands, write files, "
             "and take actions without human review.",
         ),
+        SettingDef(
+            "tools.allow_private_network",
+            "bool",
+            False,
+            "Allow web_fetch / open_preview to reach private-network addresses",
+            "tools",
+            help="When enabled, a fetch or preview whose URL points at a private or "
+            "internal address (a home-lab service, an internal dashboard, localhost) "
+            "can be approved instead of being refused outright — the approval prompt "
+            "marks it as a private-network request. A public site that redirects into "
+            "your private network is still refused either way: that address never "
+            "appeared in the approval prompt, so it is never fetched.",
+        ),
         SettingDef(
             "tools.search",
             "str",

turnstone/core/web.py

@@ -130,6 +130,7 @@ def fetch_with_ssrf_guard(
     timeout: float,
     user_agent: str = "turnstone/1.0",
     max_redirects: int = 5,
+    allow_private_origin: bool = False,
 ) -> httpx.Response:
     """GET *url* following redirects manually, SSRF-screening EVERY hop.
 
@@ -139,6 +140,14 @@ def fetch_with_ssrf_guard(
     executing the private-network request even if the response is later
     discarded.  Here each hop's URL is screened BEFORE its request is issued.
 
+    ``allow_private_origin`` is the ``[tools] allow_private_network`` lane:
+    the CALLER sets it only when the operator opted in AND the ORIGINAL
+    target itself named a private address — the approval gate then saw and
+    approved that private URL, so its redirect chain is the operator's own
+    network and hop screening is skipped.  A public origin never sets it,
+    so a public site bouncing the fetcher into private space stays blocked
+    regardless of the opt-in: that hop was never shown to the approval gate.
+
     Raises ``ValueError`` for a blocked hop or a redirect chain past
     *max_redirects* (callers already route ``ValueError`` to their
     fetch-failed lane), and lets ``httpx`` transport errors propagate
@@ -151,9 +160,10 @@ def fetch_with_ssrf_guard(
         follow_redirects=False,
     ) as client:
         for _hop in range(max_redirects + 1):
-            ssrf_err = check_ssrf(current)
-            if ssrf_err:
-                raise ValueError(ssrf_err)
+            if not allow_private_origin:
+                ssrf_err = check_ssrf(current)
+                if ssrf_err:
+                    raise ValueError(ssrf_err)
             resp = client.get(current)
             if resp.status_code in (301, 302, 303, 307, 308):
                 location = resp.headers.get("location")