by position) gave later sections the wrong anchor
+id and the trailing TOC link went dead.
+"""
+import pytest
+
+pytest.importorskip("bs4")
+
+from src.visual_report import _extract_headings
+
+
+def test_backtick_fenced_heading_is_ignored():
+ md = "## Intro\n\n```bash\n## not a heading\n```\n\n## Conclusion"
+ assert [h["text"] for h in _extract_headings(md)] == ["Intro", "Conclusion"]
+
+
+def test_tilde_fenced_heading_is_ignored():
+ md = "## A\n\n~~~\n## fake\n~~~\n\n## B"
+ assert [h["text"] for h in _extract_headings(md)] == ["A", "B"]
+
+
+def test_normal_headings_unaffected():
+ md = "## One\n\nsome text\n\n### Two"
+ out = [(h["level"], h["text"]) for h in _extract_headings(md)]
+ assert out == [(2, "One"), (3, "Two")]
diff --git a/tests/test_web_fetch_size_caps.py b/tests/test_web_fetch_size_caps.py
index 19320c6c2..a3cfa64ed 100644
--- a/tests/test_web_fetch_size_caps.py
+++ b/tests/test_web_fetch_size_caps.py
@@ -13,6 +13,57 @@ import pytest
from src.constants import WEB_FETCH_SOFT_MAX_BYTES, WEB_FETCH_HARD_MAX_BYTES
from services.search import content as content_mod
+import pytest as _pytest_for_client_stream_compat
+
+
+@_pytest_for_client_stream_compat.fixture(autouse=True)
+def _client_stream_compat_for_pinned_fetch(monkeypatch):
+ """Adapt old size-cap tests to the current pinned Client.stream path.
+
+ These tests monkeypatch httpx.stream(...) to return fake responses. The
+ production fetcher now uses httpx.Client(...).stream(...) so it can pass a
+ pinned transport. When a test has replaced httpx.stream, route Client.stream
+ through that fake. When it has not, fall back to a real Client so unrelated
+ behavior in this file is not changed.
+ """
+ import httpx
+
+ real_client_cls = httpx.Client
+ original_stream = httpx.stream
+
+ class _ClientProxy:
+ def __init__(self, *args, **kwargs):
+ self._args = args
+ self._kwargs = kwargs
+ self._real_cm = None
+ self._real_client = None
+
+ def __enter__(self):
+ if httpx.stream is original_stream:
+ self._real_cm = real_client_cls(*self._args, **self._kwargs)
+ self._real_client = self._real_cm.__enter__()
+ return self._real_client
+ return self
+
+ def __exit__(self, *args):
+ if self._real_cm is not None:
+ return self._real_cm.__exit__(*args)
+ return False
+
+ def stream(self, method, url):
+ if self._real_client is not None:
+ return self._real_client.stream(method, url)
+
+ kwargs = {
+ "headers": self._kwargs.get("headers"),
+ "timeout": self._kwargs.get("timeout"),
+ "follow_redirects": self._kwargs.get("follow_redirects"),
+ }
+ return httpx.stream(method, url, **kwargs)
+
+ monkeypatch.setattr(httpx, "Client", _ClientProxy)
+
+
class _FakeStream:
"""Stands in for the httpx.stream(...) context manager."""
diff --git a/tests/test_web_search_query_sanitization.py b/tests/test_web_search_query_sanitization.py
new file mode 100644
index 000000000..58e338233
--- /dev/null
+++ b/tests/test_web_search_query_sanitization.py
@@ -0,0 +1,209 @@
+"""Regression tests for #4547 — chat-mode web search query sanitization.
+
+Chat-mode web search (``use_web``) selects a search query via the
+generated-query flow added in #4557: an LLM extracts a concise query, falling
+back to the first non-empty line of the user message when the LLM fails or
+returns an empty result. PR #4863 layers a focused, *defensive* cleanup on top
+of that flow: whatever query is finally selected (generated or fallback) is
+passed through ``_clean_search_query()`` before reaching
+``comprehensive_web_search()``, so residual fenced/inline markdown never leaks
+into the search call.
+
+``_clean_search_query()`` renders the query to HTML via ``markdown``
+(``fenced_code`` extension), drops ```` blocks entirely, unwraps inline
+```` to its text (so ``git reset`` survives), collapses whitespace, and
+truncates.
+
+The first four tests pin the helper directly; the last three prove it is
+wired into the production path and that the combined generated-query +
+sanitization behaviour holds for all three selection outcomes (generated
+success, LLM exception, empty LLM result).
+
+This is intentionally a narrow interim/defensive fix for #4547; it does not
+replace the generated-query flow from #4557.
+"""
+from src.chat_processor import ChatProcessor, _clean_search_query
+
+
+# ── Unit tests: _clean_search_query ──
+
+
+def test_clean_search_query_removes_fenced_code_blocks():
+ """A fenced code block must be dropped entirely, including the code body
+ and the fences — only the surrounding prose survives."""
+ message = '```python\nprint("hello")\n```\nWhat is the capital of France?'
+
+ result = _clean_search_query(message)
+
+ assert result == "What is the capital of France?"
+ # Guards against the original leak: no fences, no code body.
+ assert "```" not in result
+ assert "print" not in result
+
+
+def test_clean_search_query_preserves_inline_code():
+ """Inline code text is search-relevant and must survive unwrapped; only the
+ backticks are removed. This is the ``git reset`` case the reviewer flagged
+ against the earlier regex approach (which dropped the word entirely)."""
+ message = "Is it a good idea to use `git reset` to undo my changes?"
+
+ result = _clean_search_query(message)
+
+ assert result == "Is it a good idea to use git reset to undo my changes?"
+ assert "git reset" in result
+ assert "`" not in result
+
+
+def test_clean_search_query_collapses_whitespace():
+ """Runs of whitespace (tabs, multiple spaces, newlines) collapse to a single
+ space so the query is a single clean line."""
+ message = "hello\tworld foo\n\n bar"
+
+ result = _clean_search_query(message)
+
+ assert result == "hello world foo bar"
+ assert " " not in result
+ assert "\n" not in result
+ assert "\t" not in result
+
+
+def test_clean_search_query_truncates_long_input():
+ """Long queries are capped at ``max_len`` (default 200) to stay within search
+ API limits; truncation is a strict prefix of the cleaned text."""
+ long_message = "x" * 300
+
+ result_default = _clean_search_query(long_message)
+ result_custom = _clean_search_query(long_message, max_len=10)
+
+ assert len(result_default) == 200
+ assert result_default == "x" * 200
+ assert result_custom == "x" * 10
+
+
+# ── Integration tests: the generated-query + sanitization flow ──
+#
+# These cover the combined behaviour requested in review of #4863 after #4557
+# landed: the LLM-generated query is used on success, the first-line fallback is
+# used when the LLM fails or returns empty, and in every case the *final* query
+# handed to comprehensive_web_search() is sanitized.
+
+# A messy user message whose first non-empty line (the #4557 fallback) is
+# inline-code prose followed by a fenced block. After sanitization the fallback
+# collapses to plain prose.
+_MESSY = 'Is `git reset` safe?\n```python\nprint("leaked body")\n```'
+_SANITIZED_FALLBACK = "Is git reset safe?"
+
+
+class _Session:
+ """Minimal stand-in for the session object read by the generated-query
+ flow (endpoint_url / model / headers)."""
+
+ endpoint_url = "http://example.local/v1"
+ model = "test-model"
+ headers = {"Authorization": "Bearer test"}
+
+
+class _Memory:
+ def load(self, owner=None):
+ return []
+
+
+class _Docs:
+ rag_manager = None
+
+
+def _patch_flow(monkeypatch, llm_behaviour, captured):
+ """Wire both seams of the generated-query flow: the LLM call and the
+ search call. ``llm_behaviour`` is either a string to return or an Exception
+ instance to raise."""
+
+ def _fake_search(query, *args, **kwargs):
+ captured["query"] = query
+ captured["kwargs"] = kwargs
+ return ("web context", [{"title": "src"}])
+
+ def _fake_llm(*args, **kwargs):
+ if isinstance(llm_behaviour, Exception):
+ raise llm_behaviour
+ return llm_behaviour
+
+ monkeypatch.setattr("src.chat_processor.comprehensive_web_search", _fake_search)
+ monkeypatch.setattr("src.llm_core.llm_call", _fake_llm)
+
+
+def test_generated_query_is_used_and_sanitized(monkeypatch):
+ """Requirement: on LLM success the generated query wins, and the *final*
+ query handed to comprehensive_web_search() is sanitized.
+
+ The fake LLM returns a query containing inline-code markdown so we can also
+ prove the sanitizer runs on the generated path (not just the fallback)."""
+ captured = {}
+ _patch_flow(monkeypatch, "capital of `France`", captured)
+
+ processor = ChatProcessor(memory_manager=_Memory(), personal_docs_manager=_Docs())
+ preface, _, _ = processor.build_context_preface(
+ message=_MESSY,
+ session=_Session(),
+ use_web=True,
+ use_memory=False,
+ use_rag=False,
+ )
+
+ assert "query" in captured, "comprehensive_web_search was not called"
+
+ # The generated query won (not the sanitized first-line fallback) ...
+ assert captured["query"] == "capital of France"
+ assert captured["query"] != _SANITIZED_FALLBACK
+ # ... and it was sanitized: no residual markdown fences/backticks.
+ assert "`" not in captured["query"]
+ assert "```" not in captured["query"]
+
+ # The other call-site kwargs (return_sources) are still forwarded.
+ assert captured["kwargs"].get("return_sources") is True
+ # And the retrieved context was still appended to the preface.
+ assert any("web context" in (msg.get("content") or "") for msg in preface)
+
+
+def test_falls_back_to_sanitized_first_line_when_llm_raises(monkeypatch):
+ """Requirement: when the LLM call raises, #4557's fallback (first non-empty
+ line) is used — and that fallback is sanitized before the search call."""
+ captured = {}
+ _patch_flow(monkeypatch, RuntimeError("LLM endpoint down"), captured)
+
+ processor = ChatProcessor(memory_manager=_Memory(), personal_docs_manager=_Docs())
+ processor.build_context_preface(
+ message=_MESSY,
+ session=_Session(),
+ use_web=True,
+ use_memory=False,
+ use_rag=False,
+ )
+
+ assert "query" in captured, "comprehensive_web_search was not called"
+ # Fallback was the first line ("Is `git reset` safe?"), sanitized.
+ assert captured["query"] == _SANITIZED_FALLBACK
+ assert "git reset" in captured["query"] # inline code preserved
+ assert "`" not in captured["query"] # backticks stripped
+ # The fenced body from later lines never reached the query.
+ assert "leaked body" not in captured["query"]
+
+
+def test_falls_back_to_sanitized_first_line_when_llm_returns_empty(monkeypatch):
+ """Requirement: when the LLM returns an empty/whitespace-only query, #4557
+ falls back — and that fallback is sanitized before the search call."""
+ captured = {}
+ _patch_flow(monkeypatch, " ", captured)
+
+ processor = ChatProcessor(memory_manager=_Memory(), personal_docs_manager=_Docs())
+ processor.build_context_preface(
+ message=_MESSY,
+ session=_Session(),
+ use_web=True,
+ use_memory=False,
+ use_rag=False,
+ )
+
+ assert "query" in captured, "comprehensive_web_search was not called"
+ assert captured["query"] == _SANITIZED_FALLBACK
+ assert "git reset" in captured["query"]
+ assert "`" not in captured["query"]
diff --git a/tests/test_webhook_dns_rebinding_pin.py b/tests/test_webhook_dns_rebinding_pin.py
new file mode 100644
index 000000000..144a19e96
--- /dev/null
+++ b/tests/test_webhook_dns_rebinding_pin.py
@@ -0,0 +1,156 @@
+"""Regression: webhook delivery must pin the TCP connect to the SSRF-approved IP.
+
+validate_webhook_url resolves the host to accept/reject, but the delivery
+connect previously re-resolved independently — a DNS record flipping between
+the two lookups (rebinding) could slip an internal IP past the check. _deliver
+now resolves+validates once via _validated_public_ips and pins the connect to
+that IP through _PinnedAsyncTransport. These tests drive the real transport
+against local servers so the pin is exercised end-to-end, not mocked away.
+"""
+import asyncio
+import http.server
+import ipaddress
+import socketserver
+import threading
+
+import pytest
+
+from tests.helpers.import_state import clear_module, preserve_import_state
+
+import os
+import sys
+from unittest.mock import patch
+
+with patch.dict(os.environ, {"DATABASE_URL": "sqlite:///:memory:"}), \
+ preserve_import_state("src.database", "core.database"):
+ clear_module("src.database")
+ _core_database = sys.modules.get("core.database")
+ if _core_database is not None and not getattr(_core_database, "__file__", None):
+ del sys.modules["core.database"]
+ import src.webhook_manager as wm
+
+
+# ---------------------------------------------------------------------------
+# _validated_public_ips
+# ---------------------------------------------------------------------------
+
+def test_validated_public_ips_rejects_metadata_literal():
+ with pytest.raises(ValueError):
+ wm._validated_public_ips("http://169.254.169.254/")
+
+
+def test_validated_public_ips_rejects_loopback_literal():
+ with pytest.raises(ValueError):
+ wm._validated_public_ips("http://127.0.0.1/")
+
+
+def test_validated_public_ips_returns_public_literal():
+ ips = wm._validated_public_ips("http://93.184.216.34/")
+ assert ips == [ipaddress.ip_address("93.184.216.34")]
+
+
+def test_validated_public_ips_rejects_hostname_resolving_private(monkeypatch):
+ # Rebinding shape: a hostname that (now) resolves into loopback space.
+ monkeypatch.setattr(wm, "_resolve_hostname_ips",
+ lambda h: [ipaddress.ip_address("127.0.0.1")])
+ with pytest.raises(ValueError):
+ wm._validated_public_ips("http://evil.rebind.example/")
+
+
+# ---------------------------------------------------------------------------
+# End-to-end: the pinned transport actually routes to the pinned IP
+# ---------------------------------------------------------------------------
+
+def _serve(handler):
+ srv = socketserver.TCPServer(("127.0.0.1", 0), handler)
+ port = srv.server_address[1]
+ threading.Thread(target=srv.serve_forever, daemon=True).start()
+ return srv, port
+
+
+def test_pinned_transport_connects_to_pinned_ip():
+ """A request whose URL host is a throwaway hostname is still delivered to
+ the pinned loopback IP — proving the socket destination comes from the pin,
+ not from resolving the URL host."""
+ hits = []
+
+ class _Handler(http.server.BaseHTTPRequestHandler):
+ def do_POST(self): # noqa: N802
+ length = int(self.headers.get("Content-Length", 0))
+ self.rfile.read(length)
+ hits.append(self.path)
+ self.send_response(204)
+ self.end_headers()
+
+ def log_message(self, *a):
+ pass
+
+ srv, port = _serve(_Handler)
+ try:
+ ip = ipaddress.ip_address("127.0.0.1")
+ transport = wm._PinnedAsyncTransport(ip)
+
+ async def go():
+ async with __import__("httpx").AsyncClient(
+ transport=transport, timeout=5, follow_redirects=False,
+ ) as client:
+ # Host "unresolvable.invalid" would never resolve; the pin is
+ # what makes this reach the loopback server on `port`.
+ return await client.post(
+ f"http://unresolvable.invalid:{port}/hook", content=b"{}",
+ )
+
+ resp = asyncio.run(go())
+ assert resp.status_code == 204
+ assert hits == ["/hook"]
+ finally:
+ srv.shutdown()
+
+
+def test_deliver_pins_to_validated_ip_end_to_end(monkeypatch):
+ """Full _deliver path: a hostname that validation resolves to loopback is
+ pinned to loopback and the local server receives the signed POST."""
+ received = {}
+
+ class _Handler(http.server.BaseHTTPRequestHandler):
+ def do_POST(self): # noqa: N802
+ length = int(self.headers.get("Content-Length", 0))
+ received["body"] = self.rfile.read(length)
+ received["event"] = self.headers.get("X-Odysseus-Event")
+ self.send_response(200)
+ self.end_headers()
+
+ def log_message(self, *a):
+ pass
+
+ srv, port = _serve(_Handler)
+
+ class _Query:
+ def filter(self, *a, **k): return self
+ def update(self, values): return None
+
+ class _Db:
+ def query(self, _m): return _Query()
+ def commit(self): pass
+ def rollback(self): pass
+ def close(self): pass
+
+ # Make both the validation resolve and the pin target loopback, and treat
+ # loopback as allowed for this test (production blocks it — here we only
+ # want to prove the pin routes to the validated IP).
+ monkeypatch.setattr(wm, "SessionLocal", lambda: _Db())
+ monkeypatch.setattr(wm, "_is_private_url", lambda url: False)
+ monkeypatch.setattr(wm, "_resolve_hostname_ips",
+ lambda h: [ipaddress.ip_address("127.0.0.1")])
+ monkeypatch.setattr(wm, "_ip_is_private", lambda a: False)
+
+ manager = wm.WebhookManager()
+ try:
+ asyncio.run(manager._deliver(
+ "hook-1", f"http://webhook.test:{port}/cb", "s3cret",
+ "webhook.test", {"ok": True},
+ ))
+ assert received.get("event") == "webhook.test"
+ assert b'"ok": true' in received["body"]
+ finally:
+ srv.shutdown()
diff --git a/tests/test_webhook_ssrf_resilience.py b/tests/test_webhook_ssrf_resilience.py
index e02f17a25..ca82a7595 100644
--- a/tests/test_webhook_ssrf_resilience.py
+++ b/tests/test_webhook_ssrf_resilience.py
@@ -96,26 +96,29 @@ async def test_webhook_delivery_uses_naive_utc_timestamps(monkeypatch):
class _Response:
status_code = 204
- class _Client:
- def __init__(self):
- self.content = ""
-
- async def post(self, _url, content, headers):
- self.content = content
- assert headers["X-Odysseus-Event"] == "webhook.test"
- return _Response()
-
db = _Db()
- client = _Client()
monkeypatch.setattr(wm, "SessionLocal", lambda: db)
manager = wm.WebhookManager()
- await manager._client.aclose()
- manager._client = client
+
+ # Replace the pinned-transport send seam so no real socket is opened. The
+ # public-IP literal below still exercises _validated_public_ips (which pins
+ # the connect); the captured content proves the body/headers are built.
+ captured = {}
+
+ async def _fake_send(url, body, headers, ip):
+ captured["content"] = body
+ captured["ip"] = str(ip)
+ assert headers["X-Odysseus-Event"] == "webhook.test"
+ return _Response()
+
+ monkeypatch.setattr(manager, "_send_request", _fake_send)
await manager._deliver("hook-1", "http://93.184.216.34/", None, "webhook.test", {"ok": True})
- body = json.loads(client.content)
+ # The delivery must have pinned to the literal public IP from the URL.
+ assert captured["ip"] == "93.184.216.34"
+ body = json.loads(captured["content"])
payload_timestamp = datetime.fromisoformat(body["timestamp"])
assert payload_timestamp.tzinfo is None
assert db.updates[0]["last_triggered_at"].tzinfo is None
diff --git a/tests/test_workspace_confine.py b/tests/test_workspace_confine.py
index 81bc7235c..6d90789c9 100644
--- a/tests/test_workspace_confine.py
+++ b/tests/test_workspace_confine.py
@@ -140,6 +140,65 @@ async def test_grep_and_ls_confined_e2e(ws, admin):
assert r["exit_code"] == 1 and "outside the workspace" in r["error"]
+@pytest.mark.asyncio
+async def test_glob_confined_e2e(ws, admin):
+ """glob's literal fast-path must stay inside the workspace. A pattern with
+ ../ or an absolute path outside the root would otherwise leak the existence
+ and full path of arbitrary host files (an oracle), even though read_file
+ blocks reading them."""
+ with open(os.path.join(ws, "found.py"), "w") as f:
+ f.write("x")
+ _, r = await execute_tool_block(_block("glob", json.dumps({"pattern": "found.py"})), owner="a", workspace=ws)
+ assert r["exit_code"] == 0 and "found.py" in r["output"]
+
+ # a secret outside the workspace must not be discoverable via glob
+ outside = tempfile.mkdtemp()
+ secret = os.path.join(outside, "secret.txt")
+ with open(secret, "w") as f:
+ f.write("nope")
+ # An escaping pattern must come back as "No files" (the not-found message),
+ # not as a match that returns the file's path. The not-found message echoes
+ # the pattern the model supplied, so the signal is the absence of a match,
+ # not the absence of the path string.
+ rel = os.path.relpath(secret, os.path.realpath(ws))
+ _, r = await execute_tool_block(_block("glob", json.dumps({"pattern": rel})), owner="a", workspace=ws)
+ assert r["exit_code"] == 0 and "No files" in r["output"] and secret not in r["output"]
+ _, r = await execute_tool_block(_block("glob", json.dumps({"pattern": secret})), owner="a", workspace=ws)
+ assert r["exit_code"] == 0 and "No files" in r["output"]
+
+
+@pytest.mark.asyncio
+async def test_glob_skips_sensitive_files_in_workspace(ws, admin):
+ """glob must not enumerate deny-listed sensitive files that live inside the
+ workspace. read_file/write_file/edit_file refuse them and grep skips them,
+ so glob surfacing their paths is an enumeration oracle for prompt-injection.
+ """
+ with open(os.path.join(ws, "keep.py"), "w") as f:
+ f.write("x")
+ with open(os.path.join(ws, ".env"), "w") as f:
+ f.write("AWS_SECRET=xxx")
+ with open(os.path.join(ws, "id_rsa"), "w") as f: # non-dotfile key at root
+ f.write("KEY")
+ os.makedirs(os.path.join(ws, ".ssh"), exist_ok=True)
+ with open(os.path.join(ws, ".ssh", "authorized_keys"), "w") as f:
+ f.write("ssh-rsa AAAA")
+
+ # A recursive wildcard returns ordinary files but none of the sensitive
+ # ones. The pattern "**/*" contains no secret names, so a secret basename
+ # appearing in the output is a real leak (not the echoed not-found pattern).
+ _, r = await execute_tool_block(_block("glob", json.dumps({"pattern": "**/*"})), owner="a", workspace=ws)
+ assert r["exit_code"] == 0
+ assert "keep.py" in r["output"]
+ for leak in (".env", "id_rsa", "authorized_keys"):
+ assert leak not in r["output"], f"glob leaked sensitive file: {leak}"
+
+ # Directly targeting a sensitive file (literal fast-path and wildcard) must
+ # come back as the not-found message, never a match with the file's path.
+ for pat in (".env", "**/id_rsa", "**/authorized_keys"):
+ _, r = await execute_tool_block(_block("glob", json.dumps({"pattern": pat})), owner="a", workspace=ws)
+ assert r["exit_code"] == 0 and "No files" in r["output"]
+
+
@pytest.mark.asyncio
async def test_subprocess_cwd_is_workspace_e2e(ws, admin):
"""python tool runs with cwd = workspace (OS-agnostic probe)."""