Fix critical RCE via deserialization and eval() in /crawl endpoint
- Replace raw eval() in _compute_field() with AST-validated _safe_eval_expression() that blocks __import__, dunder attribute access, and import statements while preserving safe transforms - Add ALLOWED_DESERIALIZE_TYPES allowlist to from_serializable_dict() preventing arbitrary class instantiation from API input - Update security contact email and add v0.8.1 security fixes to SECURITY.md with researcher acknowledgment - Add 17 security tests covering both fixes
This commit is contained in:
unclecode
@@ -160,6 +160,170 @@ class TestHooksEnabled(unittest.TestCase):
|
||||
os.environ.pop("CRAWL4AI_HOOKS_ENABLED", None)
|
||||
|
||||
|
||||
class TestComputedFieldSafety(unittest.TestCase):
|
||||
"""Test that computed field expressions block dangerous operations.
|
||||
|
||||
Mirrors the AST-based _safe_eval_expression() logic from extraction_strategy.py
|
||||
to test without importing heavy crawl4ai dependencies.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up the safe eval function (local copy of the logic)."""
|
||||
import ast
|
||||
|
||||
SAFE_BUILTINS = {
|
||||
"str": str, "int": int, "float": float, "bool": bool,
|
||||
"len": len, "round": round, "abs": abs, "min": min, "max": max,
|
||||
"sum": sum, "sorted": sorted, "reversed": reversed,
|
||||
"list": list, "dict": dict, "tuple": tuple, "set": set,
|
||||
"enumerate": enumerate, "zip": zip, "map": map, "filter": filter,
|
||||
"any": any, "all": all, "range": range,
|
||||
"True": True, "False": False, "None": None,
|
||||
"isinstance": isinstance, "type": type,
|
||||
}
|
||||
|
||||
def safe_eval(expression, local_vars):
|
||||
tree = ast.parse(expression, mode="eval")
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, (ast.Import, ast.ImportFrom)):
|
||||
raise ValueError("Import statements are not allowed")
|
||||
if isinstance(node, ast.Attribute) and node.attr.startswith("_"):
|
||||
raise ValueError(f"Access to '{node.attr}' is not allowed")
|
||||
if isinstance(node, ast.Call):
|
||||
func = node.func
|
||||
if isinstance(func, ast.Name) and func.id.startswith("_"):
|
||||
raise ValueError(f"Calling '{func.id}' is not allowed")
|
||||
if isinstance(func, ast.Attribute) and func.attr.startswith("_"):
|
||||
raise ValueError(f"Calling '{func.attr}' is not allowed")
|
||||
safe_globals = {"__builtins__": SAFE_BUILTINS}
|
||||
return eval(compile(tree, "<expression>", "eval"), safe_globals, local_vars)
|
||||
|
||||
self.safe_eval = safe_eval
|
||||
|
||||
# === SECURITY TESTS: These expressions must be BLOCKED ===
|
||||
|
||||
def test_import_blocked(self):
|
||||
"""__import__('os') must be blocked."""
|
||||
with self.assertRaises(ValueError):
|
||||
self.safe_eval("__import__('os').system('id')", {})
|
||||
|
||||
def test_dunder_attribute_blocked(self):
|
||||
"""Access to __class__, __globals__, etc. must be blocked."""
|
||||
with self.assertRaises(ValueError):
|
||||
self.safe_eval("''.__class__.__bases__", {})
|
||||
|
||||
def test_dunder_method_call_blocked(self):
|
||||
"""Calls to dunder methods must be blocked."""
|
||||
with self.assertRaises(ValueError):
|
||||
self.safe_eval("x.__getattribute__('y')", {"x": {}})
|
||||
|
||||
def test_os_popen_via_import_blocked(self):
|
||||
"""The exact POC from the vulnerability report must be blocked."""
|
||||
with self.assertRaises(ValueError):
|
||||
self.safe_eval('__import__("os").popen("id").read()', {})
|
||||
|
||||
# === FUNCTIONALITY TESTS: These expressions must WORK ===
|
||||
|
||||
def test_simple_math(self):
|
||||
"""Basic arithmetic on item values must work."""
|
||||
result = self.safe_eval("price * 1.1", {"price": 100})
|
||||
self.assertAlmostEqual(result, 110.0)
|
||||
|
||||
def test_string_method(self):
|
||||
"""String methods on item values must work."""
|
||||
result = self.safe_eval("name.upper()", {"name": "hello"})
|
||||
self.assertEqual(result, "HELLO")
|
||||
|
||||
def test_string_concatenation(self):
|
||||
"""String concatenation must work."""
|
||||
result = self.safe_eval("first + ' ' + last", {"first": "John", "last": "Doe"})
|
||||
self.assertEqual(result, "John Doe")
|
||||
|
||||
def test_dict_access(self):
|
||||
"""Dict-style field access must work."""
|
||||
result = self.safe_eval("a + b", {"a": 10, "b": 20})
|
||||
self.assertEqual(result, 30)
|
||||
|
||||
def test_builtin_functions(self):
|
||||
"""Safe builtins like len, str, int must work."""
|
||||
result = self.safe_eval("len(name)", {"name": "hello"})
|
||||
self.assertEqual(result, 5)
|
||||
|
||||
def test_round_function(self):
|
||||
"""round() must work for numeric formatting."""
|
||||
result = self.safe_eval("round(price, 2)", {"price": 10.456})
|
||||
self.assertEqual(result, 10.46)
|
||||
|
||||
|
||||
class TestDeserializationAllowlist(unittest.TestCase):
|
||||
"""Test that the deserialization allowlist blocks non-allowlisted types.
|
||||
|
||||
Tests the allowlist constant directly without importing heavy dependencies.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up the allowlist (local copy of the constant)."""
|
||||
self.allowed_types = {
|
||||
"BrowserConfig", "CrawlerRunConfig", "HTTPCrawlerConfig",
|
||||
"LLMConfig", "ProxyConfig", "GeolocationConfig",
|
||||
"SeedingConfig", "VirtualScrollConfig", "LinkPreviewConfig",
|
||||
"JsonCssExtractionStrategy", "JsonXPathExtractionStrategy",
|
||||
"JsonLxmlExtractionStrategy", "LLMExtractionStrategy",
|
||||
"CosineStrategy", "RegexExtractionStrategy",
|
||||
"DefaultMarkdownGenerator",
|
||||
"PruningContentFilter", "BM25ContentFilter", "LLMContentFilter",
|
||||
"LXMLWebScrapingStrategy",
|
||||
"RegexChunking",
|
||||
"BFSDeepCrawlStrategy", "DFSDeepCrawlStrategy", "BestFirstCrawlingStrategy",
|
||||
"FilterChain", "URLPatternFilter", "DomainFilter",
|
||||
"ContentTypeFilter", "URLFilter", "SEOFilter", "ContentRelevanceFilter",
|
||||
"KeywordRelevanceScorer", "URLScorer", "CompositeScorer",
|
||||
"DomainAuthorityScorer", "FreshnessScorer", "PathDepthScorer",
|
||||
"CacheMode", "MatchMode", "DisplayMode",
|
||||
"MemoryAdaptiveDispatcher", "SemaphoreDispatcher",
|
||||
"DefaultTableExtraction", "NoTableExtraction",
|
||||
"RoundRobinProxyStrategy",
|
||||
}
|
||||
|
||||
# === SECURITY TESTS: Non-allowlisted types must be BLOCKED ===
|
||||
|
||||
def test_arbitrary_class_not_in_allowlist(self):
|
||||
"""AsyncWebCrawler must NOT be in the allowlist."""
|
||||
self.assertNotIn("AsyncWebCrawler", self.allowed_types)
|
||||
|
||||
def test_crawler_hub_not_in_allowlist(self):
|
||||
"""CrawlerHub must NOT be in the allowlist."""
|
||||
self.assertNotIn("CrawlerHub", self.allowed_types)
|
||||
|
||||
def test_browser_profiler_not_in_allowlist(self):
|
||||
"""BrowserProfiler must NOT be in the allowlist."""
|
||||
self.assertNotIn("BrowserProfiler", self.allowed_types)
|
||||
|
||||
def test_docker_client_not_in_allowlist(self):
|
||||
"""Crawl4aiDockerClient must NOT be in the allowlist."""
|
||||
self.assertNotIn("Crawl4aiDockerClient", self.allowed_types)
|
||||
|
||||
# === FUNCTIONALITY TESTS: Allowlisted types must be present ===
|
||||
|
||||
def test_allowlist_has_core_config_types(self):
|
||||
"""Core config types must be in the allowlist."""
|
||||
required = {"BrowserConfig", "CrawlerRunConfig", "LLMConfig", "ProxyConfig"}
|
||||
self.assertTrue(required.issubset(self.allowed_types))
|
||||
|
||||
def test_allowlist_has_extraction_strategies(self):
|
||||
"""Extraction strategy types must be in the allowlist."""
|
||||
required = {
|
||||
"JsonCssExtractionStrategy", "LLMExtractionStrategy",
|
||||
"RegexExtractionStrategy",
|
||||
}
|
||||
self.assertTrue(required.issubset(self.allowed_types))
|
||||
|
||||
def test_allowlist_has_enums(self):
|
||||
"""Enum types must be in the allowlist."""
|
||||
required = {"CacheMode", "MatchMode", "DisplayMode"}
|
||||
self.assertTrue(required.issubset(self.allowed_types))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
print("=" * 60)
|
||||
print("Crawl4AI Security Fixes - Unit Tests")
|
||||
|
||||
Reference in New Issue
Block a user