-
+
-
-
๐จ Client Simulator Controls
-
Start traffic simulation and attack testing
-
-
Start Client Simulator
-
View Client Logs
+
+
+ ๐ Overview
+ โ๏ธ Services
+ ๐ฏ Actions
+ ๐ณ๏ธ Sinkhole
+
+
+
+
+
+
๐ System Overview
+
+
+
+
+
+
+
0ms
+
Avg Response Time
+
+
+
0
+
Attack Simulators
+
+
+
+
-
-
-
Service Logs
-
-
Close Logs
+
+
+
+
+
โ๏ธ Service Management
+
+
+
+
+
+
+
+
+
+
๐ฏ Attack Simulator Actions
+
Real-time monitoring of requests from attack simulators and clients
+
+
+
+
+
+ Timestamp
+ Source
+ Method
+ Target
+ Status
+ Response Time
+ Attack Type
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
๐ณ๏ธ Sinkhole & Blackhole Management
+
Manage malicious actor isolation and traffic redirection
+
+
+
+
+
Quarantined IPs
+
-
+
Auto-quarantine active
+
+
+
Sinkholed IPs
+
-
+
Traffic redirected
+
+
+
Blackholed IPs
+
-
+
Completely blocked
+
+
+
Violation Score
+
-
+
System average
+
+
+
+
+
+
+
+
+
๐ฏ Active Threat Management
+
+
+
+ IP/Subnet
+ Type
+ Status
+ Violations
+ Last Activity
+ Reason
+ Actions
+
+
+
+
+
+
+
+
+
+
+
-
+
๐
\ No newline at end of file
diff --git a/templates/load_balancer.html b/templates/load_balancer.html
index c0a35d5..55386c3 100644
--- a/templates/load_balancer.html
+++ b/templates/load_balancer.html
@@ -9,245 +9,148 @@
rel="stylesheet"
/>
@@ -266,22 +169,63 @@ Load Balancer Control Panel
-
- Accepted IP Addresses
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
Primary CDN
+
Status: Online
+
Weight: 3
+
Requests: 1
+
+
+
+
+
+
Secondary CDN
+
Status: Online
+
Weight: 2
+
Requests: 4
+
+
+
+
+
+
Tertiary CDN
+
Status: Online
+
Weight: 1
+
Requests: 3
+
+
+
+
+
+
Actions
+
+ Test Primary CDN
+ Test Secondary CDN
+ Test Tertiary CDN
+
+
+
@@ -321,9 +265,6 @@
Migrate CDN
-
- ยฉ 2025 CyberEdge Networks ๐
-
\ No newline at end of file
diff --git a/templates/load_balancer_enhanced.html b/templates/load_balancer_enhanced.html
new file mode 100644
index 0000000..106e1c8
--- /dev/null
+++ b/templates/load_balancer_enhanced.html
@@ -0,0 +1,468 @@
+
+
+
+
+
+
Aurora Shield Load Balancer - Round Robin Dashboard
+
+
+
+
+
+
+
+
+
๐ Load Balancer Statistics
+
+
+
Round-Robin Algorithm
+
+ Next CDN Index: 0 Last Request: Never
+
+
+
+
+
+
+
๐ CDN Health Status
+
+
+
+
+
+
+
+
๐ Request Distribution
+
+
+
+
+
+
+
+
๐ System Logs
+
+ Loading system logs...
+
+
+
+
+
+ ๐ Refresh Data
+ ๐งช Test Load Balancer
+ โฑ๏ธ Toggle Auto-refresh
+
+
+
+ Last updated: Never
+
+
+
+
+
\ No newline at end of file
diff --git a/tests/test_complete_filters.py b/tests/test_complete_filters.py
new file mode 100644
index 0000000..bca0738
--- /dev/null
+++ b/tests/test_complete_filters.py
@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+
+"""
+Comprehensive test for Aurora Shield filter functionality
+Tests the sinkholed and blackholed filter options
+"""
+
+import json
+import re
+
+def test_complete_filter_functionality():
+ """Test all aspects of the filter functionality"""
+
+ print("๐ Aurora Shield Filter Test Suite")
+ print("=" * 60)
+
+ # Test 1: HTML Filter Options
+ print("\n๐ Test 1: HTML Filter Options")
+ print("-" * 30)
+
+ with open('aurora_shield/dashboard/templates/aurora_dashboard.html', 'r', encoding='utf-8') as f:
+ html_content = f.read()
+
+ # Expected filter options
+ expected_options = [
+ ('all', 'All Actions'),
+ ('blocked', 'Blocked'),
+ ('sinkholed', 'Sinkholed'),
+ ('blackholed', 'Blackholed'),
+ ('rate-limited', 'Rate Limited'),
+ ('quarantined', 'Quarantined'),
+ ('challenged', 'Challenged'),
+ ('monitored', 'Monitored')
+ ]
+
+ # Find filter dropdown
+ filter_pattern = r'
]*id="action-filter"[^>]*>(.*?)'
+ match = re.search(filter_pattern, html_content, re.DOTALL)
+
+ if match:
+ filter_content = match.group(1)
+ print("โ
Found action filter dropdown")
+
+ for value, label in expected_options:
+ if f'value="{value}"' in filter_content and label in filter_content:
+ print(f"โ
{label} option present")
+ else:
+ print(f"โ {label} option missing")
+ else:
+ print("โ Action filter dropdown not found")
+
+ # Test 2: CSS Styles
+ print("\n๐จ Test 2: CSS Action Styles")
+ print("-" * 30)
+
+ expected_css_classes = [
+ 'action-blocked',
+ 'action-sinkholed',
+ 'action-blackholed',
+ 'action-quarantined',
+ 'action-rate-limited',
+ 'action-challenged',
+ 'action-monitored'
+ ]
+
+ for css_class in expected_css_classes:
+ if f'.{css_class} {{' in html_content:
+ print(f"โ
{css_class} style defined")
+ else:
+ print(f"โ {css_class} style missing")
+
+ # Test 3: JavaScript Functions
+ print("\n๐ง Test 3: JavaScript Functions")
+ print("-" * 30)
+
+ js_checks = [
+ ('onActionFilterChange', 'function onActionFilterChange(select)'),
+ ('Filter Update Logic', 'currentAttackFilters.action = select.value'),
+ ('Update Function Call', 'updateEnhancedAttackActivity()'),
+ ('Filter API Call', '/api/dashboard/attack-activity')
+ ]
+
+ for check_name, pattern in js_checks:
+ if pattern in html_content:
+ print(f"โ
{check_name} found")
+ else:
+ print(f"โ {check_name} missing")
+
+ # Test 4: Backend API Support
+ print("\n๐ Test 4: Backend API Support")
+ print("-" * 30)
+
+ try:
+ with open('aurora_shield/dashboard/web_dashboard.py', 'r', encoding='utf-8') as f:
+ backend_content = f.read()
+
+ backend_checks = [
+ ('Attack Activity Endpoint', '/api/dashboard/attack-activity'),
+ ('Action Filter Parameter', 'action_filter = request.args.get'),
+ ('Filter Logic', "action_taken.*lower.*replace.*==.*action_filter"),
+ ('Shield Manager Fallback', 'shield_manager.recent_requests'),
+ ('Status Mapping Functions', '_map_status_to_action'),
+ ('All Action Types', 'sinkholed.*blackholed.*quarantined')
+ ]
+
+ for check_name, pattern in backend_checks:
+ if re.search(pattern, backend_content):
+ print(f"โ
{check_name} implemented")
+ else:
+ print(f"โ {check_name} missing")
+
+ # Check helper functions for new action types
+ helper_function_checks = [
+ ('Sinkholed Mapping', "'sinkholed': 'Sinkholed'"),
+ ('Blackholed Mapping', "'blackholed': 'Blackholed'"),
+ ('Quarantined Mapping', "'quarantined': 'Quarantined'"),
+ ('Critical Severity', "'blackholed': 'critical'"),
+ ('High Severity Sinkhole', "'sinkholed': 'high'")
+ ]
+
+ for check_name, pattern in helper_function_checks:
+ if pattern in backend_content:
+ print(f"โ
{check_name} configured")
+ else:
+ print(f"โ {check_name} missing")
+
+ except FileNotFoundError:
+ print("โ Backend file not found")
+
+ # Test 5: Shield Manager Action Types
+ print("\n๐ก๏ธ Test 5: Shield Manager Action Types")
+ print("-" * 30)
+
+ try:
+ with open('aurora_shield/shield_manager.py', 'r', encoding='utf-8') as f:
+ shield_content = f.read()
+
+ shield_checks = [
+ ('Sinkholed Logging', "_log_request_realtime.*'sinkholed'"),
+ ('Blackholed Logging', "_log_request_realtime.*'blackholed'"),
+ ('Quarantined Logging', "_log_request_realtime.*'quarantined'"),
+ ('Rate Limited Logging', "_log_request_realtime.*'rate-limited'"),
+ ('Blocked Logging', "_log_request_realtime.*'blocked'")
+ ]
+
+ for check_name, pattern in shield_checks:
+ if re.search(pattern, shield_content):
+ print(f"โ
{check_name} implemented")
+ else:
+ print(f"โ {check_name} missing")
+
+ except FileNotFoundError:
+ print("โ Shield manager file not found")
+
+ # Test 6: Filter Integration Test
+ print("\n๐ Test 6: Filter Integration")
+ print("-" * 30)
+
+ # Test filter parameter processing
+ test_cases = [
+ ('blocked', 'blocked'),
+ ('sinkholed', 'sinkholed'),
+ ('blackholed', 'blackholed'),
+ ('rate-limited', 'rate-limited'),
+ ('quarantined', 'quarantined')
+ ]
+
+ for filter_value, expected_match in test_cases:
+ # Simulate the backend filter logic
+ action_taken = expected_match.title()
+ converted = action_taken.lower().replace(' ', '-')
+
+ if converted == filter_value:
+ print(f"โ
Filter '{filter_value}' matches action '{action_taken}'")
+ else:
+ print(f"โ Filter '{filter_value}' doesn't match action '{action_taken}' (got '{converted}')")
+
+ print("\n๐ Filter Test Suite Complete!")
+ print("=" * 60)
+
+if __name__ == "__main__":
+ test_complete_filter_functionality()
\ No newline at end of file
diff --git a/tests/test_config_gui.py b/tests/test_config_gui.py
new file mode 100644
index 0000000..22fcb35
--- /dev/null
+++ b/tests/test_config_gui.py
@@ -0,0 +1,214 @@
+#!/usr/bin/env python3
+"""
+Test script for Aurora Shield Configuration GUI functionality.
+"""
+
+import requests
+import json
+import time
+
+class ConfigGUITester:
+ def __init__(self):
+ self.base_url = "http://localhost:8080"
+ self.session = requests.Session()
+
+ def authenticate(self):
+ """Authenticate with the dashboard."""
+ login_data = {'username': 'admin', 'password': 'admin123'}
+ response = self.session.post(f"{self.base_url}/login", data=login_data)
+ return response.status_code == 200
+
+ def test_get_config(self):
+ """Test getting current configuration."""
+ print("๐ง Testing GET /api/dashboard/config...")
+
+ response = self.session.get(f"{self.base_url}/api/dashboard/config")
+ if response.status_code == 200:
+ config = response.json()
+ print("โ
Configuration retrieved successfully")
+ print(f" Version: {config.get('version', 'Unknown')}")
+ print(f" Rate Limiter Rate: {config.get('rate_limiter', {}).get('rate', 'Unknown')}")
+ print(f" Anomaly Threshold: {config.get('anomaly_detector', {}).get('rate_threshold', 'Unknown')}")
+ print(f" IP Initial Score: {config.get('ip_reputation', {}).get('initial_score', 'Unknown')}")
+ return config
+ else:
+ print(f"โ Failed to get config: {response.status_code} - {response.text}")
+ return None
+
+ def test_update_config(self):
+ """Test updating configuration."""
+ print("\n๐ง Testing POST /api/dashboard/config...")
+
+ # Test configuration with new values
+ test_config = {
+ 'rate_limiter': {
+ 'enabled': True,
+ 'rate': 15,
+ 'burst': 25,
+ 'window_size': 90
+ },
+ 'anomaly_detector': {
+ 'enabled': True,
+ 'request_window': 120,
+ 'rate_threshold': 150,
+ 'sensitivity': 'high'
+ },
+ 'ip_reputation': {
+ 'enabled': True,
+ 'initial_score': 90,
+ 'reputation_threshold': 60,
+ 'decay_rate': 0.15
+ },
+ 'thresholds': {
+ 'requests_per_second': 1500,
+ 'cpu_threshold': 85,
+ 'memory_threshold': 90
+ }
+ }
+
+ response = self.session.post(
+ f"{self.base_url}/api/dashboard/config",
+ json=test_config,
+ headers={'Content-Type': 'application/json'}
+ )
+
+ if response.status_code == 200:
+ result = response.json()
+ print("โ
Configuration updated successfully")
+ print(f" Message: {result.get('message', 'No message')}")
+ return True
+ else:
+ print(f"โ Failed to update config: {response.status_code} - {response.text}")
+ return False
+
+ def test_config_validation(self):
+ """Test configuration validation with invalid values."""
+ print("\n๐ง Testing configuration validation...")
+
+ # Test with invalid values
+ invalid_config = {
+ 'rate_limiter': {
+ 'rate': -5, # Invalid: negative value
+ 'burst': 50000 # Invalid: too high
+ },
+ 'anomaly_detector': {
+ 'sensitivity': 'invalid_value' # Invalid: not in choices
+ },
+ 'thresholds': {
+ 'cpu_threshold': 150 # Invalid: over 100%
+ }
+ }
+
+ response = self.session.post(
+ f"{self.base_url}/api/dashboard/config",
+ json=invalid_config,
+ headers={'Content-Type': 'application/json'}
+ )
+
+ if response.status_code == 400:
+ print("โ
Configuration validation working correctly")
+ print(f" Error: {response.json().get('error', 'No error message')}")
+ return True
+ else:
+ print(f"โ Validation should have failed but got: {response.status_code}")
+ return False
+
+ def test_config_persistence(self):
+ """Test that configuration changes persist."""
+ print("\n๐ง Testing configuration persistence...")
+
+ # Set a unique value
+ test_value = int(time.time()) % 1000 # Use timestamp for uniqueness
+ config_update = {
+ 'rate_limiter': {
+ 'rate': test_value
+ }
+ }
+
+ # Update config
+ update_response = self.session.post(
+ f"{self.base_url}/api/dashboard/config",
+ json=config_update,
+ headers={'Content-Type': 'application/json'}
+ )
+
+ if update_response.status_code != 200:
+ print(f"โ Failed to update config for persistence test")
+ return False
+
+ # Wait a moment
+ time.sleep(2)
+
+ # Retrieve config and check if value persisted
+ get_response = self.session.get(f"{self.base_url}/api/dashboard/config")
+ if get_response.status_code == 200:
+ config = get_response.json()
+ retrieved_value = config.get('rate_limiter', {}).get('rate')
+
+ if retrieved_value == test_value:
+ print("โ
Configuration persistence working correctly")
+ print(f" Set value: {test_value}, Retrieved value: {retrieved_value}")
+ return True
+ else:
+ print(f"โ Configuration not persistent. Set: {test_value}, Got: {retrieved_value}")
+ return False
+ else:
+ print(f"โ Failed to retrieve config for persistence test")
+ return False
+
+def main():
+ """Run comprehensive configuration GUI tests."""
+ print("๐ก๏ธ Aurora Shield Configuration GUI Test Suite")
+ print("=" * 60)
+
+ tester = ConfigGUITester()
+
+ # Authenticate
+ if not tester.authenticate():
+ print("โ Authentication failed")
+ return
+
+ print("โ
Authentication successful")
+
+ # Run tests
+ tests = [
+ ('Configuration Retrieval', tester.test_get_config),
+ ('Configuration Update', tester.test_update_config),
+ ('Configuration Validation', tester.test_config_validation),
+ ('Configuration Persistence', tester.test_config_persistence)
+ ]
+
+ results = []
+ for test_name, test_func in tests:
+ try:
+ result = test_func()
+ results.append((test_name, result))
+ except Exception as e:
+ print(f"โ {test_name} failed with exception: {e}")
+ results.append((test_name, False))
+
+ # Summary
+ print("\n" + "=" * 60)
+ print("๐ TEST SUMMARY")
+ print("=" * 60)
+
+ passed = 0
+ for test_name, result in results:
+ status = "โ
PASS" if result else "โ FAIL"
+ print(f"{status} {test_name}")
+ if result:
+ passed += 1
+
+ print(f"\n๐ Results: {passed}/{len(results)} tests passed")
+
+ if passed == len(results):
+ print("๐ All tests passed! Configuration GUI is working correctly.")
+ else:
+ print("โ ๏ธ Some tests failed. Check the issues above.")
+
+ print(f"\n๐ Access the Configuration GUI at: {tester.base_url}")
+ print("๐ Login: admin / admin123")
+ print("๐ Navigate to: Configuration tab")
+
+if __name__ == "__main__":
+ main()
\ No newline at end of file
diff --git a/test_dashboard.py b/tests/test_dashboard.py
similarity index 100%
rename from test_dashboard.py
rename to tests/test_dashboard.py
diff --git a/tests/test_direct_shield.py b/tests/test_direct_shield.py
new file mode 100644
index 0000000..f1d4906
--- /dev/null
+++ b/tests/test_direct_shield.py
@@ -0,0 +1,213 @@
+#!/usr/bin/env python3
+"""
+Test Aurora Shield's automated sinkhole via the shield API endpoint.
+"""
+
+import requests
+import time
+import random
+import json
+
+class DirectShieldTest:
+ def __init__(self):
+ self.shield_url = "http://localhost:8080"
+ self.session = requests.Session()
+
+ def authenticate(self):
+ """Authenticate with the dashboard."""
+ login_data = {'username': 'admin', 'password': 'admin123'}
+ response = self.session.post(f"{self.shield_url}/login", data=login_data)
+ return response.status_code == 200
+
+ def send_request_for_protection_check(self, source_ip, is_malicious=True, request_id=None):
+ """Send a request to Aurora Shield for protection check."""
+
+ # Prepare request data for shield processing
+ request_data = {
+ 'ip': source_ip,
+ 'method': 'GET',
+ 'url': '/test-endpoint',
+ 'headers': {
+ 'User-Agent': 'AttackBot/1.0' if is_malicious else 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
+ 'X-Forwarded-For': source_ip,
+ 'X-Real-IP': source_ip
+ },
+ 'timestamp': int(time.time())
+ }
+
+ if is_malicious:
+ # Add malicious patterns
+ attack_types = [
+ {'params': {'id': "1' OR '1'='1"}, 'type': 'sql_injection'},
+ {'params': {'search': ''}, 'type': 'xss'},
+ {'params': {'file': '../../../etc/passwd'}, 'type': 'path_traversal'},
+ {'params': {'cmd': 'rm -rf /'}, 'type': 'command_injection'}
+ ]
+
+ attack = random.choice(attack_types)
+ request_data['params'] = attack['params']
+ request_data['attack_type'] = attack['type']
+ else:
+ # Legitimate request
+ request_data['params'] = {'page': 'home', 'user_id': f'user_{random.randint(1000, 9999)}'}
+
+ try:
+ response = self.session.post(
+ f"{self.shield_url}/api/shield/check-request",
+ json=request_data,
+ timeout=5
+ )
+
+ return {
+ 'status_code': response.status_code,
+ 'response': response.json() if response.status_code == 200 else response.text,
+ 'source_ip': source_ip,
+ 'malicious': is_malicious
+ }
+ except Exception as e:
+ return {
+ 'status_code': 0,
+ 'response': str(e),
+ 'source_ip': source_ip,
+ 'malicious': is_malicious
+ }
+
+ def simulate_zero_reputation_attacks(self):
+ """Simulate multiple attacks from IPs to trigger zero reputation."""
+ print("๐ฏ [TEST 1] Simulating zero-reputation attacks...")
+
+ attack_ips = ['192.168.100.50', '10.0.0.100', '172.16.0.200']
+ results = []
+
+ for ip in attack_ips:
+ print(f"\n Attacking from {ip}...")
+
+ # Send multiple malicious requests to trigger reputation drop
+ for i in range(15):
+ result = self.send_request_for_protection_check(ip, is_malicious=True, request_id=i)
+ results.append(result)
+
+ if i % 5 == 0:
+ print(f" Request {i+1}: Status {result['status_code']}")
+ if result['status_code'] == 200:
+ response = result['response']
+ print(f" Action: {response.get('action', 'unknown')}")
+ if 'sinkhole' in str(response).lower():
+ print(f" ๐ณ๏ธ SINKHOLED!")
+
+ time.sleep(0.1)
+
+ return results
+
+ def simulate_legitimate_traffic(self):
+ """Simulate legitimate traffic."""
+ print("\nโ
[TEST 2] Simulating legitimate traffic...")
+
+ legit_ips = ['203.0.113.25', '198.51.100.50']
+ results = []
+
+ for ip in legit_ips:
+ print(f"\n Legitimate requests from {ip}...")
+
+ for i in range(8):
+ result = self.send_request_for_protection_check(ip, is_malicious=False, request_id=i)
+ results.append(result)
+
+ if i % 3 == 0:
+ print(f" Request {i+1}: Status {result['status_code']}")
+ if result['status_code'] == 200:
+ response = result['response']
+ print(f" Action: {response.get('action', 'unknown')}")
+
+ time.sleep(0.2)
+
+ return results
+
+ def check_sinkhole_and_attacks(self):
+ """Check current sinkhole status and attacking IPs."""
+ if not self.authenticate():
+ print("โ Failed to authenticate")
+ return
+
+ print("\n๐ Checking Aurora Shield Status...")
+
+ try:
+ # Check attacking IPs
+ attacking_response = self.session.get(f"{self.shield_url}/api/dashboard/attacking-ips")
+ if attacking_response.status_code == 200:
+ data = attacking_response.json()
+ attacking_data = data.get('data', {}) if isinstance(data, dict) else data
+
+ print(f"\n๐ Attacking IP Analysis:")
+ recent_attacks = attacking_data.get('recent_attacks', [])
+ print(f" Recent attacks: {len(recent_attacks)}")
+
+ sinkhole_summary = attacking_data.get('sinkhole_summary', {})
+ print(f" Sinkholed IPs: {sinkhole_summary.get('sinkholed_ips', 0)}")
+ print(f" Quarantined IPs: {sinkhole_summary.get('quarantined_ips', 0)}")
+
+ sinkholed_ips = attacking_data.get('sinkholed_ips', [])
+ if sinkholed_ips:
+ print(f"\n๐ Currently Sinkholed:")
+ for ip in sinkholed_ips:
+ print(f" - {ip}")
+
+ # Check general stats
+ stats_response = self.session.get(f"{self.shield_url}/api/dashboard/stats")
+ if stats_response.status_code == 200:
+ stats = stats_response.json()
+ print(f"\n๐ Overall Aurora Shield Stats:")
+ print(f" Total requests processed: {stats.get('total_requests', 0)}")
+ print(f" Blocked requests: {stats.get('blocked_requests', 0)}")
+ print(f" Active threats: {stats.get('active_threats', 0)}")
+
+ except Exception as e:
+ print(f"โ Status check failed: {e}")
+
+def main():
+ """Run direct Aurora Shield API test."""
+ print("๐ก๏ธ Aurora Shield Direct API Test")
+ print("Testing automated sinkhole via /api/shield/check-request")
+ print("=" * 60)
+
+ tester = DirectShieldTest()
+
+ try:
+ # Test authentication first
+ if not tester.authenticate():
+ print("โ Authentication failed, but continuing with tests...")
+
+ # Test 1: Zero reputation attacks
+ attack_results = tester.simulate_zero_reputation_attacks()
+ time.sleep(2)
+
+ # Check status after attacks
+ tester.check_sinkhole_and_attacks()
+ time.sleep(2)
+
+ # Test 2: Legitimate traffic
+ legit_results = tester.simulate_legitimate_traffic()
+ time.sleep(2)
+
+ # Final comprehensive status check
+ print("\n" + "="*60)
+ print("๐ FINAL COMPREHENSIVE STATUS")
+ print("="*60)
+ tester.check_sinkhole_and_attacks()
+
+ # Summary
+ print(f"\n๐ Test Summary:")
+ attack_count = len([r for r in attack_results if r['malicious']])
+ legit_count = len([r for r in legit_results if not r['malicious']])
+ print(f" Attack requests sent: {attack_count}")
+ print(f" Legitimate requests sent: {legit_count}")
+ print(f" โ
Test completed!")
+ print(f"\n๐ Check the dashboard at http://localhost:8080")
+
+ except KeyboardInterrupt:
+ print("\nโน๏ธ Test interrupted")
+ except Exception as e:
+ print(f"\nโ Test failed: {e}")
+
+if __name__ == "__main__":
+ main()
\ No newline at end of file
diff --git a/tests/test_dummy_basic.py b/tests/test_dummy_basic.py
new file mode 100644
index 0000000..0d262dc
--- /dev/null
+++ b/tests/test_dummy_basic.py
@@ -0,0 +1,5 @@
+def test_always_passes():
+ assert True
+
+def test_simple_math():
+ assert 1 + 1 == 2
diff --git a/tests/test_dummy_compat.py b/tests/test_dummy_compat.py
new file mode 100644
index 0000000..2b3ad0b
--- /dev/null
+++ b/tests/test_dummy_compat.py
@@ -0,0 +1,10 @@
+import sys
+
+def test_python_version_supported():
+ # ensure test is compatible with the project's supported Python versions
+ major = sys.version_info.major
+ assert major in (3,)
+
+def test_string_operations():
+ s = "hello"
+ assert s.upper() == "HELLO"
diff --git a/tests/test_emergency_mode.py b/tests/test_emergency_mode.py
new file mode 100644
index 0000000..4915c36
--- /dev/null
+++ b/tests/test_emergency_mode.py
@@ -0,0 +1,143 @@
+#!/usr/bin/env python3
+
+"""
+Test the Enhanced Emergency Mode functionality
+"""
+
+def test_emergency_mode_enhancement():
+ """Test that Emergency Mode has been enhanced with convincing shutdown description"""
+
+ print("๐จ Emergency Mode Enhancement Test")
+ print("=" * 60)
+
+ with open('aurora_shield/dashboard/templates/aurora_dashboard.html', 'r', encoding='utf-8') as f:
+ content = f.read()
+
+ # Test 1: Enhanced Description
+ print("\n๐ Test 1: Enhanced Emergency Mode Description")
+ print("-" * 45)
+
+ description_elements = [
+ ('CRITICAL SECURITY PROTOCOL', 'Critical protocol warning'),
+ ('infrastructure shutdown', 'Infrastructure shutdown mention'),
+ ('emergency maintenance', 'Emergency maintenance purpose'),
+ ('multi-vector attacks', 'Multi-vector attack context'),
+ ('gracefully terminated', 'Graceful termination process'),
+ ('CDN nodes', 'CDN nodes shutdown'),
+ ('load balancers', 'Load balancer shutdown'),
+ ('demo applications', 'Demo application shutdown'),
+ ('Aurora Shield core dashboard remains operational', 'Dashboard persistence'),
+ ('temporary service unavailability', 'Service impact warning'),
+ ('preserve system integrity', 'System integrity justification')
+ ]
+
+ for element, description in description_elements:
+ if element in content:
+ print(f"โ
{description} included")
+ else:
+ print(f"โ {description} missing")
+
+ # Test 2: Enhanced Button Text
+ print("\n๐ด Test 2: Emergency Button Enhancement")
+ print("-" * 45)
+
+ button_elements = [
+ ('Activate Emergency Shutdown', 'Enhanced button text'),
+ ('btn btn-danger', 'Danger button styling'),
+ ('toggleEmergencyMode()', 'Emergency function call')
+ ]
+
+ for element, description in button_elements:
+ if element in content:
+ print(f"โ
{description} present")
+ else:
+ print(f"โ {description} missing")
+
+ # Test 3: Enhanced JavaScript Function
+ print("\n๐ป Test 3: Enhanced JavaScript Functionality")
+ print("-" * 45)
+
+ js_features = [
+ ('CRITICAL SECURITY ALERT', 'Enhanced alert message'),
+ ('Emergency infrastructure shutdown', 'Infrastructure shutdown warning'),
+ ('Load balancer containers', 'Load balancer shutdown detail'),
+ ('CDN distribution nodes', 'CDN shutdown detail'),
+ ('Demo application instances', 'Demo app shutdown detail'),
+ ('Attack orchestrator services', 'Orchestrator shutdown detail'),
+ ('Aurora Shield core dashboard will remain', 'Dashboard persistence assurance'),
+ ('Estimated downtime: 2-5 minutes', 'Downtime estimate'),
+ ('manual restart after threat assessment', 'Manual restart requirement'),
+ ('showEmergencyShutdownProgress', 'Progress function'),
+ ('EMERGENCY SHUTDOWN INITIATED', 'Shutdown confirmation'),
+ ('Phase 1: Gracefully stopping', 'Shutdown phases'),
+ ('MAINTENANCE MODE', 'Maintenance mode status'),
+ ('updateEmergencyModeUI', 'UI update function')
+ ]
+
+ for feature, description in js_features:
+ if feature in content:
+ print(f"โ
{description} implemented")
+ else:
+ print(f"โ {description} missing")
+
+ # Test 4: Progress Overlay Features
+ print("\n๐ Test 4: Emergency Shutdown Progress Overlay")
+ print("-" * 45)
+
+ overlay_features = [
+ ('emergency-overlay', 'Progress overlay container'),
+ ('EMERGENCY SHUTDOWN IN PROGRESS', 'Progress overlay title'),
+ ('Initiating emergency protocols', 'Protocol initiation step'),
+ ('Analyzing threat severity', 'Threat analysis step'),
+ ('Notifying system administrators', 'Admin notification step'),
+ ('DO NOT CLOSE THIS WINDOW', 'User warning'),
+ ('progress-line', 'Progress line styling'),
+ ('rgba(0,0,0,0.9)', 'Dark overlay background')
+ ]
+
+ for feature, description in overlay_features:
+ if feature in content:
+ print(f"โ
{description} present")
+ else:
+ print(f"โ {description} missing")
+
+ # Test 5: Emergency Status Styling
+ print("\n๐จ Test 5: Emergency Status Styling")
+ print("-" * 45)
+
+ styling_features = [
+ ('status-emergency', 'Emergency status class'),
+ ('emergency-pulse', 'Pulsing animation'),
+ ('@keyframes emergency-pulse', 'Animation definition'),
+ ('System in Maintenance Mode', 'Maintenance mode button text'),
+ ('btn btn-warning', 'Warning button style for maintenance')
+ ]
+
+ for feature, description in styling_features:
+ if feature in content:
+ print(f"โ
{description} defined")
+ else:
+ print(f"โ {description} missing")
+
+ # Test 6: Post-Shutdown State
+ print("\n๐ ๏ธ Test 6: Post-Shutdown State Management")
+ print("-" * 45)
+
+ post_shutdown_features = [
+ ('Contact system administrator', 'Admin contact message'),
+ ('security@aurorashield.com', 'Emergency contact email'),
+ ('Manual intervention required', 'Manual restart requirement'),
+ ('emergency maintenance mode', 'Maintenance mode description')
+ ]
+
+ for feature, description in post_shutdown_features:
+ if feature in content:
+ print(f"โ
{description} included")
+ else:
+ print(f"โ {description} missing")
+
+ print("\n๐ Emergency Mode Enhancement Test Complete!")
+ print("=" * 60)
+
+if __name__ == "__main__":
+ test_emergency_mode_enhancement()
\ No newline at end of file
diff --git a/tests/test_emergency_shutdown.py b/tests/test_emergency_shutdown.py
new file mode 100644
index 0000000..63ac6fb
--- /dev/null
+++ b/tests/test_emergency_shutdown.py
@@ -0,0 +1,84 @@
+#!/usr/bin/env python3
+"""
+Test script for Emergency Shutdown API
+Tests the new Docker shutdown functionality
+"""
+import requests
+import json
+
+def test_emergency_shutdown():
+ """Test the emergency shutdown API endpoint"""
+
+ # First, try to login to get a session
+ login_url = "http://localhost:8080/login"
+ dashboard_url = "http://localhost:8080/api/emergency/shutdown"
+
+ # Create a session to maintain cookies
+ session = requests.Session()
+
+ try:
+ # Login first
+ print("๐ Attempting to log in...")
+ login_data = {
+ 'username': 'admin',
+ 'password': 'admin123'
+ }
+
+ login_response = session.post(login_url, data=login_data)
+ print(f"Login status: {login_response.status_code}")
+
+ if login_response.status_code not in [200, 302]:
+ print("โ Login failed")
+ return
+
+ print("โ
Login successful")
+
+ # Test emergency shutdown
+ print("\n๐จ Testing Emergency Shutdown API...")
+
+ shutdown_data = {
+ 'reason': 'Testing emergency shutdown functionality from script'
+ }
+
+ response = session.post(dashboard_url,
+ json=shutdown_data,
+ headers={'Content-Type': 'application/json'})
+
+ print(f"Response Status: {response.status_code}")
+ print(f"Response Headers: {dict(response.headers)}")
+
+ try:
+ result = response.json()
+ print(f"\n๐ Response Data:")
+ print(json.dumps(result, indent=2))
+
+ if result.get('success'):
+ print(f"\nโ
Emergency shutdown successful!")
+ print(f"Containers stopped: {result.get('containers_stopped', 0)}")
+ print(f"Containers failed: {result.get('containers_failed', 0)}")
+
+ if 'results' in result:
+ print(f"\n๐ Container shutdown details:")
+ for container in result['results']:
+ status_icon = "โ
" if container['status'] == 'stopped' else "โ"
+ print(f"{status_icon} {container['name']} ({container['id']}) - {container['status']}")
+ if container.get('error'):
+ print(f" Error: {container['error']}")
+
+ else:
+ print(f"โ Emergency shutdown failed: {result.get('error', 'Unknown error')}")
+
+ except json.JSONDecodeError:
+ print(f"โ Invalid JSON response: {response.text}")
+
+ except requests.exceptions.ConnectionError:
+ print("โ Could not connect to Aurora Shield dashboard. Is it running?")
+ print(" Run: docker-compose up -d")
+
+ except Exception as e:
+ print(f"โ Unexpected error: {e}")
+
+if __name__ == "__main__":
+ print("๐ฅ Aurora Shield Emergency Shutdown Test")
+ print("="*50)
+ test_emergency_shutdown()
\ No newline at end of file
diff --git a/tests/test_filter_options.py b/tests/test_filter_options.py
new file mode 100644
index 0000000..2cf1f82
--- /dev/null
+++ b/tests/test_filter_options.py
@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+
+"""
+Quick test to verify the dashboard filter options are working correctly
+"""
+
+import re
+
+def test_dashboard_filter_options():
+ """Test that the new filter options are properly added to the dashboard"""
+
+ # Read the dashboard HTML template
+ with open('aurora_shield/dashboard/templates/aurora_dashboard.html', 'r', encoding='utf-8') as f:
+ content = f.read()
+
+ # Check if the new filter options are present
+ filter_options = [
+ 'blocked',
+ 'sinkholed',
+ 'blackholed',
+ 'quarantined',
+ 'rate-limited',
+ 'challenged',
+ 'monitored'
+ ]
+
+ print("๐ Testing Filter Options in Dashboard...")
+ print("=" * 50)
+
+ # Find the action filter dropdown
+ filter_pattern = r']*id="action-filter"[^>]*>(.*?)'
+ match = re.search(filter_pattern, content, re.DOTALL)
+
+ if match:
+ filter_dropdown = match.group(1)
+ print("โ
Found action filter dropdown")
+
+ # Check each option
+ missing_options = []
+ for option in filter_options:
+ if f'value="{option}"' in filter_dropdown:
+ print(f"โ
Found {option} option")
+ else:
+ missing_options.append(option)
+ print(f"โ Missing {option} option")
+
+ if not missing_options:
+ print("\n๐ All filter options are present!")
+ else:
+ print(f"\nโ ๏ธ Missing options: {missing_options}")
+ else:
+ print("โ Could not find action filter dropdown")
+
+ # Check CSS styles for action types
+ print("\n๐จ Testing CSS Styles...")
+ print("=" * 50)
+
+ css_classes = [
+ 'action-blocked',
+ 'action-sinkholed',
+ 'action-blackholed',
+ 'action-quarantined',
+ 'action-rate-limited',
+ 'action-challenged',
+ 'action-monitored'
+ ]
+
+ missing_styles = []
+ for css_class in css_classes:
+ if f'.{css_class} {{' in content:
+ print(f"โ
Found {css_class} style")
+ else:
+ missing_styles.append(css_class)
+ print(f"โ Missing {css_class} style")
+
+ if not missing_styles:
+ print("\n๐ All CSS styles are present!")
+ else:
+ print(f"\nโ ๏ธ Missing styles: {missing_styles}")
+
+ # Test filter JavaScript function
+ print("\n๐ง Testing JavaScript Function...")
+ print("=" * 50)
+
+ if 'function onActionFilterChange(select)' in content:
+ print("โ
Found onActionFilterChange function")
+ else:
+ print("โ Missing onActionFilterChange function")
+
+ if 'currentAttackFilters.action = select.value' in content:
+ print("โ
Found filter update logic")
+ else:
+ print("โ Missing filter update logic")
+
+if __name__ == "__main__":
+ test_dashboard_filter_options()
\ No newline at end of file
diff --git a/tests/test_logs_export.py b/tests/test_logs_export.py
new file mode 100644
index 0000000..b050fa0
--- /dev/null
+++ b/tests/test_logs_export.py
@@ -0,0 +1,116 @@
+#!/usr/bin/env python3
+"""
+Test script for Logs Export API
+Tests the new attack logs export functionality
+"""
+import requests
+import json
+
+def test_logs_export():
+ """Test the logs export API endpoint"""
+
+ # First, try to login to get a session
+ login_url = "http://localhost:8080/login"
+ export_url = "http://localhost:8080/api/export/logs"
+
+ # Create a session to maintain cookies
+ session = requests.Session()
+
+ try:
+ # Login first
+ print("๐ Attempting to log in...")
+ login_data = {
+ 'username': 'admin',
+ 'password': 'admin123'
+ }
+
+ login_response = session.post(login_url, data=login_data)
+ print(f"Login status: {login_response.status_code}")
+
+ if login_response.status_code not in [200, 302]:
+ print("โ Login failed")
+ return
+
+ print("โ
Login successful")
+
+ # Test logs export
+ print("\n๐ Testing Logs Export API...")
+
+ response = session.get(export_url)
+
+ print(f"Response Status: {response.status_code}")
+ print(f"Response Headers: {dict(response.headers)}")
+
+ if response.status_code == 200:
+ # Check if it's JSON content
+ content_type = response.headers.get('content-type', '')
+ if 'application/json' in content_type:
+ # Parse the JSON to validate structure
+ try:
+ log_data = response.json()
+ print(f"\nโ
Logs export successful!")
+ print(f"๐ Export Summary:")
+
+ if 'export_info' in log_data:
+ export_info = log_data['export_info']
+ print(f" Generated at: {export_info.get('generated_at', 'N/A')}")
+ print(f" Exported by: {export_info.get('exported_by', 'N/A')}")
+ print(f" System uptime: {export_info.get('uptime', 'N/A')}")
+
+ if 'attack_logs' in log_data:
+ attack_count = len(log_data['attack_logs'])
+ print(f" Attack logs: {attack_count} entries")
+
+ if attack_count > 0:
+ print(f" Sample attack log:")
+ sample = log_data['attack_logs'][0]
+ print(f" - IP: {sample.get('ip', 'N/A')}")
+ print(f" - Status: {sample.get('status', 'N/A')}")
+ print(f" - Timestamp: {sample.get('timestamp', 'N/A')}")
+
+ if 'blocked_requests' in log_data:
+ blocked_info = log_data['blocked_requests']
+ print(f" Blocked requests: {blocked_info.get('total_blocked', 0)}")
+ print(f" Block rate: {blocked_info.get('block_rate', 'N/A')}")
+
+ if 'reputation_scores' in log_data:
+ ip_count = len(log_data['reputation_scores'])
+ print(f" IP reputation scores: {ip_count} IPs tracked")
+
+ if 'mitigation_actions' in log_data:
+ mitigation_count = len(log_data['mitigation_actions'])
+ print(f" Mitigation actions: {mitigation_count} active")
+
+ # Show file size
+ content_length = len(response.content)
+ print(f" Export file size: {content_length:,} bytes")
+
+ # Save sample file for verification
+ with open('sample_export.json', 'w', encoding='utf-8') as f:
+ json.dump(log_data, f, indent=2, ensure_ascii=False)
+ print(f" ๐ Sample saved as: sample_export.json")
+
+ except json.JSONDecodeError as e:
+ print(f"โ Invalid JSON response: {e}")
+ print(f"Raw content (first 500 chars): {response.text[:500]}")
+ else:
+ print(f"โ Unexpected content type: {content_type}")
+
+ else:
+ try:
+ error_data = response.json()
+ print(f"โ Export failed: {error_data.get('error', 'Unknown error')}")
+ except:
+ print(f"โ Export failed with status {response.status_code}: {response.text}")
+
+ except requests.exceptions.ConnectionError:
+ print("โ Could not connect to Aurora Shield dashboard. Is it running?")
+ print(" Run: docker-compose up -d aurora-shield")
+
+ except Exception as e:
+ print(f"โ Unexpected error: {e}")
+
+if __name__ == "__main__":
+ print("๐ Aurora Shield Logs Export Test")
+ print("="*50)
+ test_logs_export()
\ No newline at end of file
diff --git a/tests/test_monitoring_cleanup.py b/tests/test_monitoring_cleanup.py
new file mode 100644
index 0000000..d6b9f7b
--- /dev/null
+++ b/tests/test_monitoring_cleanup.py
@@ -0,0 +1,81 @@
+#!/usr/bin/env python3
+
+"""
+Test to verify Kibana and Grafana buttons have been removed
+"""
+
+def test_monitoring_buttons_removed():
+ """Test that Kibana and Grafana buttons have been removed from monitoring tab"""
+
+ print("๐ Monitoring Tab Button Removal Test")
+ print("=" * 50)
+
+ with open('aurora_shield/dashboard/templates/aurora_dashboard.html', 'r', encoding='utf-8') as f:
+ content = f.read()
+
+ # Test 1: Check removed buttons
+ print("\n๐๏ธ Test 1: Button Removal")
+ print("-" * 30)
+
+ removed_buttons = [
+ ('Open Grafana', 'Grafana button'),
+ ('Open Kibana', 'Kibana button'),
+ ('localhost:3000', 'Grafana URL'),
+ ('localhost:5601', 'Kibana URL')
+ ]
+
+ for text, description in removed_buttons:
+ if text in content:
+ print(f"โ {description} still present")
+ else:
+ print(f"โ
{description} removed")
+
+ # Test 2: Check preserved functionality
+ print("\n๐ Test 2: Preserved Elements")
+ print("-" * 30)
+
+ preserved_elements = [
+ ('Export Logs', 'Export logs button'),
+ ('exportLogs()', 'Export logs function'),
+ ('๐พ', 'Export logs icon')
+ ]
+
+ for element, description in preserved_elements:
+ if element in content:
+ print(f"โ
{description} preserved")
+ else:
+ print(f"โ {description} missing (should be preserved)")
+
+ # Test 3: Check monitoring tab structure
+ print("\n๐ Test 3: Monitoring Tab Structure")
+ print("-" * 30)
+
+ # Count buttons in the monitoring section
+ import re
+
+ # Find the monitoring tab content - more specific pattern
+ monitoring_section = re.search(r'\s*
\s*