Harden security: sanitize errors, validate URL schemes, protect against ReDoS

- Sanitize streaming error messages to prevent internal hostname leakage
- Add URL scheme allowlist (http/https only) in extractor and JS fallback
- Wrap regex filter_query in async timeout (2s) to prevent ReDoS
- Tighten max_content_length limit from 500K to 200K chars
- Unify filter_query max_length to 200 in schema and function
- Fix test fixtures: FakeExtractor, FakeCache signatures, disable rerank in tests

.gitignore

@@ -138,3 +138,10 @@ Thumbs.db
 .copilot
 .gpt
 .chatgpt
+CLAUDE.md
+
+# Secrets & certificates
+*.pem
+*.key
+.secrets/
+config.local.yaml

app/models/schemas.py

@@ -62,7 +62,7 @@ class SearchRequest(BaseModel):
     time_range: TimeRange | None = Field(default=None, description="Filter results by time range")
     filter_query: str | None = Field(
         default=None,
-        max_length=500,
+        max_length=200,
         description="Regex pattern to filter results — only results whose title, URL, or content match are returned",
     )
     user_location: UserLocation | None = Field(default=None, description="User location for localizing search results")
@@ -97,7 +97,7 @@ class ExtractRequest(BaseModel):
     extract_depth: ExtractDepth = Field(default=ExtractDepth.basic, description="Extraction depth")
     format: ContentFormat = Field(default=ContentFormat.markdown, description="Output format")
     max_content_length: int | None = Field(
-        default=None, ge=100, le=500000, description="Max chars per extracted page (overrides server default)"
+        default=None, ge=100, le=200000, description="Max chars per extracted page (overrides server default)"
     )
 
 

app/routers/search.py

@@ -185,8 +185,8 @@ async def _enrich_snippets(
         await _enrich_batch(remaining[:retry_count])
 
 
-def _safe_regex_filter(pattern_str: str, results: list[SearchResult]) -> list[SearchResult]:
-    """Filter results by regex, compiling with a size guard against ReDoS."""
+def _regex_filter_sync(pattern_str: str, results: list[SearchResult]) -> list[SearchResult]:
+    """Synchronous regex filter (runs in a thread via _safe_regex_filter)."""
     if len(pattern_str) > 200:
         logger.warning("filter_query_too_long", length=len(pattern_str))
         return results
@@ -198,6 +198,18 @@ def _safe_regex_filter(pattern_str: str, results: list[SearchResult]) -> list[Se
     return [r for r in results if pattern.search(r.title) or pattern.search(r.url) or pattern.search(r.content)]
 
 
+async def _safe_regex_filter(pattern_str: str, results: list[SearchResult]) -> list[SearchResult]:
+    """Filter results by regex with ReDoS protection via timeout."""
+    try:
+        return await asyncio.wait_for(
+            asyncio.to_thread(_regex_filter_sync, pattern_str, results),
+            timeout=2.0,
+        )
+    except TimeoutError:
+        logger.warning("filter_query_timeout", filter_query=pattern_str)
+        return results
+
+
 def _country_to_searxng_lang(country: str) -> str:
     code = country.upper().strip()
     if lang := _COUNTRY_LANG_MAP.get(code):
@@ -299,7 +311,7 @@ async def _do_search(request: Request, body: SearchRequest) -> SearchResponse:
 
     # Dynamic filtering: keep only results matching the filter regex
     if body.filter_query and results:
-        results = _safe_regex_filter(body.filter_query, results)
+        results = await _safe_regex_filter(body.filter_query, results)
 
     # Snippet enrichment: extract content for top-N results in basic mode
     if (

app/routers/search_stream.py

@@ -56,8 +56,9 @@ async def search_stream(
                 include_images=body.include_images,
                 language=language,
             )
-        except Exception as e:
-            yield {"event": "error", "data": json.dumps({"error": str(e)})}
+        except Exception:
+            logger.exception("search_stream_backend_failed", query=body.query)
+            yield {"event": "error", "data": json.dumps({"error": "Search service temporarily unavailable"})}
             return
 
         # Build results, separate infobox entries, and deduplicate
@@ -87,7 +88,7 @@ async def search_stream(
 
         # Dynamic filtering
         if body.filter_query and results:
-            results = _safe_regex_filter(body.filter_query, results)
+            results = await _safe_regex_filter(body.filter_query, results)
 
         # Snippet enrichment for basic mode (shared logic with sync endpoint)
         if (

app/services/extractor.py

@@ -18,6 +18,8 @@ from app.services.resilience import extract_circuit, retry_async
 
 logger = structlog.get_logger(__name__)
 
+_ALLOWED_SCHEMES = frozenset({"http", "https"})
+
 
 @dataclass
 class ExtractionResult:
@@ -66,6 +68,9 @@ class ContentExtractor:
     async def extract_url(
         self, url: str, *, output_format: str = "markdown", max_content_length: int | None = None
     ) -> ExtractionResult:
+        scheme = urlparse(url).scheme.lower()
+        if scheme not in _ALLOWED_SCHEMES:
+            return ExtractionResult(url=url, error=f"Unsupported URL scheme: {scheme}")
         domain_sem = self._get_domain_semaphore(url)
         async with self._global_semaphore, domain_sem:
             return await self._fetch_and_extract(url, output_format, max_content_length)
@@ -260,6 +265,8 @@ class ContentExtractor:
 
     async def _extract_js(self, url: str, output_format: str) -> str | None:
         """Tier 3: render page with headless browser via Browserless, then extract."""
+        if urlparse(url).scheme.lower() not in _ALLOWED_SCHEMES:
+            return None
         try:
             from playwright.async_api import async_playwright
         except ImportError:

tests/conftest.py

@@ -7,7 +7,6 @@ import pytest_asyncio
 from httpx import ASGITransport, AsyncClient
 
 from app.services.extractor import ExtractionResult
-from app.services.reranker import RerankerService
 from app.services.search_backend import (
     BackendSearchResponse,
     RawImageResult,
@@ -48,13 +47,19 @@ class FakeSearchBackend(SearchBackend):
 
 
 class FakeExtractor:
-    async def extract_url(self, url: str, *, output_format: str = "markdown") -> ExtractionResult:
+    async def extract_url(
+        self, url: str, *, output_format: str = "markdown", max_content_length: int | None = None
+    ) -> ExtractionResult:
         if "fail" in url:
             return ExtractionResult(url=url, error="Extraction failed")
         return ExtractionResult(url=url, content=f"# Extracted content from {url}\n\nFull text here.")
 
-    async def extract_urls(self, urls: list[str], *, output_format: str = "markdown") -> list[ExtractionResult]:
-        return await asyncio.gather(*[self.extract_url(url, output_format=output_format) for url in urls])
+    async def extract_urls(
+        self, urls: list[str], *, output_format: str = "markdown", max_content_length: int | None = None
+    ) -> list[ExtractionResult]:
+        return await asyncio.gather(
+            *[self.extract_url(url, output_format=output_format, max_content_length=max_content_length) for url in urls]
+        )
 
 
 # ---- Fake cache ----
@@ -66,7 +71,7 @@ class FakeCache:
         self._extract: dict[str, str] = {}
         self.enabled = True
 
-    async def get_search(self, query: str, params_hash: str) -> dict | None:
+    async def get_search(self, query: str, params_hash: str, *, allow_stale: bool = False) -> dict | None:
         return self._search.get(f"{query}|{params_hash}")
 
     async def set_search(self, query: str, params_hash: str, data: dict) -> None:
@@ -162,12 +167,17 @@ async def _reset_sse_app_status():
 async def app():
     """Create a test app with faked services."""
     # Import here to avoid module-level config loading issues
+    from app.config import settings
     from app.main import app as _app
 
+    # Disable rerank and snippet enrichment in tests to keep result counts predictable
+    settings.rerank.enabled = False
+    settings.extraction.snippet_enrichment = False
+
     _app.state.search_backend = FakeSearchBackend()
     _app.state.extractor = FakeExtractor()
     _app.state.cache = FakeCache()
-    _app.state.reranker = RerankerService()  # disabled by default
+    _app.state.reranker = None
     _app.state.llm = FakeLLMService()
 
     yield _app