Skip to content

fix(docker): report a refused seed as one failed result, not a dead batch #2299

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wippa-studios wants to merge 1 commit into from
Closed

fix(docker): report a refused seed as one failed result, not a dead batch #2299

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
173 changes: 157 additions & 16 deletions deploy/docker/api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import os
import json
import asyncio
from typing import List, Tuple, Dict
from typing import List, Tuple, Dict, NamedTuple
from functools import partial
from uuid import uuid4
from datetime import datetime, timezone
Expand All @@ -18,6 +18,7 @@
from crawl4ai import (
AsyncWebCrawler,
CrawlerRunConfig,
CrawlResult,
LLMExtractionStrategy,
CacheMode,
BrowserConfig,
Expand Down Expand Up @@ -647,14 +648,119 @@ async def stream_results(crawler: AsyncWebCrawler, results_gen: AsyncGenerator)
await _dispose_crawler(crawler)


def _normalize_and_validate_seeds(urls: List[str]) -> List[str]:
class _SeedBatch(NamedTuple):
"""The outcome of normalizing and destination-checking one batch of seeds.

`crawlable` are the seeds to fetch, in the caller's order. `refused` holds
`(url, detail)` for each seed the destination check turned away, in the
caller's order, where `detail` is the message the 400 used to carry.
"""
crawlable: List[str]
refused: List[Tuple[str, str]]


def _normalize_and_validate_seeds(urls: List[str]) -> _SeedBatch:
"""Prefix bare hosts with https:// and SSRF-validate every seed URL's
destination. Shared by the streaming and non-streaming crawl handlers so a
new entry point cannot silently skip the destination check."""
urls = [('https://' + url) if not url.startswith(('http://', 'https://')) and not url.startswith(("raw:", "raw://")) else url for url in urls]
new entry point cannot silently skip the destination check.

A refused seed does not fail the whole request. It is reported back as one
failed result among the batch, the way a robots.txt refusal already is, so
one bad URL in a sitemap-derived list cannot take a batch down (#2288). The
detail is carried through verbatim, which is what keeps this from becoming
a resolution oracle: an internal address and a name that does not resolve
both still produce the same opaque "URL blocked", and the caller already
knows the hostname it sent. A request with nothing left to crawl is still a
rejected request, and keeps the 400 (see _require_crawlable_seeds)."""
crawlable: List[str] = []
refused: List[Tuple[str, str]] = []
for url in urls:
validate_url_destination(url)
return urls
url = ('https://' + url) if not url.startswith(('http://', 'https://')) and not url.startswith(("raw:", "raw://")) else url
try:
validate_url_destination(url)
except HTTPException as e:
refused.append((url, e.detail))
continue
crawlable.append(url)
if refused:
# One line an operator can grep, so a refused seed is visible server-side
# and not just inferable from the response. The URLs are the caller's own
# and come straight back to it in the failed results. Each is truncated
# because a seed can be arbitrarily long, and a scanner probing blocked
# URLs should not be able to write megabytes into the log.
shown = ", ".join(url[:200] for url, _detail in refused[:5])
if len(refused) > 5:
shown += f", +{len(refused) - 5} more"
logger.warning(
"[seeds] %d of %d seed(s) refused by the destination check, "
"reported as failed results: %s",
len(refused), len(urls), shown,
)
return _SeedBatch(crawlable, refused)


def _require_crawlable_seeds(batch: _SeedBatch) -> None:
"""Refuse a request that has no crawlable seed left.

With every seed refused there is no batch to report per-URL failures in, so
this stays the 400 it has always been. That also keeps a single blocked URL
a 400, which is the contract the /md, /llm and /crawl/stream callers (and
the existing SSRF tests) rely on.
"""
if batch.refused and not batch.crawlable:
raise HTTPException(status_code=400, detail=batch.refused[0][1])


def _refused_seed_result(url: str, detail: str):
"""A refused seed as a failed CrawlResult.

Shaped like the robots.txt refusal in async_webcrawler.arun() (success
False, 403, a marker header) so a caller handles one refused seed and one
robots-refused seed the same way. Built as a real CrawlResult so the key
set matches every other entry in `results` -- the SDK client rebuilds each
one with `CrawlResult(**result)`, so a partial dict would break it.
"""
return CrawlResult(
url=url,
html="",
success=False,
status_code=403,
error_message=detail,
response_headers={"X-Egress-Status": "Blocked by egress policy"},
)


def _append_refused_results(results: List, batch: _SeedBatch) -> List:
"""Add one failed result per refused seed to `results`.

Appended, not spliced into place: MemoryAdaptiveDispatcher.run_urls returns
in completion order, so `results` never lined up with the caller's `urls` to
begin with. Callers therefore already have to match a result to its seed by
`result.url` -- which is exactly the field that makes a refused seed
identifiable, and the thing the 400 gave them no way to recover.
"""
if not batch.refused:
return results
return list(results) + [
_refused_seed_result(url, detail) for url, detail in batch.refused
]


async def _prepend_refused_results(results_gen, batch: _SeedBatch):
"""Yield one failed result per refused seed, then the crawled results.

A refusal is known before the first byte is fetched, so it goes first. A
stream is completion-ordered like the batch path, so a caller matches
results to seeds by `result.url` here too.

The results are built before the first yield on purpose: this generator is
consumed inside the response, where an exception would truncate a 200 the
client had already been given, with no error frame and no completion marker.
"""
for result in [_refused_seed_result(url, detail) for url, detail in batch.refused]:
yield result
async for result in results_gen:
yield result


async def handle_crawl_request(
Expand Down Expand Up @@ -684,7 +790,12 @@ async def handle_crawl_request(
peak_mem_mb = start_mem_mb

try:
urls = _normalize_and_validate_seeds(urls)
seeds = _normalize_and_validate_seeds(urls)
_require_crawlable_seeds(seeds)
# A refused seed comes back as a failed result rather than a 400, so the
# rest of this handler only ever sees the seeds it may actually fetch.
seed_count = len(urls)
urls = seeds.crawlable
browser_config = BrowserConfig.load(browser_config, provenance=Provenance.UNTRUSTED)
crawler_config = CrawlerRunConfig.load(crawler_config, provenance=Provenance.UNTRUSTED)
from egress_broker import enforce_egress
Expand Down Expand Up @@ -722,8 +833,14 @@ async def handle_crawl_request(

base_config = config["crawler"]["base_config"]

# Build the config(s) to pass to arun/arun_many
if crawler_configs and len(urls) > 1:
# Build the config(s) to pass to arun/arun_many.
# A caller that sent per-URL crawler_configs still means them when a
# seed is refused, so the branch keys off the ORIGINAL url count, not
# the number that survived. The list itself is not filtered or
# reordered: the dispatcher pairs a config to a URL by url_matcher
# (see BaseDispatcher.select_config), not by position.
per_url_configs = bool(crawler_configs) and seed_count > 1
if per_url_configs:
# Per-URL config list: deserialize each and apply base_config
config_list = [CrawlerRunConfig.load(cc, provenance=Provenance.UNTRUSTED) for cc in crawler_configs]
for cfg in config_list:
Expand All @@ -746,9 +863,15 @@ async def handle_crawl_request(
effective_config = crawler_config

results = []
func = getattr(crawler, "arun" if len(urls) == 1 else "arun_many")
# arun_many unless exactly one seed is left, or the caller sent per-URL
# configs (arun_many is what pairs them to their URLs). The `!= 1` shape
# is deliberate: /crawl and /crawl/stream reject an empty urls list, but
# CrawlJobPayload.urls has no min_length, so arun_many([]) is still
# reachable here and must not become an arun(urls[0]) IndexError.
use_many = len(urls) != 1 or per_url_configs
func = getattr(crawler, "arun_many" if use_many else "arun")
partial_func = partial(func,
urls[0] if len(urls) == 1 else urls,
urls if use_many else urls[0],
config=effective_config,
dispatcher=dispatcher)
# Optional per-crawl wall-clock deadline (config limits.wall_clock_s; 0 = none).
Expand All @@ -763,6 +886,9 @@ async def handle_crawl_request(
if not isinstance(results, list):
results = [results]

# Report one failed result for each seed the destination check refused.
results = _append_refused_results(results, seeds)

end_mem_mb = _get_memory_mb() # <--- Get memory after
end_time = time.time()

Expand Down Expand Up @@ -900,7 +1026,10 @@ async def handle_stream_crawl_request(
# SSRF guard: validate every seed URL's destination before fetching,
# mirroring handle_crawl_request. The streaming path previously skipped
# this, leaving /crawl/stream (and /crawl with stream=true) unguarded.
urls = _normalize_and_validate_seeds(urls)
seeds = _normalize_and_validate_seeds(urls)
_require_crawlable_seeds(seeds)
seed_count = len(urls)
urls = seeds.crawlable
browser_config = BrowserConfig.load(
browser_config, provenance=Provenance.UNTRUSTED
)
Expand All @@ -917,13 +1046,16 @@ async def handle_stream_crawl_request(
clamp_deep_crawl(crawler_config)
crawler_config.stream = True

# Deep crawl streaming supports exactly one start URL
if crawler_config.deep_crawl_strategy is not None and len(urls) != 1:
# Deep crawl streaming supports exactly one start URL. Keyed off the
# caller's own count: a request for several seeds is refused for that
# reason whether or not one of them was also refused by egress, and the
# deep-crawl limit is not this handler's business to relax.
if crawler_config.deep_crawl_strategy is not None and seed_count != 1:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=(
"Deep crawling with stream currently supports exactly one URL per request. "
f"Received {len(urls)} URLs."
f"Received {seed_count} URLs."
),
)

Expand All @@ -948,7 +1080,9 @@ async def handle_stream_crawl_request(
hooks_info = {'status': hooks_status}

# Deep crawl with single URL: use arun() which returns an async generator
# mirroring the Python library's streaming behavior
# mirroring the Python library's streaming behavior.
# The guard above rejects any request that did not send exactly one seed,
# so reaching here means seed_count == 1 and therefore len(urls) == 1.
if crawler_config.deep_crawl_strategy is not None and len(urls) == 1:
results_gen = await crawler.arun(
urls[0],
Expand All @@ -968,6 +1102,13 @@ async def handle_stream_crawl_request(
dispatcher=dispatcher
)

# Yield the refused seeds first, so a streamed batch reports a refusal
# exactly like the non-streaming one does instead of dropping it. Their
# position in the stream is not the caller's `urls` position, because a
# stream's order is completion order, not request order.
if seeds.refused:
results_gen = _prepend_refused_results(results_gen, seeds)

return crawler, results_gen, hooks_info

except (UntrustedConfigError, HookValidationError) as e:
Expand Down
Loading