Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
22 changes: 20 additions & 2 deletions crawl4ai/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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():
Expand Down
38 changes: 38 additions & 0 deletions tests/general/test_robot_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -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...")

Expand Down
92 changes: 92 additions & 0 deletions tests/unit/test_robots_query_rules.py
Original file line number Diff line number Diff line change
@@ -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