diff --git a/crawl4ai/utils.py b/crawl4ai/utils.py index 279c27708..f4cb7a62c 100644 --- a/crawl4ai/utils.py +++ b/crawl4ai/utils.py @@ -249,6 +249,22 @@ def needs_update(self): return installed is None or installed < current +def _preserve_bare_query(rules_text: str) -> str: + """Append '*' to Allow/Disallow values ending in a bare '?' (e.g. '/*?'). + + '/*?' and '/*?*' allow/deny exactly the same URLs; the rewrite only + survives parsers that would otherwise drop the trailing '?'. + """ + fixed = [] + for raw_line in rules_text.splitlines(): + body, _, _ = raw_line.partition("#") + key, sep, value = body.partition(":") + if sep and key.strip().lower() in ("allow", "disallow") and value.strip().endswith("?"): + raw_line = f"{key.strip()}: {value.strip()}*" + fixed.append(raw_line) + return "\n".join(fixed) + + class RobotsParser: # Default 7 days cache TTL CACHE_TTL = 7 * 24 * 60 * 60 @@ -355,8 +371,10 @@ async def can_fetch(self, url: str, user_agent: str = "*") -> bool: return True # Create parser for this check - parser = RobotFileParser() - parser.parse(rules.splitlines()) + parser = RobotFileParser() + # Old Pythons drop a trailing '?' from rules, so '/*?' becomes + # '/*' and blocks the whole site. '/*?*' matches the same URLs. + parser.parse(_preserve_bare_query(rules).splitlines()) # If parser can't read rules, allow access if not parser.mtime(): diff --git a/tests/general/test_robot_parser.py b/tests/general/test_robot_parser.py index a2fc30f1a..6ed20fbc9 100644 --- a/tests/general/test_robot_parser.py +++ b/tests/general/test_robot_parser.py @@ -123,6 +123,44 @@ async def giant_robots(request): finally: await runner.cleanup() + # 4b. Test query-string disallow (Disallow: /*?) on a separate host port. + # Plain URLs must stay crawlable while query URLs are denied (RFC 9309). + async def start_query_server(): + query_app = web.Application() + + async def query_robots(request): + return web.Response(text="User-agent: *\nDisallow: /*?\n") + + query_app.router.add_get('/robots.txt', query_robots) + query_runner = web.AppRunner(query_app) + await query_runner.setup() + query_site = web.TCPSite(query_runner, 'localhost', 8081) + await query_site.start() + return query_runner + + query_runner = await start_query_server() + try: + print("\n4b. Testing query-string robots.txt rules...") + query_base = "http://localhost:8081" + + result = await parser.can_fetch(f"{query_base}/", "bot") + print(f"Plain root (/): {'allowed' if result else 'denied'}") + assert result, "Plain root should be allowed with Disallow: /*?" + + result = await parser.can_fetch(f"{query_base}/article", "bot") + print(f"Plain page (/article): {'allowed' if result else 'denied'}") + assert result, "Plain page should be allowed with Disallow: /*?" + + result = await parser.can_fetch(f"{query_base}/?page=2", "bot") + print(f"Query root (/?page=2): {'allowed' if result else 'denied'}") + assert not result, "Query root should be denied with Disallow: /*?" + + result = await parser.can_fetch(f"{query_base}/article?ref=x", "bot") + print(f"Query page (/article?ref=x): {'allowed' if result else 'denied'}") + assert not result, "Query page should be denied with Disallow: /*?" + finally: + await query_runner.cleanup() + # 5. Cache manipulation print("\n5. Testing cache manipulation...") diff --git a/tests/unit/test_robots_query_rules.py b/tests/unit/test_robots_query_rules.py new file mode 100644 index 000000000..90b1c61bf --- /dev/null +++ b/tests/unit/test_robots_query_rules.py @@ -0,0 +1,92 @@ +"""Unit tests for robots.txt rules ending in a bare '?' (e.g. 'Disallow: /*?'). + +urllib drops the trailing '?', which the wildcard patch in crawl4ai.utils turns +into the regex '^/.*' - disallowing the whole site. _preserve_bare_query rewrites +the rule to the equivalent '/*?*', which survives the round trip. +""" + +import asyncio + +import pytest + +from crawl4ai.utils import RobotsParser, _preserve_bare_query + +QUERY_RULES = "User-agent: *\nDisallow: /*?\n" + + +@pytest.mark.parametrize( + "line, expected", + [ + # A bare trailing '?' gains an explicit '*' + ("Disallow: /*?", "Disallow: /*?*"), + ("Allow: /*?", "Allow: /*?*"), + ("Disallow: /search?", "Disallow: /search?*"), + # Case and spacing are normalised, not required + ("disallow: /*?", "disallow: /*?*"), + ("DISALLOW:/*?", "DISALLOW: /*?*"), + ("Disallow: /*? ", "Disallow: /*?*"), + # Already explicit, or no trailing '?': left alone + ("Disallow: /*?*", "Disallow: /*?*"), + ("Disallow: /private/", "Disallow: /private/"), + ("Allow: /public/", "Allow: /public/"), + # Non-rule directives are never rewritten, even ending in '?' + ("User-agent: *", "User-agent: *"), + ("Sitemap: https://example.com/sitemap.xml?", "Sitemap: https://example.com/sitemap.xml?"), + ("", ""), + ("# just a comment", "# just a comment"), + ], +) +def test_preserve_bare_query_line_rewriting(line, expected): + assert _preserve_bare_query(line) == expected + + +def test_preserve_bare_query_keeps_document_structure(): + """Untouched lines, blank lines and ordering survive verbatim. + + The rewrite is splitlines()-based, so a trailing newline is not preserved. + That is harmless: the only caller re-splits the result immediately. + """ + source = "User-agent: *\nDisallow: /private/\n\nDisallow: /*?\nAllow: /public/\n" + assert _preserve_bare_query(source) == ( + "User-agent: *\nDisallow: /private/\n\nDisallow: /*?*\nAllow: /public/" + ) + # What the caller actually consumes is unaffected by the missing newline. + assert _preserve_bare_query(source).splitlines() == [ + "User-agent: *", "Disallow: /private/", "", "Disallow: /*?*", "Allow: /public/", + ] + + +def test_preserve_bare_query_is_idempotent(): + once = _preserve_bare_query(QUERY_RULES) + assert _preserve_bare_query(once) == once + + +def _can_fetch(rules, path, tmp_path): + """Answer can_fetch for a host whose rules are pre-seeded in the cache. + + Nothing listens on the host, and can_fetch falls back to 'allowed' whenever a + fetch fails, so any denial below can only have come from the cached rules. + """ + host = "localhost:8098" + parser = RobotsParser(cache_dir=str(tmp_path)) + parser._cache_rules(host, rules) + assert parser._get_cached_rules(host)[1], "seeded rules should be fresh" + return asyncio.run(parser.can_fetch(f"http://{host}{path}", "bot")) + + +@pytest.mark.parametrize("path", ["/", "/article", "/a/b/c"]) +def test_query_disallow_keeps_plain_urls_crawlable(path, tmp_path): + """'Disallow: /*?' must not take the whole site down.""" + assert _can_fetch(QUERY_RULES, path, tmp_path) is True + + +@pytest.mark.parametrize("path", ["/?page=2", "/article?ref=x", "/a/b?x=1&y=2"]) +def test_query_disallow_denies_query_urls(path, tmp_path): + assert _can_fetch(QUERY_RULES, path, tmp_path) is False + + +def test_ordinary_rules_still_apply(tmp_path): + """The rewrite must not disturb rules that never had a trailing '?'.""" + rules = "User-agent: *\nDisallow: /private/\nAllow: /public/\n" + assert _can_fetch(rules, "/public/page", tmp_path) is True + assert _can_fetch(rules, "/private/secret", tmp_path) is False