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
59 changes: 46 additions & 13 deletions crawl4ai/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,21 +55,43 @@
import re

original_applies_to = RuleLine.applies_to
original_ruleline_init = RuleLine.__init__


def _has_query_marker(value: str) -> bool:
"""True if a quoted or raw robots path/URL still carries a '?' query marker."""
if not value:
return False
return "?" in value or "%3F" in value.upper()


def patched_ruleline_init(self, path, allowance):
# urlparse/urlunparse drops an empty query, so '/*?' collapses to '/*'.
# Preserve a trailing '?' so query-only rules keep their meaning (#2225).
keep_query_marker = isinstance(path, str) and path.endswith("?")
original_ruleline_init(self, path, allowance)
if keep_query_marker and not _has_query_marker(self.path):
self.path += "%3F"


def patched_applies_to(self, filename):
# Handle wildcards in paths
if '*' in self.path or '%2A' in self.path or self.path in ("*", "%2A"):
pattern = self.path.replace('%2A', '*')
pattern = re.escape(pattern).replace('\\*', '.*')
pattern = '^' + pattern
if pattern.endswith('\\$'):
pattern = pattern[:-2] + '$'
try:
return bool(re.match(pattern, filename))
except re.error:
return original_applies_to(self, filename)
return original_applies_to(self, filename)
# Disallow: /*? (and similar) must only match URLs that actually carry a query.
if _has_query_marker(self.path) and not _has_query_marker(filename):
return False
# Handle wildcards in paths
if '*' in self.path or '%2A' in self.path or self.path in ("*", "%2A"):
pattern = self.path.replace('%2A', '*')
pattern = re.escape(pattern).replace('\\*', '.*')
pattern = '^' + pattern
if pattern.endswith('\\$'):
pattern = pattern[:-2] + '$'
try:
return bool(re.match(pattern, filename))
except re.error:
return original_applies_to(self, filename)
return original_applies_to(self, filename)

RuleLine.__init__ = patched_ruleline_init
RuleLine.applies_to = patched_applies_to
# Monkey patch ends

Expand Down Expand Up @@ -361,7 +383,18 @@ async def can_fetch(self, url: str, user_agent: str = "*") -> bool:
# If parser can't read rules, allow access
if not parser.mtime():
return True


# urllib.robotparser drops an empty query component, so a URL like
# https://host/page? is normalized to /page. Keep a query marker so
# rules such as Disallow: /*? still see that the URL carries '?'.
# Only look at the pre-fragment part so "? in fragment" is not treated
# as a query, and rebuild via urlunparse so page?#frag becomes page?=#frag
# instead of corrupting the fragment. Bare RobotFileParser.can_fetch
# still allows page? — this rewrite is RobotsParser-only.
before_hash = url.split("#", 1)[0]
if "?" in before_hash and not parsed.query:
url = urlunparse(parsed._replace(query="="))

return parser.can_fetch(user_agent, url)

def clear_cache(self):
Expand Down
45 changes: 44 additions & 1 deletion tests/general/test_robot_parser.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,19 @@
from crawl4ai.utils import RobotsParser

from urllib.robotparser import RobotFileParser

import asyncio
import aiohttp
from aiohttp import web
import tempfile
import shutil
import os, sys, time, json
import pytest


# Ecommerce-style rule: block URLs that carry a query string, not the whole site.
DISALLOW_QUERY_WILDCARD_ROBOTS = """User-agent: *
Disallow: /*?
"""


async def test_robots_parser():
Expand Down Expand Up @@ -148,8 +156,43 @@ async def giant_robots(request):
shutil.rmtree(temp_dir)
print("\nTest cleanup completed")

def test_robotfileparser_disallow_query_wildcard():
"""RuleLine patch: Disallow: /*? only matches URLs that carry a query string."""
parser = RobotFileParser()
parser.parse(DISALLOW_QUERY_WILDCARD_ROBOTS.splitlines())
assert parser.can_fetch("*", "https://shop.example/page") is True
assert parser.can_fetch("*", "https://shop.example/") is True
assert parser.can_fetch("*", "https://shop.example/products/shoes") is True
assert parser.can_fetch("*", "https://shop.example/page?q=1") is False
assert parser.can_fetch("*", "https://shop.example/?s=search") is False


@pytest.mark.asyncio
async def test_disallow_query_wildcard_only_blocks_query_urls():
"""RobotsParser.can_fetch: Disallow: /*? must not collapse to Disallow: /* (#2225)."""
temp_dir = tempfile.mkdtemp()
try:
parser = RobotsParser(cache_dir=temp_dir)
parser._cache_rules("shop.example", DISALLOW_QUERY_WILDCARD_ROBOTS)

assert await parser.can_fetch("https://shop.example/page", "*") is True
assert await parser.can_fetch("https://shop.example/", "*") is True
assert await parser.can_fetch("https://shop.example/products/shoes", "*") is True
assert await parser.can_fetch("https://shop.example/page?q=1", "*") is False
assert await parser.can_fetch("https://shop.example/?s=search", "*") is False
# Empty-but-present query still counts as carrying a query string
assert await parser.can_fetch("https://shop.example/page?", "*") is False
# page?#frag has a query marker before the fragment and must stay blocked
assert await parser.can_fetch("https://shop.example/page?#frag", "*") is False
print("✓ Disallow: /*? blocks query URLs only (issue #2225)")
finally:
shutil.rmtree(temp_dir)


async def main():
try:
test_robotfileparser_disallow_query_wildcard()
await test_disallow_query_wildcard_only_blocks_query_urls()
await test_robots_parser()
except Exception as e:
print(f"Test failed: {str(e)}")
Expand Down