Detect Google Scholar's block page, and unwrap Bing's redirect links

Both found by running the live smoke test against the deployed MCP server,
not by the suite - the division of labour CONTRIBUTING.md already
describes.

Google Scholar serves an anti-bot interstitial to the Space's IP instead of
results. That page has no `div.gs_ri`, so the scraper sat in
`wait_for_selector` for the full 30 seconds and then reported a bare
"Timeout 30000ms exceeded" - slow for every caller, and worded like a
broken selector rather than a blocked deployment.

It now waits for the results selector and Google's challenge elements
together, so a block is recognised as soon as the page loads and raises
GoogleScholarBlockedException with a message that says what actually
happened. A challenge page carrying none of those elements falls back to
matching Google's wording, but only after the selector wait has failed:
sniffing the text up front would be faster and wrong, since a legitimate
Scholar search for "unusual traffic" returns papers whose snippets contain
that phrase. A page with neither results nor a recognisable challenge still
raises the underlying timeout, so a genuine selector regression can't hide
behind the block excuse.

Bing answers normally but does not put the destination in a result's href.
Every link is wrapped in `https://www.bing.com/ck/a?...&u=a1<base64url>`,
so callers were handed tracking links rather than the pages they point at -
useless for citation, deduplication, or deciding whether a result is worth
fetching. decode_bing_redirect unwraps them, verified against a redirect
captured from the live deployment.

Anything that isn't a decodable Bing redirect is returned unchanged: a link
we can't decode is still a working link, and losing it would be worse than
leaving it wrapped. The decoded payload comes from a scraped page, so only
http(s) destinations are let out - a javascript: or data: URL keeps the
original link instead of being handed back as if it were a result.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MSNSYnceqvdVz7Csis4K9e

CONTRIBUTING.md

@@ -30,13 +30,22 @@ ruff check .   # same lint CI runs
 pytest
 ```
 
-238 tests, a few seconds, fully offline. All outbound HTTP is mocked with
+257 tests, a few seconds, fully offline. All outbound HTTP is mocked with
 `respx`; the Playwright-driven scrapers (Bing, Brave, Google Scholar,
 Google Patents search) run against a real headless Chromium but navigate to
 local fixture HTML instead of the live sites — see `tests/helpers.py` for
 how and why. Nothing here depends on, or can trip, any external site's
 anti-bot measures.
 
+A note on the two backends that are blocked from the deployed Space's IP
+at the time of writing: Google Scholar serves an anti-bot interstitial
+rather than results, and `query_google_scholar` now detects that and raises
+`GoogleScholarBlockedException` instead of waiting out a 30s selector
+timeout. Bing answers normally but wraps every result link in its `/ck/a`
+redirector, which `decode_bing_redirect` unwraps. Both were found by
+running the live smoke test below against the deployed MCP server, not by
+the suite - which is exactly the division of labour described next.
+
 **Important limitation to keep in mind:** this suite protects against
 regressions in this code. It cannot catch a live site changing its markup,
 because the fixtures are frozen snapshots of the selector contract as of

serp.py

@@ -1,10 +1,12 @@
+import base64
 from contextlib import asynccontextmanager
 from typing import Optional
 from duckduckgo_search import DDGS
 import httpx
 from pydantic import BaseModel, Field, field_validator
 from playwright.async_api import Browser, BrowserContext, Page
-from urllib.parse import quote_plus
+from playwright.async_api import TimeoutError as PlaywrightTimeoutError
+from urllib.parse import parse_qs, quote_plus, urlparse
 import re
 from lxml import etree
 import asyncio
@@ -53,6 +55,22 @@ class BraveSearchBlockedException(Exception):
     pass
 
 
+class GoogleScholarBlockedException(Exception):
+    """Raised when Google Scholar serves its anti-bot interstitial.
+
+    Scholar refuses datacenter IPs with a /sorry challenge page rather than
+    results. That page simply has no `div.gs_ri`, so without this the
+    scraper waits out the full selector timeout and then reports a bare
+    "Timeout exceeded" - slow, and indistinguishable from a genuine
+    selector regression.
+    """
+
+    def __init__(self, *args):
+        super().__init__(
+            "Google Scholar blocked the request (anti-bot interstitial served "
+            "instead of results); the deployment's IP is likely rate-limited.")
+
+
 class BrowserUnavailableError(Exception):
     """Raised when a Playwright-backed query is attempted but the browser failed to start."""
 
@@ -119,11 +137,49 @@ async def _block_stylesheet_and_image_resources(route, request):
         await route.continue_()
 
 
-async def _extract_google_scholar_results(page: Page, n_results: int) -> list[dict]:
-    """Extract results from an already-loaded Google Scholar results page."""
-    await page.wait_for_selector("div.gs_ri")
+# How long to wait for Scholar to show either results or a challenge.
+SCHOLAR_SELECTOR_TIMEOUT_MS = 30_000
+
+_SCHOLAR_RESULT_SELECTOR = "div.gs_ri"
+# Elements Google's challenge page carries. Waiting for these *alongside*
+# the results selector is what makes a block fail fast instead of costing a
+# full timeout.
+_SCHOLAR_BLOCK_SELECTOR = "form#captcha-form, #recaptcha, .g-recaptcha, #infoDiv"
+# Fallback for a challenge page carrying none of the above - Google varies
+# the markup, so match on the wording it has used consistently.
+_SCHOLAR_BLOCK_TEXT_MARKERS = (
+    "unusual traffic",
+    "not a robot",
+    "/sorry/index",
+)
+
+
+def _looks_like_scholar_block(page_content: str) -> bool:
+    lowered = page_content.lower()
+    return any(marker in lowered for marker in _SCHOLAR_BLOCK_TEXT_MARKERS)
+
 
-    items = await page.locator("div.gs_ri").all()
+async def _extract_google_scholar_results(page: Page, n_results: int,
+                                          timeout_ms: int = SCHOLAR_SELECTOR_TIMEOUT_MS) -> list[dict]:
+    """Extract results from an already-loaded Google Scholar results page.
+
+    Raises GoogleScholarBlockedException when Google served its anti-bot
+    interstitial instead of results. A page with neither results nor a
+    recognisable challenge still raises the underlying timeout, so a real
+    selector regression can't hide behind the block excuse.
+    """
+    try:
+        await page.wait_for_selector(
+            f"{_SCHOLAR_RESULT_SELECTOR}, {_SCHOLAR_BLOCK_SELECTOR}", timeout=timeout_ms)
+    except PlaywrightTimeoutError:
+        if _looks_like_scholar_block(await page.content()):
+            raise GoogleScholarBlockedException() from None
+        raise
+
+    if await page.locator(_SCHOLAR_BLOCK_SELECTOR).count():
+        raise GoogleScholarBlockedException()
+
+    items = await page.locator(_SCHOLAR_RESULT_SELECTOR).all()
     results = []
     for item in items[:n_results]:
         title = await item.locator("h3").inner_text(timeout=1000)
@@ -275,6 +331,39 @@ async def query_brave_search(browser: Browser, q: str, n_results: int = 10):
         return await _extract_brave_results(page, n_results)
 
 
+def decode_bing_redirect(url: Optional[str]) -> Optional[str]:
+    """Resolve a Bing `/ck/a` tracking link to the page it points at.
+
+    Bing doesn't put the destination in a result's href; it wraps every link
+    in `https://www.bing.com/ck/a?...&u=a1<base64url>&ntb=1`. Handing that
+    back means callers get a tracking link instead of the page - useless for
+    citation, deduplication, or deciding whether a result is worth fetching.
+
+    Anything that isn't a decodable Bing redirect is returned unchanged: a
+    link we can't decode is still a working link, and losing it would be
+    worse than leaving it wrapped. The decoded value comes from a scraped
+    page, so only http(s) destinations are let out.
+    """
+    if not url or "bing.com/ck/a" not in url:
+        return url
+
+    encoded = parse_qs(urlparse(url).query).get("u", [""])[0]
+    if not encoded.startswith("a1"):
+        return url
+
+    payload = encoded[2:]
+    try:
+        # Bing strips base64 padding; restore however much is missing.
+        decoded = base64.urlsafe_b64decode(
+            payload + "=" * (-len(payload) % 4)).decode("utf-8")
+    except (ValueError, UnicodeDecodeError):
+        return url
+
+    if not decoded.startswith(("http://", "https://")):
+        return url
+    return decoded
+
+
 async def _extract_bing_results(page: Page, n_results: int) -> list[dict]:
     """Extract results from an already-loaded Bing results page."""
     await page.wait_for_selector("li.b_algo")
@@ -306,7 +395,7 @@ async def _extract_bing_results(page: Page, n_results: int) -> list[dict]:
         if title and url:
             results.append({
                 "title": title.strip(),
-                "href": url.strip(),
+                "href": decode_bing_redirect(url.strip()),
                 "body": snippet.strip()
             })
 

tests/fixtures/bing_results_redirect.html

@@ -0,0 +1,13 @@
+<!DOCTYPE html>
+<html>
+<body>
+<li class="b_algo">
+  <h2><a href="https://www.bing.com/ck/a?!&amp;&amp;p=355fd6a450a5&amp;ptn=3&amp;ver=2&amp;hsh=4&amp;u=a1aHR0cHM6Ly9kb2NzLnB5dGhvbi5vcmcvMy9saWJyYXJ5L2FzeW5jaW8uaHRtbA&amp;ntb=1">asyncio documentation</a></h2>
+  <div class="b_caption"><p>Snippet A</p></div>
+</li>
+<li class="b_algo">
+  <h2><a href="https://example.org/direct">Direct Result</a></h2>
+  <div class="b_caption"><p>Snippet B</p></div>
+</li>
+</body>
+</html>

tests/fixtures/scholar_blocked.html

@@ -0,0 +1,14 @@
+<!DOCTYPE html>
+<html>
+<head><title>https://scholar.google.com/sorry/index</title></head>
+<body>
+<div id="infoDiv">
+  <p>Our systems have detected unusual traffic from your computer network.
+     This page checks to see if it's really you sending the requests, and
+     not a robot.</p>
+</div>
+<form id="captcha-form" action="/sorry/index" method="post">
+  <div class="g-recaptcha" data-sitekey="placeholder"></div>
+</form>
+</body>
+</html>

tests/fixtures/scholar_blocked_no_markers.html

@@ -0,0 +1,6 @@
+<!DOCTYPE html>
+<html>
+<body>
+<p>Our systems have detected unusual traffic from your computer network.</p>
+</body>
+</html>

tests/test_bing_redirect.py

@@ -0,0 +1,98 @@
+"""Decoding Bing's result-link redirector.
+
+Bing does not put the destination in a result's href. It wraps every link
+in `https://www.bing.com/ck/a?...&u=a1<base64url>&ntb=1`, so the URLs this
+API handed back were Bing tracking links rather than the pages they point
+at. An agent following one goes through Bing's redirector, and the href is
+useless for deduplication, citation, or deciding whether a result is worth
+fetching.
+
+Verified against a redirect captured from the live deployment: the `u`
+parameter is the literal prefix "a1" followed by the destination, base64url
+encoded with its padding stripped.
+"""
+
+import pytest
+
+from serp import decode_bing_redirect
+
+# Captured from a real /serp/search response.
+REAL_REDIRECT = (
+    "https://www.bing.com/ck/a?!&&p=355fd6a450a5a451290a8e71334947199bc81cf4"
+    "c6084e0f8daa8d5a0debc698JmltdHM9MTc4Nzk2MTYwMA&ptn=3&ver=2&hsh=4"
+    "&fclid=12f2c74a-55e9-6296-1923-d08f542d635f"
+    "&u=a1aHR0cHM6Ly9kb2NzLnB5dGhvbi5vcmcvMy9saWJyYXJ5L2FzeW5jaW8uaHRtbA&ntb=1"
+)
+
+
+def test_decodes_a_real_captured_redirect():
+    assert decode_bing_redirect(REAL_REDIRECT) == "https://docs.python.org/3/library/asyncio.html"
+
+
+def test_decodes_a_url_whose_base64_needs_padding():
+    """Bing strips base64 padding, so the decoder has to restore it - and
+    the amount needed varies with the length of the destination."""
+    import base64
+
+    for destination in [
+        "https://example.org/a",
+        "https://example.org/ab",
+        "https://example.org/abc",
+        "https://example.org/abcd",
+    ]:
+        payload = base64.urlsafe_b64encode(destination.encode()).decode().rstrip("=")
+        assert decode_bing_redirect(f"https://www.bing.com/ck/a?u=a1{payload}") == destination
+
+
+# ------------------------------ leaving links alone ------------------------------
+# Every failure mode below must return the original URL. A link that can't be
+# decoded is still a working link; replacing it with None or "" would lose it.
+
+
+def test_a_direct_url_passes_through_unchanged():
+    assert decode_bing_redirect("https://example.org/page") == "https://example.org/page"
+
+
+def test_a_redirect_without_a_u_parameter_passes_through():
+    url = "https://www.bing.com/ck/a?!&&p=abc&ntb=1"
+    assert decode_bing_redirect(url) == url
+
+
+def test_a_u_parameter_without_the_a1_prefix_passes_through():
+    url = "https://www.bing.com/ck/a?u=zzsomethingelse"
+    assert decode_bing_redirect(url) == url
+
+
+def test_undecodable_base64_passes_through():
+    url = "https://www.bing.com/ck/a?u=a1!!!not-base64!!!"
+    assert decode_bing_redirect(url) == url
+
+
+def test_base64_that_is_not_utf8_passes_through():
+    import base64
+    payload = base64.urlsafe_b64encode(b"\xff\xfe\xfd").decode().rstrip("=")
+    url = f"https://www.bing.com/ck/a?u=a1{payload}"
+    assert decode_bing_redirect(url) == url
+
+
+@pytest.mark.parametrize("hostile", [
+    "javascript:alert(1)",
+    "data:text/html;base64,PHNjcmlwdD4=",
+    "file:///etc/passwd",
+    "not a url at all",
+])
+def test_a_decoded_non_http_scheme_is_rejected(hostile):
+    """The decoded payload is attacker-influenceable content from a scraped
+    page, so only http(s) destinations are allowed out. Anything else keeps
+    the original Bing link rather than being handed to a caller as if it
+    were a result URL.
+    """
+    import base64
+    payload = base64.urlsafe_b64encode(hostile.encode()).decode().rstrip("=")
+    url = f"https://www.bing.com/ck/a?u=a1{payload}"
+
+    assert decode_bing_redirect(url) == url
+
+
+def test_none_is_handled():
+    assert decode_bing_redirect(None) is None

tests/test_search_service.py

@@ -388,3 +388,23 @@ async def test_search_names_every_failed_query_not_just_the_last(monkeypatch):
 
     assert result.results == []
     assert "'a'" in result.error and "'b'" in result.error
+
+
+async def test_a_scholar_block_is_reported_as_such_to_the_caller(monkeypatch):
+    """The point of the dedicated exception is the message the caller sees:
+    "Google Scholar blocked the request" tells an operator their IP is
+    rate-limited, where the bare "Timeout 30000ms exceeded" it replaced read
+    like a broken selector.
+    """
+    from serp import GoogleScholarBlockedException
+
+    async def blocked(browser, q, n):
+        raise GoogleScholarBlockedException()
+
+    monkeypatch.setattr(services_module, "query_google_scholar", blocked)
+
+    result = await make_service().google_scholar(SerpQuery(queries=["anything"]))
+
+    assert result.results == []
+    assert "blocked" in result.error.lower()
+    assert "timeout" not in result.error.lower()

tests/test_serp_playwright.py

@@ -20,6 +20,7 @@ from helpers import load_fixture
 from serp import (
     BraveSearchBlockedException,
     BrowserUnavailableError,
+    GoogleScholarBlockedException,
     _block_stylesheet_and_image_resources,
     _extract_bing_results,
     _extract_brave_results,
@@ -173,3 +174,87 @@ async def test_everything_else_is_allowed_through(resource_type):
     await _block_stylesheet_and_image_resources(route, _FakeRequest(resource_type))
 
     assert route.continued and not route.aborted
+
+
+# ----------------------------- Google Scholar blocking -----------------------------
+# Scholar serves an anti-bot interstitial to datacenter IPs instead of
+# results. Because that page simply has no `div.gs_ri`, the scraper used to
+# sit in `wait_for_selector` for the full 30s timeout and then report a bare
+# "Timeout 30000ms exceeded" - slow, and it reads like a broken selector
+# rather than a blocked deployment. Observed live on the deployed Space.
+
+
+async def test_google_scholar_block_page_raises(page_factory):
+    page = await page_factory(load_fixture("scholar_blocked.html"))
+
+    with pytest.raises(GoogleScholarBlockedException):
+        await _extract_google_scholar_results(page, 10)
+
+
+async def test_google_scholar_block_is_detected_without_waiting_for_the_timeout(page_factory):
+    """The whole point: a challenge page is recognised immediately rather
+    than costing every caller a full selector timeout."""
+    import time
+
+    page = await page_factory(load_fixture("scholar_blocked.html"))
+
+    started = time.perf_counter()
+    with pytest.raises(GoogleScholarBlockedException):
+        await _extract_google_scholar_results(page, 10)
+    elapsed = time.perf_counter() - started
+
+    assert elapsed < 5, f"took {elapsed:.1f}s to notice a block page"
+
+
+async def test_google_scholar_block_is_detected_from_page_text_alone(page_factory):
+    """Google varies its interstitial. When none of the known challenge
+    elements are present, fall back to the wording before giving up.
+
+    This path is deliberately reached only *after* the selector wait times
+    out, so the test passes a short timeout rather than sitting through the
+    real one. Sniffing the text up front would be faster but wrong: a
+    legitimate Scholar search for "unusual traffic" returns papers whose
+    snippets contain that phrase, and would be misreported as a block. The
+    fuzzy check is only safe once the precise one has already failed.
+    """
+    page = await page_factory(load_fixture("scholar_blocked_no_markers.html"))
+
+    with pytest.raises(GoogleScholarBlockedException):
+        await _extract_google_scholar_results(page, 10, timeout_ms=1000)
+
+
+async def test_a_page_with_neither_results_nor_a_block_still_times_out(page_factory):
+    """An unrecognised empty page must not be silently reported as a block -
+    that would hide a genuine selector regression behind a plausible excuse.
+    """
+    from playwright.async_api import TimeoutError as PlaywrightTimeoutError
+
+    page = await page_factory("<html><body><p>something else entirely</p></body></html>")
+
+    with pytest.raises(PlaywrightTimeoutError):
+        await _extract_google_scholar_results(page, 10, timeout_ms=1000)
+
+
+async def test_normal_scholar_results_are_unaffected(page_factory):
+    """The block check must not cost the happy path anything."""
+    page = await page_factory(load_fixture("scholar_results.html"))
+
+    results = await _extract_google_scholar_results(page, 10)
+
+    assert [r["title"] for r in results] == ["Scholar Paper A", "Scholar Paper B"]
+
+
+# ------------------------------- Bing redirect links -------------------------------
+
+
+async def test_bing_result_hrefs_are_decoded(page_factory):
+    """Bing wraps result links in its /ck/a redirector; the extractor should
+    hand back the destination, not the tracking link."""
+    page = await page_factory(load_fixture("bing_results_redirect.html"))
+
+    results = await _extract_bing_results(page, 10)
+
+    assert [r["href"] for r in results] == [
+        "https://docs.python.org/3/library/asyncio.html",
+        "https://example.org/direct",
+    ]